AMQP-849: RT and DRTMLC - add ErrorHandler

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

Support the configuration of an error handler for exceptions when
delivering replies (e.g. late replies).

**cherry-pick to 2.0.x with adjustments to assertJ and docs**

* Polishing - PR Comments

* More polishing

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java

@@ -89,6 +89,7 @@
 import org.springframework.retry.RetryCallback;
 import org.springframework.retry.support.RetryTemplate;
 import org.springframework.util.Assert;
+import org.springframework.util.ErrorHandler;
 import org.springframework.util.StringUtils;
 
 import com.rabbitmq.client.AMQP;
@@ -186,63 +187,53 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware,
 
 	private final AtomicInteger containerInstance = new AtomicInteger();
 
-	private volatile String exchange = DEFAULT_EXCHANGE;
+	private String exchange = DEFAULT_EXCHANGE;
 
-	private volatile String routingKey = DEFAULT_ROUTING_KEY;
+	private String routingKey = DEFAULT_ROUTING_KEY;
 
 	// The default queue name that will be used for synchronous receives.
-	private volatile String defaultReceiveQueue;
+	private String defaultReceiveQueue;
 
-	private volatile long receiveTimeout = 0;
+	private long receiveTimeout = 0;
 
-	private volatile long replyTimeout = DEFAULT_REPLY_TIMEOUT;
+	private long replyTimeout = DEFAULT_REPLY_TIMEOUT;
 
-	private volatile MessageConverter messageConverter = new SimpleMessageConverter();
+	private MessageConverter messageConverter = new SimpleMessageConverter();
 
-	private volatile MessagePropertiesConverter messagePropertiesConverter = new DefaultMessagePropertiesConverter();
+	private MessagePropertiesConverter messagePropertiesConverter = new DefaultMessagePropertiesConverter();
 
-	private volatile String encoding = DEFAULT_ENCODING;
+	private String encoding = DEFAULT_ENCODING;
 
-	private volatile String replyAddress;
+	private String replyAddress;
 
 	@Nullable
-	private volatile ConfirmCallback confirmCallback;
+	private ConfirmCallback confirmCallback;
 
-	private volatile ReturnCallback returnCallback;
+	private ReturnCallback returnCallback;
 
-	private volatile Boolean confirmsOrReturnsCapable;
-
-	private volatile boolean publisherConfirms;
+	private Expression mandatoryExpression = new ValueExpression<Boolean>(false);
 
-	private volatile Expression mandatoryExpression = new ValueExpression<Boolean>(false);
+	private String correlationKey = null;
 
-	private volatile String correlationKey = null;
+	private RetryTemplate retryTemplate;
 
-	private volatile RetryTemplate retryTemplate;
+	private RecoveryCallback<?> recoveryCallback;
 
-	private volatile RecoveryCallback<?> recoveryCallback;
+	private Expression sendConnectionFactorySelectorExpression;
 
-	private volatile Expression sendConnectionFactorySelectorExpression;
+	private Expression receiveConnectionFactorySelectorExpression;
 
-	private volatile Expression receiveConnectionFactorySelectorExpression;
+	private boolean useDirectReplyToContainer = true;
 
-	private volatile boolean usingFastReplyTo;
-
-	private volatile boolean useDirectReplyToContainer = true;
-
-	private volatile boolean evaluatedFastReplyTo;
+	private boolean useTemporaryReplyQueues;
 
-	private volatile boolean useTemporaryReplyQueues;
+	private Collection<MessagePostProcessor> beforePublishPostProcessors;
 
-	private volatile Collection<MessagePostProcessor> beforePublishPostProcessors;
+	private Collection<MessagePostProcessor> afterReceivePostProcessors;
 
-	private volatile Collection<MessagePostProcessor> afterReceivePostProcessors;
+	private CorrelationDataPostProcessor correlationDataPostProcessor;
 
