org.apache.storm.kafka.spout

Class KafkaSpoutConfig.Builder<K,V>

java.lang.Object

org.apache.storm.kafka.spout.KafkaSpoutConfig.Builder<K,V>

Enclosing class:

KafkaSpoutConfig<K,V>

public static class KafkaSpoutConfig.Builder<K,V>
extends Object

Constructor Summary

Constructors

Constructor and Description
Builder(String bootstrapServers, Class<? extends org.apache.kafka.common.serialization.Deserializer<K>> keyDes, Class<? extends org.apache.kafka.common.serialization.Deserializer<V>> valDes, Collection<String> topics) Deprecated. Please use #Builder(java.lang.String, java.util.Collection) instead, and set the deserializer with setProp(java.lang.String, java.lang.Object), ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG and ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG
Builder(String bootstrapServers, Class<? extends org.apache.kafka.common.serialization.Deserializer<K>> keyDes, Class<? extends org.apache.kafka.common.serialization.Deserializer<V>> valDes, Pattern topics) Deprecated. Please use #Builder(java.lang.String, java.util.regex.Pattern) instead, and set the deserializer with setProp(java.lang.String, java.lang.Object), ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG and ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG
Builder(String bootstrapServers, Class<? extends org.apache.kafka.common.serialization.Deserializer<K>> keyDes, Class<? extends org.apache.kafka.common.serialization.Deserializer<V>> valDes, String... topics) Deprecated. Please use #Builder(java.lang.String, java.lang.String...) instead, and set the deserializer with setProp(java.lang.String, java.lang.Object), ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG and ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG
Builder(String bootstrapServers, Class<? extends org.apache.kafka.common.serialization.Deserializer<K>> keyDes, Class<? extends org.apache.kafka.common.serialization.Deserializer<V>> valDes, Subscription subscription) Deprecated. Please use #Builder(java.lang.String, org.apache.storm.kafka.spout.Subscription) instead, and set the deserializer with setProp(java.lang.String, java.lang.Object), ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG and ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG
Builder(String bootstrapServers, Collection<String> topics)
Builder(String bootstrapServers, Pattern topics)
Builder(String bootstrapServers, SerializableDeserializer<K> keyDes, SerializableDeserializer<V> valDes, Collection<String> topics) Deprecated. Please use #Builder(java.lang.String, java.util.Collection) instead, and set the deserializer with setProp(java.lang.String, java.lang.Object), ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG and ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG
Builder(String bootstrapServers, SerializableDeserializer<K> keyDes, SerializableDeserializer<V> valDes, Pattern topics) Deprecated. Please use #Builder(java.lang.String, java.util.regex.Pattern) instead, and set the deserializer with setProp(java.lang.String, java.lang.Object), ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG and ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG
Builder(String bootstrapServers, SerializableDeserializer<K> keyDes, SerializableDeserializer<V> valDes, String... topics) Deprecated. Please use #Builder(java.lang.String, java.lang.String...) instead, and set the deserializer with setProp(java.lang.String, java.lang.Object), ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG and ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG
Builder(String bootstrapServers, SerializableDeserializer<K> keyDes, SerializableDeserializer<V> valDes, Subscription subscription) Deprecated. Please use #Builder(java.lang.String, org.apache.storm.kafka.spout.Subscription) instead, and set the deserializer with setProp(java.lang.String, java.lang.Object), ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG and ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG
Builder(String bootstrapServers, String... topics)
Builder(String bootstrapServers, Subscription subscription) Create a KafkaSpoutConfig builder with default property values and no key/value deserializers.

Method Summary

