From 73bc3fedf50480a60c2b6b8fd2f73558f5206eab Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 19 Nov 2015 11:14:25 -0500 Subject: [PATCH] INT-3889: Fix AMQP PubSub Channel Test Race Cond. JIRA: https://jira.spring.io/browse/INT-3889 The test case `pubSubLostConnectionTest` forces the connection closed and then sends another message. It is testing recovery. However, since the pub/sub queue is an auto-delete queue, we need to wait until the new consumer is established and the queue exists, before sending the message after the test. Otherwise, since the exchange is fanout, the message is simply dropped. Also set `@DirtiesContext` on other tests. Also remove non-tests. I have __not__ removed the diagnostics because I don't believe this addresses AMQP-543. --- .../amqp/channel/ChannelTests.java | 39 +++++++++++++-- .../amqp/config/AmqpChannelParserTests.java | 2 + .../AmqpInboundChannelAdapterParserTests.java | 4 +- .../config/AmqpInboundGatewayParserTests.java | 2 + .../AmqpOutboundGatewayParserTests.java | 2 +- .../amqp/config/ChannelSample-context.xml | 28 ----------- .../amqp/config/ChannelSample.java | 31 ------------ .../amqp/config/EchoSample-context.xml | 49 ------------------- .../integration/amqp/config/EchoSample.java | 31 ------------ .../amqp/config/GatewayEchoTests-context.xml | 41 ---------------- .../amqp/config/GatewayEchoTests.java | 37 -------------- ...utboundGatewayIntegrationTests-context.xml | 12 ++--- .../OutboundGatewayIntegrationTests.java | 4 +- .../amqp/config/OutboundGatewayTests.java | 10 +++- .../config/PubSubChannelSample-context.xml | 28 ----------- .../amqp/config/PubSubChannelSample.java | 31 ------------ 16 files changed, 61 insertions(+), 290 deletions(-) delete mode 100644 spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/ChannelSample-context.xml delete mode 100644 spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/ChannelSample.java delete mode 100644 spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/EchoSample-context.xml delete mode 100644 spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/EchoSample.java delete mode 100644 spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/GatewayEchoTests-context.xml delete mode 100644 spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/GatewayEchoTests.java delete mode 100644 spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/PubSubChannelSample-context.xml delete mode 100644 spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/PubSubChannelSample.java 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 3658f3f076..ec295b3065 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 @@ -20,12 +20,15 @@ import static org.hamcrest.Matchers.instanceOf; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; import static org.mockito.Mockito.mock; import java.util.Collection; +import java.util.Map; import java.util.concurrent.CyclicBarrier; import java.util.concurrent.TimeUnit; +import org.junit.After; import org.junit.ClassRule; import org.junit.Test; import org.junit.runner.RunWith; @@ -34,6 +37,7 @@ import org.springframework.amqp.core.AmqpTemplate; import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitAdmin; +import org.springframework.amqp.rabbit.listener.BlockingQueueConsumer; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.annotation.Autowired; @@ -74,6 +78,11 @@ public class ChannelTests extends LogAdjustingTestSupport { super("org.springframework.integration", "org.springframework.integration.amqp", "org.springframework.amqp"); } + @After + public void tearDown() { + new RabbitAdmin(this.factory).deleteExchange("si.fanout.foo"); + } + @Test public void pubSubLostConnectionTest() throws Exception { final CyclicBarrier latch = new CyclicBarrier(2); @@ -88,16 +97,36 @@ public class ChannelTests extends LogAdjustingTestSupport { } } }); - channel.send(new GenericMessage("foo")); + this.channel.send(new GenericMessage("foo")); latch.await(10, TimeUnit.SECONDS); latch.reset(); + BlockingQueueConsumer consumer = (BlockingQueueConsumer) TestUtils + .getPropertyValue(this.channel, "container.consumers", Map.class).keySet().iterator().next(); factory.destroy(); - channel.send(new GenericMessage("bar")); + waitForNewConsumer(this.channel, consumer); + this.channel.send(new GenericMessage("bar")); latch.await(10, TimeUnit.SECONDS); - channel.destroy(); + this.channel.destroy(); assertEquals(0, TestUtils.getPropertyValue(factory, "connectionListener.delegates", Collection.class).size()); } + private void waitForNewConsumer(PublishSubscribeAmqpChannel channel, BlockingQueueConsumer consumer) + throws Exception { + BlockingQueueConsumer newConsumer = (BlockingQueueConsumer) TestUtils + .getPropertyValue(channel, "container.consumers", Map.class).keySet().iterator().next(); + int n = 0; + boolean newConsumerIsConsuming = newConsumer != consumer && TestUtils.getPropertyValue(newConsumer, + "consumerTags", Map.class).size() > 0; + while (n++ < 100 && !newConsumerIsConsuming) { + Thread.sleep(100); + newConsumer = (BlockingQueueConsumer) TestUtils + .getPropertyValue(channel, "container.consumers", Map.class).keySet().iterator().next(); + newConsumerIsConsuming = newConsumer != consumer && TestUtils.getPropertyValue(newConsumer, + "consumerTags", Map.class).size() > 0; + } + assertTrue("Failed to restart consumer", n < 100); + } + /* * Verify queue is declared if not present and not declared if it is already present. */ @@ -144,6 +173,10 @@ public class ChannelTests extends LogAdjustingTestSupport { channelFactoryBean.afterPropertiesSet(); channel = channelFactoryBean.getObject(); assertThat(channel, instanceOf(PublishSubscribeAmqpChannel.class)); + + RabbitAdmin rabbitAdmin = new RabbitAdmin(this.factory); + rabbitAdmin.deleteQueue("testChannel"); + rabbitAdmin.deleteExchange("si.fanout.testChannel"); } } diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests.java index 152e2d5f3d..2d3f808c55 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests.java @@ -32,6 +32,7 @@ import org.springframework.integration.support.utils.IntegrationUtils; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.support.ChannelInterceptorAdapter; +import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -42,6 +43,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; */ @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext public class AmqpChannelParserTests { @Autowired diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParserTests.java index 1efb02dfd0..22017d5745 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParserTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundChannelAdapterParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -39,6 +39,7 @@ import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.test.util.TestUtils; +import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -50,6 +51,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; */ @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext public class AmqpInboundChannelAdapterParserTests { @Autowired diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests.java index 50eca8de85..6d7f9dfe49 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests.java @@ -49,6 +49,7 @@ import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessagingException; +import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import org.springframework.util.ReflectionUtils; @@ -63,6 +64,7 @@ import org.springframework.util.ReflectionUtils; */ @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext public class AmqpInboundGatewayParserTests { @Autowired diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java index f22d041eb9..da61bcb811 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java @@ -329,7 +329,7 @@ public class AmqpOutboundGatewayParserTests { public void testInt2971HeaderMapperAndMappedHeadersExclusivity() { try { new ClassPathXmlApplicationContext("AmqpOutboundGatewayParserTests-headerMapper-fail-context.xml", - this.getClass()).close();; + this.getClass()).close(); } catch (BeanDefinitionParsingException e) { assertTrue(e.getMessage().startsWith("Configuration problem: The 'header-mapper' attribute " + diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/ChannelSample-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/ChannelSample-context.xml deleted file mode 100644 index d078fbdfed..0000000000 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/ChannelSample-context.xml +++ /dev/null @@ -1,28 +0,0 @@ - - - - - - - - - - - - - - - - - - diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/ChannelSample.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/ChannelSample.java deleted file mode 100644 index af63d8dca1..0000000000 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/ChannelSample.java +++ /dev/null @@ -1,31 +0,0 @@ -/* - * Copyright 2002-2011 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.integration.amqp.config; - -import org.springframework.context.support.ClassPathXmlApplicationContext; - -/** - * @author Mark Fisher - * @since 2.1 - */ -public class ChannelSample { - - public static void main(String[] args) { - new ClassPathXmlApplicationContext("ChannelSample-context.xml", ChannelSample.class); - } - -} diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/EchoSample-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/EchoSample-context.xml deleted file mode 100644 index b2d2687d09..0000000000 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/EchoSample-context.xml +++ /dev/null @@ -1,49 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/EchoSample.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/EchoSample.java deleted file mode 100644 index 7b2ade3541..0000000000 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/EchoSample.java +++ /dev/null @@ -1,31 +0,0 @@ -/* - * Copyright 2002-2011 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.integration.amqp.config; - -import org.springframework.context.support.ClassPathXmlApplicationContext; - -/** - * @author Mark Fisher - * @since 2.1 - */ -public class EchoSample { - - public static void main(String[] args) { - new ClassPathXmlApplicationContext("EchoSample-context.xml", EchoSample.class); - } - -} diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/GatewayEchoTests-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/GatewayEchoTests-context.xml deleted file mode 100644 index 2b4c568dfa..0000000000 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/GatewayEchoTests-context.xml +++ /dev/null @@ -1,41 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/GatewayEchoTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/GatewayEchoTests.java deleted file mode 100644 index 2f7c8231ce..0000000000 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/GatewayEchoTests.java +++ /dev/null @@ -1,37 +0,0 @@ -/* - * Copyright 2002-2011 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.integration.amqp.config; - -import org.junit.Test; -import org.springframework.context.support.ClassPathXmlApplicationContext; - -/** - * @author Mark Fisher - * @since 2.1 - */ -public class GatewayEchoTests { - - public static void main(String[] args) { - new ClassPathXmlApplicationContext("GatewayEchoTests-context.xml", GatewayEchoTests.class); - } - - @Test - /** - * Dummy test to satisfy maven in some environments. - */ - public void test() {} -} diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayIntegrationTests-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayIntegrationTests-context.xml index 59f2fb35a5..d45044815f 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayIntegrationTests-context.xml +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayIntegrationTests-context.xml @@ -12,11 +12,11 @@ - + - + - + @@ -28,13 +28,13 @@ + queue-names="si.ob.gw.integration"/> diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayIntegrationTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayIntegrationTests.java index c052ba81df..c2bd39bda8 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayIntegrationTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayIntegrationTests.java @@ -29,6 +29,7 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.GenericMessage; +import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -40,10 +41,11 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; */ @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext public class OutboundGatewayIntegrationTests { @ClassRule - public static final BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueues("si.test.queue"); + public static final BrokerRunning brokerIsRunning = BrokerRunning.isRunning(); @Autowired private MessageChannel toRabbit; diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayTests.java index a49ac86bfc..7a5261df92 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2015 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -24,6 +24,7 @@ import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; +import org.junit.After; import org.junit.Test; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; @@ -37,9 +38,9 @@ import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint; import org.springframework.integration.config.IntegrationEvaluationContextFactoryBean; import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.handler.MessageProcessor; -import org.springframework.messaging.support.GenericMessage; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; +import org.springframework.messaging.support.GenericMessage; import org.springframework.test.util.ReflectionTestUtils; import org.springframework.util.ClassUtils; @@ -56,6 +57,11 @@ public class OutboundGatewayTests { private final ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(getClass().getSimpleName() + "-context.xml", getClass()); + @After + public void tearDown() { + context.close(); + } + @Test public void testVanillaConfiguration() throws Exception { assertTrue(context.getBeanFactory().containsBeanDefinition("vanilla")); diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/PubSubChannelSample-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/PubSubChannelSample-context.xml deleted file mode 100644 index cf731eacf0..0000000000 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/PubSubChannelSample-context.xml +++ /dev/null @@ -1,28 +0,0 @@ - - - - - - - - - - - - - - - - - - diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/PubSubChannelSample.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/PubSubChannelSample.java deleted file mode 100644 index afdf37d904..0000000000 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/PubSubChannelSample.java +++ /dev/null @@ -1,31 +0,0 @@ -/* - * Copyright 2002-2011 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.integration.amqp.config; - -import org.springframework.context.support.ClassPathXmlApplicationContext; - -/** - * @author Mark Fisher - * @since 2.1 - */ -public class PubSubChannelSample { - - public static void main(String[] args) { - new ClassPathXmlApplicationContext("PubSubChannelSample-context.xml", PubSubChannelSample.class); - } - -}