GH-1339: Fix RLErrorHandler with Conversion Ex. (#1346)

Resolves #1339

Error handler was not called for conversion exceptions, preventing
the application from returning some error to the caller for request/reply processing.

**cherry-pick to 2.2.x**

Author: garyrussell

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -131,8 +131,50 @@ public void setMessageConverter(MessageConverter messageConverter) {
 
 	@Override
 	public void onMessage(org.springframework.amqp.core.Message amqpMessage, Channel channel) throws Exception { // NOSONAR
-		Message<?> message = toMessagingMessage(amqpMessage);
-		invokeHandlerAndProcessResult(amqpMessage, channel, message);
+		Message<?> message = null;
+		try {
+			message = toMessagingMessage(amqpMessage);
+			invokeHandlerAndProcessResult(amqpMessage, channel, message);
+		}
+		catch (ListenerExecutionFailedException ex) {
+			handleException(amqpMessage, channel, message, ex);
+		}
+		catch (ReplyFailureException ex) {
+			throw ex;
+		}
+		catch (Exception ex) {
+			handleException(amqpMessage, channel, message, new ListenerExecutionFailedException(
+					"Failed to convert message", ex, amqpMessage));
+		}
+	}
+
+	private void handleException(org.springframework.amqp.core.Message amqpMessage, Channel channel,
+			@Nullable Message<?> message, ListenerExecutionFailedException e) throws Exception {
+
+		if (this.errorHandler != null) {
+			try {
+				Message<?> messageWithChannel = null;
+				if (message != null) {
+					messageWithChannel = MessageBuilder.fromMessage(message)
+							.setHeader(AmqpHeaders.CHANNEL, channel)
+							.build();
+				}
+				Object errorResult = this.errorHandler.handleError(amqpMessage, messageWithChannel, e);
+				if (errorResult != null) {
+					handleResult(this.handlerAdapter.getInvocationResultFor(errorResult, message.getPayload()),
+							amqpMessage, channel, message);
+				}
+				else {
+					logger.trace("Error handler returned no result");
+				}
+			}
+			catch (Exception ex) {
+				returnOrThrow(amqpMessage, channel, message, ex, ex);
+			}
+		}
+		else {
+			returnOrThrow(amqpMessage, channel, message, e.getCause(), e);
+		}
 	}
 
 	protected void invokeHandlerAndProcessResult(@Nullable org.springframework.amqp.core.Message amqpMessage,
@@ -142,41 +184,16 @@ protected void invokeHandlerAndProcessResult(@Nullable org.springframework.amqp.
 			logger.debug("Processing [" + message + "]");
 		}
 		InvocationResult result = null;
-		try {
-			if (this.messagingMessageConverter.method == null && amqpMessage != null) {
-				amqpMessage.getMessageProperties()
-						.setTargetMethod(this.handlerAdapter.getMethodFor(message.getPayload()));
-			}
-			result = invokeHandler(amqpMessage, channel, message);
-			if (result.getReturnValue() != null) {
-				handleResult(result, amqpMessage, channel, message);
-			}
-			else {
-				logger.trace("No result object given - no result to handle");
-			}
+		if (this.messagingMessageConverter.method == null && amqpMessage != null) {
+			amqpMessage.getMessageProperties()
+					.setTargetMethod(this.handlerAdapter.getMethodFor(message.getPayload()));
 		}
-		catch (ListenerExecutionFailedException e) {
-			if (this.errorHandler != null) {
-				try {
-					Message<?> messageWithChannel = MessageBuilder.fromMessage(message)
-							.setHeader(AmqpHeaders.CHANNEL, channel)
-							.build();
-					Object errorResult = this.errorHandler.handleError(amqpMessage, messageWithChannel, e);
-					if (errorResult != null) {
-						handleResult(this.handlerAdapter.getInvocationResultFor(errorResult, message.getPayload()),
-								amqpMessage, channel, message);
-					}
-					else {
-						logger.trace("Error handler returned no result");
-					}
-				}
-				catch (Exception ex) {
-					returnOrThrow(amqpMessage, channel, message, ex, ex);
-				}
-			}
-			else {
-				returnOrThrow(amqpMessage, channel, message, e.getCause(), e);
-			}
+		result = invokeHandler(amqpMessage, channel, message);
+		if (result.getReturnValue() != null) {
+			handleResult(result, amqpMessage, channel, message);
+		}
+		else {
+			logger.trace("No result object given - no result to handle");
 		}
 	}
 

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/api/RabbitListenerErrorHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2021 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.
@@ -18,6 +18,7 @@
 
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.rabbit.support.ListenerExecutionFailedException;
+import org.springframework.lang.Nullable;
 
 /**
  * An error handler which is called when a {code @RabbitListener} method
@@ -35,13 +36,13 @@ public interface RabbitListenerErrorHandler {
 	 * Handle the error. If an exception is not thrown, the return value is returned to
 	 * the sender using normal {@code replyTo/@SendTo} semantics.
 	 * @param amqpMessage the raw message received.
-	 * @param message the converted spring-messaging message.
+	 * @param message the converted spring-messaging message (if available).
 	 * @param exception the exception the listener threw, wrapped in a
 	 * {@link ListenerExecutionFailedException}.
 	 * @return the return value to be sent to the sender.
 	 * @throws Exception an exception which may be the original or different.
 	 */
-	Object handleError(Message amqpMessage, org.springframework.messaging.Message<?> message,
+	Object handleError(Message amqpMessage, @Nullable org.springframework.messaging.Message<?> message,
 			ListenerExecutionFailedException exception) throws Exception; // NOSONAR
 
 }

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapterTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2020 the original author or authors.
+ * Copyright 2014-2021 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.
@@ -29,16 +29,20 @@
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler;
 import org.springframework.amqp.rabbit.support.ListenerExecutionFailedException;
 import org.springframework.amqp.rabbit.test.MessageTestUtils;
 import org.springframework.amqp.support.AmqpHeaders;
 import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
+import org.springframework.amqp.support.converter.MessageConversionException;
+import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.amqp.support.converter.SimpleMessageConverter;
 import org.springframework.amqp.utils.test.TestUtils;
 import org.springframework.beans.factory.support.StaticListableBeanFactory;
@@ -273,9 +277,50 @@ public void batchTypedObjectTest() {
 		assertThat(this.sample.batchPayloads.get(0).getClass()).isEqualTo(Foo.class);
 	}
 
+	@Test
+	void errorHandlerAfterConversionEx() throws Exception {
+		org.springframework.amqp.core.Message message = MessageTestUtils.createTextMessage("foo");
+		Channel channel = mock(Channel.class);
+		AtomicBoolean ehCalled = new AtomicBoolean();
+		MessagingMessageListenerAdapter listener = getSimpleInstance("fail",
+				new RabbitListenerErrorHandler() {
+
+			@Override
+			public Object handleError(org.springframework.amqp.core.Message amqpMessage, Message<?> message,
+					ListenerExecutionFailedException exception) throws Exception {
+
+				ehCalled.set(true);
+				return null;
+			}
+
+		}, String.class);
+		listener.setMessageConverter(new MessageConverter() {
+
+			@Override
+			public org.springframework.amqp.core.Message toMessage(Object object, MessageProperties messageProperties)
+					throws MessageConversionException {
+
+				return null;
+			}
+
+			@Override
+			public Object fromMessage(org.springframework.amqp.core.Message message) throws MessageConversionException {
+				throw new MessageConversionException("test");
+			}
+		});
+		listener.onMessage(message, channel);
+		assertThat(ehCalled.get()).isTrue();
+	}
+
 	protected MessagingMessageListenerAdapter getSimpleInstance(String methodName, Class<?>... parameterTypes) {
+		return getSimpleInstance(methodName, null, parameterTypes);
+	}
+
+	protected MessagingMessageListenerAdapter getSimpleInstance(String methodName, RabbitListenerErrorHandler eh,
+			Class<?>... parameterTypes) {
+
 		Method m = ReflectionUtils.findMethod(SampleBean.class, methodName, parameterTypes);
-		return createInstance(m, false);
+		return createInstance(m, false, eh);
 	}
 
 	protected MessagingMessageListenerAdapter getSimpleInstance(String methodName, boolean returnExceptions,
@@ -285,7 +330,13 @@ protected MessagingMessageListenerAdapter getSimpleInstance(String methodName, b
 	}
 
 	protected MessagingMessageListenerAdapter createInstance(Method m, boolean returnExceptions) {
-		MessagingMessageListenerAdapter adapter = new MessagingMessageListenerAdapter(null, m, returnExceptions, null);
+		return createInstance(m, returnExceptions, null);
+	}
+
+	protected MessagingMessageListenerAdapter createInstance(Method m, boolean returnExceptions,
+			RabbitListenerErrorHandler eh) {
+
+		MessagingMessageListenerAdapter adapter = new MessagingMessageListenerAdapter(null, m, returnExceptions, eh);
 		adapter.setHandlerAdapter(new HandlerAdapter(factory.createInvocableHandlerMethod(sample, m)));
 		return adapter;
 	}

src/reference/asciidoc/amqp.adoc

@@ -3276,7 +3276,7 @@ static class TxServiceImpl implements TxService<Foo> {
 
     @Override
     @RabbitListener(...)
-    public String handle(Foo foo, String rk) {
+    public String handle(Thing thing, String rk) {
          ...
     }
 
@@ -3359,6 +3359,10 @@ public Object handleError(Message amqpMessage, org.springframework.messaging.Mes
 ----
 ====
 
+Starting with version 2.2.18, if a message conversion exception is thrown, the error handler will be called, with `null` in the `message` argument.
+This allows the application to send some result to the caller, indicating that a badly-formed message was received.
+Previously, such errors were thrown and handled by the container.
+
 ====== Container Management
 
 Containers created for annotations are not registered with the application context.