Modifier and Type	Method and Description
KafkaSpoutConfig<K,V>	build()
KafkaSpoutConfig.Builder<K,V>	setBootstrapServers(String servers) Deprecated. Please use setProp(java.lang.String, java.lang.Object) with ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG instead
KafkaSpoutConfig.Builder<K,V>	setEmitNullTuples(boolean emitNullTuples) Specifies if the spout should emit null tuples to the component downstream, or rather not emit and directly ack them.
KafkaSpoutConfig.Builder<K,V>	setFetchMinBytes(int bytes) Deprecated. Please use setProp(java.lang.String, java.lang.Object) with ConsumerConfig.FETCH_MIN_BYTES_CONFIG instead
KafkaSpoutConfig.Builder<K,V>	setFirstPollOffsetStrategy(KafkaSpoutConfig.FirstPollOffsetStrategy firstPollOffsetStrategy) Sets the offset used by the Kafka spout in the first poll to Kafka broker upon process start.
KafkaSpoutConfig.Builder<K,V>	setGroupId(String id) Deprecated. Please use setProp(java.lang.String, java.lang.Object) with ConsumerConfig.GROUP_ID_CONFIG instead
<NK> KafkaSpoutConfig.Builder<NK,V>	setKey(Class<? extends org.apache.kafka.common.serialization.Deserializer<NK>> clazz) Deprecated. Please use setProp(java.lang.String, java.lang.Object) with ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG instead
<NK> KafkaSpoutConfig.Builder<NK,V>	setKey(SerializableDeserializer<NK> keyDeserializer) Deprecated. Please use setProp(java.lang.String, java.lang.Object) with ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG instead
KafkaSpoutConfig.Builder<K,V>	setMaxPartitionFectchBytes(int bytes) Deprecated. Please use setProp(java.lang.String, java.lang.Object) with ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG instead
KafkaSpoutConfig.Builder<K,V>	setMaxPollRecords(int records) Deprecated. Please use setProp(java.lang.String, java.lang.Object) with ConsumerConfig.MAX_POLL_RECORDS_CONFIG instead
KafkaSpoutConfig.Builder<K,V>	setMaxUncommittedOffsets(int maxUncommittedOffsets) Defines the max number of polled offsets (records) that can be pending commit, before another poll can take place.
KafkaSpoutConfig.Builder<K,V>	setMetricsTimeBucketSizeInSecs(int metricsTimeBucketSizeInSecs) The time period that metrics data in bucketed into.
KafkaSpoutConfig.Builder<K,V>	setOffsetCommitPeriodMs(long offsetCommitPeriodMs) Specifies the period, in milliseconds, the offset commit task is periodically called.
KafkaSpoutConfig.Builder<K,V>	setPartitionRefreshPeriodMs(long partitionRefreshPeriodMs) Sets partition refresh period in milliseconds.
KafkaSpoutConfig.Builder<K,V>	setPollTimeoutMs(long pollTimeoutMs) Specifies the time, in milliseconds, spent waiting in poll if data is not available.
KafkaSpoutConfig.Builder<K,V>	setProcessingGuarantee(KafkaSpoutConfig.ProcessingGuarantee processingGuarantee) Specifies which processing guarantee the spout should offer.
KafkaSpoutConfig.Builder<K,V>	setProp(Map<String,Object> props) Set multiple KafkaConsumer properties.
KafkaSpoutConfig.Builder<K,V>	setProp(Properties props) Set multiple KafkaConsumer properties.
KafkaSpoutConfig.Builder<K,V>	setProp(String key, Object value) Set a KafkaConsumer property.
KafkaSpoutConfig.Builder<K,V>	setRecordTranslator(Func<org.apache.kafka.clients.consumer.ConsumerRecord<K,V>,List<Object>> func, Fields fields) Configure a translator with tuples to be emitted on the default stream.
KafkaSpoutConfig.Builder<K,V>	setRecordTranslator(Func<org.apache.kafka.clients.consumer.ConsumerRecord<K,V>,List<Object>> func, Fields fields, String stream) Configure a translator with tuples to be emitted to a given stream.
KafkaSpoutConfig.Builder<K,V>	setRecordTranslator(RecordTranslator<K,V> translator)
KafkaSpoutConfig.Builder<K,V>	setRetry(KafkaSpoutRetryService retryService) Sets the retry service for the spout to use.
KafkaSpoutConfig.Builder<K,V>	setSecurityProtocol(String protocol) Deprecated. Please use setProp(java.lang.String, java.lang.Object) with “security.protocol” instead
KafkaSpoutConfig.Builder<K,V>	setSSLKeystore(String location, String password) Deprecated. Please use setProp(java.lang.String, java.lang.Object) with “ssl.keystore.location” and “ssl.keystore.password” instead
KafkaSpoutConfig.Builder<K,V>	setSSLKeystore(String location, String password, String keyPassword) Deprecated. Please use setProp(java.lang.String, java.lang.Object) with “ssl.keystore.location”, “ssl.keystore.password” and “ssl.key.password” instead
KafkaSpoutConfig.Builder<K,V>	setSSLTruststore(String location, String password) Deprecated. Please use setProp(java.lang.String, java.lang.Object) with “security.protocol”, “ssl.truststore.location” and “ssl.truststore.password” instead
KafkaSpoutConfig.Builder<K,V>	setTupleListener(KafkaTupleListener tupleListener) Sets the tuple listener for the spout to use.
KafkaSpoutConfig.Builder<K,V>	setTupleTrackingEnforced(boolean tupleTrackingEnforced) Specifies whether the spout should require Storm to track emitted tuples when using a KafkaSpoutConfig.ProcessingGuarantee other than KafkaSpoutConfig.ProcessingGuarantee.AT_LEAST_ONCE.
<NV> KafkaSpoutConfig.Builder<K,NV>	setValue(Class<? extends org.apache.kafka.common.serialization.Deserializer<NV>> clazz) Deprecated. Please use setProp(java.lang.String, java.lang.Object) with ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG instead
<NV> KafkaSpoutConfig.Builder<K,NV>	setValue(SerializableDeserializer<NV> valueDeserializer) Deprecated. Please use setProp(java.lang.String, java.lang.Object) with ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG instead

