Package 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
  Description
  `KafkaItemReader(Properties consumerProperties, String topicName, Integer... partitions)`
  Create a new `KafkaItemReader`.
  `KafkaItemReader(Properties consumerProperties, String topicName, List<Integer> partitions)`
  Create a new `KafkaItemReader`.

- ## Method Summary

  Modifier and Type
  Method
  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(Map<org.apache.kafka.common.TopicPartition,Long> partitionOffsets)`
  Setter for partition offsets.
  `void`
  `setPollTimeout(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, setExecutionContextName, setName`

  ### Methods inherited from class java.lang.Object

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

- ## Constructor Details

  - ### KafkaItemReader

    public KafkaItemReader(Properties consumerProperties, String topicName, 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(Properties consumerProperties, String topicName, List<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 Details

  - ### setPollTimeout

    public void setPollTimeout(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(Map<org.apache.kafka.common.TopicPartition,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()`