Use RetrySynchronizationManager

Use the `RetrySynchronizationManager` instead of a `RetryListener`.

Fix `setAttributesIfNecessary` for gateway conversion errors (this, at least,
should be cherry-picked).

Author: garyrussell

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

@@ -39,9 +39,7 @@
 import org.springframework.integration.support.ErrorMessageStrategy;
 import org.springframework.integration.support.ErrorMessageUtils;
 import org.springframework.retry.RecoveryCallback;
-import org.springframework.retry.RetryCallback;
-import org.springframework.retry.RetryContext;
-import org.springframework.retry.RetryListener;
+import org.springframework.retry.support.RetrySynchronizationManager;
 import org.springframework.retry.support.RetryTemplate;
 import org.springframework.util.Assert;
 
@@ -132,9 +130,6 @@ protected void onInit() {
 					+ "send an error message when retries are exhausted");
 		}
 		Listener messageListener = new Listener();
-		if (this.retryTemplate != null) {
-			this.retryTemplate.registerListener(messageListener);
-		}
 		this.messageListenerContainer.setMessageListener(messageListener);
 		this.messageListenerContainer.afterPropertiesSet();
 		super.onInit();
@@ -177,7 +172,9 @@ private void setAttributesIfNecessary(Message amqpMessage, org.springframework.m
 			attributesHolder.set(ErrorMessageUtils.getAttributeAccessor(null, null));
 		}
 		if (needAttributes) {
-			AttributeAccessor attributes = attributesHolder.get();
+			AttributeAccessor attributes = this.retryTemplate != null
+					? RetrySynchronizationManager.getContext()
+					: attributesHolder.get();
 			if (attributes != null) {
 				attributes.setAttribute(ErrorMessageUtils.INPUT_MESSAGE_CONTEXT_KEY, message);
 				attributes.setAttribute(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE, amqpMessage);
@@ -196,7 +193,7 @@ protected AttributeAccessor getErrorMessageAttributes(org.springframework.messag
 		}
 	}
 
-	protected class Listener implements ChannelAwareMessageListener, RetryListener {
+	protected class Listener implements ChannelAwareMessageListener {
 
 		@SuppressWarnings("unchecked")
 		@Override
@@ -259,26 +256,6 @@ private org.springframework.messaging.Message<Object> createMessage(Message mess
 			return messagingMessage;
 		}
 
-		@Override
-		public <T, E extends Throwable> boolean open(RetryContext context, RetryCallback<T, E> callback) {
-			if (AmqpInboundChannelAdapter.this.recoveryCallback != null) {
-				attributesHolder.set(context);
-			}
-			return true;
-		}
-
-		@Override
-		public <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback,
-				Throwable throwable) {
-			attributesHolder.remove();
-		}
-
-		@Override
-		public <T, E extends Throwable> void onError(RetryContext context, RetryCallback<T, E> callback,
-				Throwable throwable) {
-			// Empty
-		}
-
 	}
 
 }

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

@@ -296,6 +296,7 @@ private org.springframework.messaging.Message<Object> convert(Message message, C
 			}
 			catch (RuntimeException e) {
 				if (getErrorChannel() != null) {
+					setAttributesIfNecessary(message, null);
 					AmqpInboundGateway.this.messagingTemplate.send(getErrorChannel(), buildErrorMessage(null,
 							new ListenerExecutionFailedException("Message conversion failed", e, message)));
 				}

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

@@ -248,9 +248,12 @@ public Object fromMessage(org.springframework.amqp.core.Message message) throws
 
 		});
 		adapter.afterPropertiesSet();
-		((ChannelAwareMessageListener) container.getMessageListener()).onMessage(null, null);
+		((ChannelAwareMessageListener) container.getMessageListener())
+				.onMessage(mock(org.springframework.amqp.core.Message.class), null);
 		assertNull(outputChannel.receive(0));
-		assertNotNull(errorChannel.receive(0));
+		Message<?> received = errorChannel.receive(0);
+		assertNotNull(received);
+		assertNotNull(received.getHeaders().get(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE));
 	}
 
 	@Test
@@ -276,14 +279,17 @@ public org.springframework.amqp.core.Message toMessage(Object object, MessagePro
 
 			@Override
 			public Object fromMessage(org.springframework.amqp.core.Message message) throws MessageConversionException {
-				return null;
+				throw new MessageConversionException("intended");
 			}
 
 		});
 		adapter.afterPropertiesSet();
-		((ChannelAwareMessageListener) container.getMessageListener()).onMessage(null, null);
+		((ChannelAwareMessageListener) container.getMessageListener())
+				.onMessage(mock(org.springframework.amqp.core.Message.class), null);
 		assertNull(outputChannel.receive(0));
-		assertNotNull(errorChannel.receive(0));
+		Message<?> received = errorChannel.receive(0);
+		assertNotNull(received);
+		assertNotNull(received.getHeaders().get(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE));
 	}
 
 	@Test