GH-1339: Fix NPEs with returns after conversion ex

Resolves https://github.com/spring-projects/spring-amqp/issues/1339

Previous fix calls the error handler after a conversion exception, but
sending a reply returned by the eh (or returning the exception) caused
NPEs because the return logic referenced the inbound converted message.

**cherry-pick to 2.2.x**

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.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.
@@ -176,9 +176,11 @@ public abstract class AbstractAdaptableMessageListener implements ChannelAwareMe
 	 * that return result objects, which will be wrapped in
 	 * a response message and sent to a response destination.
 	 * <p>
+	 * It is parsed in {@link Address} so should be of the form exchange/rk.
+	 * <p>
 	 * It can be a string surrounded by "!{...}" in which case the expression is
 	 * evaluated at runtime; see the reference manual for more information.
-	 * @param defaultReplyTo The exchange.
+	 * @param defaultReplyTo The replyTo address.
 	 * @since 1.6
 	 */
 	public void setResponseAddress(String defaultReplyTo) {

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

@@ -161,8 +161,11 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis
 				}
 				Object errorResult = this.errorHandler.handleError(amqpMessage, messageWithChannel, e);
 				if (errorResult != null) {
-					handleResult(this.handlerAdapter.getInvocationResultFor(errorResult, message.getPayload()),
-							amqpMessage, channel, message);
+					Object payload = message == null ? null : message.getPayload();
+					InvocationResult invResult = payload == null
+							? new InvocationResult(errorResult, null, null, null, null)
+							: this.handlerAdapter.getInvocationResultFor(errorResult, payload);
+					handleResult(invResult, amqpMessage, channel, message);
 				}
 				else {
 					logger.trace("Error handler returned no result");
@@ -203,15 +206,17 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis
 		if (!this.returnExceptions) {
 			throw exceptionToThrow;
 		}
+		Object payload = message == null ? null : message.getPayload();
 		try {
+
 			handleResult(new InvocationResult(new RemoteInvocationResult(throwableToReturn), null,
-						this.handlerAdapter.getReturnTypeFor(message.getPayload()),
+						payload == null ? Object.class : this.handlerAdapter.getReturnTypeFor(payload),
 						this.handlerAdapter.getBean(),
-						this.handlerAdapter.getMethodFor(message.getPayload())),
+						payload == null ? null : this.handlerAdapter.getMethodFor(payload)),
 					amqpMessage, channel, message);
 		}
 		catch (ReplyFailureException rfe) {
-			if (void.class.equals(this.handlerAdapter.getReturnTypeFor(message.getPayload()))) {
+			if (payload == null || void.class.equals(this.handlerAdapter.getReturnTypeFor(payload))) {
 				throw exceptionToThrow;
 			}
 			else {

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

@@ -19,6 +19,7 @@ package org.springframework.amqp.rabbit.listener.adapter;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.fail;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
@@ -293,7 +294,7 @@ public class MessagingMessageListenerAdapterTests {
 				return null;
 			}
 
-		}, String.class);
+		}, false, String.class);
 		listener.setMessageConverter(new MessageConverter() {
 
 			@Override
@@ -312,15 +313,71 @@ public class MessagingMessageListenerAdapterTests {
 		assertThat(ehCalled.get()).isTrue();
 	}
 
+	@Test
+	void errorHandlerAfterConversionExWithResult() 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 "foo";
+			}
+
+		}, false, String.class);
+		listener.setMessageConverter(new MessageConverter() {
+
+			@Override
+			public org.springframework.amqp.core.Message toMessage(Object object, MessageProperties messageProperties)
+					throws MessageConversionException {
+
+				return new org.springframework.amqp.core.Message(((String) object).getBytes(), messageProperties);
+			}
+
+			@Override
+			public Object fromMessage(org.springframework.amqp.core.Message message) throws MessageConversionException {
+				throw new MessageConversionException("test");
+			}
+		});
+		listener.setResponseAddress("foo/bar");
+		listener.onMessage(message, channel);
+		verify(channel).basicPublish(any(), any(), anyBoolean(), any(), any());
+		assertThat(ehCalled.get()).isTrue();
+	}
+
+	@Test
+	void errorHandlerAfterConversionExWithReturnEx() throws Exception {
+		org.springframework.amqp.core.Message message = MessageTestUtils.createTextMessage("foo");
+		Channel channel = mock(Channel.class);
+		MessagingMessageListenerAdapter listener = getSimpleInstance("fail", null, true, String.class);
+		class MC extends SimpleMessageConverter {
+
+			@Override
+			public Object fromMessage(org.springframework.amqp.core.Message message) throws MessageConversionException {
+				throw new MessageConversionException("test");
+			}
+
+		}
+		listener.setMessageConverter(new MC());
+		listener.setResponseAddress("foo/bar");
+		listener.onMessage(message, channel);
+		verify(channel).basicPublish(any(), any(), anyBoolean(), any(), any());
+	}
+
 	protected MessagingMessageListenerAdapter getSimpleInstance(String methodName, Class<?>... parameterTypes) {
-		return getSimpleInstance(methodName, null, parameterTypes);
+		return getSimpleInstance(methodName, null, false, parameterTypes);
 	}
 
 	protected MessagingMessageListenerAdapter getSimpleInstance(String methodName, RabbitListenerErrorHandler eh,
-			Class<?>... parameterTypes) {
+			boolean returnEx, Class<?>... parameterTypes) {
 
 		Method m = ReflectionUtils.findMethod(SampleBean.class, methodName, parameterTypes);
-		return createInstance(m, false, eh);
+		return createInstance(m, returnEx, eh);
 	}
 
 	protected MessagingMessageListenerAdapter getSimpleInstance(String methodName, boolean returnExceptions,