org.thales.punch.kafka.api

Class KafkaContinuousConsumerBuilder<K,V>

java.lang.Object

org.thales.punch.kafka.api.KafkaContinuousConsumerBuilder<K,V>

Type Parameters:

K - the class of the record key

V - the class of the record value

public class KafkaContinuousConsumerBuilder<K,V>
extends Object

The IKafkaContinuousConsumer builder. Refer the the kafka KafkaConsumer javadoc. Here is an example :

 kafkaReader = new KafkaConsumerBuilder<byte[], byte[]>()
				.setBootstrapServer("localhost:9092")
				.setGroupId("mytenant_apache_processing")
				.setTopic("apache")
				.setStartOffsetStrategy("earliest")
				.setMetricContext(yourMetricContext)
				.build();
 

It is handy to pass in additional kafka standard properties. You can do that using the {setProperty(String, Object) method. For example :

 kafkaReader = new KafkaReaderBuilder<byte[], byte[]>()
				.setProperty("fetch.max.bytes", 1048576)
				...
				.build();
 

Watchout by default the auto commit is disabled, you are in charge of inovking the commit method whevever need. You can use the setAutoCommit()

Author:

dimi

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	KafkaContinuousConsumerBuilder.START_OFFSET_STRATEGY One of "earliest", "last_committed", "latest", "from_offset" or "from_timestamp".

Constructor Summary

Constructors

Constructor and Description
KafkaContinuousConsumerBuilder() public ctor

Method Summary

Modifier and Type	Method and Description
IKafkaContinuousConsumer	forReceiver(IRecordHandler<K,V> receiver)
KafkaContinuousConsumerBuilder<K,V>	setAutoCommit() Start reading from the earliest offset.
KafkaContinuousConsumerBuilder<K,V>	setAutoCommitInterval(long interval) Set the auto commit interval
KafkaContinuousConsumerBuilder<K,V>	setBootstrapServer(String bootstrapServers)
KafkaContinuousConsumerBuilder<K,V>	setClientId(String consumerId)
KafkaContinuousConsumerBuilder<K,V>	setEarliestStartOffsetStrategy() Start reading from the earliest offset.
KafkaContinuousConsumerBuilder<K,V>	setFromDatetimeStartOffsetStrategy(String datetime, String format)
KafkaContinuousConsumerBuilder<K,V>	setFromDatetimeStartOffsetStrategy(String datetime, String format, boolean continueWhenNoRecords) Start reading from a specific datetime.
KafkaContinuousConsumerBuilder<K,V>	setFromOffsetStartStrategy(long offset) Start reading from an offset
KafkaContinuousConsumerBuilder<K,V>	setGroupId(String groupId)
KafkaContinuousConsumerBuilder<K,V>	setLastCommittedStartOffsetStrategy() Use a last commit strategy.
KafkaContinuousConsumerBuilder<K,V>	setLatestStartOffsetStrategy() Start reading from the latest offset.
KafkaContinuousConsumerBuilder<K,V>	setMetricContext(org.thales.punch.libraries.metrics.api.IMetricContext metricContext) Associate a metricContext to the kafka reader.
KafkaContinuousConsumerBuilder<K,V>	setOffsetTreeAckStrategy() Enable the ack and fail method of your IPartition handle to be handled using the underlyinh offset tree strategy.
KafkaContinuousConsumerBuilder<K,V>	setOffsetWatchDogTimeout(Long timeout) Set offset watchdog timeout.
KafkaContinuousConsumerBuilder<K,V>	setPartitionNumber(int num) Set a partition number if you want to consume only a given partition.
KafkaContinuousConsumerBuilder<K,V>	setProperties(Properties properties) Set additional properties, typically some defined in the kafka consumer javadocs.
KafkaContinuousConsumerBuilder<K,V>	setProperty(String name, Object value) Set additional property, typically some defined in the kafka consumer javadocs.
KafkaContinuousConsumerBuilder<K,V>	setSecurityStores(org.thales.punch.platform.api.pojo.JavaStoreSecurity securityStores)
KafkaContinuousConsumerBuilder<K,V>	setSmartFailureHandling(boolean b) If set to true, after a failure already acked or pending records will not be reemitted.
KafkaContinuousConsumerBuilder<K,V>	setTickInterval(long i)
KafkaContinuousConsumerBuilder<K,V>	setTopic(List<String> topics) Set the topics to be consumed by this consumer.
KafkaContinuousConsumerBuilder<K,V>	setTopic(String... topic)

Methods inherited from class java.lang.Object

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

Constructor Detail

KafkaContinuousConsumerBuilder

public KafkaContinuousConsumerBuilder()

public ctor

Method Detail

setBootstrapServer

public KafkaContinuousConsumerBuilder<K,V> setBootstrapServer(String bootstrapServers)

Parameters:

bootstrapServers - the bootstrap server(s) address

Returns:

this builder

setGroupId

public KafkaContinuousConsumerBuilder<K,V> setGroupId(String groupId)

Parameters:

groupId -

Returns:

this builder

setClientId

public KafkaContinuousConsumerBuilder<K,V> setClientId(String consumerId)

Parameters:

consumerId -

Returns:

this builder

setSmartFailureHandling

public KafkaContinuousConsumerBuilder<K,V> setSmartFailureHandling(boolean b)

If set to true, after a failure already acked or pending records will not be reemitted.

Parameters:

b - true or false

Returns:

this builder

setMetricContext

public KafkaContinuousConsumerBuilder<K,V> setMetricContext(org.thales.punch.libraries.metrics.api.IMetricContext metricContext)

Associate a metricContext to the kafka reader. If not provided, no metric will be published.

The metric context will be derivated for each partition so as to publish a list of per partition metrics.

Parameters:

metricContext - your metric context.

Returns:

this builder

setTopic

public KafkaContinuousConsumerBuilder<K,V> setTopic(String... topic)

Parameters:

topic -

Returns:

this builder

setPartitionNumber

public KafkaContinuousConsumerBuilder<K,V> setPartitionNumber(int num)

Set a partition number if you want to consume only a given partition.

Parameters:

num - the partition number. If you call this with -1 it has no effect, and the reader will consume all partitions.

Returns:

this builder

setAutoCommit

public KafkaContinuousConsumerBuilder<K,V> setAutoCommit()

Start reading from the earliest offset.

Returns:

this builder

setAutoCommitInterval

public KafkaContinuousConsumerBuilder<K,V> setAutoCommitInterval(long interval)

Set the auto commit interval

Parameters:

interval -

Returns:

setEarliestStartOffsetStrategy

public KafkaContinuousConsumerBuilder<K,V> setEarliestStartOffsetStrategy()

Start reading from the earliest offset.

Returns:

this builder

setLatestStartOffsetStrategy

public KafkaContinuousConsumerBuilder<K,V> setLatestStartOffsetStrategy()

Start reading from the latest offset.

Returns:

this builder

setLastCommittedStartOffsetStrategy

public KafkaContinuousConsumerBuilder<K,V> setLastCommittedStartOffsetStrategy()

Use a last commit strategy. The KafkaReader will automatically commit the offset you acknowledge. It will do that regularly every autoCommitIntervalMs.

Returns:

this builder

setOffsetWatchDogTimeout

public KafkaContinuousConsumerBuilder<K,V> setOffsetWatchDogTimeout(Long timeout)

Set offset watchdog timeout. if too much time goes by without a partition committed offset going up, then the JVM should suicide.

Returns:

this builder

setOffsetTreeAckStrategy

public KafkaContinuousConsumerBuilder<K,V> setOffsetTreeAckStrategy()

Enable the ack and fail method of your IPartition handle to be handled using the underlyinh offset tree strategy. What this means is the acked and failed record will be dealt in a way not to skipp unacknowledged record.

This mode is mandatory ofr storm or storm like streaming processing i f you care about at least once semantics.

Returns:

setSecurityStores

public KafkaContinuousConsumerBuilder<K,V> setSecurityStores(org.thales.punch.platform.api.pojo.JavaStoreSecurity securityStores)

setFromDatetimeStartOffsetStrategy

public KafkaContinuousConsumerBuilder<K,V> setFromDatetimeStartOffsetStrategy(String datetime,
                                                                              String format,
                                                                              boolean continueWhenNoRecords)

Start reading from a specific datetime. We find the latest message written to Kafka with this timestamp (or earlier) and we start reading from there.

Parameters:

datetime - the date value

format - the date format. Accepted values are "ISO" and "epoch_ms".

continueWhenNoRecords - indicates whether to throw an error or to continue to initialize partitions when no records are found for the given datetime

Returns:

this builder

setFromDatetimeStartOffsetStrategy

public KafkaContinuousConsumerBuilder<K,V> setFromDatetimeStartOffsetStrategy(String datetime,
                                                                              String format)

setFromOffsetStartStrategy

public KafkaContinuousConsumerBuilder<K,V> setFromOffsetStartStrategy(long offset)

Start reading from an offset

Parameters:

offset - the starting offset

Returns:

this builder

forReceiver

public IKafkaContinuousConsumer forReceiver(IRecordHandler<K,V> receiver)

Parameters:

receiver - the record receiver

Returns:

the final kafka consumer

setProperty

public KafkaContinuousConsumerBuilder<K,V> setProperty(String name,
                                                       Object value)

Set additional property, typically some defined in the kafka consumer javadocs.

Parameters:

name - the property name

value - the property value

Returns:

this builder

setProperties

public KafkaContinuousConsumerBuilder<K,V> setProperties(Properties properties)

Set additional properties, typically some defined in the kafka consumer javadocs.

Be careful this method simply put all the properties you pass as is. If a topic is present in there it will be considered.

Parameters:

properties - the properties

Returns:

this builder

setTopic

public KafkaContinuousConsumerBuilder<K,V> setTopic(List<String> topics)

Set the topics to be consumed by this consumer.

Parameters:

topics - the list of topics

Returns:

this builder

setTickInterval

public KafkaContinuousConsumerBuilder<K,V> setTickInterval(long i)