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

Fix Race in Test

Browse Source
This commit is contained in:
Gary Russell
2023-10-25 14:11:15 -04:00
parent fc69f85662
commit d2e213a1d3
1 changed files with 3 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java
View File

@@ -2790,12 +2790,14 @@ public class KafkaMessageListenerContainerTests {
return null;
}).given(consumer).pause(any());
given(consumer.paused()).willReturn(pausedParts);
CountDownLatch firstPoll = new CountDownLatch(1);
given(consumer.poll(any(Duration.class))).willAnswer(i -> {
if (paused.get()) {
pauseLatch1.countDown();
// hold up the consumer thread while we revoke/assign partitions on the test thread
suspendConsumerThread.await(10, TimeUnit.SECONDS);
}
firstPoll.countDown();
Thread.sleep(50);
return ConsumerRecords.empty();
});
@@ -2819,6 +2821,7 @@ public class KafkaMessageListenerContainerTests {
new KafkaMessageListenerContainer<>(cf, containerProps);
container.start();
InOrder inOrder = inOrder(consumer);
assertThat(firstPoll.await(10, TimeUnit.SECONDS)).isNotNull();
container.pausePartition(tp0);
container.pausePartition(tp1);
assertThat(pauseLatch1.await(10, TimeUnit.SECONDS)).isTrue();