From 2c96f6a9582c1dc6aaba3ec46b5fe04918be8861 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 8 May 2017 10:28:56 -0400 Subject: [PATCH] Fix race condition in the AMQP ChannelTests https://build.spring.io/browse/INT-AT42SIO-502 There is a small time window when we remove the current consumer from the local store, but there is no a new one yet. So, we have to check the `Set`(`Map`) size before calling its `iterator` **Cherry-pick to 4.3.x** --- .../amqp/channel/ChannelTests.java | 23 +++++++++++-------- 1 file changed, 14 insertions(+), 9 deletions(-) diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java index 78e55925a0..181b7aba87 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java @@ -149,19 +149,24 @@ public class ChannelTests extends LogAdjustingTestSupport { assertEquals(0, TestUtils.getPropertyValue(factory, "connectionListener.delegates", Collection.class).size()); } + @SuppressWarnings("unchecked") private void waitForNewConsumer(PublishSubscribeAmqpChannel channel, BlockingQueueConsumer consumer) throws Exception { - BlockingQueueConsumer newConsumer = (BlockingQueueConsumer) TestUtils.getPropertyValue(channel, - "container.consumers", Set.class).iterator().next(); + + final Object consumersMonitor = TestUtils.getPropertyValue(channel, "container.consumersMonitor"); int n = 0; - boolean newConsumerIsConsuming = newConsumer != consumer && TestUtils.getPropertyValue(newConsumer, - "consumerTags", Map.class).size() > 0; - while (n++ < 100 && !newConsumerIsConsuming) { + while (n++ < 100) { + Set consumers = TestUtils.getPropertyValue(channel, "container.consumers", Set.class); + synchronized (consumersMonitor) { + if (!consumers.isEmpty()) { + BlockingQueueConsumer newConsumer = consumers.iterator().next(); + if (newConsumer != consumer && TestUtils.getPropertyValue(newConsumer, + "consumerTags", Map.class).size() > 0) { + break; + } + } + } Thread.sleep(100); - newConsumer = (BlockingQueueConsumer) TestUtils.getPropertyValue(channel, - "container.consumers", Set.class).iterator().next(); - newConsumerIsConsuming = newConsumer != consumer && TestUtils.getPropertyValue(newConsumer, - "consumerTags", Map.class).size() > 0; } assertTrue("Failed to restart consumer", n < 100); }