Methods inherited from class java.lang.Object

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

Constructor Detail

Builder

public Builder(String bootstrapServers,
               String... topics)

Builder

public Builder(String bootstrapServers,
               Collection<String> topics)

Builder

public Builder(String bootstrapServers,
               Pattern topics)

Builder

@Deprecated
public Builder(String bootstrapServers,
                           SerializableDeserializer<K> keyDes,
                           SerializableDeserializer<V> valDes,
                           String... topics)

Deprecated. Please use #Builder(java.lang.String, java.lang.String...) instead, and set the deserializer with setProp(java.lang.String, java.lang.Object), ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG and ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG

Builder

@Deprecated
public Builder(String bootstrapServers,
                           SerializableDeserializer<K> keyDes,
                           SerializableDeserializer<V> valDes,
                           Collection<String> topics)

Deprecated. Please use #Builder(java.lang.String, java.util.Collection) instead, and set the deserializer with setProp(java.lang.String, java.lang.Object), ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG and ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG

Builder

@Deprecated
public Builder(String bootstrapServers,
                           SerializableDeserializer<K> keyDes,
                           SerializableDeserializer<V> valDes,
                           Pattern topics)

Deprecated. Please use #Builder(java.lang.String, java.util.regex.Pattern) instead, and set the deserializer with setProp(java.lang.String, java.lang.Object), ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG and ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG

Builder

@Deprecated
public Builder(String bootstrapServers,
                           SerializableDeserializer<K> keyDes,
                           SerializableDeserializer<V> valDes,
                           Subscription subscription)

Deprecated. Please use #Builder(java.lang.String, org.apache.storm.kafka.spout.Subscription) instead, and set the deserializer with setProp(java.lang.String, java.lang.Object), ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG and ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG

Builder

@Deprecated
public Builder(String bootstrapServers,
                           Class<? extends org.apache.kafka.common.serialization.Deserializer<K>> keyDes,
                           Class<? extends org.apache.kafka.common.serialization.Deserializer<V>> valDes,
                           String... topics)

Deprecated. Please use #Builder(java.lang.String, java.lang.String...) instead, and set the deserializer with setProp(java.lang.String, java.lang.Object), ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG and ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG

Builder

@Deprecated
public Builder(String bootstrapServers,
                           Class<? extends org.apache.kafka.common.serialization.Deserializer<K>> keyDes,
                           Class<? extends org.apache.kafka.common.serialization.Deserializer<V>> valDes,
                           Collection<String> topics)

Deprecated. Please use #Builder(java.lang.String, java.util.Collection) instead, and set the deserializer with setProp(java.lang.String, java.lang.Object), ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG and ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG

Builder

