AMQP-817: Synthetic NACKs on separate thread

JIRA: https://jira.spring.io/browse/AMQP-817

Previously, synthetic nacks were processed on the client thread, which caused
a deadlock if the user attempted some operation such as creating a new channel
on that thread.

Also, when generated from a application-level `close()` a similar deadlock between
the connection factory and `PublisherCallbackChannelImpl` could occur.

Synthetic nacks are now always processed on a separate thread.

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java

@@ -623,7 +623,7 @@ public class CachingConnectionFactory extends AbstractConnectionFactory
 		}
 		if (this.publisherConfirms || this.publisherReturns) {
 			if (!(channel instanceof PublisherCallbackChannelImpl)) {
-				channel = new PublisherCallbackChannelImpl(channel);
+				channel = new PublisherCallbackChannelImpl(channel, getExecutorService());
 			}
 		}
 		if (channel != null) {

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/PublisherCallbackChannelImpl.java

@@ -31,6 +31,8 @@ import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.TimeoutException;
 
 import org.apache.commons.logging.Log;
@@ -79,6 +81,8 @@ import com.rabbitmq.client.ShutdownSignalException;
 public class PublisherCallbackChannelImpl
 		implements PublisherCallbackChannel, ConfirmListener, ReturnListener, ShutdownListener {
 
+	private static final ExecutorService DEFAULT_EXECUTOR = Executors.newSingleThreadExecutor();
+
 	private final Log logger = LogFactory.getLog(this.getClass());
 
 	private final Channel delegate;
@@ -91,9 +95,16 @@ public class PublisherCallbackChannelImpl
 
 	private volatile java.util.function.Consumer<Channel> afterAckCallback;
 
+	private final ExecutorService executor;
+
 	public PublisherCallbackChannelImpl(Channel delegate) {
+		this(delegate, null);
+	}
+
+	public PublisherCallbackChannelImpl(Channel delegate, ExecutorService executor) {
 		delegate.addShutdownListener(this);
 		this.delegate = delegate;
+		this.executor = executor != null ? executor : DEFAULT_EXECUTOR;
 	}
 
 	@Override
@@ -781,7 +792,11 @@ public class PublisherCallbackChannelImpl
 				this.logger.trace(this.delegate + " is already closed");
 			}
 		}
-		generateNacksForPendingAcks("Channel closed by application");
+		shutdownCompleted("Channel closed by application");
+	}
+
+	private void shutdownCompleted(String cause) {
+		this.executor.execute(() -> generateNacksForPendingAcks(cause));
 	}
 
 	private synchronized void generateNacksForPendingAcks(String cause) {
@@ -997,7 +1012,7 @@ public class PublisherCallbackChannelImpl
 
 	@Override
 	public void shutdownCompleted(ShutdownSignalException cause) {
-		generateNacksForPendingAcks(cause.getMessage());
+		shutdownCompleted(cause.getMessage());
 	}
 
 // Object

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

@@ -401,6 +401,10 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
 		exec.shutdown();
 		assertTrue(exec.awaitTermination(10, TimeUnit.SECONDS));
 		ccf.destroy();
+		int n = 0;
+		while (n++ < 100 && pendingConfirms.size() > 0) {
+			Thread.sleep(100);
+		}
 		assertEquals(0, pendingConfirms.size());
 	}
 
@@ -739,7 +743,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
 		when(mockConnection.createChannel()).thenReturn(mockChannel1, mockChannel2);
 
 		CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory);
-		ccf.setExecutor(mock(ExecutorService.class));
+		ccf.setExecutor(Executors.newSingleThreadExecutor());
 		ccf.setPublisherConfirms(true);
 		final RabbitTemplate template = new RabbitTemplate(ccf);
 
@@ -768,13 +772,12 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
 	@Test
 	public void testPublisherCallbackChannelImplCloseWithPending() throws Exception {
 
-		final AtomicInteger nacks = new AtomicInteger();
-
 		Listener listener = mock(Listener.class);
+		final CountDownLatch latch = new CountDownLatch(2);
 		doAnswer(invocation -> {
 			boolean ack = invocation.getArgument(1);
 			if (!ack) {
-				nacks.incrementAndGet();
+				latch.countDown();
 			}
 			return null;
 		}).when(listener).handleConfirm(any(PendingConfirm.class), anyBoolean());
@@ -795,8 +798,13 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
 
 		channel.close();
 
+		assertTrue(latch.await(10, TimeUnit.SECONDS));
+
+		int n = 0;
+		while (n++ < 100 && TestUtils.getPropertyValue(channel, "pendingConfirms", Map.class).size() > 0) {
+			Thread.sleep(100);
+		}
 		assertEquals(0, TestUtils.getPropertyValue(channel, "pendingConfirms", Map.class).size());
-		assertEquals(2, nacks.get());
 
 	}
 

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

@@ -23,7 +23,6 @@ import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
@@ -37,17 +36,18 @@ import com.rabbitmq.client.Channel;
 
 /**
  * @author Gary Russell
- *
  * @since 2.1
  *
  */
-@RabbitAvailable(queues = RabbitTemplatePublisherCallbacksIntegrationTests3.QUEUE)
+@RabbitAvailable(queues = { RabbitTemplatePublisherCallbacksIntegrationTests3.QUEUE1,
+		RabbitTemplatePublisherCallbacksIntegrationTests3.QUEUE2 })
 public class RabbitTemplatePublisherCallbacksIntegrationTests3 {
 
-	public static final String QUEUE = "defer.close";
+	public static final String QUEUE1 = "synthetic.nack";
+
+	public static final String QUEUE2 = "defer.close";
 
 	@Test
-	@Disabled
 	public void testRepublishOnNackThreadNoExchange() throws Exception {
 		CachingConnectionFactory cf = new CachingConnectionFactory(
 				RabbitAvailableCondition.getBrokerRunning().getConnectionFactory());
@@ -56,13 +56,13 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests3 {
 		final CountDownLatch confirmLatch = new CountDownLatch(2);
 		template.setConfirmCallback((cd, a, c) -> {
 			if (confirmLatch.getCount() == 2) {
-				template.convertAndSend(QUEUE, ((MyCD) cd).payload); // deadlock creating new channel
+				template.convertAndSend(QUEUE1, ((MyCD) cd).payload);
 			}
 			confirmLatch.countDown();
 		});
 		template.convertAndSend("bad.exchange", "junk", "foo", new MyCD("foo"));
 		assertThat(confirmLatch.await(10, TimeUnit.SECONDS)).isTrue();
-		assertThat(template.receive(QUEUE, 10_000)).isNotNull();
+		assertThat(template.receive(QUEUE1, 10_000)).isNotNull();
 	}
 
 	@Test
@@ -90,7 +90,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests3 {
 		channel2.close();
 		conn.close();
 		assertThat(TestUtils.getPropertyValue(cf, "cachedChannelsNonTransactional", List.class).size()).isEqualTo(2);
-		template.convertAndSend("", QUEUE + "junk", "foo", new MyCD("foo"));
+		template.convertAndSend("", QUEUE2 + "junk", "foo", new MyCD("foo"));
 		assertThat(returnLatch.await(10, TimeUnit.SECONDS)).isTrue();
 		assertThat(confirmLatch.await(10, TimeUnit.SECONDS)).isTrue();
 		assertThat(cacheCount.get()).isEqualTo(1);