Package org.voltdb.stream.api.kafka

Class KafkaStreamSinkConfigurator<T>

java.lang.Object

org.voltdb.stream.api.kafka.KafkaStreamSinkConfigurator<T>

All Implemented Interfaces:

OperatorConfigurator, VoltStreamSinkConfigurator<T>

public class KafkaStreamSinkConfigurator<T>
extends Object
implements VoltStreamSinkConfigurator<T>

this configurator can be autoconfigured using helm's values under streaming.pipeline.configuration. By default, source configuration is lookup by `kafka.sink` config path, for example:

 kafka:
   sink:
     topicName: topicA
     bootstrapServers: serverA,serverB
     # optional
     startingOffset: EARLIEST [LATEST | NONE]
     pollTimeout: PT0.25S
     maxCommitTimeout: PT10S
     maxCommitRetries: 3
     properties:
       key1: value1
       key2: value2
 

Method Summary

Modifier and Type	Method	Description
static <P> KafkaStreamSinkConfigurator<P>	accepting(Class<P> vc)
KafkaStreamSinkConfigurator<T>	configure(String path)	configures the configurator using custom config path.
ExceptionHandler	getExceptionHandler()
KafkaStreamSinkConfigurator<T>	withBootstrapServers(String bootstrapServers)
VoltStreamSinkConfigurator<T>	withExceptionHandler(ExceptionHandler exceptionHandler)
<K> KafkaStreamSinkConfigurator<T>	withHeaders(Function<T,Iterable<org.apache.kafka.common.header.Header>> extractorFunction)
KafkaStreamSinkConfigurator<T>	withKeyExtractor(Function<T,Long> extractorFunction)
<K> KafkaStreamSinkConfigurator<T>	withKeyExtractor(Function<T,K> extractorFunction, Class<? extends org.apache.kafka.common.serialization.Serializer<K>> serializerClass)
KafkaStreamSinkConfigurator<T>	withProperty(String key, boolean value)
KafkaStreamSinkConfigurator<T>	withProperty(String key, int value)
KafkaStreamSinkConfigurator<T>	withProperty(String key, long value)
KafkaStreamSinkConfigurator<T>	withProperty(String key, String value)
KafkaStreamSinkConfigurator<T>	withSchemaRegistry(String schemaRegistryUrl)
KafkaStreamSinkConfigurator<T>	withSSL(KafkaStreamSslConfiguration sslConfigurator)
KafkaStreamSinkConfigurator<T>	withTopicName(String topicName)
<K> KafkaStreamSinkConfigurator<T>	withValueExtractor(Function<T,K> extractorFunction, Class<? extends org.apache.kafka.common.serialization.Serializer<?>> serializerClass)
KafkaStreamSinkConfigurator<T>	withValueSerializer(Class<? extends org.apache.kafka.common.serialization.Serializer<?>> serializerClass)
KafkaStreamSinkConfigurator<T>	withValueSerializer(String serializerClass)

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface org.voltdb.stream.api.extension.OperatorConfigurator

configure, getConfiguration

Method Details

accepting

public static <P> KafkaStreamSinkConfigurator<P> accepting(Class<P> vc)

withTopicName

public KafkaStreamSinkConfigurator<T> withTopicName(String topicName)

withBootstrapServers

public KafkaStreamSinkConfigurator<T> withBootstrapServers(String bootstrapServers)

withSSL

public KafkaStreamSinkConfigurator<T> withSSL(KafkaStreamSslConfiguration sslConfigurator)

withProperty

public KafkaStreamSinkConfigurator<T> withProperty(String key,
 String value)

withProperty

public KafkaStreamSinkConfigurator<T> withProperty(String key,
 int value)

withProperty

public KafkaStreamSinkConfigurator<T> withProperty(String key,
 long value)

withProperty

public KafkaStreamSinkConfigurator<T> withProperty(String key,
 boolean value)

withValueSerializer

public KafkaStreamSinkConfigurator<T> withValueSerializer(Class<? extends org.apache.kafka.common.serialization.Serializer<?>> serializerClass)

withValueSerializer

public KafkaStreamSinkConfigurator<T> withValueSerializer(String serializerClass)

withKeyExtractor

public KafkaStreamSinkConfigurator<T> withKeyExtractor(Function<T,Long> extractorFunction)

withKeyExtractor

public <K> KafkaStreamSinkConfigurator<T> withKeyExtractor(Function<T,K> extractorFunction,
 Class<? extends org.apache.kafka.common.serialization.Serializer<K>> serializerClass)

withValueExtractor

public <K> KafkaStreamSinkConfigurator<T> withValueExtractor(Function<T,K> extractorFunction,
 Class<? extends org.apache.kafka.common.serialization.Serializer<?>> serializerClass)

withHeaders

public <K> KafkaStreamSinkConfigurator<T> withHeaders(Function<T,Iterable<org.apache.kafka.common.header.Header>> extractorFunction)

withSchemaRegistry

public KafkaStreamSinkConfigurator<T> withSchemaRegistry(String schemaRegistryUrl)

withExceptionHandler

public VoltStreamSinkConfigurator<T> withExceptionHandler(ExceptionHandler exceptionHandler)

getExceptionHandler

public ExceptionHandler getExceptionHandler()

configure

public KafkaStreamSinkConfigurator<T> configure(String path)

configures the configurator using custom config path. for example having a yaml file

 my:
   custom:
     sink:
       kafka:
         key: A,B,C
    other:
      key: value
 

a path is "my.custom.sink.kafka", this will return a map of keys and values. the default path is "kafka.sink"

Note for more info see helm configuration under streaming.pipeline.configuration.kafka.sink

Parameters:

path - to a config

Returns:

kafka configurator