@Deprecated
public Builder(String bootstrapServers,
                           Class<? extends org.apache.kafka.common.serialization.Deserializer<K>> keyDes,
                           Class<? extends org.apache.kafka.common.serialization.Deserializer<V>> valDes,
                           Pattern topics)

Deprecated. Please use #Builder(java.lang.String, java.util.regex.Pattern) instead, and set the deserializer with setProp(java.lang.String, java.lang.Object), ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG and ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG

Builder

@Deprecated
public Builder(String bootstrapServers,
                           Class<? extends org.apache.kafka.common.serialization.Deserializer<K>> keyDes,
                           Class<? extends org.apache.kafka.common.serialization.Deserializer<V>> valDes,
                           Subscription subscription)

Deprecated. Please use #Builder(java.lang.String, org.apache.storm.kafka.spout.Subscription) instead, and set the deserializer with setProp(java.lang.String, java.lang.Object), ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG and ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG

Builder

public Builder(String bootstrapServers,
               Subscription subscription)

Create a KafkaSpoutConfig builder with default property values and no key/value deserializers.

Parameters:

bootstrapServers - The bootstrap servers the consumer will use

subscription - The subscription defining which topics and partitions each spout instance will read.

Method Detail

setKey

@Deprecated
public <NK> KafkaSpoutConfig.Builder<NK,V> setKey(SerializableDeserializer<NK> keyDeserializer)

Deprecated. Please use setProp(java.lang.String, java.lang.Object) with ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG instead

Specifying this key deserializer overrides the property key.deserializer. If you have set a custom RecordTranslator before calling this it may result in class cast exceptions at runtime.

setKey

@Deprecated
public <NK> KafkaSpoutConfig.Builder<NK,V> setKey(Class<? extends org.apache.kafka.common.serialization.Deserializer<NK>> clazz)

Deprecated. Please use setProp(java.lang.String, java.lang.Object) with ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG instead

Specify a class that can be instantiated to create a key.deserializer This is the same as setting key.deserializer, but overrides it. If you have set a custom RecordTranslator before calling this it may result in class cast exceptions at runtime.

setValue

@Deprecated
public <NV> KafkaSpoutConfig.Builder<K,NV> setValue(SerializableDeserializer<NV> valueDeserializer)

Deprecated. Please use setProp(java.lang.String, java.lang.Object) with ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG instead

Specifying this value deserializer overrides the property value.deserializer. If you have set a custom RecordTranslator before calling this it may result in class cast exceptions at runtime.

setValue

@Deprecated
public <NV> KafkaSpoutConfig.Builder<K,NV> setValue(Class<? extends org.apache.kafka.common.serialization.Deserializer<NV>> clazz)

Deprecated. Please use setProp(java.lang.String, java.lang.Object) with ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG instead

Specify a class that can be instantiated to create a value.deserializer This is the same as setting value.deserializer, but overrides it. If you have set a custom RecordTranslator before calling this it may result in class cast exceptions at runtime.

setProp

public KafkaSpoutConfig.Builder<K,V> setProp(String key,
                                             Object value)

Set a KafkaConsumer property.

setProp

public KafkaSpoutConfig.Builder<K,V> setProp(Map<String,Object> props)

Set multiple KafkaConsumer properties.

setProp

public KafkaSpoutConfig.Builder<K,V> setProp(Properties props)

Set multiple KafkaConsumer properties.

setGroupId

@Deprecated
public KafkaSpoutConfig.Builder<K,V> setGroupId(String id)

Deprecated. Please use setProp(java.lang.String, java.lang.Object) with ConsumerConfig.GROUP_ID_CONFIG instead

Set the group.id for the consumers

setBootstrapServers

@Deprecated
public KafkaSpoutConfig.Builder<K,V> setBootstrapServers(String servers)

Deprecated. Please use setProp(java.lang.String, java.lang.Object) with ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG instead

reset the bootstrap servers for the Consumer

setFetchMinBytes

@Deprecated
public KafkaSpoutConfig.Builder<K,V> setFetchMinBytes(int bytes)

Deprecated. Please use setProp(java.lang.String, java.lang.Object) with ConsumerConfig.FETCH_MIN_BYTES_CONFIG instead

The minimum amount of data the broker should return for a fetch request.

