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

Fix race condition in the MessageDrivenAdapterTests

Browse Source 
The `pause` variable is set in the container already after a `consumer.pause()` call.
Therefore it's state might not be changed after acquiring the latch from the mock answer
This commit is contained in:
Artem Bilan
2020-08-18 11:17:41 -04:00
parent 4c11a61c99
commit f45206a013
2 changed files with 16 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
build.gradle
View File

@@ -607,6 +607,7 @@ project('spring-integration-kafka') {
testImplementation "com.willowtreeapps.assertk:assertk-jvm:$assertkVersion"
testImplementation 'org.jetbrains.kotlin:kotlin-reflect'
testImplementation 'org.jetbrains.kotlin:kotlin-stdlib-jdk8'
testImplementation "org.hamcrest:hamcrest-core:$hamcrestVersion"
testRuntimeOnly 'com.fasterxml.jackson.core:jackson-core'
testRuntimeOnly 'com.fasterxml.jackson.core:jackson-databind'

29
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java
View File

@@ -17,6 +17,7 @@
package org.springframework.integration.kafka.inbound;
import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyMap;
import static org.mockito.ArgumentMatchers.eq;
@@ -34,7 +35,9 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -64,6 +67,9 @@ import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.event.ConsumerPausedEvent;
import org.springframework.kafka.event.ConsumerResumedEvent;
import org.springframework.kafka.event.KafkaEvent;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.KafkaMessageListenerContainer;
import org.springframework.kafka.listener.SeekToCurrentErrorHandler;
@@ -578,25 +584,20 @@ class MessageDrivenAdapterTests {
return null;
}).given(consumer).commitSync(anyMap(), any());
given(consumer.assignment()).willReturn(records.keySet());
final CountDownLatch pauseLatch = new CountDownLatch(1);
willAnswer(i -> {
pauseLatch.countDown();
return null;
}).given(consumer).pause(records.keySet());
given(consumer.paused()).willReturn(records.keySet());
final CountDownLatch resumeLatch = new CountDownLatch(1);
willAnswer(i -> {
resumeLatch.countDown();
return null;
}).given(consumer).resume(records.keySet());
TopicPartitionOffset[] topicPartition = new TopicPartitionOffset[] {
new TopicPartitionOffset("foo", 0) };
TopicPartitionOffset[] topicPartition = { new TopicPartitionOffset("foo", 0) };
ContainerProperties containerProps = new ContainerProperties(topicPartition);
containerProps.setAckMode(ContainerProperties.AckMode.RECORD);
containerProps.setClientId("clientId");
containerProps.setIdleEventInterval(100L);
BlockingQueue<KafkaEvent> containerEvents = new LinkedBlockingQueue<>();
KafkaMessageListenerContainer<Integer, String> container =
new KafkaMessageListenerContainer<>(cf, containerProps);
container.setApplicationEventPublisher(event -> {
if (event instanceof ConsumerPausedEvent || event instanceof ConsumerResumedEvent) {
containerEvents.offer((KafkaEvent) event);
}
});
KafkaMessageDrivenChannelAdapter adapter = new KafkaMessageDrivenChannelAdapter(container);
QueueChannel outputChannel = new QueueChannel();
adapter.setOutputChannel(outputChannel);
@@ -606,10 +607,10 @@ class MessageDrivenAdapterTests {
verify(consumer, times(2)).commitSync(anyMap(), any());
assertThat(outputChannel.getQueueSize()).isEqualTo(2);
adapter.pause();
assertThat(pauseLatch.await(10, TimeUnit.SECONDS)).isTrue();
await().until(containerEvents::take, ConsumerPausedEvent.class::isInstance);
assertThat(adapter.isPaused()).isTrue();
adapter.resume();
assertThat(resumeLatch.await(10, TimeUnit.SECONDS)).isTrue();
await().until(containerEvents::take, ConsumerResumedEvent.class::isInstance);
assertThat(adapter.isPaused()).isFalse();
adapter.stop();
}