-	private volatile CorrelationDataPostProcessor correlationDataPostProcessor;
-
-	private volatile boolean isListener;
-
-	private volatile Expression userIdExpression;
+	private Expression userIdExpression;
 
 	private String beanName = "rabbitTemplate";
 
@@ -254,6 +245,18 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware,
 
 	private boolean noLocalReplyConsumer;
 
+	private ErrorHandler replyErrorHandler;
+
+	private volatile Boolean confirmsOrReturnsCapable;
+
+	private volatile boolean publisherConfirms;
+
+	private volatile boolean usingFastReplyTo;
+
+	private volatile boolean evaluatedFastReplyTo;
+
+	private volatile boolean isListener;
+
 	/**
 	 * Convenient constructor for use with setter injection. Don't forget to set the connection factory.
 	 */
@@ -722,9 +725,20 @@ public void setNoLocalReplyConsumer(boolean noLocalReplyConsumer) {
 		this.noLocalReplyConsumer = noLocalReplyConsumer;
 	}
 
+	/**
+	 * When using a direct reply-to container for request/reply operations, set an error
+	 * handler to be invoked when a reply delivery fails (e.g. due to a late reply).
+	 * @param replyErrorHandler the reply error handler
+	 * @since 2.0.11
+	 * @see #setUseDirectReplyToContainer(boolean)
+	 */
+	public void setReplyErrorHandler(ErrorHandler replyErrorHandler) {
+		this.replyErrorHandler = replyErrorHandler;
+	}
+
 	/**
 	 * Invoked by the container during startup so it can verify the queue is correctly
-	 * configured (if a simple reply queue name is used instead of exchange/routingKey.
+	 * configured (if a simple reply queue name is used instead of exchange/routingKey).
 	 * @return the queue name, if configured.
 	 * @since 1.5
 	 */
@@ -1782,6 +1796,9 @@ private Message doSendAndReceiveWithDirect(String exchange, String routingKey, M
 								.toArray(new MessagePostProcessor[this.afterReceivePostProcessors.size()]));
 					}
 					container.setNoLocal(this.noLocalReplyConsumer);
+					if (this.replyErrorHandler != null) {
+						container.setErrorHandler(this.replyErrorHandler);
+					}
 					container.start();
 					this.directReplyToContainers.put(connectionFactory, container);
 					this.replyAddress = Address.AMQ_RABBITMQ_REPLY_TO;
@@ -2404,8 +2421,7 @@ public void onMessage(Message message) {
 					.getHeaders().get(this.correlationKey);
 		}
 		if (messageTag == null) {
-			logger.error("No correlation header in reply");
-			return;
+			throw new AmqpRejectAndDontRequeueException("No correlation header in reply");
 		}
 
 		PendingReply pendingReply = this.replyHolder.get(messageTag);

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

@@ -17,14 +17,26 @@
 package org.springframework.amqp.rabbit.core;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.Assert.assertNotNull;
 
 import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.junit.Test;
 
+import org.springframework.amqp.AmqpRejectAndDontRequeueException;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageProperties;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
 import org.springframework.amqp.utils.test.TestUtils;