setMaxPartitionFectchBytes

@Deprecated
public KafkaSpoutConfig.Builder<K,V> setMaxPartitionFectchBytes(int bytes)

Deprecated. Please use setProp(java.lang.String, java.lang.Object) with ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG instead

The maximum amount of data per-partition the broker will return.

setMaxPollRecords

@Deprecated
public KafkaSpoutConfig.Builder<K,V> setMaxPollRecords(int records)

Deprecated. Please use setProp(java.lang.String, java.lang.Object) with ConsumerConfig.MAX_POLL_RECORDS_CONFIG instead

The maximum number of records a poll will return.

setSSLKeystore

@Deprecated
public KafkaSpoutConfig.Builder<K,V> setSSLKeystore(String location,
                                                                String password)

Deprecated. Please use setProp(java.lang.String, java.lang.Object) with “ssl.keystore.location” and “ssl.keystore.password” instead

Configure the SSL Keystore for mutual authentication

setSSLKeystore

@Deprecated
public KafkaSpoutConfig.Builder<K,V> setSSLKeystore(String location,
                                                                String password,
                                                                String keyPassword)

Deprecated. Please use setProp(java.lang.String, java.lang.Object) with “ssl.keystore.location”, “ssl.keystore.password” and “ssl.key.password” instead

Configure the SSL Keystore for mutual authentication

setSSLTruststore

@Deprecated
public KafkaSpoutConfig.Builder<K,V> setSSLTruststore(String location,
                                                                  String password)

Deprecated. Please use setProp(java.lang.String, java.lang.Object) with “security.protocol”, “ssl.truststore.location” and “ssl.truststore.password” instead

Configure the SSL Truststore to authenticate with the brokers

setSecurityProtocol

@Deprecated
public KafkaSpoutConfig.Builder<K,V> setSecurityProtocol(String protocol)

Deprecated. Please use setProp(java.lang.String, java.lang.Object) with “security.protocol” instead

Protocol used to communicate with brokers. Valid values are: PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL.

setPollTimeoutMs

public KafkaSpoutConfig.Builder<K,V> setPollTimeoutMs(long pollTimeoutMs)

Specifies the time, in milliseconds, spent waiting in poll if data is not available. Default is 2s.

Parameters:

pollTimeoutMs - time in ms

setOffsetCommitPeriodMs

public KafkaSpoutConfig.Builder<K,V> setOffsetCommitPeriodMs(long offsetCommitPeriodMs)

Specifies the period, in milliseconds, the offset commit task is periodically called. Default is 15s.

This setting only has an effect if the configured KafkaSpoutConfig.ProcessingGuarantee is KafkaSpoutConfig.ProcessingGuarantee.AT_LEAST_ONCE or KafkaSpoutConfig.ProcessingGuarantee.NO_GUARANTEE.

Parameters:

offsetCommitPeriodMs - time in ms

setMaxUncommittedOffsets

public KafkaSpoutConfig.Builder<K,V> setMaxUncommittedOffsets(int maxUncommittedOffsets)

Defines the max number of polled offsets (records) that can be pending commit, before another poll can take place. Once this limit is reached, no more offsets (records) can be polled until the next successful commit(s) sets the number of pending offsets below the threshold. The default is KafkaSpoutConfig.DEFAULT_MAX_UNCOMMITTED_OFFSETS. This limit is per partition and may in some cases be exceeded, but each partition cannot exceed this limit by more than maxPollRecords - 1.

This setting only has an effect if the configured KafkaSpoutConfig.ProcessingGuarantee is KafkaSpoutConfig.ProcessingGuarantee.AT_LEAST_ONCE.

Parameters:

maxUncommittedOffsets - max number of records that can be be pending commit

setFirstPollOffsetStrategy

public KafkaSpoutConfig.Builder<K,V> setFirstPollOffsetStrategy(KafkaSpoutConfig.FirstPollOffsetStrategy firstPollOffsetStrategy)

Sets the offset used by the Kafka spout in the first poll to Kafka broker upon process start. Please refer to to the documentation in KafkaSpoutConfig.FirstPollOffsetStrategy

Parameters:

firstPollOffsetStrategy - Offset used by Kafka spout first poll

setRetry

