From f59a3687a97706d1ddb6fea46cc931b2337760a5 Mon Sep 17 00:00:00 2001
From: Gary Russell <grussell@pivotal.io>
Date: Tue, 2 Apr 2019 13:05:57 -0400
Subject: [PATCH] Fix race in KafkaProducerMessageHandlerTests.

---
 .../kafka/outbound/KafkaProducerMessageHandlerTests.java    | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java
index 74018802b2..f04b93e930 100644
--- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java
+++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java
@@ -44,6 +44,7 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
@@ -128,8 +129,9 @@ public class KafkaProducerMessageHandlerTests {
 
 	@BeforeClass
 	public static void setUp() {
-		ConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(
-				KafkaTestUtils.consumerProps("testOut", "true", embeddedKafka));
+		Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("testOut", "true", embeddedKafka);
+		consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+		ConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
 		consumer = cf.createConsumer();
 		embeddedKafka.consumeFromAllEmbeddedTopics(consumer);
 	}