org.thales.punch.kafka.impl

Class KafkaConsumerImpl<K,V>

java.lang.Object

org.thales.punch.kafka.impl.KafkaConsumerImpl<K,V>

Type Parameters:

K - the class of the records key

V - the class of the records value

All Implemented Interfaces:

IKafkaContinuousConsumer

public class KafkaConsumerImpl<K,V>
extends Object
implements IKafkaContinuousConsumer

The IKafkaReader implementation. It is extremely simple. It basically encapsulates a KafkaConsumer. What it adds is the commit handling so that committing is only performed for offsets acked by the calling application.

Author:

dimi

Constructor Summary

Constructors

Constructor and Description
KafkaConsumerImpl(Properties props, String[] topics, int partitionNumber, KafkaContinuousConsumerBuilder.START_OFFSET_STRATEGY startOffsetStrategy, long startupOffset, long fromTimestamp, Long offsetWatchDogTimeout, boolean continueWhenNoRecords, org.thales.punch.platform.api.pojo.JavaStoreSecurity securityStores) Ctor.

Method Summary

Modifier and Type	Method and Description
void	close() Close this consumer
void	commit() Commit all the offsets for all the partitions managed by your reader.
int	nextMessages(Duration timeout) Fetch incoming messages, if any.
KafkaConsumerImpl<K,V>	setMetricContext(org.thales.punch.libraries.metrics.api.IMetricContext metricContext) Provide a metric context to this consumer
KafkaConsumerImpl<K,V>	setOffsetTree(boolean useOffsetTree) Request an offset tree t=strategy.
KafkaConsumerImpl<K,V>	setReceiver(IRecordHandler<K,V> receiver) PSet the receiver to handle incoming records.
KafkaConsumerImpl<K,V>	setSmartFailureHandling(boolean smartFailureHandling) Set the smart failure handling.
KafkaConsumerImpl<K,V>	setTickInterval(long tickInterval) Set the tick interval.

Methods inherited from class java.lang.Object

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

Constructor Detail

KafkaConsumerImpl

public KafkaConsumerImpl(Properties props,
                         String[] topics,
                         int partitionNumber,
                         KafkaContinuousConsumerBuilder.START_OFFSET_STRATEGY startOffsetStrategy,
                         long startupOffset,
                         long fromTimestamp,
                         Long offsetWatchDogTimeout,
                         boolean continueWhenNoRecords,
                         org.thales.punch.platform.api.pojo.JavaStoreSecurity securityStores)

Ctor.

Parameters:

props - the properties

topics - the topics

partitionNumber - -1 to read all partitions, gte 0 to read a given partition

startOffsetStrategy - one of EARLIEST, LATEST, LAST_COMMITTED or FROM_OFFSET

startupOffset - only relevant with the FROM_OFFSET strategy. Indicates the offset to start with.

fromTimestamp - only relevant with the FROM_TIMESTAMP strategy. Indicates the timestamp to begin with.

offsetWatchDogTimeout -

continueWhenNoRecords - only relevant with the FROM_DATETIME strategy. Indicates whether to throw an error or to continue to initialize partitions when no records are found for the given datetime

securityStores - the Object containing the truststore and keystore information. If not null, activate ssl.

Method Detail

setReceiver

public KafkaConsumerImpl<K,V> setReceiver(IRecordHandler<K,V> receiver)

PSet the receiver to handle incoming records.

Parameters:

receiver - your receiver

Returns:

this consumer

setMetricContext

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

Provide a metric context to this consumer

Parameters:

metricContext - your metric context

Returns:

this consumer

setOffsetTree

public KafkaConsumerImpl<K,V> setOffsetTree(boolean useOffsetTree)

Request an offset tree t=strategy. This strategy only makes sense if you are working with a consumer that ack or fail records continuously.

Parameters:

useOffsetTree - true or false

Returns:

this consumer

setSmartFailureHandling

public KafkaConsumerImpl<K,V> setSmartFailureHandling(boolean smartFailureHandling)

Set the smart failure handling. If true, records are not replayed if already in the pipe of your application. It only makes sense with the offset tree strategy.

Parameters:

smartFailureHandling -

Returns:

setTickInterval

public KafkaConsumerImpl<K,V> setTickInterval(long tickInterval)

Set the tick interval. It is used to upcall your receiver periodically.

Parameters:

tickInterval -

Returns:

close

public void close()

Description copied from interface: IKafkaContinuousConsumer

Close this consumer

Specified by:

close in interface IKafkaContinuousConsumer

nextMessages

public int nextMessages(Duration timeout)

Description copied from interface: IKafkaContinuousConsumer

Fetch incoming messages, if any.

This method is blocking. It encapsulate a call to the underlying KafkaConsumer poll. If some records are fetched from your topic. It will invoke your IRecordHandler in a reentrant manner.

Specified by:

nextMessages in interface IKafkaContinuousConsumer

Parameters:

timeout - the blocking timeout in milliseconds

Returns:

the number of records read on all partitions.

commit

public void commit()

Description copied from interface: IKafkaContinuousConsumer

Commit all the offsets for all the partitions managed by your reader. If you need to work at partition level, use the IPartition interface instead.

Specified by:

commit in interface IKafkaContinuousConsumer