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

Author: garyrussell

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

@@ -25,6 +25,7 @@
 import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
 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;
@@ -200,14 +201,10 @@ 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 {
-						createAndSend(message, channel);
-					}
-					finally {
-						attributesHolder.remove();
-					}
+				if (retryDisabled) {
+					createAndSend(message, channel);
 				}
 				else {
 					final org.springframework.messaging.Message<Object> toSend = createMessage(message, channel);
@@ -220,15 +217,21 @@ public void onMessage(final Message message, final Channel channel) throws Excep
 							(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 createAndSend(Message message, Channel channel) {

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/dsl/AmqpTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2017 the original author or authors.
+ * Copyright 2014-2018 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.
@@ -16,11 +16,16 @@
 
 package org.springframework.integration.amqp.dsl;
 
+import static org.hamcrest.Matchers.instanceOf;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 
+import java.util.concurrent.atomic.AtomicReference;
+
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.Test;
 
@@ -37,21 +42,27 @@
 import org.springframework.amqp.rabbit.junit.RabbitAvailableCondition;
 import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+import org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException;
+import org.springframework.amqp.support.converter.MessageConversionException;
+import org.springframework.amqp.support.converter.SimpleMessageConverter;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.Lifecycle;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.integration.amqp.channel.AbstractAmqpChannel;
 import org.springframework.integration.amqp.inbound.AmqpInboundGateway;
 import org.springframework.integration.amqp.support.AmqpHeaderMapper;
+import org.springframework.integration.amqp.support.AmqpMessageHeaderErrorMessageStrategy;
 import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlowBuilder;
 import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.support.StringObjectMapBuilder;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
@@ -67,7 +78,8 @@
  */
 @SpringJUnitConfig
 @RabbitAvailable(queues = { "amqpOutboundInput", "amqpReplyChannel", "asyncReplies",
-							"defaultReplyTo", "si.dsl.test", "testTemplateChannelTransacted" })
+							"defaultReplyTo", "si.dsl.test", "si.dsl.exception.test.dlq",
+							"si.dsl.conv.exception.test.dlq", "testTemplateChannelTransacted" })
 @DirtiesContext
 public class AmqpTests {
 
@@ -92,8 +104,10 @@ public class AmqpTests {
 	private Lifecycle asyncOutboundGateway;
 
 	@AfterAll
-	public static void tearDown() {
-		RabbitAvailableCondition.getBrokerRunning().removeTestQueues();
+	public static void tearDown(ConfigurableApplicationContext context) {
+		context.stop(); // prevent queues from being redeclared after deletion
+		RabbitAvailableCondition.getBrokerRunning().removeTestQueues("si.dsl.exception.test",
+				"si.dsl.conv.exception.test");
 	}
 
 	@Test
@@ -173,6 +187,33 @@ public void testAmqpAsyncOutboundGatewayFlow() throws Exception {
 		this.asyncOutboundGateway.stop();
 	}
 
+	@Autowired
+	private AtomicReference<ListenerExecutionFailedException> lefe;
+
+	@Autowired
+	private AtomicReference<?> raw;
+
+
+	@Test
+	public void testInboundMessagingExceptionFlow() {
+		this.amqpTemplate.convertAndSend("si.dsl.exception.test", "foo");
+		assertNotNull(this.amqpTemplate.receive("si.dsl.exception.test.dlq", 30_000));
+		assertNull(this.lefe.get());
+		assertNotNull(this.raw.get());
+		this.raw.set(null);
+	}
+
+	@Test
+	public void testInboundConversionExceptionFlow() {
+		this.amqpTemplate.convertAndSend("si.dsl.conv.exception.test", "foo");
+		assertNotNull(this.amqpTemplate.receive("si.dsl.conv.exception.test.dlq", 30_000));
+		assertNotNull(this.lefe.get());
+		assertThat(this.lefe.get().getCause(), instanceOf(MessageConversionException.class));
+		assertNotNull(this.raw.get());
+		this.raw.set(null);
+		this.lefe.set(null);
+	}
+
 	@Autowired
 	private AbstractAmqpChannel unitChannel;
 
@@ -277,6 +318,84 @@ public IntegrationFlow amqpInboundFlow(ConnectionFactory rabbitConnectionFactory
 					.get();
 		}
 
+		@Bean
+		public AtomicReference<ListenerExecutionFailedException> lefe() {
+			return new AtomicReference<>();
+		}
+
+		@Bean
+		public AtomicReference<org.springframework.amqp.core.Message> raw() {
+			return new AtomicReference<>();
+		}
+
+		@Bean
+		public Queue exQueue() {
+			return new Queue("si.dsl.exception.test", true, false, false,
+					new StringObjectMapBuilder()
+						.put("x-dead-letter-exchange", "")
+						.put("x-dead-letter-routing-key", exDLQ().getName())
+						.get());
+		}
+
+		@Bean
+		public Queue exDLQ() {
+			return new Queue("si.dsl.exception.test.dlq");
+		}
+
+		@Bean
+		public IntegrationFlow inboundWithExceptionFlow(ConnectionFactory cf) {
+			return IntegrationFlows.from(Amqp.inboundAdapter(cf, exQueue())
+						.configureContainer(c -> c.defaultRequeueRejected(false))
+						.errorChannel("errors.input"))
+					.handle(m -> {
+						throw new RuntimeException("fail");
+					})
+					.get();
+		}
+
+		@Bean
+		public IntegrationFlow errors() {
+			return f -> f.handle(m -> {
+					raw().set(m.getHeaders().get(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE,
+							org.springframework.amqp.core.Message.class));
+					if (m.getPayload() instanceof ListenerExecutionFailedException) {
+						lefe().set((ListenerExecutionFailedException) m.getPayload());
+					}
+					throw (RuntimeException) m.getPayload();
+				});
+		}
+
+		@Bean
+		public Queue exConvQueue() {
+			return new Queue("si.dsl.conv.exception.test", true, false, false,
+					new StringObjectMapBuilder()
+						.put("x-dead-letter-exchange", "")
+						.put("x-dead-letter-routing-key", exConvDLQ().getName())
+						.get());
+		}
+
+		@Bean
+		public Queue exConvDLQ() {
+			return new Queue("si.dsl.conv.exception.test.dlq");
+		}
+
+		@Bean
+		public IntegrationFlow inboundWithConvExceptionFlow(ConnectionFactory cf) {
+			return IntegrationFlows.from(Amqp.inboundAdapter(cf, exConvQueue())
+						.configureContainer(c -> c.defaultRequeueRejected(false))
+						.messageConverter(new SimpleMessageConverter() {
+
+							@Override
+							public Object fromMessage(org.springframework.amqp.core.Message message)
+									throws MessageConversionException {
+								throw new MessageConversionException("fail");
+							}
+
+						})
+						.errorChannel("errors.input"))
+					.get();
+		}
+
 		@Bean
 		public Queue asyncReplies() {
 			return new Queue("asyncReplies");

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

@@ -239,17 +239,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");
 			}
 
 		});

spring-integration-core/src/main/java/org/springframework/integration/support/StringObjectMapBuilder.java

@@ -0,0 +1,28 @@
+/*
+ * Copyright 2018 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.support;
+
+/**
+ * A map builder creating a map with String keys and values.
+ *
+ * @author Gary Russell
+ *
+ * @since 5.0.6
+ */
+public class StringObjectMapBuilder extends MapBuilder<StringObjectMapBuilder, String, Object> {
+
+}