From f45206a0132e2a39fdfe6d8bacaa687764214eab Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 18 Aug 2020 11:17:41 -0400 Subject: [PATCH] Fix race condition in the `MessageDrivenAdapterTests` 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 --- build.gradle | 1 + .../inbound/MessageDrivenAdapterTests.java | 29 ++++++++++--------- 2 files changed, 16 insertions(+), 14 deletions(-) diff --git a/build.gradle b/build.gradle index 53ab8daba5..a462b89e86 100644 --- a/build.gradle +++ b/build.gradle @@ -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' diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java index f271832948..8029039bda 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java @@ -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 containerEvents = new LinkedBlockingQueue<>(); KafkaMessageListenerContainer 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(); }