diff --git a/build.gradle b/build.gradle
index 25428e95c..5f3c0f229 100644
--- a/build.gradle
+++ b/build.gradle
@@ -102,6 +102,7 @@ allprojects {
 		micrometerVersion = '1.1.4'
 		prometheusPushgatewayVersion = '0.6.0'
 		docResourcesVersion = '0.1.1.RELEASE'
+		assertjVersion='3.12.2'
 	}
 
 	apply plugin: 'idea'
@@ -325,6 +326,8 @@ project('spring-batch-infrastructure') {
 		testCompile "org.mockito:mockito-core:$mockitoVersion"
 		testCompile "org.xerial:sqlite-jdbc:$sqliteVersion"
 		testCompile "javax.xml.bind:jaxb-api:$jaxbApiVersion"
+		testCompile "org.springframework.kafka:spring-kafka-test:$springKafkaVersion"
+		testCompile "org.assertj:assertj-core:$assertjVersion"
 
 		testRuntime "com.sun.mail:javax.mail:$javaMailVersion"
 		testRuntime "org.codehaus.groovy:groovy-jsr223:$groovyVersion"
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/KafkaItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/KafkaItemReader.java
index 54ba0e7c0..a2cc6aca5 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/KafkaItemReader.java
+++ b/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();
 		}
 	}
 }
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/OffsetsProvider.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/OffsetsProvider.java
deleted file mode 100644
index 7c898624b..000000000
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/OffsetsProvider.java
+++ /dev/null
@@ -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);
-}
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/builder/KafkaItemReaderBuilder.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/builder/KafkaItemReaderBuilder.java
index f2fbe4f5b..901ce2bbd 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/builder/KafkaItemReaderBuilder.java
+++ b/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;
 	}
 }
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/support/AutoCommitOffsetsProvider.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/support/AutoCommitOffsetsProvider.java
deleted file mode 100644
index 630d2858e..000000000
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/support/AutoCommitOffsetsProvider.java
+++ /dev/null
@@ -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) {
-	}
-}
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/support/BeginningOffsetsProvider.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/support/BeginningOffsetsProvider.java
deleted file mode 100644
index 2a518fb16..000000000
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/support/BeginningOffsetsProvider.java
+++ /dev/null
@@ -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;
-	}
-}
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/support/SimpleOffsetsProvider.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/support/SimpleOffsetsProvider.java
deleted file mode 100644
index b780e7025..000000000
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/support/SimpleOffsetsProvider.java
+++ /dev/null
@@ -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;
-	}
-}
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/support/TimestampOffsetsProvider.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/support/TimestampOffsetsProvider.java
deleted file mode 100644
index d5a81f24a..000000000
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/kafka/support/TimestampOffsetsProvider.java
+++ /dev/null
@@ -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;
-	}
-}
diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/kafka/KafkaItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/kafka/KafkaItemReaderTests.java
index d03dacb7c..6e419e689 100644
--- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/kafka/KafkaItemReaderTests.java
+++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/kafka/KafkaItemReaderTests.java
@@ -1,167 +1,312 @@
+/*
+ * 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
+ *
+ *          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,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
 package org.springframework.batch.item.kafka;
 
 import java.time.Duration;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Properties;
 
-import org.apache.kafka.clients.consumer.Consumer;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.StringDeserializer;
 import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
 import org.junit.Test;
-import org.mockito.Mock;
-import org.mockito.MockitoAnnotations;
+
 import org.springframework.batch.item.ExecutionContext;
-import org.springframework.kafka.core.ConsumerFactory;
-
-import static java.util.Arrays.asList;
-import static java.util.Collections.singletonList;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
 
+import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.fail;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.BDDMockito.given;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
 
 /**
  * @author Mathieu Ouellet
+ * @author Mahmoud Ben Hassine
  */
 public class KafkaItemReaderTests {
 
-	private static final TopicPartition TOPIC_PARTITION = new TopicPartition("topic", 0);
-
-	@Mock
-	private ConsumerFactory<String, String> consumerFactory;
-
-	@Mock
-	private Consumer<String, String> consumer;
-
-	@Mock
-	private OffsetsProvider offsetsProvider;
+	@ClassRule
+	public static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1);
 
 	private KafkaItemReader<String, String> reader;
