From 076a35210427ef90e29b7aeb40e268eb2308f496 Mon Sep 17 00:00:00 2001
From: Artem Bilan <abilan@pivotal.io>
Date: Mon, 10 Oct 2016 13:00:35 -0400
Subject: [PATCH] Fix `ConcurrentMLCTests` race condition

https://build.spring.io/browse/SK-MAS-117/
---
 ...oncurrentMessageListenerContainerTests.java | 18 ++++++++++++++++++
 1 file changed, 18 insertions(+)

diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java
index 8fc14cd2..7af7109f 100644
--- a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java
+++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java
@@ -501,10 +501,28 @@ public class ConcurrentMessageListenerContainerTests {
 		// this consumer is positioned at 1, the next offset after the successfully
 		// processed 'foo'
 		// it has not been updated because 'bar' failed
+		// Since there is no simple ability to hook into 'commitSync()' action with concurrent containers,
+		// ping partition until success through some sleep period
+		for (int i = 0; i < 100; i++) {
+			if (consumer.position(new TopicPartition(topic9, 0)) == 1) {
+				break;
+			}
+			else {
+				Thread.sleep(100);
+			}
+		}
 		assertThat(consumer.position(new TopicPartition(topic9, 0))).isEqualTo(1);
 		// this consumer is positioned at 1, the next offset after the successfully
 		// processed 'qux'
 		// it has been updated even 'baz' failed
+		for (int i = 0; i < 100; i++) {
+			if (consumer.position(new TopicPartition(topic9, 1)) == 2) {
+				break;
+			}
+			else {
+				Thread.sleep(100);
+			}
+		}
 		assertThat(consumer.position(new TopicPartition(topic9, 1))).isEqualTo(2);
 		consumer.close();
 		logger.info("Stop ack on error");