From 37c92a128b41803f91052596a34a1f730080df54 Mon Sep 17 00:00:00 2001
From: Gary Russell <grussell@pivotal.io>
Date: Tue, 5 Jun 2018 16:34:58 -0400
Subject: [PATCH] INT-4482: AMQP: Fix Double ErrorMessage

JIRA: https://jira.spring.io/browse/INT-4482

The outer try/catch sends an `ErrorMessage` for all exceptions; it should
only do so for `MessageConversionException`. Integration flow exceptions
will have been already handled by `MessageProducerSupport`.

Also, populate the raw message header consistently - previously it only
was populated for flow exceptions. Although the LEFE contains the raw
message, it should be in the `ErrorMessage` header for consistency.

**cherry-pick to 5.0.x, 4.3.x**

* Polishing - PR Comments

# Conflicts:
#	spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java
#	spring-integration-amqp/src/test/java/org/springframework/integration/amqp/dsl/AmqpTests.java
---
 .../inbound/AmqpInboundChannelAdapter.java    | 20 +++++++++++--------
 .../amqp/inbound/InboundEndpointTests.java    | 10 ++--------
 2 files changed, 14 insertions(+), 16 deletions(-)

diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java
index ecd7e1ec5d..d8482044ec 100644
--- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java
+++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java
@@ -24,6 +24,7 @@ import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
 import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
 import org.springframework.amqp.support.AmqpHeaders;
+import org.springframework.amqp.support.converter.MessageConversionException;
 import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.amqp.support.converter.SimpleMessageConverter;
 import org.springframework.core.AttributeAccessor;
@@ -197,14 +198,11 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
 		@SuppressWarnings("unchecked")
 		@Override
 		public void onMessage(final Message message, final Channel channel) throws Exception {
+			boolean retryDisabled = AmqpInboundChannelAdapter.this.retryTemplate == null;
 			try {
-				if (AmqpInboundChannelAdapter.this.retryTemplate == null) {
-					try {
-						processMessage(message, channel);
-					}
-					finally {
-						attributesHolder.remove();
-					}
+
+				if (retryDisabled) {
+					processMessage(message, channel);
 				}
 				else {
 					AmqpInboundChannelAdapter.this.retryTemplate.execute(new RetryCallback<Object, RuntimeException>() {
@@ -218,8 +216,9 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
 					}, (RecoveryCallback<Object>) AmqpInboundChannelAdapter.this.recoveryCallback);
 				}
 			}
-			catch (RuntimeException e) {
+			catch (MessageConversionException e) {
 				if (getErrorChannel() != null) {
+					setAttributesIfNecessary(message, null);
 					getMessagingTemplate().send(getErrorChannel(), buildErrorMessage(null,
 							new ListenerExecutionFailedException("Message conversion failed", e, message)));
 				}
@@ -227,6 +226,11 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
 					throw e;
 				}
 			}
+			finally {
+				if (retryDisabled) {
+					attributesHolder.remove();
+				}
+			}
 		}
 
 		private void processMessage(Message message, Channel channel) {
diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java
index aa5843a826..3982c9d4fa 100644
--- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java
+++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java
@@ -251,17 +251,11 @@ public class InboundEndpointTests {
 		adapter.setOutputChannel(outputChannel);
 		QueueChannel errorChannel = new QueueChannel();
 		adapter.setErrorChannel(errorChannel);
-		adapter.setMessageConverter(new MessageConverter() {
-
-			@Override
-			public org.springframework.amqp.core.Message toMessage(Object object, MessageProperties messageProperties)
-					throws MessageConversionException {
-				throw new MessageConversionException("intended");
-			}
+		adapter.setMessageConverter(new SimpleMessageConverter() {
 
 			@Override
 			public Object fromMessage(org.springframework.amqp.core.Message message) throws MessageConversionException {
-				return null;
+				throw new MessageConversionException("intended");
 			}
 
 		});