+	private KafkaTemplate<String, String> template;
+	private Properties consumerProperties;
+
+	@BeforeClass
+	public static void setUpTopics() {
+		embeddedKafka.getEmbeddedKafka().addTopics(
+				new NewTopic("topic1", 1, (short) 1),
+				new NewTopic("topic2", 2, (short) 1),
+				new NewTopic("topic3", 1, (short) 1),
+				new NewTopic("topic4", 2, (short) 1)
+		);
+	}
 
 	@Before
-	public void setUp() throws Exception {
-		MockitoAnnotations.initMocks(this);
-		Map<String, Object> config = new HashMap<>();
-		config.put("max.poll.records", 2);
-		config.put("enable.auto.commit", false);
-		when(consumerFactory.getConfigurationProperties()).thenReturn(config);
-		when(consumerFactory.createConsumer()).thenReturn(consumer);
-		reader = new KafkaItemReader<>();
-		reader.setConsumerFactory(consumerFactory);
-		reader.setOffsetsProvider(offsetsProvider);
-		reader.setTopicPartitions(singletonList(TOPIC_PARTITION));
-		reader.setSaveState(true);
-		reader.setPollTimeout(50L);
-		reader.afterPropertiesSet();
+	public void setUp() {
+		Map<String, Object> producerProperties = KafkaTestUtils.producerProps(embeddedKafka.getEmbeddedKafka());
+		ProducerFactory<String, String> producerFactory = new DefaultKafkaProducerFactory<>(producerProperties);
+		this.template = new KafkaTemplate<>(producerFactory);
+
+		this.consumerProperties = new Properties();
+		this.consumerProperties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
+				embeddedKafka.getEmbeddedKafka().getBrokersAsString());
+		this.consumerProperties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "1");
+		this.consumerProperties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
+				StringDeserializer.class.getName());
+		this.consumerProperties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
+				StringDeserializer.class.getName());
 	}
 
 	@Test
