Fix race in test

CorrelationData must have an id when correlating returns and confirms.

With no id; there is a race to deliver the confirm and return.

Author: garyrussell

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests3.java

@@ -62,10 +62,10 @@ public void testRepublishOnNackThreadNoExchange() throws Exception {
 		template.setConfirmCallback((cd, a, c) -> {
 			confirmLatch.countDown();
 			if (confirmLatch.getCount() == 1) {
-				template.convertAndSend(QUEUE1, ((MyCD) cd).payload);
+				template.convertAndSend(QUEUE1, cd.getId());
 			}
 		});
-		template.convertAndSend("bad.exchange", "junk", "foo", new MyCD("foo"));
+		template.convertAndSend("bad.exchange", "junk", "foo", new CorrelationData("foo"));
 		assertThat(confirmLatch.await(10, TimeUnit.SECONDS)).isTrue();
 		assertThat(template.receive(QUEUE1, 10_000)).isNotNull();
 	}
@@ -97,7 +97,8 @@ public void testDeferredChannelCacheNack() throws Exception {
 		channel2.close();
 		conn.close();
 		assertThat(TestUtils.getPropertyValue(cf, "cachedChannelsNonTransactional", List.class).size()).isEqualTo(2);
-		template.convertAndSend("", QUEUE2 + "junk", "foo", new MyCD("foo"));
+		CorrelationData correlationData = new CorrelationData("foo");
+		template.convertAndSend("", QUEUE2 + "junk", "foo", correlationData);
 		assertThat(returnLatch.await(10, TimeUnit.SECONDS)).isTrue();
 		assertThat(confirmLatch.await(10, TimeUnit.SECONDS)).isTrue();
 		int n = 0;
@@ -106,6 +107,7 @@ public void testDeferredChannelCacheNack() throws Exception {
 		}
 		assertThat(cacheCount.get()).isEqualTo(1);
 		assertThat(returnCalledFirst.get()).isTrue();
+		assertThat(correlationData.getReturnedMessage()).isNotNull();
 		cf.destroy();
 	}
 
@@ -129,7 +131,7 @@ public void testDeferredChannelCacheAck() throws Exception {
 		channel2.close();
 		conn.close();
 		assertThat(TestUtils.getPropertyValue(cf, "cachedChannelsNonTransactional", List.class).size()).isEqualTo(2);
-		template.convertAndSend("", QUEUE2, "foo", new MyCD("foo"));
+		template.convertAndSend("", QUEUE2, "foo", new CorrelationData("foo"));
 		assertThat(confirmLatch.await(10, TimeUnit.SECONDS)).isTrue();
 		assertThat(cacheCount.get()).isEqualTo(1);
 		cf.destroy();
@@ -146,22 +148,11 @@ public void testTwoSendsAndReceivesDRTMLC() throws Exception {
 		template.setConfirmCallback((cd, a, c) -> {
 			confirmLatch.countDown();
 		});
-		template.convertSendAndReceive("", QUEUE3, "foo", new MyCD("foo"));
-		template.convertSendAndReceive("", QUEUE3, "foo", new MyCD("foo")); // listener not registered
+		template.convertSendAndReceive("", QUEUE3, "foo", new CorrelationData("foo"));
+		template.convertSendAndReceive("", QUEUE3, "foo", new CorrelationData("foo")); // listener not registered
 		assertThat(confirmLatch.await(10, TimeUnit.SECONDS)).isTrue();
 		assertThat(template.receive(QUEUE3, 10_000)).isNotNull();
 		assertThat(template.receive(QUEUE3, 10_000)).isNotNull();
 	}
 
-
-	private static class MyCD extends CorrelationData {
-
-		final String payload;
-
-		MyCD(String payload) {
-			this.payload = payload;
-		}
-
-	}
-
 }