From da3d9c700d70fc38d8675a32e6c66b137bf3b65e Mon Sep 17 00:00:00 2001
From: Gary Russell <grussell@pivotal.io>
Date: Mon, 20 Feb 2017 12:40:18 -0500
Subject: [PATCH] Fix Test Race Conditions

- AsyncRabbitTemplateTests - tests expecting 2 consumers when both sends could use
  the same one. Add a latch to ensure 2 consumers are used.
- Wrong matcher used to check exclusive log message - with timing we might have 2 or more
  logs. Use hasItem() instead of contains().
---
 .../amqp/rabbit/AsyncRabbitTemplateTests.java | 24 ++++++++++++++++++-
 ...ageListenerContainerIntegration2Tests.java |  6 ++---
 2 files changed, 26 insertions(+), 4 deletions(-)

diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplateTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplateTests.java
index 3a790541..01b1c418 100644
--- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplateTests.java
+++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/AsyncRabbitTemplateTests.java
@@ -88,6 +88,9 @@ public class AsyncRabbitTemplateTests {
 	@Autowired
 	private Queue requests;
 
+	@Autowired
+	private AtomicReference<CountDownLatch> latch;
+
 	private final Message fooMessage = new SimpleMessageConverter().toMessage("foo", new MessageProperties());
 
 	@Test
@@ -103,10 +106,13 @@ public class AsyncRabbitTemplateTests {
 
 	@Test
 	public void testConvert1ArgDirect() throws Exception {
+		this.latch.set(new CountDownLatch(1));
 		ListenableFuture<String> future1 = this.asyncDirectTemplate.convertSendAndReceive("foo");
 		ListenableFuture<String> future2 = this.asyncDirectTemplate.convertSendAndReceive("bar");
+		this.latch.get().countDown();
 		checkConverterResult(future1, "FOO");
 		checkConverterResult(future2, "BAR");
+		this.latch.set(null);
 		waitForZeroInUseConsumers();
 		assertThat(TestUtils
 						.getPropertyValue(this.asyncDirectTemplate, "directReplyToContainer.consumerCount", Integer.class),
@@ -155,12 +161,15 @@ public class AsyncRabbitTemplateTests {
 
 	@Test
 	public void testMessage1ArgDirect() throws Exception {
+		this.latch.set(new CountDownLatch(1));
 		ListenableFuture<Message> future1 = this.asyncDirectTemplate.sendAndReceive(getFooMessage());
 		ListenableFuture<Message> future2 = this.asyncDirectTemplate.sendAndReceive(getFooMessage());
+		this.latch.get().countDown();
 		Message reply1 = checkMessageResult(future1, "FOO");
 		assertEquals(Address.AMQ_RABBITMQ_REPLY_TO, reply1.getMessageProperties().getConsumerQueue());
 		Message reply2 = checkMessageResult(future2, "FOO");
 		assertEquals(Address.AMQ_RABBITMQ_REPLY_TO, reply2.getMessageProperties().getConsumerQueue());
+		this.latch.set(null);
 		waitForZeroInUseConsumers();
 		assertThat(TestUtils
 					.getPropertyValue(this.asyncDirectTemplate, "directReplyToContainer.consumerCount", Integer.class),
@@ -441,6 +450,11 @@ public class AsyncRabbitTemplateTests {
 	@Configuration
 	public static class Config {
 
+		@Bean
+		public AtomicReference<CountDownLatch> latch() {
+			return new AtomicReference<>();
+		}
+
 		@Bean
 		public ConnectionFactory connectionFactory() {
 			CachingConnectionFactory connectionFactory = new CachingConnectionFactory("localhost");
@@ -503,6 +517,15 @@ public class AsyncRabbitTemplateTests {
 			container.setMessageListener(
 					new MessageListenerAdapter((ReplyingMessageListener<String, String>)
 							message -> {
+								CountDownLatch countDownLatch = latch().get();
+								if (countDownLatch != null) {
+									try {
+										countDownLatch.await(10, TimeUnit.SECONDS);
+									}
+									catch (InterruptedException e) {
+										Thread.currentThread().interrupt();
+									}
+								}
 								if ("sleep".equals(message)) {
 									try {
 										Thread.sleep(500); // time for confirm to be delivered, or timeout to occur
@@ -515,7 +538,6 @@ public class AsyncRabbitTemplateTests {
 									return null;
 								}
 								return message.toUpperCase();
-
 							}));
 			return container;
 		}
diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java
index 70b57199..7753b53b 100644
--- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java
+++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java
@@ -16,9 +16,9 @@
 
 package org.springframework.amqp.rabbit.listener;
 
-import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasItem;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
@@ -368,11 +368,11 @@ public class SimpleMessageListenerContainerIntegration2Tests {
 		container2.stop();
 		ArgumentCaptor<String> captor = ArgumentCaptor.forClass(String.class);
 		verify(logger, atLeastOnce()).info(captor.capture());
-		assertThat(captor.getAllValues(), contains(containsString("exclusive")));
+		assertThat(captor.getAllValues(), hasItem(containsString("exclusive")));
 		assertEquals("Consumer raised exception, attempting restart", eventRef.get().getReason());
 		assertFalse(eventRef.get().isFatal());
 		assertThat(eventRef.get().getThrowable(), instanceOf(AmqpIOException.class));
-		verify(containerLogger).warn(any());
+		verify(containerLogger, atLeastOnce()).warn(any());
 	}
 
 	@Test