-	public void testAfterPropertiesSet() throws Exception {
-		reader = new KafkaItemReader<>();
-
+	public void testValidation() {
 		try {
-			reader.afterPropertiesSet();
+			new KafkaItemReader<>(null, "topic", 0);
 			fail("Expected exception was not thrown");
 		}
-		catch (IllegalStateException ignore) {
+		catch (IllegalArgumentException exception) {
+			assertEquals("Consumer properties must not be null", exception.getMessage());
 		}
 
-		reader.setTopicPartitions(singletonList(new TopicPartition("topic", 0)));
-		reader.setTopics(singletonList("topic"));
 		try {
-			reader.afterPropertiesSet();
+			new KafkaItemReader<>(new Properties(), "topic", 0);
 			fail("Expected exception was not thrown");
 		}
-		catch (IllegalStateException ignore) {
+		catch (IllegalArgumentException exception) {
+			assertEquals("bootstrap.servers property must be provided", exception.getMessage());
 		}
 
-		reader.setTopics(null);
+		Properties consumerProperties = new Properties();
+		consumerProperties.put("bootstrap.servers", embeddedKafka.getEmbeddedKafka());
 		try {
-			reader.afterPropertiesSet();
+			new KafkaItemReader<>(consumerProperties, "topic", 0);
 			fail("Expected exception was not thrown");
 		}
-		catch (IllegalArgumentException ignore) {
+		catch (IllegalArgumentException exception) {
+			assertEquals("group.id property must be provided", exception.getMessage());
 		}
 
-		reader.setConsumerFactory(consumerFactory);
+		consumerProperties.put("group.id", "1");
 		try {
-			reader.afterPropertiesSet();
+			new KafkaItemReader<>(consumerProperties, "topic", 0);
 			fail("Expected exception was not thrown");
 		}
-		catch (IllegalArgumentException ignore) {
+		catch (IllegalArgumentException exception) {
+			assertEquals("key.deserializer property must be provided", exception.getMessage());
 		}
 
-		reader.setOffsetsProvider(offsetsProvider);
-		reader.afterPropertiesSet();
+		consumerProperties.put("key.deserializer", StringDeserializer.class.getName());
+		try {
+			new KafkaItemReader<>(consumerProperties, "topic", 0);
+			fail("Expected exception was not thrown");
+		}
+		catch (IllegalArgumentException exception) {
+			assertEquals("value.deserializer property must be provided", exception.getMessage());
+		}
+
+		consumerProperties.put("value.deserializer", StringDeserializer.class.getName());
+		try {
+			new KafkaItemReader<>(consumerProperties, "", 0);
+			fail("Expected exception was not thrown");
+		}
+		catch (IllegalArgumentException exception) {
+			assertEquals("Topic name must not be null or empty", exception.getMessage());
+		}
+
+		try {
+			this.reader = new KafkaItemReader<>(consumerProperties, "topic");
+			fail("Expected exception was not thrown");
+		}
+		catch (Exception exception) {
+			assertEquals("At least one partition must be provided", exception.getMessage());
+		}
+
+		try {
+			this.reader = new KafkaItemReader<>(consumerProperties, "topic", 0);
+		}
+		catch (Exception exception) {
+			fail("Must not throw an exception when configuration is valid");
+		}
+
+		try {
+			this.reader.setPollTimeout(null);
+			fail("Expected exception was not thrown");
+		}
+		catch (IllegalArgumentException exception) {
+			assertEquals("pollTimeout must not be null", exception.getMessage());
+		}
+
+		try {
+			this.reader.setPollTimeout(Duration.ZERO);
+			fail("Expected exception was not thrown");
+		}
+		catch (IllegalArgumentException exception) {
+			assertEquals("pollTimeout must not be zero", exception.getMessage());
+		}
+
+		try {
+			this.reader.setPollTimeout(Duration.ofSeconds(-1));
+			fail("Expected exception was not thrown");
+		}
+		catch (IllegalArgumentException exception) {
+			assertEquals("pollTimeout must not be negative", exception.getMessage());
+		}
 	}
 
 	@Test
-	public void testAssignTopicPartitions() {
-		reader.open(new ExecutionContext());
-		verify(consumer).assign(singletonList(TOPIC_PARTITION));
+	public void testReadFromSinglePartition() {
+		this.template.setDefaultTopic("topic1");
+		this.template.sendDefault("val0");
+		this.template.sendDefault("val1");
+		this.template.sendDefault("val2");
+		this.template.sendDefault("val3");
+
+		this.reader = new KafkaItemReader<>(this.consumerProperties, "topic1", 0);
+		this.reader.setPollTimeout(Duration.ofSeconds(1));
+		this.reader.open(new ExecutionContext());
+
+		String item = this.reader.read();
+		assertThat(item, is("val0"));
+
+		item = this.reader.read();
+		assertThat(item, is("val1"));
+
+		item = this.reader.read();
+		assertThat(item, is("val2"));
+
+		item = this.reader.read();
+		assertThat(item, is("val3"));
+
+		item = this.reader.read();
+		assertNull(item);
+
+		this.reader.close();
 	}
 
 	@Test
-	public void testRead() throws Exception {
-		Map<TopicPartition, List<ConsumerRecord<String, String>>> records = new HashMap<>();
-		records.put(TOPIC_PARTITION, singletonList(
-				new ConsumerRecord<>(TOPIC_PARTITION.topic(), TOPIC_PARTITION.partition(), 0L, "key0", "val0")));
-		when(consumer.poll(any())).thenReturn(new ConsumerRecords<>(records));
+	public void testReadFromMultiplePartitions() {
+		this.template.setDefaultTopic("topic2");
+		this.template.sendDefault("val0");
+		this.template.sendDefault("val1");
+		this.template.sendDefault("val2");
+		this.template.sendDefault("val3");
 
-		reader.open(new ExecutionContext());
-		String read = reader.read();
-		assertThat(read, is("val0"));
+		this.reader = new KafkaItemReader<>(this.consumerProperties, "topic2", 0, 1);
+		this.reader.setPollTimeout(Duration.ofSeconds(1));
+		this.reader.open(new ExecutionContext());
+
+		List<String> items = new ArrayList<>();
+		items.add(this.reader.read());
+		items.add(this.reader.read());
+		items.add(this.reader.read());
+		items.add(this.reader.read());
+		assertThat(items, containsInAnyOrder("val0", "val1", "val2", "val3"));
+		String item = this.reader.read();
+		assertNull(item);
+
+		this.reader.close();
 	}
 
 	@Test
-	public void testPollRecords() throws Exception {
-		Map<TopicPartition, List<ConsumerRecord<String, String>>> firstPoll = new HashMap<>();
-		firstPoll.put(TOPIC_PARTITION, asList(new ConsumerRecord<>("topic", 0, 0L, "key0", "val0"),
-				new ConsumerRecord<>("topic", 0, 1L, "key1", "val1")));
-		when(consumer.poll(Duration.ofMillis(2000L))).thenReturn(new ConsumerRecords<>(firstPoll));
+	public void testReadFromSinglePartitionAfterRestart() {
+		this.template.setDefaultTopic("topic3");
+		this.template.sendDefault("val0");
+		this.template.sendDefault("val1");
+		this.template.sendDefault("val2");
+		this.template.sendDefault("val3");
+		this.template.sendDefault("val4");
 
-		Map<TopicPartition, List<ConsumerRecord<String, String>>> secondPoll = new HashMap<>();
-		secondPoll.put(TOPIC_PARTITION, singletonList(new ConsumerRecord<>("topic", 0, 2L, "key2", "val2")));
-		when(consumer.poll(Duration.ofMillis(50L))).thenReturn(new ConsumerRecords<>(secondPoll));
-
-		reader.open(new ExecutionContext());
-
-		String read = reader.read();
-		assertThat(read, is("val0"));
-
-		read = reader.read();
-		assertThat(read, is("val1"));
-
-		read = reader.read();
-		assertThat(read, is("val2"));
-	}
-
-	@Test
-	public void testSeekOnSavedState() {
-		long offset = 100L;
-		Map<TopicPartition, Long> offsets = new HashMap<>();
-		offsets.put(TOPIC_PARTITION, offset);
 		ExecutionContext executionContext = new ExecutionContext();
-		executionContext.put("topic.partition.offset", offsets);
-		reader.open(executionContext);
-		verify(consumer).seek(TOPIC_PARTITION, offset);
+		Map<TopicPartition, Long> offsets = new HashMap<>();
+		offsets.put(new TopicPartition("topic3", 0), 1L);
+		executionContext.put("topic.partition.offsets", offsets);
+
+		// topic3-0: val0, val1, val2, val3, val4
+		//                  ^
+		//                  |
+		//   last committed offset = 1  (should restart from offset = 2)
+
+		this.reader = new KafkaItemReader<>(this.consumerProperties, "topic3", 0);
+		this.reader.setPollTimeout(Duration.ofSeconds(1));
+		this.reader.open(executionContext);
+
+		List<String> items = new ArrayList<>();
+		items.add(this.reader.read());
+		items.add(this.reader.read());
+		items.add(this.reader.read());
+		assertThat(items, containsInAnyOrder("val2", "val3", "val4"));
+		String item = this.reader.read();
+		assertNull(item);
+
+		this.reader.close();
 	}
 
 	@Test
-	public void testSeekToProvidedOffsets() {
-		long offset = 100L;
+	public void testReadFromMultiplePartitionsAfterRestart() {
+		this.template.send("topic4", 0, null, "val0");
+		this.template.send("topic4", 0, null, "val2");
+		this.template.send("topic4", 0, null, "val4");
+		this.template.send("topic4", 0, null, "val6");
+		this.template.send("topic4", 1, null, "val1");
+		this.template.send("topic4", 1, null, "val3");
+		this.template.send("topic4", 1, null, "val5");
+		this.template.send("topic4", 1, null, "val7");
+
+		ExecutionContext executionContext = new ExecutionContext();
 		Map<TopicPartition, Long> offsets = new HashMap<>();
-		offsets.put(TOPIC_PARTITION, offset);
-		given(offsetsProvider.get(singletonList(TOPIC_PARTITION))).willReturn(offsets);
-		reader.open(new ExecutionContext());
-		verify(consumer).seek(TOPIC_PARTITION, offset);
+		offsets.put(new TopicPartition("topic4", 0), 1L);
+		offsets.put(new TopicPartition("topic4", 1), 2L);
+		executionContext.put("topic.partition.offsets", offsets);
+
+		// topic4-0: val0, val2, val4, val6
+		//                   ^
+		//                   |
+		//   last committed  offset = 1  (should restart from offset = 2)
+		// topic4-1: val1, val3, val5, val7
+		//                         ^
+		//                         |
+		//         last committed  offset = 2  (should restart from offset = 3)
+
+		this.reader = new KafkaItemReader<>(this.consumerProperties, "topic4", 0, 1);
+		this.reader.setPollTimeout(Duration.ofSeconds(1));
+		this.reader.open(executionContext);
+
+		List<String> items = new ArrayList<>();
+		items.add(this.reader.read());
+		items.add(this.reader.read());
+		items.add(this.reader.read());
+		assertThat(items, containsInAnyOrder("val4", "val6", "val7"));
+		String item = this.reader.read();
+		assertNull(item);
+
+		this.reader.close();
 	}
 
-}
\ No newline at end of file
+}
diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/kafka/builder/KafkaItemReaderBuilderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/kafka/builder/KafkaItemReaderBuilderTests.java
index 1bedccd6d..4eed03850 100644
--- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/kafka/builder/KafkaItemReaderBuilderTests.java
+++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/kafka/builder/KafkaItemReaderBuilderTests.java
@@ -1,143 +1,238 @@
+/*
+ * 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
+ *
+ *          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,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
 package org.springframework.batch.item.kafka.builder;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.mockito.Mockito.when;
-
 import java.time.Duration;
-import java.util.Collections;
-import java.util.HashMap;
+import java.util.Arrays;
 import java.util.List;
-import java.util.Map;
+import java.util.Properties;
 
+import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.StringDeserializer;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
-import org.mockito.Mock;
-import org.mockito.MockitoAnnotations;
+
 import org.springframework.batch.item.kafka.KafkaItemReader;
-import org.springframework.batch.item.kafka.OffsetsProvider;
-import org.springframework.batch.item.kafka.support.BeginningOffsetsProvider;
-import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.test.util.ReflectionTestUtils;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.fail;
+
 /**
  * @author Mathieu Ouellet
+ * @author Mahmoud Ben Hassine
  */
 public class KafkaItemReaderBuilderTests {
 
 	@Rule
 	public ExpectedException thrown = ExpectedException.none();
 
-	@Mock
-	private ConsumerFactory<Object, Object> consumerFactory;
-	@Mock
-	private OffsetsProvider offsetsProvider;
-
-	private List<TopicPartition> topicPartitions = Collections.singletonList(new TopicPartition("topic", 0));
+	private Properties consumerProperties;
 
 	@Before
-	public void setUp() {
-		MockitoAnnotations.initMocks(this);
-		Map<String, Object> config = new HashMap<>();
-		config.put("max.poll.records", 2);
-		config.put("enable.auto.commit", false);
-		when(consumerFactory.getConfigurationProperties()).thenReturn(config);
+	public void setUp() throws Exception {
+		this.consumerProperties = new Properties();
+		this.consumerProperties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+		this.consumerProperties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "1");
+		this.consumerProperties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
+				StringDeserializer.class.getName());
+		this.consumerProperties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
+				StringDeserializer.class.getName());
 	}
 
 	@Test
-	public void testNullConsumerFactory() {
+	public void testNullConsumerProperties() {
 		this.thrown.expect(IllegalArgumentException.class);
-		this.thrown.expectMessage("'consumerFactory' must not be null.");
+		this.thrown.expectMessage("Consumer properties must not be null");
 
 		new KafkaItemReaderBuilder<>()
 				.name("kafkaItemReader")
-				.topicPartitions(topicPartitions)
-				.consumerFactory(null)
-				.offsetsProvider(offsetsProvider)
+				.consumerProperties(null)
 				.build();
 	}
 
 	@Test
-	public void testNullTopicsAndTopicPartitions() {
-		this.thrown.expect(IllegalStateException.class);
-		this.thrown.expectMessage("Either 'topicPartitions' or 'topics' must be provided.");
+	public void testConsumerPropertiesValidation() {
+		try {
+			new KafkaItemReaderBuilder<>()
+					.name("kafkaItemReader")
+					.consumerProperties(new Properties())
+					.build();
+			fail("Expected exception was not thrown");
+		} catch (IllegalArgumentException exception) {
+			assertEquals("bootstrap.servers property must be provided", exception.getMessage());
+		}
 
-		new KafkaItemReaderBuilder<>()
-				.name("kafkaItemReader")
-				.topicPartitions(null)
-				.consumerFactory(consumerFactory)
-				.offsetsProvider(offsetsProvider)
-				.build();
+		Properties consumerProperties = new Properties();
+		consumerProperties.put("bootstrap.servers", "foo");
+		try {
+			new KafkaItemReaderBuilder<>()
+					.name("kafkaItemReader")
+					.consumerProperties(consumerProperties)
+					.build();
+			fail("Expected exception was not thrown");
+		} catch (IllegalArgumentException exception) {
+			assertEquals("group.id property must be provided", exception.getMessage());
+		}
+
+		consumerProperties.put("group.id", "1");
+		try {
+			new KafkaItemReaderBuilder<>()
+					.name("kafkaItemReader")
+					.consumerProperties(consumerProperties)
+					.build();
+			fail("Expected exception was not thrown");
+		} catch (IllegalArgumentException exception) {
+			assertEquals("key.deserializer property must be provided", exception.getMessage());
+		}
+
+		consumerProperties.put("key.deserializer", StringDeserializer.class.getName());
+		try {
+			new KafkaItemReaderBuilder<>()
+					.name("kafkaItemReader")
+					.consumerProperties(consumerProperties)
+					.build();
+			fail("Expected exception was not thrown");
+		} catch (IllegalArgumentException exception) {
+			assertEquals("value.deserializer property must be provided", exception.getMessage());
+		}
+
+		consumerProperties.put("value.deserializer", StringDeserializer.class.getName());
+		try {
+			new KafkaItemReaderBuilder<>()
+					.name("kafkaItemReader")
+					.consumerProperties(consumerProperties)
+					.topic("test")
+					.partitions(0, 1)
+					.build();
+		} catch (Exception exception) {
+			fail("Must not throw an exception when configuration is valid");
+		}
 	}
 
 	@Test
-	public void testPollTimeoutNegative() {
+	public void testNullTopicName() {
 		this.thrown.expect(IllegalArgumentException.class);
-		this.thrown.expectMessage("pollTimeout must not be negative.");
+		this.thrown.expectMessage("Topic name must not be null or empty");
 
 		new KafkaItemReaderBuilder<>()
 				.name("kafkaItemReader")
-				.consumerFactory(consumerFactory)
-				.topicPartitions(topicPartitions)
-				.offsetsProvider(offsetsProvider)
-				.pollTimeout(-1)
+				.consumerProperties(this.consumerProperties)
+				.topic(null)
 				.build();
 	}
 
 	@Test
-	public void testNullOffsetsProvider() {
+	public void testEmptyTopicName() {
 		this.thrown.expect(IllegalArgumentException.class);
-		this.thrown.expectMessage("'offsetsProvider' must not be null.");
+		this.thrown.expectMessage("Topic name must not be null or empty");
 
 		new KafkaItemReaderBuilder<>()
 				.name("kafkaItemReader")
-				.consumerFactory(consumerFactory)
-				.topicPartitions(topicPartitions)
-				.offsetsProvider(null)
+				.consumerProperties(this.consumerProperties)
+				.topic("")
 				.build();
 	}
 
 	@Test
-	public void testOffsetsProviderWithAutoCommitConsumerFactory() {
-		this.thrown.expect(IllegalStateException.class);
-		this.thrown.expectMessage("'AutoCommitOffsetsProvider' must be used if 'consumerFactory' is set to auto commit.");
-
-		when(consumerFactory.isAutoCommit()).thenReturn(true);
+	public void testNullPollTimeout() {
+		this.thrown.expect(IllegalArgumentException.class);
+		this.thrown.expectMessage("pollTimeout must not be null");
 
 		new KafkaItemReaderBuilder<>()
 				.name("kafkaItemReader")
-				.consumerFactory(consumerFactory)
-				.topicPartitions(topicPartitions)
-				.offsetsProvider(new BeginningOffsetsProvider())
+				.consumerProperties(this.consumerProperties)
+				.topic("test")
+				.pollTimeout(null)
 				.build();
 	}
 
 	@Test
-	public void testKafkaItemReaderBuild() {
+	public void testNegativePollTimeout() {
+		this.thrown.expect(IllegalArgumentException.class);
+		this.thrown.expectMessage("pollTimeout must not be negative");
+
+		new KafkaItemReaderBuilder<>()
+				.name("kafkaItemReader")
+				.consumerProperties(this.consumerProperties)
+				.topic("test")
+				.pollTimeout(Duration.ofSeconds(-1))
+				.build();
+	}
+
+	@Test
+	public void testZeroPollTimeout() {
+		this.thrown.expect(IllegalArgumentException.class);
+		this.thrown.expectMessage("pollTimeout must not be zero");
+
+		new KafkaItemReaderBuilder<>()
+				.name("kafkaItemReader")
+				.consumerProperties(this.consumerProperties)
+				.topic("test")
+				.pollTimeout(Duration.ZERO)
+				.build();
+	}
+
+	@Test
+	public void testEmptyPartitions() {
+		this.thrown.expect(IllegalArgumentException.class);
+		this.thrown.expectMessage("At least one partition must be provided");
+
+		new KafkaItemReaderBuilder<>()
+				.name("kafkaItemReader")
+				.consumerProperties(this.consumerProperties)
+				.topic("test")
+				.pollTimeout(Duration.ofSeconds(10))
+				.build();
+	}
+
+	@Test
+	@SuppressWarnings("unchecked")
+	public void testKafkaItemReaderCreation() {
 		// given
 		boolean saveState = false;
-		long pollTimeout = 100;
-		int maxItemCount = 100;
+		Duration pollTimeout = Duration.ofSeconds(100);
+		String topic = "test";
+		List<Integer> partitions = Arrays.asList(0, 1);
 
 		// when
-		KafkaItemReader<Object, Object> reader = new KafkaItemReaderBuilder<>()
-				.consumerFactory(consumerFactory)
-				.topicPartitions(topicPartitions)
-				.offsetsProvider(offsetsProvider)
+		KafkaItemReader<String, String> reader = new KafkaItemReaderBuilder<String, String>()
+				.name("kafkaItemReader")
+				.consumerProperties(this.consumerProperties)
+				.topic(topic)
+				.partitions(partitions)
 				.pollTimeout(pollTimeout)
 				.saveState(saveState)
-				.name("kafkaItemReader")
-				.maxItemCount(maxItemCount)
 				.build();
 
 		// then
-		assertEquals(consumerFactory, ReflectionTestUtils.getField(reader, "consumerFactory"));
-		assertEquals(topicPartitions, ReflectionTestUtils.getField(reader, "topicPartitions"));
-		assertEquals(Duration.ofMillis(pollTimeout), ReflectionTestUtils.getField(reader, "pollTimeout"));
-		assertEquals(saveState, ReflectionTestUtils.getField(reader, "saveState"));
-		assertEquals(maxItemCount, ReflectionTestUtils.getField(reader, "maxItemCount"));
+		assertNotNull(reader);
+		assertFalse((Boolean) ReflectionTestUtils.getField(reader, "saveState"));
+		assertEquals(pollTimeout, ReflectionTestUtils.getField(reader, "pollTimeout"));
+		List<TopicPartition> topicPartitions = (List<TopicPartition>) ReflectionTestUtils.getField(reader, "topicPartitions");
+		assertEquals(2, topicPartitions.size());
+		assertEquals(topic, topicPartitions.get(0).topic());
+		assertEquals(partitions.get(0).intValue(), topicPartitions.get(0).partition());
+		assertEquals(topic, topicPartitions.get(1).topic());
+		assertEquals(partitions.get(1).intValue(), topicPartitions.get(1).partition());
 	}
-}
\ No newline at end of file
+}
diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/kafka/support/BeginningOffsetsProviderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/kafka/support/BeginningOffsetsProviderTests.java
deleted file mode 100644
index d04204f05..000000000
--- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/kafka/support/BeginningOffsetsProviderTests.java
+++ /dev/null
@@ -1,36 +0,0 @@
-package org.springframework.batch.item.kafka.support;
-
-import org.apache.kafka.clients.consumer.Consumer;
-import org.apache.kafka.common.TopicPartition;
-import org.junit.Before;
-import org.junit.Test;
-import org.mockito.Mock;
-import org.mockito.MockitoAnnotations;
-
-import static java.util.Collections.singletonList;
-
-import static org.mockito.Mockito.verify;
-
-public class BeginningOffsetsProviderTests {
-
-	private static final TopicPartition TOPIC_PARTITION = new TopicPartition("topic", 0);
-
-	@Mock
-	private Consumer<?, ?> consumer;
-
-	private BeginningOffsetsProvider offsetsProvider;
-
-	@Before
-	public void setUp() throws Exception {
-		MockitoAnnotations.initMocks(this);
-		offsetsProvider = new BeginningOffsetsProvider();
-		offsetsProvider.setConsumer(consumer);
-	}
-
-	@Test
-	public void testGetBeginningOffsets() {
-		offsetsProvider.get(singletonList(TOPIC_PARTITION));
-
-		verify(consumer).beginningOffsets(singletonList(TOPIC_PARTITION));
-	}
-}
\ No newline at end of file
diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/kafka/support/SimpleOffsetsProviderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/kafka/support/SimpleOffsetsProviderTests.java
deleted file mode 100644
index 77fdbe58c..000000000
--- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/kafka/support/SimpleOffsetsProviderTests.java
+++ /dev/null
@@ -1,36 +0,0 @@
-package org.springframework.batch.item.kafka.support;
-
-import java.util.Map;
-
-import org.apache.kafka.common.TopicPartition;
-import org.junit.Before;
-import org.junit.Test;
-import org.mockito.MockitoAnnotations;
-
-import static java.util.Collections.singletonList;
-import static java.util.Collections.singletonMap;
-
-import static org.junit.Assert.assertEquals;
-
-public class SimpleOffsetsProviderTests {
-
-	private static final TopicPartition TOPIC_PARTITION = new TopicPartition("topic", 0);
-
-	private Map<TopicPartition, Long> offsets;
-
-	private SimpleOffsetsProvider offsetsProvider;
-
-	@Before
-	public void setUp() throws Exception {
-		MockitoAnnotations.initMocks(this);
-		offsets = singletonMap(TOPIC_PARTITION, 0L);
-		offsetsProvider = new SimpleOffsetsProvider();
-		offsetsProvider.setOffsets(offsets);
-	}
-
-	@Test
-	public void testGetProvidedOffsets() {
-		assertEquals(offsets, offsetsProvider.get(singletonList(TOPIC_PARTITION)));
-	}
-
-}
\ No newline at end of file
diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/kafka/support/TimestampOffsetsProviderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/kafka/support/TimestampOffsetsProviderTests.java
deleted file mode 100644
index 6bfafec9e..000000000
--- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/kafka/support/TimestampOffsetsProviderTests.java
+++ /dev/null
@@ -1,57 +0,0 @@
-package org.springframework.batch.item.kafka.support;
-
-import java.util.Map;
-
-import org.apache.kafka.clients.consumer.Consumer;
-import org.apache.kafka.common.TopicPartition;
-import org.junit.Before;
-import org.junit.Test;
-import org.mockito.Mock;
-import org.mockito.MockitoAnnotations;
-
-import static java.util.Collections.singletonList;
-import static java.util.Collections.singletonMap;
-
-import static org.junit.Assert.assertTrue;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-
-public class TimestampOffsetsProviderTests {
-
-	private static final TopicPartition TOPIC_PARTITION = new TopicPartition("topic", 0);
-
-	@Mock
-	private Consumer<?, ?> consumer;
-
-	private Long timestampToSearch = 0L;
-
-	private TimestampOffsetsProvider offsetsProvider;
-
-	@Before
-	public void setUp() throws Exception {
-		MockitoAnnotations.initMocks(this);
-
-		offsetsProvider = new TimestampOffsetsProvider(timestampToSearch);
-		offsetsProvider.setConsumer(consumer);
-	}
-
-	@Test
-	public void testFilterNullOffsetsForTimestamp() {
-		// given
-		when(consumer.offsetsForTimes(singletonMap(TOPIC_PARTITION, timestampToSearch)))
-				.thenReturn(singletonMap(TOPIC_PARTITION, null));
-
-		// when
-		Map<TopicPartition, Long> offsets = offsetsProvider.get(singletonList(TOPIC_PARTITION));
-
-		// then
-		assertTrue(offsets.isEmpty());
-	}
-
-	@Test
-	public void testGetOffsetsForTimes() {
-		offsetsProvider.get(singletonList(TOPIC_PARTITION));
-
-		verify(consumer).offsetsForTimes(singletonMap(TOPIC_PARTITION, timestampToSearch));
-	}
-}
\ No newline at end of file