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

    All Methods Instance Methods Concrete Methods
    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, 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()`