Fix KafkaItemReader

This commit introduces the changes discussed in PR #664:

* Remove `OffsetsProvider`: We only need offsets to be retrieved from
the execution context in a restart scenario
* Make the reader inherit from `AbstractItemStreamItemReader` instead
of `AbstractItemCountingItemStreamItemReader`. With Kafka offsets, item
counting does not make sense
* Remove `assignTimeoutProvider`. A configurable poll timeout with
a default value is enough.
* Update the constructor of the reader to accept consumer properties,
a topic name and a list of partitions instead of leaking Kafka APIs like
`TopicPartition` and `ConsumerFactory`.

Resolves BATCH-2764

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/KafkaItemReader.java

@@ -1,11 +1,11 @@
 /*
- * Copyright 2018 the original author or authors.
+ * Copyright 2019 the original author or authors.
  *
  *  Licensed under the Apache License, Version 2.0 (the "License");
  *  you may not use this file except in compliance with the License.
  *  You may obtain a copy of the License at
  *
- *          http://www.apache.org/licenses/LICENSE-2.0
+ *          https://www.apache.org/licenses/LICENSE-2.0
  *
  *  Unless required by applicable law or agreed to in writing, software
  *  distributed under the License is distributed on an "AS IS" BASIS,
@@ -17,172 +17,172 @@
 package org.springframework.batch.item.kafka;
 
 import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.function.Supplier;
-import java.util.stream.Collectors;
+import java.util.Properties;
 
-import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.common.TopicPartition;
+
 import org.springframework.batch.item.ExecutionContext;
-import org.springframework.batch.item.ItemReader;
-import org.springframework.batch.item.ItemStreamException;
-import org.springframework.batch.item.kafka.support.AutoCommitOffsetsProvider;
-import org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader;
-import org.springframework.beans.factory.InitializingBean;
-import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.batch.item.support.AbstractItemStreamItemReader;
 import org.springframework.util.Assert;
-import org.springframework.util.ClassUtils;
 
 /**
  * <p>
- * An {@link ItemReader} implementation for Apache Kafka.
+ * An {@link org.springframework.batch.item.ItemReader} implementation for Apache Kafka.
+ * Uses a {@link KafkaConsumer} to read data from a given topic.
+ * Multiple partitions within the same topic can be assigned to this reader.
+ * </p>
+ *
+ * <p>
+ * Since {@link KafkaConsumer} is not thread-safe, this reader is not thead-safe.
  * </p>
  *
  * @author Mathieu Ouellet
+ * @author Mahmoud Ben Hassine
  * @since 4.2
- *
  */