+import org.springframework.util.ErrorHandler;
 
 /**
  * @author Gary Russell
@@ -43,17 +55,56 @@ protected RabbitTemplate createSendAndReceiveRabbitTemplate(ConnectionFactory co
 
 	@SuppressWarnings("unchecked")
 	@Test
-	public void channelReleasedOnTimeout() {
+	public void channelReleasedOnTimeout() throws Exception {
 		final CachingConnectionFactory connectionFactory = new CachingConnectionFactory("localhost");
 		RabbitTemplate template = createSendAndReceiveRabbitTemplate(connectionFactory);
 		template.setReplyTimeout(1);
+		AtomicReference<Throwable> exception = new AtomicReference<>();
+		CountDownLatch latch = new CountDownLatch(1);
+		ErrorHandler replyErrorHandler = t -> {
+			exception.set(t);
+			latch.countDown();
+		};
+		template.setReplyErrorHandler(replyErrorHandler);
 		Object reply = template.convertSendAndReceive(ROUTE, "foo");
 		assertThat(reply).isNull();
 		Object container = TestUtils.getPropertyValue(template, "directReplyToContainers", Map.class)
 				.get(template.isUsePublisherConnection()
 						? connectionFactory.getPublisherConnectionFactory()
 						: connectionFactory);
 		assertThat(TestUtils.getPropertyValue(container, "inUseConsumerChannels", Map.class)).hasSize(0);
+		assertThat(TestUtils.getPropertyValue(container, "errorHandler")).isSameAs(replyErrorHandler);
+		Message replyMessage = new Message("foo".getBytes(), new MessageProperties());
+		assertThatThrownBy(() -> template.onMessage(replyMessage))
+			.isInstanceOf(AmqpRejectAndDontRequeueException.class)
+			.hasMessage("No correlation header");
+		replyMessage.getMessageProperties().setCorrelationId("foo");
+		assertThatThrownBy(() -> template.onMessage(replyMessage))
+			.isInstanceOf(AmqpRejectAndDontRequeueException.class)
+			.hasMessage("Reply received after timeout");
+
+		ExecutorService executor = Executors.newFixedThreadPool(1);
+		// Set up a consumer to respond to our producer
+		executor.submit(() -> {
+			Message message = template.receive(ROUTE, 10_000);
+			assertNotNull("No message received", message);
+			template.send(message.getMessageProperties().getReplyTo(), replyMessage);
+			return message;
+		});
+		while (template.receive(ROUTE, 100) != null) {
+			// empty
+		}
+		reply = template.convertSendAndReceive(ROUTE, "foo");
+		assertThat(reply).isNull();
+		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(exception.get()).isInstanceOf(ListenerExecutionFailedException.class);
+		assertThat(exception.get().getCause().getMessage()).isEqualTo("Reply received after timeout");
+		assertThat(((ListenerExecutionFailedException) exception.get()).getFailedMessage().getBody())
+			.isEqualTo(replyMessage.getBody());
+		assertThat(TestUtils.getPropertyValue(container, "inUseConsumerChannels", Map.class)).hasSize(0);
+		executor.shutdownNow();
+		template.stop();
+		connectionFactory.destroy();
 	}
 
 }

src/reference/asciidoc/amqp.adoc

@@ -3220,6 +3220,10 @@ Also, you must not have registered your own `ReturnCallback` with the `RabbitTem
 
 Starting with version 2.1.2, a method `replyTimedOut` has been added, allowing subclasses to be informed of the timeout so they can clean up any retained state.
 
+Starting with versions 2.0.11, 2.1.3, when using the default `DirectReplyToMessageListenerContainer`, you can add an error handler by setting the template's `replyErrorHandler` property.
+This error handler will be invoked for any failed deliveries, such as late replies and messages received without a correlation header.
+The exception passed in is a `ListenerExecutionFailedException` which has a `failedMessage` property.
+
 [[direct-reply-to]]
 ===== RabbitMQ Direct reply-to
 

src/reference/asciidoc/whats-new.adoc

@@ -58,6 +58,10 @@ See <<template-confirms>> for more information.
 A method `replyTimedOut` is now provided to notify subclasses that a reply has timed out, allowing for any state cleanup.
 See <<reply-timeout>> for more information.
 
+You can now specify an `ErrorHandler` to be invoked when using request/reply with a `DirectReplyToMessageListenerContainer` (the default) when exceptions occur when replies are delivered (e.g. late replies).
+See `setReplyErrorHandler` on the `RabbitTemplate`.
+(Also since 2.0.11).
+
 ===== Message Conversion
 
 A new `Jackson2XmlMessageConverter` is introduced to support converting messages from/to XML format.