Fix race condition in RabbitTemplatePublisherCallbacksIntegration1Tests

The `CachingConnectionFactory` has an ability to wait for async channels close when it is destroyed.
However, that happens only if an `ApplicationContext` is stopped.

* Supply a mock `ApplicationContext` to the `CachingConnectionFactory` of the test
and emit respective `ContextClosedEvent` in the test before calling `cf.destroy()`

Author: artembilan

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2024 the original author or authors.
+ * Copyright 2002-2025 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.
@@ -79,6 +79,8 @@
 import org.springframework.amqp.support.converter.SimpleMessageConverter;
 import org.springframework.amqp.utils.test.TestUtils;
 import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.event.ContextClosedEvent;
 import org.springframework.expression.Expression;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 
@@ -100,6 +102,8 @@ public class RabbitTemplatePublisherCallbacksIntegration1Tests {
 
 	public static final String ROUTE = "test.queue.RabbitTemplatePublisherCallbacksIntegrationTests";
 
+	private static final ApplicationContext APPLICATION_CONTEXT = mock();
+
 	private final ExecutorService executorService = Executors.newSingleThreadExecutor();
 
 	private CachingConnectionFactory connectionFactory;
@@ -122,25 +126,35 @@ public void create() {
 		connectionFactory.setHost("localhost");
 		connectionFactory.setChannelCacheSize(10);
 		connectionFactory.setPort(BrokerTestUtils.getPort());
+		connectionFactory.setApplicationContext(APPLICATION_CONTEXT);
+
 		connectionFactoryWithConfirmsEnabled = new CachingConnectionFactory();
 		connectionFactoryWithConfirmsEnabled.setHost("localhost");
 		connectionFactoryWithConfirmsEnabled.setChannelCacheSize(100);
 		connectionFactoryWithConfirmsEnabled.setPort(BrokerTestUtils.getPort());
 		connectionFactoryWithConfirmsEnabled.setPublisherConfirmType(ConfirmType.CORRELATED);
+		connectionFactoryWithConfirmsEnabled.setApplicationContext(APPLICATION_CONTEXT);
+
 		templateWithConfirmsEnabled = new RabbitTemplate(connectionFactoryWithConfirmsEnabled);
+
 		connectionFactoryWithReturnsEnabled = new CachingConnectionFactory();
 		connectionFactoryWithReturnsEnabled.setHost("localhost");
 		connectionFactoryWithReturnsEnabled.setChannelCacheSize(1);
 		connectionFactoryWithReturnsEnabled.setPort(BrokerTestUtils.getPort());
 		connectionFactoryWithReturnsEnabled.setPublisherReturns(true);
+		connectionFactoryWithReturnsEnabled.setApplicationContext(APPLICATION_CONTEXT);
+
 		templateWithReturnsEnabled = new RabbitTemplate(connectionFactoryWithReturnsEnabled);
 		templateWithReturnsEnabled.setMandatory(true);
+
 		connectionFactoryWithConfirmsAndReturnsEnabled = new CachingConnectionFactory();
 		connectionFactoryWithConfirmsAndReturnsEnabled.setHost("localhost");
 		connectionFactoryWithConfirmsAndReturnsEnabled.setChannelCacheSize(100);
 		connectionFactoryWithConfirmsAndReturnsEnabled.setPort(BrokerTestUtils.getPort());
 		connectionFactoryWithConfirmsAndReturnsEnabled.setPublisherConfirmType(ConfirmType.CORRELATED);
 		connectionFactoryWithConfirmsAndReturnsEnabled.setPublisherReturns(true);
+		connectionFactoryWithConfirmsAndReturnsEnabled.setApplicationContext(APPLICATION_CONTEXT);
+
 		templateWithConfirmsAndReturnsEnabled = new RabbitTemplate(connectionFactoryWithConfirmsAndReturnsEnabled);
 		templateWithConfirmsAndReturnsEnabled.setMandatory(true);
 	}
@@ -149,17 +163,27 @@ public void create() {
 	public void cleanUp() {
 		this.templateWithConfirmsEnabled.stop();
 		this.templateWithReturnsEnabled.stop();
+
+		this.connectionFactory.onApplicationEvent(new ContextClosedEvent(APPLICATION_CONTEXT));
 		this.connectionFactory.destroy();
+
+		this.connectionFactoryWithConfirmsEnabled.onApplicationEvent(new ContextClosedEvent(APPLICATION_CONTEXT));
 		this.connectionFactoryWithConfirmsEnabled.destroy();
+
+		this.connectionFactoryWithReturnsEnabled.onApplicationEvent(new ContextClosedEvent(APPLICATION_CONTEXT));
 		this.connectionFactoryWithReturnsEnabled.destroy();
+
+		this.connectionFactoryWithConfirmsAndReturnsEnabled.onApplicationEvent(new ContextClosedEvent(APPLICATION_CONTEXT));
+		this.connectionFactoryWithConfirmsAndReturnsEnabled.destroy();
+
 		this.executorService.shutdown();
 	}
 
 	@Test
 	public void testPublisherConfirmReceived() throws Exception {
 		final CountDownLatch latch = new CountDownLatch(10000);
 		final AtomicInteger acks = new AtomicInteger();
-		final AtomicReference<CorrelationData> confirmCorrelation = new AtomicReference<CorrelationData>();
+		final AtomicReference<CorrelationData> confirmCorrelation = new AtomicReference<>();
 		AtomicReference<String> callbackThreadName = new AtomicReference<>();
 		this.templateWithConfirmsEnabled.setConfirmCallback((correlationData, ack, cause) -> {
 			acks.incrementAndGet();
@@ -208,7 +232,7 @@ public Message postProcessMessage(Message message, Correlation correlation, Stri
 		this.templateWithConfirmsEnabled.execute(channel -> {
 			Map<?, ?> listenerMap = TestUtils.getPropertyValue(((ChannelProxy) channel).getTargetChannel(),
 					"listenerForSeq", Map.class);
-			await().until(() -> listenerMap.size() == 0);
+			await().until(listenerMap::isEmpty);
 			return null;
 		});
 
@@ -227,7 +251,8 @@ public void testPublisherConfirmWithSendAndReceive() throws Exception {
 			confirmCD.set(correlationData);
 			latch.countDown();
 		});
-		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(this.connectionFactoryWithConfirmsEnabled);
+		SimpleMessageListenerContainer container =
+				new SimpleMessageListenerContainer(this.connectionFactoryWithConfirmsEnabled);
 		container.setQueueNames(ROUTE);
 		container.setReceiveTimeout(10);
 		container.setMessageListener(
@@ -643,7 +668,6 @@ public void testConcurrentConfirms() throws Exception {
 		assertThat(waitForAll3AcksLatch.await(10, TimeUnit.SECONDS)).isTrue();
 		assertThat(acks.get()).isEqualTo(3);
 
-
 		channel.basicConsume("foo", false, (Map) null, null);
 		verify(mockChannel).basicConsume("foo", false, (Map) null, null);