-public class KafkaItemReader<K, V> extends AbstractItemCountingItemStreamItemReader<V> implements InitializingBean {
+public class KafkaItemReader<K, V> extends AbstractItemStreamItemReader<V> {
 
-	private static final String TOPIC_PARTITION_OFFSET = "topic.partition.offset";
+	private static final String TOPIC_PARTITION_OFFSETS = "topic.partition.offsets";
 
-	private static final long DEFAULT_POLL_TIMEOUT = 50L;
-
-	private static final long MIN_ASSIGN_TIMEOUT = 2000L;
-
-	private final Supplier<Duration> assignTimeoutProvider = () -> Duration
-			.ofMillis(Math.max(this.pollTimeout.toMillis() * 20, MIN_ASSIGN_TIMEOUT));
-
-	private Duration pollTimeout = Duration.ofMillis(DEFAULT_POLL_TIMEOUT);
+	private static final long DEFAULT_POLL_TIMEOUT = 30L;
 
 	private List<TopicPartition> topicPartitions;
 
-	private List<String> topics;
+	private Map<TopicPartition, Long> partitionOffsets;
 
-	private ConsumerFactory<K, V> consumerFactory;
+	private KafkaConsumer<K, V> kafkaConsumer;
 
-	private Consumer<K, V> consumer;
+	private Properties consumerProperties;
 
-	private OffsetsProvider offsetsProvider;
+	private Iterator<ConsumerRecord<K, V>> consumerRecords;
 
-	private AtomicBoolean assigned = new AtomicBoolean(false);
+	private Duration pollTimeout = Duration.ofSeconds(DEFAULT_POLL_TIMEOUT);
 
-	private Map<TopicPartition, Long> offsets;
+	private boolean saveState = true;
 
-	private Iterator<ConsumerRecord<K, V>> records;
-
-	public KafkaItemReader() {
-		super();
-		setName(ClassUtils.getShortName(KafkaItemReader.class));
+	/**
+	 * Create a new {@link KafkaItemReader}.
+	 * <p><strong>{@code consumerProperties} must contain the following keys:
+	 * 'bootstrap.servers', 'group.id', 'key.deserializer' and 'value.deserializer' </strong></p>.
+	 * @param consumerProperties properties of the consumer
+	 * @param topicName name of the topic to read data from
+	 * @param partitions list of partitions to read data from
+	 */
+	public KafkaItemReader(Properties consumerProperties, String topicName, Integer... partitions) {
+		this(consumerProperties, topicName, Arrays.asList(partitions));
 	}
 
-	public void setPollTimeout(long pollTimeout) {
-		Assert.isTrue(pollTimeout >= 0, "'pollTimeout' must no be negative.");
-		this.pollTimeout = Duration.ofMillis(pollTimeout);
-	}
-
-	public void setTopicPartitions(List<TopicPartition> topicPartitions) {
-		this.topicPartitions = topicPartitions;
-	}
-
-	public void setTopics(List<String> topics) {
-		this.topics = topics;
-	}
-
-	public void setConsumerFactory(ConsumerFactory<K, V> consumerFactory) {
-		this.consumerFactory = consumerFactory;
-	}
-
-	public void setOffsetsProvider(OffsetsProvider offsetsProvider) {
-		this.offsetsProvider = offsetsProvider;
-	}
-
-	@Override
-	public void afterPropertiesSet() throws Exception {
-		Assert.state(topicPartitions != null || topics != null, "Either 'topicPartitions' or 'topics' must be provided.");
-		Assert.state(topicPartitions == null || topics == null, "Both 'topicPartitions' and 'topics' cannot be specified together.");
-		Assert.notNull(consumerFactory, "'consumerFactory' must not be null.");
-		Assert.notNull(offsetsProvider, "'offsetsProvider' must not be null.");
-		if  (consumerFactory.isAutoCommit()) {
-			Assert.state(offsetsProvider instanceof AutoCommitOffsetsProvider, "'AutoCommitOffsetsProvider' must be used if 'consumerFactory' is set to auto commit.");
+	/**
+	 * Create a new {@link KafkaItemReader}.
+	 * <p><strong>{@code consumerProperties} must contain the following keys:
+	 * 'bootstrap.servers', 'group.id', 'key.deserializer' and 'value.deserializer' </strong></p>.
+	 * @param consumerProperties properties of the consumer
+	 * @param topicName name of the topic to read data from
+	 * @param partitions list of partitions to read data from
+	 */
+	public KafkaItemReader(Properties consumerProperties, String topicName, List<Integer> partitions) {
+		Assert.notNull(consumerProperties, "Consumer properties must not be null");
+		Assert.isTrue(consumerProperties.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG),
+				ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG + " property must be provided");
+		Assert.isTrue(consumerProperties.containsKey(ConsumerConfig.GROUP_ID_CONFIG),
+				ConsumerConfig.GROUP_ID_CONFIG + " property must be provided");
+		Assert.isTrue(consumerProperties.containsKey(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG),
+				ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG + " property must be provided");
+		Assert.isTrue(consumerProperties.containsKey(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG),
+				ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG + " property must be provided");
+		this.consumerProperties = consumerProperties;
+		Assert.hasLength(topicName, "Topic name must not be null or empty");
+		Assert.isTrue(!partitions.isEmpty(), "At least one partition must be provided");
+		this.topicPartitions = new ArrayList<>();
+		for (Integer partition : partitions) {
+			this.topicPartitions.add(new TopicPartition(topicName, partition));
 		}
 	}
 
-	@Override
-	@SuppressWarnings("unchecked")
-	public void open(ExecutionContext executionContext) throws ItemStreamException {
-		super.open(executionContext);
-		try {
-			if (isSaveState() && executionContext.containsKey(TOPIC_PARTITION_OFFSET)) {
-				offsets = (Map<TopicPartition, Long>) executionContext.get(TOPIC_PARTITION_OFFSET);
-			}
-			else {
-				offsets = offsetsProvider.get(topicPartitions);
-			}
+	/**
+	 * Set a timeout for the consumer topic polling duration. Default to 30 seconds.
+	 * @param pollTimeout for the consumer poll operation
+	 */
+	public void setPollTimeout(Duration pollTimeout) {
+		Assert.notNull(pollTimeout, "pollTimeout must not be null");
+		Assert.isTrue(!pollTimeout.isZero(), "pollTimeout must not be zero");
+		Assert.isTrue(!pollTimeout.isNegative(), "pollTimeout must not be negative");
+		this.pollTimeout = pollTimeout;
+	}
 
-			if (offsets != null && !offsets.isEmpty()) {
-				offsets.forEach(consumer::seek);
+	/**
+	 * Set the flag that determines whether to save internal data for
+	 * {@link 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.
+	 * @param saveState flag value (default true).
+	 */
+	public void setSaveState(boolean saveState) {
+		this.saveState = saveState;
+	}
+
+	/**
+	 * The flag that determines whether to save internal state for restarts.
+	 * @return true if the flag was set
+	 */
+	public boolean isSaveState() {
+		return this.saveState;
+	}
+
+	@Override
+	public void open(ExecutionContext executionContext) {
+		this.kafkaConsumer = new KafkaConsumer<>(this.consumerProperties);
+		this.partitionOffsets = new HashMap<>();
+		for (TopicPartition topicPartition : this.topicPartitions) {
+			this.partitionOffsets.put(topicPartition, 0L);
+		}
+		if (this.saveState && executionContext.containsKey(TOPIC_PARTITION_OFFSETS)) {
+			Map<TopicPartition, Long> offsets = (Map<TopicPartition, Long>) executionContext.get(TOPIC_PARTITION_OFFSETS);
+			for (Map.Entry<TopicPartition, Long> entry : offsets.entrySet()) {
+				this.partitionOffsets.put(entry.getKey(), entry.getValue() == 0 ? 0 : entry.getValue() + 1);
 			}
 		}
-		catch (Exception e) {
-			throw new ItemStreamException("Failed to initialize the reader", e);
-		}
+		this.kafkaConsumer.assign(this.topicPartitions);
+		this.partitionOffsets.forEach(this.kafkaConsumer::seek);
 	}
 
 	@Override
-	protected void doOpen() throws Exception {
-		consumer = consumerFactory.createConsumer();
-		if (topics != null) {
-			topicPartitions = topics.stream()
-					.flatMap(topic -> consumer.partitionsFor(topic).stream())
-					.map(partitionInfo -> new TopicPartition(partitionInfo.topic(), partitionInfo.partition()))
-					.collect(Collectors.toList());
+	public V read() {
+		if (this.consumerRecords == null || !this.consumerRecords.hasNext()) {
+			this.consumerRecords = this.kafkaConsumer.poll(this.pollTimeout).iterator();
 		}
-		consumer.assign(topicPartitions);
-		offsetsProvider.setConsumer(consumer);
-	}
-
-	@Override
-	protected void jumpToItem(int itemIndex) throws Exception {
-	}
-
-	@Override
-	protected V doRead() throws Exception {
-		if (records == null || !records.hasNext()) {
-			records = doPoll();
-		}
-		if (records.hasNext()) {
-			ConsumerRecord<K, V> record = records.next();
-			offsets.put(new TopicPartition(record.topic(), record.partition()), record.offset());
+		if (this.consumerRecords.hasNext()) {
+			ConsumerRecord<K, V> record = this.consumerRecords.next();
+			this.partitionOffsets.put(new TopicPartition(record.topic(), record.partition()), record.offset());
 			return record.value();
 		}
-		return null;
-	}
-
-	protected Iterator<ConsumerRecord<K, V>> doPoll() {
-		return consumer.poll(assigned.getAndSet(true) ? pollTimeout : assignTimeoutProvider.get()).iterator();
-	}
-
-	@Override
-	public void update(ExecutionContext executionContext) throws ItemStreamException {
-		super.update(executionContext);
-		if (isSaveState()) {
-			Assert.notNull(executionContext, "ExecutionContext must not be null");
-			executionContext.put(TOPIC_PARTITION_OFFSET, offsets);
+		else {
+			return null;
 		}
 	}
 
 	@Override
-	protected void doClose() throws Exception {
-		records = null;
-		offsets = null;
-		assigned.set(false);
-		if (consumer != null) {
-			consumer.close();
+	public void update(ExecutionContext executionContext) {
+		if (this.saveState) {
+			executionContext.put(TOPIC_PARTITION_OFFSETS, new HashMap<>(this.partitionOffsets));
+		}
+		this.kafkaConsumer.commitSync();
+	}
+
+	@Override
+	public void close() {
+		if (this.kafkaConsumer != null) {
+			this.kafkaConsumer.close();
 		}
 	}
 }

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/OffsetsProvider.java

@@ -1,37 +0,0 @@
-package org.springframework.batch.item.kafka;
-
-import java.util.List;
-import java.util.Map;
-
-import org.apache.kafka.clients.consumer.Consumer;
-import org.apache.kafka.common.TopicPartition;
-
-/**
- * A convenient strategy for providing initial externally stored offsets for the {@link KafkaItemReader} to
- * seeks to.
- *
- * @author Mathieu Ouellet
- * @since 4.2
- */
-public interface OffsetsProvider {
-
-	/**
-	 * <p>
-	 * Map of assigned topic-partitions and the offset, or read position, at which the {@link KafkaItemReader} should
-	 * start.
-	 * </p>
-	 *
-	 * @param topicPartitions list of assigned topic-partitions to get offset for
-	 * @return map of offset by topic-partition
-	 */
-	Map<TopicPartition, Long> get(List<TopicPartition> topicPartitions);
-
-	/**
-	 * <p>
-	 * Inject a {@link Consumer} that can be used to fetch internally stored offsets and/or topic-partition assignment.
-	 * </p>
-	 *
-	 * @param consumer the {@link Consumer} to set
-	 */
-	void setConsumer(Consumer<?, ?> consumer);
-}

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/builder/KafkaItemReaderBuilder.java

@@ -1,11 +1,11 @@
 /*
- * Copyright 2018 the original author or authors.
+ * Copyright 2019 the original author or authors.
  *
  *  Licensed under the Apache License, Version 2.0 (the "License");
  *  you may not use this file except in compliance with the License.
  *  You may obtain a copy of the License at
  *
- *          http://www.apache.org/licenses/LICENSE-2.0
+ *          https://www.apache.org/licenses/LICENSE-2.0
  *
  *  Unless required by applicable law or agreed to in writing, software
  *  distributed under the License is distributed on an "AS IS" BASIS,
@@ -16,45 +16,43 @@
 
 package org.springframework.batch.item.kafka.builder;
 
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
+import java.util.Properties;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
 
-import org.apache.kafka.clients.consumer.Consumer;
-import org.apache.kafka.common.TopicPartition;
 import org.springframework.batch.item.kafka.KafkaItemReader;
-import org.springframework.batch.item.kafka.OffsetsProvider;
-import org.springframework.batch.item.kafka.support.AutoCommitOffsetsProvider;
-import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.util.Assert;
 
 /**
- * A builder implementation for the {@link KafkaItemReader}
+ * A builder implementation for the {@link KafkaItemReader}.
  *
  * @author Mathieu Ouellet
+ * @author Mahmoud Ben Hassine
  * @since 4.2
  * @see KafkaItemReader
  */
 public class KafkaItemReaderBuilder<K, V> {
 
-	private ConsumerFactory<K, V> consumerFactory;
+	private Properties consumerProperties;
 
-	private List<TopicPartition> topicPartitions;
+	private String topic;
 
-	private List<String> topics;
+	private List<Integer> partitions = new ArrayList<>();
 
-	private OffsetsProvider offsetsProvider = new AutoCommitOffsetsProvider();
-
-	private long pollTimeout = 50L;
+	private Duration pollTimeout = Duration.ofSeconds(30L);
 
 	private boolean saveState = true;
 
 	private String name;
 
-	private int maxItemCount = Integer.MAX_VALUE;
-
 	/**
-	 * Configure if the state of the {@link org.springframework.batch.item.ItemStreamSupport} should be persisted within
-	 * the {@link org.springframework.batch.item.ExecutionContext} for restart purposes.
-	 *
+	 * Configure if the state of the {@link org.springframework.batch.item.ItemStreamSupport}
+	 * should be persisted within the {@link org.springframework.batch.item.ExecutionContext}
+	 * for restart purposes.
 	 * @param saveState defaults to true
 	 * @return The current instance of the builder.
 	 */
@@ -64,9 +62,8 @@ public class KafkaItemReaderBuilder<K, V> {
 	}
 
 	/**
-	 * The name used to calculate the key within the {@link org.springframework.batch.item.ExecutionContext}. Required
-	 * if {@link #saveState(boolean)} is set to true.
-	 *
+	 * The name used to calculate the key within the {@link org.springframework.batch.item.ExecutionContext}.
+	 * Required if {@link #saveState(boolean)} is set to true.
 	 * @param name name of the reader instance
 	 * @return The current instance of the builder.
 	 * @see org.springframework.batch.item.ItemStreamSupport#setName(String)
@@ -77,73 +74,53 @@ public class KafkaItemReaderBuilder<K, V> {
 	}
 
 	/**
-	 * Configure the max number of items to be read.
-	 *
-	 * @param maxItemCount the max items to be read
+	 * Configure the underlying consumer properties.
+	 * <p><strong>{@code consumerProperties} must contain the following keys:
+	 * 'bootstrap.servers', 'group.id', 'key.deserializer' and 'value.deserializer' </strong></p>.
+	 * @param consumerProperties properties of the consumer
 	 * @return The current instance of the builder.
-	 * @see org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader#setMaxItemCount(int)
 	 */
-	public KafkaItemReaderBuilder<K, V> maxItemCount(int maxItemCount) {
-		this.maxItemCount = maxItemCount;
+	public KafkaItemReaderBuilder<K, V> consumerProperties(Properties consumerProperties) {
+		this.consumerProperties = consumerProperties;
 		return this;
 	}
 
 	/**
-	 * The {@link ConsumerFactory} implementation to produce a new {@link Consumer} instance for the reader.
-	 *
-	 * @param consumerFactory
+	 * A list of partitions to manually assign to the consumer.
+	 * @param partitions list of partitions to assign to the consumer
 	 * @return The current instance of the builder.
-	 * @see KafkaItemReader#setConsumerFactory(ConsumerFactory)
 	 */
-	public KafkaItemReaderBuilder<K, V> consumerFactory(ConsumerFactory<K, V> consumerFactory) {
-		this.consumerFactory = consumerFactory;
+	public KafkaItemReaderBuilder<K, V> partitions(Integer... partitions) {
+		return partitions(Arrays.asList(partitions));
+	}
+
+	/**
+	 * A list of partitions to manually assign to the consumer.
+	 * @param partitions list of partitions to assign to the consumer
+	 * @return The current instance of the builder.
+	 */
+	public KafkaItemReaderBuilder<K, V> partitions(List<Integer> partitions) {
+		this.partitions = partitions;
 		return this;
 	}
 
 	/**
-	 * A list of {@link TopicPartition}s to manually assign the consumer.
-	 *
-	 * @param topicPartitions list of partitions to assign the consumer
+	 * A topic name to manually assign to the consumer.
+	 * @param topic name to assign to the consumer
 	 * @return The current instance of the builder.
-	 * @see KafkaItemReader#setTopicPartitions(List)
 	 */
-	public KafkaItemReaderBuilder<K, V> topicPartitions(List<TopicPartition> topicPartitions) {
-		this.topicPartitions = topicPartitions;
+	public KafkaItemReaderBuilder<K, V> topic(String topic) {
+		this.topic = topic;
 		return this;
 	}
 
 	/**
-	 * A list of topics to manually assign the consumer.
-	 *
-	 * @param topics list of topics to assign the consumer
+	 * Set the pollTimeout for the poll() operations. Default to 30 seconds.
+	 * @param pollTimeout timeout for the poll operation
 	 * @return The current instance of the builder.
-	 * @see KafkaItemReader#setTopics(List)
+	 * @see KafkaItemReader#setPollTimeout(Duration)
 	 */
-	public KafkaItemReaderBuilder<K, V> topics(List<String> topics) {
-		this.topics = topics;
-		return this;
-	}
-
-	/**
-	 * The {@link OffsetsProvider} implementation to provide initial offsets.
-	 *
-	 * @param offsetsProvider
-	 * @return The current instance of the builder.
-	 * @see KafkaItemReader#setOffsetsProvider(OffsetsProvider)
-	 */
-	public KafkaItemReaderBuilder<K,V> offsetsProvider(OffsetsProvider offsetsProvider)  {
-		this.offsetsProvider = offsetsProvider;
-		return this;
-	}
-
-	/**
-	 * Set the pollTimeout for the poll() operations.
-	 *
-	 * @param pollTimeout default to 50ms
-	 * @return The current instance of the builder.
-	 * @see KafkaItemReader#setPollTimeout(long)
-	 */
-	public KafkaItemReaderBuilder<K, V> pollTimeout(long pollTimeout) {
+	public KafkaItemReaderBuilder<K, V> pollTimeout(Duration pollTimeout) {
 		this.pollTimeout = pollTimeout;
 		return this;
 	}
@@ -152,24 +129,25 @@ public class KafkaItemReaderBuilder<K, V> {
 		if (this.saveState) {
 			Assert.hasText(this.name, "A name is required when saveState is set to true");
 		}
-		Assert.state(this.topicPartitions != null || this.topics != null, "Either 'topicPartitions' or 'topics' must be provided.");
-		Assert.state(this.topicPartitions == null || this.topics == null, "Both 'topicPartitions' and 'topics' cannot be specified together.");
-		Assert.isTrue(this.pollTimeout >= 0, "pollTimeout must not be negative.");
-		Assert.notNull(this.consumerFactory, "'consumerFactory' must not be null.");
-		Assert.notNull(this.offsetsProvider, "'offsetsProvider' must not be null.");
-		if  (this.consumerFactory.isAutoCommit()) {
-			Assert.state(this.offsetsProvider instanceof AutoCommitOffsetsProvider, "'AutoCommitOffsetsProvider' must be used if 'consumerFactory' is set to auto commit.");
-		}
+		Assert.notNull(consumerProperties, "Consumer properties must not be null");
+		Assert.isTrue(consumerProperties.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG),
+				ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG + " property must be provided");
+		Assert.isTrue(consumerProperties.containsKey(ConsumerConfig.GROUP_ID_CONFIG),
+				ConsumerConfig.GROUP_ID_CONFIG + " property must be provided");
+		Assert.isTrue(consumerProperties.containsKey(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG),
+				ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG + " property must be provided");
+		Assert.isTrue(consumerProperties.containsKey(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG),
+				ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG + " property must be provided");
+		Assert.hasLength(topic, "Topic name must not be null or empty");
+		Assert.notNull(pollTimeout, "pollTimeout must not be null");
+		Assert.isTrue(!pollTimeout.isZero(), "pollTimeout must not be zero");
+		Assert.isTrue(!pollTimeout.isNegative(), "pollTimeout must not be negative");
+		Assert.isTrue(!partitions.isEmpty(), "At least one partition must be provided");
 
-		KafkaItemReader<K, V> reader = new KafkaItemReader<>();
-		reader.setConsumerFactory(this.consumerFactory);
-		reader.setTopicPartitions(this.topicPartitions);
-		reader.setTopics(this.topics);
-		reader.setOffsetsProvider(this.offsetsProvider);
+		KafkaItemReader<K, V> reader = new KafkaItemReader<>(this.consumerProperties, this.topic, this.partitions);
 		reader.setPollTimeout(this.pollTimeout);
 		reader.setSaveState(this.saveState);
 		reader.setName(this.name);
-		reader.setMaxItemCount(this.maxItemCount);
 		return reader;
 	}
 }

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/support/AutoCommitOffsetsProvider.java

@@ -1,32 +0,0 @@
-package org.springframework.batch.item.kafka.support;
-
-import java.util.List;
-import java.util.Map;
-
-import org.apache.kafka.clients.consumer.Consumer;
-import org.apache.kafka.common.TopicPartition;
-import org.springframework.batch.item.kafka.OffsetsProvider;
-
-/**
- * Noop implementation of {@link OffsetsProvider} to use with automatic offset committing. This is the only
- * implementation allowed when the KafkaConsumer has 'enable.auto.commit' set to true. See 'auto.offset.reset' config
- * for initial offset strategy options.
- *
- * @author Mathieu Ouellet
- * @since 4.2
- */
-public class AutoCommitOffsetsProvider implements OffsetsProvider {
-
-	/**
-	 * @return null, preventing a call to
-	 * {@link org.apache.kafka.clients.consumer.Consumer#seek(org.apache.kafka.common.TopicPartition, long)}
-	 */
-	@Override
-	public Map<TopicPartition, Long> get(List<TopicPartition> topicPartitions) {
-		return null;
-	}
-
-	@Override
-	public void setConsumer(Consumer<?, ?> consumer) {
-	}
-}

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/support/BeginningOffsetsProvider.java

@@ -1,34 +0,0 @@
-package org.springframework.batch.item.kafka.support;
-
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
-
-import org.apache.kafka.clients.consumer.Consumer;
-import org.apache.kafka.common.TopicPartition;
-import org.springframework.batch.item.kafka.OffsetsProvider;
-
-/**
- * <p>
- * Implementation of {@link OffsetsProvider} that returns the earliest offsets for the given topic-partitions.
- * Equivalent of using 'auto.offset.reset' set to 'earliest' with 'enable.auto.commit' set to false.
- * </p>
- *
- * @author Mathieu Ouellet
- * @see org.apache.kafka.clients.consumer.KafkaConsumer#beginningOffsets(Collection)
- * @since 4.2
- */
-public class BeginningOffsetsProvider implements OffsetsProvider {
-
-	private Consumer<?, ?> consumer;
-
-	@Override
-	public Map<TopicPartition, Long> get(List<TopicPartition> topicPartitions) {
-		return this.consumer.beginningOffsets(topicPartitions);
-	}
-
-	@Override
-	public void setConsumer(Consumer<?, ?> consumer) {
-		this.consumer = consumer;
-	}
-}

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/support/SimpleOffsetsProvider.java

@@ -1,34 +0,0 @@
-package org.springframework.batch.item.kafka.support;
-
-import java.util.List;
-import java.util.Map;
-
-import org.apache.kafka.clients.consumer.Consumer;
-import org.apache.kafka.common.TopicPartition;
-import org.springframework.batch.item.kafka.OffsetsProvider;
-
-/**
- * <p>
- * Implementation of {@link OffsetsProvider} that returns default or provided offsets for the given topic-partitions.
- * </p>
- *
- * @author Mathieu Ouellet
- * @since 4.2
- */
-public class SimpleOffsetsProvider implements OffsetsProvider {
-
-	private Map<TopicPartition, Long> offsets;
-
-	@Override
-	public Map<TopicPartition, Long> get(List<TopicPartition> topicPartitions) {
-		return this.offsets;
-	}
-
-	@Override
-	public void setConsumer(Consumer<?, ?> consumer) {
-	}
-
-	public void setOffsets(Map<TopicPartition, Long> offsets) {
-		this.offsets = offsets;
-	}
-}

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/support/TimestampOffsetsProvider.java

@@ -1,45 +0,0 @@
-package org.springframework.batch.item.kafka.support;
-
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.function.Function;
-import java.util.stream.Collectors;
-
-import org.apache.kafka.clients.consumer.Consumer;
-import org.apache.kafka.common.TopicPartition;
-import org.springframework.batch.item.kafka.OffsetsProvider;
-
-/**
- * <p>
- * Implementation of {@link OffsetsProvider} that returns offsets for the given topic-partitions by timestamp.
- * </p>
- *
- * @author Mathieu Ouellet
- * @see org.apache.kafka.clients.consumer.KafkaConsumer#offsetsForTimes(java.util.Map)
- * @since 4.2
- */
-public class TimestampOffsetsProvider implements OffsetsProvider {
-
-	private final Long timestampToSearch;
-
-	private Consumer<?, ?> consumer;
-
-	public TimestampOffsetsProvider(Long timestampToSearch) {
-		this.timestampToSearch = timestampToSearch;
-	}
-
-	@Override
-	public Map<TopicPartition, Long> get(List<TopicPartition> topicPartitions) {
-		Map<TopicPartition, Long> timestampsToSearch = topicPartitions.stream()
-				.collect(Collectors.toMap(Function.identity(), topicPartition -> timestampToSearch));
-		return consumer.offsetsForTimes(timestampsToSearch).entrySet().stream()
-				.filter(entry -> entry.getValue() != null)
-				.collect(Collectors.toMap(Entry::getKey, entry -> entry.getValue().offset()));
-	}
-
-	@Override
-	public void setConsumer(Consumer<?, ?> consumer) {
-		this.consumer = consumer;
-	}
-}