org.springframework.batch.item.kafka

Class KafkaItemReader<K,V>

java.lang.Object

org.springframework.batch.item.ItemStreamSupport

org.springframework.batch.item.support.AbstractItemStreamItemReader<V>

org.springframework.batch.item.kafka.KafkaItemReader<K,V>

All Implemented Interfaces:

ItemReader<V>, ItemStream, ItemStreamReader<V>

public class KafkaItemReader<K,V>
extends AbstractItemStreamItemReader<V>

An ItemReader implementation for Apache Kafka. Uses a KafkaConsumer to read data from a given topic. Multiple partitions within the same topic can be assigned to this reader.

Since KafkaConsumer is not thread-safe, this reader is not thread-safe.

Since:

4.2

Author:

Mathieu Ouellet, Mahmoud Ben Hassine

Constructor Summary

Constructors

Constructor and Description
KafkaItemReader(java.util.Properties consumerProperties, java.lang.String topicName, java.lang.Integer... partitions) Create a new KafkaItemReader.
KafkaItemReader(java.util.Properties consumerProperties, java.lang.String topicName, java.util.List<java.lang.Integer> partitions) Create a new KafkaItemReader.

Method Summary

Modifier and Type	Method and Description
void	close() No-op.
boolean	isSaveState() The flag that determines whether to save internal state for restarts.
void	open(ExecutionContext executionContext) No-op.
V	read() Reads a piece of input data and advance to the next one.
void	setPartitionOffsets(java.util.Map<org.apache.kafka.common.TopicPartition,java.lang.Long> partitionOffsets) Setter for partition offsets.
void	setPollTimeout(java.time.Duration pollTimeout) Set a timeout for the consumer topic polling duration.
void	setSaveState(boolean saveState) Set the flag that determines whether to save internal data for ExecutionContext.
void	update(ExecutionContext executionContext) Return empty ExecutionContext.

Methods inherited from class org.springframework.batch.item.ItemStreamSupport

getExecutionContextKey, getName, setExecutionContextName, setName

Methods inherited from class java.lang.Object

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

Constructor Detail

KafkaItemReader

public KafkaItemReader(java.util.Properties consumerProperties,
                       java.lang.String topicName,
                       java.lang.Integer... partitions)

Create a new KafkaItemReader.

consumerProperties must contain the following keys: 'bootstrap.servers', 'group.id', 'key.deserializer' and 'value.deserializer'

.

Parameters:

consumerProperties - properties of the consumer

topicName - name of the topic to read data from

partitions - list of partitions to read data from

KafkaItemReader

public KafkaItemReader(java.util.Properties consumerProperties,
                       java.lang.String topicName,
                       java.util.List<java.lang.Integer> partitions)

Create a new KafkaItemReader.

consumerProperties must contain the following keys: 'bootstrap.servers', 'group.id', 'key.deserializer' and 'value.deserializer'

.

Parameters:

consumerProperties - properties of the consumer

topicName - name of the topic to read data from

partitions - list of partitions to read data from

Method Detail

setPollTimeout

public void setPollTimeout(java.time.Duration pollTimeout)

Set a timeout for the consumer topic polling duration. Default to 30 seconds.

Parameters:

pollTimeout - for the consumer poll operation

setSaveState

public void setSaveState(boolean saveState)

Set the flag that determines whether to save internal data for ExecutionContext. Only switch this to false if you don't want to save any state from this stream, and you don't need it to be restartable. Always set it to false if the reader is being used in a concurrent environment.

Parameters:

saveState - flag value (default true).

isSaveState

public boolean isSaveState()

The flag that determines whether to save internal state for restarts.

Returns:

true if the flag was set

setPartitionOffsets

public void setPartitionOffsets(java.util.Map<org.apache.kafka.common.TopicPartition,java.lang.Long> partitionOffsets)

Setter for partition offsets. This mapping tells the reader the offset to start reading from in each partition. This is optional, defaults to starting from offset 0 in each partition. Passing an empty map makes the reader start from the offset stored in Kafka for the consumer group ID.

In case of a restart, offsets stored in the execution context will take precedence.

Parameters:

partitionOffsets - mapping of starting offset in each partition

open

public void open(ExecutionContext executionContext)

Description copied from class: ItemStreamSupport

No-op.

Specified by:

open in interface ItemStream

Overrides:

open in class ItemStreamSupport

Parameters:

executionContext - current step's ExecutionContext. Will be the executionContext from the last run of the step on a restart.

See Also:

ItemStream.open(ExecutionContext)

read

@Nullable
public V read()

Description copied from interface: ItemReader

Reads a piece of input data and advance to the next one. Implementations must return null at the end of the input data set. In a transactional setting, caller might get the same item twice from successive calls (or otherwise), if the first call was in a transaction that rolled back.

Returns:

T the item to be processed or null if the data source is exhausted

update

public void update(ExecutionContext executionContext)

Description copied from class: ItemStreamSupport

Return empty ExecutionContext.

Specified by:

update in interface ItemStream

Overrides:

update in class ItemStreamSupport

Parameters:

executionContext - to be updated

See Also:

ItemStream.update(ExecutionContext)

close

public void close()

Description copied from class: ItemStreamSupport

No-op.

Specified by:

close in interface ItemStream

Overrides:

close in class ItemStreamSupport

See Also:

ItemStream.close()