GH-1230: Fix return/confirm delivery order

Resolves #1230

We must guarantee that returns are delivered before confirms.
#1055 changed the logic
to deliver on a different thread to avoid a deadlock in the amqp-client.

Since the executor service might have multiple threads, we added a latch to
ensure the return is delivered first.

Unfortunately, this was a one-shot latch and subsequent deliveries using the
same channel might arrive in the wrong order.

Move the latch to the `PendingConfirm` to reinstate in-order delivery.

**cherry-pick to 2.2.x**

Author: garyrussell

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -16,6 +16,9 @@
 
 package org.springframework.amqp.rabbit.connection;
 
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
 import org.springframework.lang.Nullable;
 
 /**
@@ -29,13 +32,19 @@
  */
 public class PendingConfirm {
 
+	static final long RETURN_CALLBACK_TIMEOUT = 60;
+
 	@Nullable
 	private final CorrelationData correlationData;
 
 	private final long timestamp;
 
+	private final CountDownLatch latch = new CountDownLatch(1);
+
 	private String cause;
 
+	private boolean returned;
+
 	/**
 	 * @param correlationData The correlation data.
 	 * @param timestamp The timestamp.
@@ -80,6 +89,43 @@ public String getCause() {
 		return this.cause;
 	}
 
+	/**
+	 * True if a returned message has been received.
+	 * @return true if there is a return.
+	 * @since 2.2.10
+	 */
+	public boolean isReturned() {
+		return this.returned;
+	}
+
+	/**
+	 * Indicate that a returned message has been received.
+	 * @param isReturned true if there is a return.
+	 * @since 2.2.10
+	 */
+	public void setReturned(boolean isReturned) {
+		this.returned = isReturned;
+	}
+
+	/**
+	 * Return true if a return has been passed to the listener or if no return has been
+	 * received.
+	 * @return false if an expected returned message has not been passed to the listener.
+	 * @throws InterruptedException if interrupted.
+	 * @since 2.2.10
+	 */
+	public boolean waitForReturnIfNeeded() throws InterruptedException {
+		return this.returned ? this.latch.await(RETURN_CALLBACK_TIMEOUT, TimeUnit.SECONDS) : true;
+	}
+
+	/**
+	 * Count down the returned message latch; call after the listener has been called.
+	 * @since 2.2.10
+	 */
+	public void countDown() {
+		this.latch.countDown();
+	}
+
 	@Override
 	public String toString() {
 		return "PendingConfirm [correlationData=" + this.correlationData + (this.cause == null ? "" : " cause=" + this.cause) + "]";

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

@@ -32,9 +32,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ConcurrentSkipListMap;
-import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
 import org.apache.commons.logging.Log;
@@ -96,8 +94,6 @@ public class PublisherCallbackChannelImpl
 
 	private static final MessagePropertiesConverter CONVERTER  = new DefaultMessagePropertiesConverter();
 
-	private static final long RETURN_CALLBACK_TIMEOUT = 60;
-
 	private final Log logger = LogFactory.getLog(this.getClass());
 
 	private final Channel delegate;
@@ -112,12 +108,8 @@ public class PublisherCallbackChannelImpl
 
 	private final ExecutorService executor;
 
-	private final CountDownLatch returnLatch = new CountDownLatch(1);
-
 	private volatile java.util.function.Consumer<Channel> afterAckCallback;
 
-	private boolean hasReturned;
-
 	/**
 	 * Create a {@link PublisherCallbackChannelImpl} instance based on the provided
 	 * delegate and executor.
@@ -1016,9 +1008,9 @@ private void doHandleConfirm(boolean ack, Listener listener, PendingConfirm pend
 		this.executor.execute(() -> {
 			try {
 				if (listener.isConfirmListener()) {
-					if (this.hasReturned && !this.returnLatch.await(RETURN_CALLBACK_TIMEOUT, TimeUnit.SECONDS)) {
-						this.logger
-								.error("Return callback failed to execute in " + RETURN_CALLBACK_TIMEOUT + " seconds");
+					if (pendingConfirm.isReturned() && !pendingConfirm.waitForReturnIfNeeded()) {
+						this.logger.error("Return callback failed to execute in "
+								+ PendingConfirm.RETURN_CALLBACK_TIMEOUT + " seconds");
 					}
 					if (this.logger.isDebugEnabled()) {
 						this.logger.debug("Sending confirm " + pendingConfirm);
@@ -1067,9 +1059,11 @@ public void handleReturn(int replyCode,
 			String routingKey,
 			AMQP.BasicProperties properties,
 			byte[] body) {
+
 		LongString returnCorrelation = (LongString) properties.getHeaders().get(RETURNED_MESSAGE_CORRELATION_KEY);
+		PendingConfirm confirm = null;
 		if (returnCorrelation != null) {
-			PendingConfirm confirm = this.pendingReturns.remove(returnCorrelation.toString());
+			confirm = this.pendingReturns.remove(returnCorrelation.toString());
 			if (confirm != null) {
 				MessageProperties messageProperties = CONVERTER.toMessageProperties(properties,
 						new Envelope(0L, false, exchange, routingKey), StandardCharsets.UTF_8.name());
@@ -1096,8 +1090,11 @@ public void handleReturn(int replyCode,
 			}
 		}
 		else {
-			this.hasReturned = true;
+			if (confirm != null) {
+				confirm.setReturned(true);
+			}
 			Listener listenerToInvoke = listener;
+			PendingConfirm toCountDown = confirm;
 			this.executor.execute(() -> {
 				try {
 					listenerToInvoke.handleReturn(replyCode, replyText, exchange, routingKey, properties, body);
@@ -1106,7 +1103,9 @@ public void handleReturn(int replyCode,
 					this.logger.error("Exception delivering returned message ", e);
 				}
 				finally {
-					this.returnLatch.countDown();
+					if (toCountDown != null) {
+						toCountDown.countDown();
+					}
 				}
 			});
 		}

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelTests.java

@@ -18,13 +18,20 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.BDDMockito.given;
 import static org.mockito.BDDMockito.willAnswer;
 import static org.mockito.Mockito.mock;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
 import java.util.Properties;
+import java.util.UUID;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -33,12 +40,15 @@
 import org.junit.jupiter.api.Test;
 
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.ConfirmType;
+import org.springframework.amqp.rabbit.connection.PublisherCallbackChannel.Listener;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.junit.RabbitAvailable;
 import org.springframework.amqp.rabbit.junit.RabbitAvailableCondition;
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
 
+import com.rabbitmq.client.AMQP.BasicProperties;
 import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.LongString;
 import com.rabbitmq.client.Method;
 import com.rabbitmq.client.ShutdownListener;
 import com.rabbitmq.client.ShutdownSignalException;
@@ -138,4 +148,82 @@ void testNotCached() throws Exception {
 		assertThat(cacheProperties.getProperty("idleChannelsNotTx")).isEqualTo("2");
 	}
 
+	@Test
+	void confirmAlwaysAfterReturn() throws InterruptedException {
+		Channel delegate = mock(Channel.class);
+		ExecutorService executor = Executors.newCachedThreadPool();
+		PublisherCallbackChannelImpl channel = new PublisherCallbackChannelImpl(delegate, executor);
+		TheListener listener = new TheListener();
+		channel.addListener(listener);
+		channel.addPendingConfirm(listener, 1, new PendingConfirm(new CorrelationData("1"), 0L));
+		HashMap<String, Object> headers = new HashMap<>();
+		LongString correlation = mock(LongString.class);
+		given(correlation.getBytes()).willReturn("1".getBytes());
+		given(correlation.toString()).willReturn("1");
+		headers.put(PublisherCallbackChannelImpl.RETURNED_MESSAGE_CORRELATION_KEY, correlation);
+		headers.put(PublisherCallbackChannelImpl.RETURN_LISTENER_CORRELATION_KEY, listener.getUUID());
+		BasicProperties properties = new BasicProperties.Builder().headers(headers).build();
+		channel.handleReturn(0, "", "", "", properties, new byte[0]);
+		channel.handleAck(1, false);
+		assertThat(listener.latch1.await(10, TimeUnit.SECONDS)).isTrue();
+		channel.addPendingConfirm(listener, 2, new PendingConfirm(new CorrelationData("1"), 0L));
+		channel.handleReturn(0, "", "", "", properties, new byte[0]);
+		channel.handleAck(2, false);
+		assertThat(listener.latch2.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(listener.calls).containsExactly("return", "confirm", "return", "confirm");
+	}
+
+	private static class TheListener implements Listener {
+
+		private final UUID uuid = UUID.randomUUID();
+
+		final List<String> calls = Collections.synchronizedList(new ArrayList<>());
+
+		final CountDownLatch latch1 = new CountDownLatch(2);
+
+		final CountDownLatch latch2 = new CountDownLatch(4);
+
+		@Override
+		public void handleConfirm(PendingConfirm pendingConfirm, boolean ack) {
+			this.calls.add("confirm");
+			this.latch1.countDown();
+			this.latch2.countDown();
+		}
+
+		@Override
+		public void handleReturn(int replyCode, String replyText, String exchange, String routingKey,
+				BasicProperties properties, byte[] body) throws IOException {
+
+			try {
+				Thread.sleep(500);
+			}
+			catch (InterruptedException e) {
+				Thread.currentThread().interrupt();
+			}
+			this.calls.add("return");
+			this.latch1.countDown();
+			this.latch2.countDown();
+		}
+
+		@Override
+		public void revoke(Channel channel) {
+		}
+
+		@Override
+		public String getUUID() {
+			return this.uuid.toString();
+		}
+
+		@Override
+		public boolean isConfirmListener() {
+			return true;
+		}
+
+		@Override
+		public boolean isReturnListener() {
+			return true;
+		}
+
+	}
+
 }