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

Author: garyrussell

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java

@@ -24,6 +24,7 @@
 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 @@ protected class Listener implements ChannelAwareMessageListener, RetryListener {
 		@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,15 +216,21 @@ public Void doWithRetry(RetryContext context) throws RuntimeException {
 					}, (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)));
 				}
 				else {
 					throw e;
 				}
 			}
+			finally {
+				if (retryDisabled) {
+					attributesHolder.remove();
+				}
+			}
 		}
 
 		private void processMessage(Message message, Channel channel) {

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java

@@ -251,17 +251,11 @@ public void testAdapterConversionError() throws Exception {
 		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");
 			}
 
 		});