public KafkaSpoutConfig.Builder<K,V> setRetry(KafkaSpoutRetryService retryService)

Sets the retry service for the spout to use.

This setting only has an effect if the configured KafkaSpoutConfig.ProcessingGuarantee is KafkaSpoutConfig.ProcessingGuarantee.AT_LEAST_ONCE.

Parameters:

retryService - the new retry service

Returns:

the builder (this).

setTupleListener

public KafkaSpoutConfig.Builder<K,V> setTupleListener(KafkaTupleListener tupleListener)

Sets the tuple listener for the spout to use.

Parameters:

tupleListener - the tuple listener

Returns:

the builder (this).

setRecordTranslator

public KafkaSpoutConfig.Builder<K,V> setRecordTranslator(RecordTranslator<K,V> translator)

setRecordTranslator

public KafkaSpoutConfig.Builder<K,V> setRecordTranslator(Func<org.apache.kafka.clients.consumer.ConsumerRecord<K,V>,List<Object>> func,
                                                         Fields fields)

Configure a translator with tuples to be emitted on the default stream.

Parameters:

func - extracts and turns a Kafka ConsumerRecord into a list of objects to be emitted

fields - the names of the fields extracted

Returns:

this to be able to chain configuration

setRecordTranslator

public KafkaSpoutConfig.Builder<K,V> setRecordTranslator(Func<org.apache.kafka.clients.consumer.ConsumerRecord<K,V>,List<Object>> func,
                                                         Fields fields,
                                                         String stream)

Configure a translator with tuples to be emitted to a given stream.

Parameters:

func - extracts and turns a Kafka ConsumerRecord into a list of objects to be emitted

fields - the names of the fields extracted

stream - the stream to emit the tuples on

Returns:

this to be able to chain configuration

setPartitionRefreshPeriodMs

public KafkaSpoutConfig.Builder<K,V> setPartitionRefreshPeriodMs(long partitionRefreshPeriodMs)

Sets partition refresh period in milliseconds. This is how often kafka will be polled to check for new topics and/or new partitions. This is mostly for Subscription implementations that manually assign partitions. NamedSubscription and PatternSubscription rely on kafka to handle this instead.

Parameters:

partitionRefreshPeriodMs - time in milliseconds

Returns:

the builder (this)

setEmitNullTuples

public KafkaSpoutConfig.Builder<K,V> setEmitNullTuples(boolean emitNullTuples)

Specifies if the spout should emit null tuples to the component downstream, or rather not emit and directly ack them. By default this parameter is set to false, which means that null tuples are not emitted.

Parameters:

emitNullTuples - sets if null tuples should or not be emitted downstream

setProcessingGuarantee

public KafkaSpoutConfig.Builder<K,V> setProcessingGuarantee(KafkaSpoutConfig.ProcessingGuarantee processingGuarantee)

Specifies which processing guarantee the spout should offer. Refer to the documentation for KafkaSpoutConfig.ProcessingGuarantee.

Parameters:

processingGuarantee - The processing guarantee the spout should offer.

setTupleTrackingEnforced

public KafkaSpoutConfig.Builder<K,V> setTupleTrackingEnforced(boolean tupleTrackingEnforced)

Specifies whether the spout should require Storm to track emitted tuples when using a KafkaSpoutConfig.ProcessingGuarantee other than KafkaSpoutConfig.ProcessingGuarantee.AT_LEAST_ONCE. The spout will always track emitted tuples when offering at-least-once guarantees regardless of this setting. This setting is false by default.

Enabling tracking can be useful even in cases where reliability is not a concern, because it allows Config.TOPOLOGY_MAX_SPOUT_PENDING to have an effect, and enables some spout metrics (e.g. complete-latency) that would otherwise be disabled.

Parameters:

tupleTrackingEnforced - true if Storm should track emitted tuples, false otherwise

setMetricsTimeBucketSizeInSecs

public KafkaSpoutConfig.Builder<K,V> setMetricsTimeBucketSizeInSecs(int metricsTimeBucketSizeInSecs)

The time period that metrics data in bucketed into.

Parameters:

metricsTimeBucketSizeInSecs - time in seconds

build

public KafkaSpoutConfig<K,V> build()