Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix ConcurrentMLCTests race condition

Browse Source 
https://build.spring.io/browse/SK-MAS-117/
This commit is contained in:
Artem Bilan
2016-10-10 13:00:35 -04:00
parent 0802f631c4
commit 076a352104
1 changed files with 18 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java
View File

@@ -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");