GH-3437: Add MessageRecoverer to AMQP Inbounds (#3465)

* GH-3437: Add `MessageRecoverer` to AMQP Inbounds

Fixes https://github.com/spring-projects/spring-integration/issues/3437

For better end-user experience with AMQP Inbound Endpoints, expose
a `MessageRecoverer` option
* Test the feature and document this new option

* * Fix Checkstyle & update Copyright

* Fix language in Docs

Co-authored-by: Gary Russell <grussell@vmware.com>

Co-authored-by: Gary Russell <grussell@vmware.com>

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpBaseInboundChannelAdapterSpec.java

@@ -16,6 +16,7 @@
 
 package org.springframework.integration.amqp.dsl;
 
+import org.springframework.amqp.rabbit.retry.MessageRecoverer;
 import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter;
 import org.springframework.integration.amqp.support.AmqpHeaderMapper;
@@ -102,4 +103,16 @@ public class AmqpBaseInboundChannelAdapterSpec<S extends AmqpBaseInboundChannelA
 		return _this();
 	}
 
+	/**
+	 * Set a {@link MessageRecoverer} when using retry within the adapter.
+	 * @param messageRecoverer the callback.
+	 * @return the spec.
+	 * @since 5.5
+	 * @see AmqpInboundChannelAdapter#setMessageRecoverer(MessageRecoverer)
+	 */
+	public S messageRecoverer(MessageRecoverer messageRecoverer) {
+		this.target.setMessageRecoverer(messageRecoverer);
+		return _this();
+	}
+
 }

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpBaseInboundGatewaySpec.java

@@ -17,7 +17,9 @@
 package org.springframework.integration.amqp.dsl;
 
 import org.springframework.amqp.rabbit.batch.BatchingStrategy;
+import org.springframework.amqp.rabbit.retry.MessageRecoverer;
 import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter;
 import org.springframework.integration.amqp.inbound.AmqpInboundGateway;
 import org.springframework.integration.amqp.support.AmqpHeaderMapper;
 import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
@@ -176,4 +178,16 @@ public class AmqpBaseInboundGatewaySpec<S extends AmqpBaseInboundGatewaySpec<S>>
 		return _this();
 	}
 
+	/**
+	 * Set a {@link MessageRecoverer} when using retry within the adapter.
+	 * @param messageRecoverer the callback.
+	 * @return the spec.
+	 * @since 5.5
+	 * @see AmqpInboundChannelAdapter#setMessageRecoverer(MessageRecoverer)
+	 */
+	public S messageRecoverer(MessageRecoverer messageRecoverer) {
+		this.target.setMessageRecoverer(messageRecoverer);
+		return _this();
+	}
+
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-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.
@@ -29,6 +29,8 @@ import org.springframework.amqp.rabbit.batch.SimpleBatchingStrategy;
 import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.api.ChannelAwareBatchMessageListener;
 import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
+import org.springframework.amqp.rabbit.retry.MessageBatchRecoverer;
+import org.springframework.amqp.rabbit.retry.MessageRecoverer;
 import org.springframework.amqp.support.AmqpHeaders;
 import org.springframework.amqp.support.converter.MessageConversionException;
 import org.springframework.amqp.support.converter.MessageConverter;
@@ -110,6 +112,8 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
 
 	private RecoveryCallback<?> recoveryCallback;
 
+	private MessageRecoverer messageRecoverer;
+
 	private BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(0, 0, 0L);
 
 	private boolean bindSourceMessage;
@@ -154,6 +158,7 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
 
 	/**
 	 * Set a {@link RecoveryCallback} when using retry within the adapter.
+	 * Mutually exclusive with {@link #setMessageRecoverer(MessageRecoverer)}.
 	 * @param recoveryCallback the callback.
 	 * @since 4.3.10
 	 * @see #setRetryTemplate(RetryTemplate)
@@ -162,6 +167,16 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
 		this.recoveryCallback = recoveryCallback;
 	}
 
+	/**
+	 * Configure a {@link MessageRecoverer} for retry operations.
+	 * A more AMQP-specific convenience instead of {@link #setRecoveryCallback(RecoveryCallback)}.
+	 * @param messageRecoverer the {@link MessageRecoverer} to use.
+	 * @since 5.5
+	 */
+	public void setMessageRecoverer(MessageRecoverer messageRecoverer) {
+		this.messageRecoverer = messageRecoverer;
+	}
+
 	/**
 	 * Set a batching strategy to use when de-batching messages created by a batching
 	 * producer (such as the BatchingRabbitTemplate).
@@ -206,6 +221,7 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
 			Assert.state(getErrorChannel() == null, "Cannot have an 'errorChannel' property when a 'RetryTemplate' is "
 					+ "provided; use an 'ErrorMessageSendingRecoverer' in the 'recoveryCallback' property to "
 					+ "send an error message when retries are exhausted");
+			setupRecoveryCallbackIfAny();
 		}
 		Listener messageListener;
 		if (this.messageListenerContainer.isConsumerBatchEnabled()) {
@@ -219,6 +235,38 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
 		super.onInit();
 	}
 
+	private void setupRecoveryCallbackIfAny() {
+		Assert.state(this.recoveryCallback == null || this.messageRecoverer == null,
+				"Only one of 'recoveryCallback' or 'messageRecoverer' may be provided, but not both");
+		if (this.messageRecoverer != null) {
+			if (this.messageListenerContainer.isConsumerBatchEnabled()) {
+				Assert.isInstanceOf(MessageBatchRecoverer.class, this.messageRecoverer,
+						"The 'messageRecoverer' must be an instance of MessageBatchRecoverer " +
+								"when consumer configured for batch mode");
+				this.recoveryCallback =
+						context -> {
+							@SuppressWarnings("unchecked")
+							List<Message> messagesToRecover =
+									(List<Message>) RetrySynchronizationManager.getContext()
+											.getAttribute(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE);
+							((MessageBatchRecoverer) this.messageRecoverer).recover(messagesToRecover,
+									context.getLastThrowable());
+							return null;
+						};
+			}
+			else {
+				this.recoveryCallback =
+						context -> {
+							Message messageToRecover =
+									(Message) RetrySynchronizationManager.getContext()
+											.getAttribute(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE);
+							this.messageRecoverer.recover(messageToRecover, context.getLastThrowable());
+							return null;
+						};
+			}
+		}
+	}
+
 	@Override
 	protected void doStart() {
 		this.messageListenerContainer.start();

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-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.
@@ -30,6 +30,7 @@ import org.springframework.amqp.rabbit.batch.SimpleBatchingStrategy;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
+import org.springframework.amqp.rabbit.retry.MessageRecoverer;
 import org.springframework.amqp.support.AmqpHeaders;
 import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.amqp.support.converter.SimpleMessageConverter;
@@ -83,7 +84,9 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
 
 	private RetryTemplate retryTemplate;
 
-	private RecoveryCallback<? extends Object> recoveryCallback;
+	private RecoveryCallback<?> recoveryCallback;
+
+	private MessageRecoverer messageRecoverer;
 
 	private BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(0, 0, 0L);
 
@@ -181,6 +184,7 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
 
 	/**
 	 * Set a {@link RecoveryCallback} when using retry within the gateway.
+	 * Mutually exclusive with {@link #setMessageRecoverer(MessageRecoverer)}.
 	 * @param recoveryCallback the callback.
 	 * @since 4.3.10
 	 * @see #setRetryTemplate(RetryTemplate)
@@ -189,6 +193,16 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
 		this.recoveryCallback = recoveryCallback;
 	}
 
+	/**
+	 * Configure a {@link MessageRecoverer} for retry operations.
+	 * A more AMQP-specific convenience instead of {@link #setRecoveryCallback(RecoveryCallback)}.
+	 * @param messageRecoverer the {@link MessageRecoverer} to use.
+	 * @since 5.5
+	 */
+	public void setMessageRecoverer(MessageRecoverer messageRecoverer) {
+		this.messageRecoverer = messageRecoverer;
+	}
+
 	/**
 	 * Set a batching strategy to use when de-batching messages.
 	 * Default is {@link SimpleBatchingStrategy}.
@@ -239,6 +253,7 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
 			Assert.state(getErrorChannel() == null, "Cannot have an 'errorChannel' property when a 'RetryTemplate' is "
 					+ "provided; use an 'ErrorMessageSendingRecoverer' in the 'recoveryCallback' property to "
 					+ "send an error message when retries are exhausted");
+			setupRecoveryCallbackIfAny();
 		}
 		Listener messageListener = new Listener();
 		this.messageListenerContainer.setMessageListener(messageListener);
@@ -254,6 +269,21 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
 		}
 	}
 
+	private void setupRecoveryCallbackIfAny() {
+		Assert.state(this.recoveryCallback == null || this.messageRecoverer == null,
+				"Only one of 'recoveryCallback' or 'messageRecoverer' may be provided, but not both");
+		if (this.messageRecoverer != null) {
+				this.recoveryCallback =
+						context -> {
+							Message messageToRecover =
+									(Message) RetrySynchronizationManager.getContext()
+											.getAttribute(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE);
+							this.messageRecoverer.recover(messageToRecover, context.getLastThrowable());
+							return null;
+						};
+		}
+	}
+
 	@Override
 	protected void doStart() {
 		super.doStart();

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2020 the original author or authors.
+ * Copyright 2013-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.
@@ -17,6 +17,8 @@
 package org.springframework.integration.amqp.inbound;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
+import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyBoolean;
 import static org.mockito.ArgumentMatchers.anyString;
@@ -30,10 +32,13 @@ import static org.mockito.Mockito.when;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 
-import org.junit.Test;
+import org.assertj.core.api.InstanceOfAssertFactories;
+import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
 
 import org.springframework.amqp.core.AcknowledgeMode;
@@ -47,6 +52,7 @@ import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
 import org.springframework.amqp.rabbit.listener.api.ChannelAwareBatchMessageListener;
 import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
+import org.springframework.amqp.rabbit.retry.MessageBatchRecoverer;
 import org.springframework.amqp.rabbit.support.ListenerExecutionFailedException;
 import org.springframework.amqp.support.AmqpHeaders;
 import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
@@ -362,6 +368,39 @@ public class InboundEndpointTests {
 		assertThat(errors.receive(0)).isNull();
 	}
 
+	@Test
+	public void testRetryWithMessageRecovererOnMessageAdapter() throws Exception {
+		ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
+		AbstractMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
+		AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(container);
+		adapter.setOutputChannel(new DirectChannel());
+		adapter.setRetryTemplate(new RetryTemplate());
+		AtomicReference<org.springframework.amqp.core.Message> recoveredMessage = new AtomicReference<>();
+		AtomicReference<Throwable> recoveredError = new AtomicReference<>();
+		CountDownLatch recoveredLatch = new CountDownLatch(1);
+		adapter.setMessageRecoverer((message, cause) -> {
+			recoveredMessage.set(message);
+			recoveredError.set(cause);
+			recoveredLatch.countDown();
+		});
+		adapter.afterPropertiesSet();
+		ChannelAwareMessageListener listener = (ChannelAwareMessageListener) container.getMessageListener();
+		org.springframework.amqp.core.Message amqpMessage =
+				org.springframework.amqp.core.MessageBuilder.withBody("foo".getBytes())
+						.andProperties(new MessageProperties())
+						.build();
+		listener.onMessage(amqpMessage, null);
+
+		assertThat(recoveredLatch.await(10, TimeUnit.SECONDS)).isTrue();
+
+		assertThat(recoveredError.get())
+				.isInstanceOf(MessagingException.class)
+				.extracting(Throwable::getMessage, InstanceOfAssertFactories.STRING)
+				.contains("Dispatcher has no");
+
+		assertThat(recoveredMessage.get()).isSameAs(amqpMessage);
+	}
+
 	@Test
 	public void testRetryWithinOnMessageGateway() throws Exception {
 		ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
@@ -390,6 +429,39 @@ public class InboundEndpointTests {
 		assertThat(errors.receive(0)).isNull();
 	}
 
+	@Test
+	public void testRetryWithMessageRecovererOnMessageGateway() throws Exception {
+		ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
+		AbstractMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
+		AmqpInboundGateway adapter = new AmqpInboundGateway(container);
+		adapter.setRequestChannel(new DirectChannel());
+		adapter.setRetryTemplate(new RetryTemplate());
+		AtomicReference<org.springframework.amqp.core.Message> recoveredMessage = new AtomicReference<>();
+		AtomicReference<Throwable> recoveredError = new AtomicReference<>();
+		CountDownLatch recoveredLatch = new CountDownLatch(1);
+		adapter.setMessageRecoverer((message, cause) -> {
+			recoveredMessage.set(message);
+			recoveredError.set(cause);
+			recoveredLatch.countDown();
+		});
+		adapter.afterPropertiesSet();
+		ChannelAwareMessageListener listener = (ChannelAwareMessageListener) container.getMessageListener();
+		org.springframework.amqp.core.Message amqpMessage =
+				org.springframework.amqp.core.MessageBuilder.withBody("foo".getBytes())
+						.andProperties(new MessageProperties())
+						.build();
+		listener.onMessage(amqpMessage, null);
+
+		assertThat(recoveredLatch.await(10, TimeUnit.SECONDS)).isTrue();
+
+		assertThat(recoveredError.get())
+				.isInstanceOf(MessagingException.class)
+				.extracting(Throwable::getMessage, InstanceOfAssertFactories.STRING)
+				.contains("Dispatcher has no");
+
+		assertThat(recoveredMessage.get()).isSameAs(amqpMessage);
+	}
+
 	@SuppressWarnings({ "unchecked" })
 	@Test
 	public void testBatchAdapter() throws Exception {
@@ -443,7 +515,7 @@ public class InboundEndpointTests {
 
 	@SuppressWarnings({ "unchecked" })
 	@Test
-	public void testConsumerBatchExtract() throws Exception {
+	public void testConsumerBatchExtract() {
 		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(mock(ConnectionFactory.class));
 		container.setConsumerBatchEnabled(true);
 		AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(container);
@@ -462,12 +534,12 @@ public class InboundEndpointTests {
 		assertThat(received).isNotNull();
 		assertThat(((List<String>) received.getPayload())).contains("test1", "test2");
 		assertThat(received.getHeaders().get(AmqpInboundChannelAdapter.CONSOLIDATED_HEADERS, List.class))
-			.hasSize(2);
+				.hasSize(2);
 	}
 
 	@SuppressWarnings({ "unchecked" })
 	@Test
-	public void testConsumerBatch() throws Exception {
+	public void testConsumerBatch() {
 		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(mock(ConnectionFactory.class));
 		container.setConsumerBatchEnabled(true);
 		AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(container);
@@ -484,12 +556,37 @@ public class InboundEndpointTests {
 		Message<?> received = out.receive(0);
 		assertThat(received).isNotNull();
 		assertThat(((List<Message<String>>) received.getPayload()))
-			.extracting(message -> message.getPayload())
-			.contains("test1", "test2");
+				.extracting(message -> message.getPayload())
+				.contains("test1", "test2");
 	}
 
 	@Test
-	public void testAdapterConversionErrorConsumerBatchExtract() throws Exception {
+	public void testConsumerBatchAndWrongMessageRecoverer() {
+		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(mock(ConnectionFactory.class));
+		container.setConsumerBatchEnabled(true);
+		AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(container);
+		adapter.setRetryTemplate(new RetryTemplate());
+		adapter.setMessageRecoverer((message, cause) -> { });
+		assertThatIllegalArgumentException()
+				.isThrownBy(adapter::afterPropertiesSet)
+				.withMessageStartingWith("The 'messageRecoverer' must be an instance of MessageBatchRecoverer " +
+						"when consumer configured for batch mode");
+	}
+
+	@Test
+	public void testExclusiveRecover() {
+		AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(mock(AbstractMessageListenerContainer.class));
+		adapter.setRetryTemplate(new RetryTemplate());
+		adapter.setMessageRecoverer((message, cause) -> { });
+		adapter.setRecoveryCallback(context -> null);
+		assertThatIllegalStateException()
+				.isThrownBy(adapter::afterPropertiesSet)
+				.withMessageStartingWith("Only one of 'recoveryCallback' or 'messageRecoverer' may be provided, " +
+						"but not both");
+	}
+
+	@Test
+	public void testAdapterConversionErrorConsumerBatchExtract() {
 		Connection connection = mock(Connection.class);
 		doAnswer(invocation -> mock(Channel.class)).when(connection).createChannel(anyBoolean());
 		ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
@@ -546,7 +643,7 @@ public class InboundEndpointTests {
 	}
 
 	@Test
-	public void testAdapterConversionErrorConsumerBatch() throws Exception {
+	public void testAdapterConversionErrorConsumerBatch() {
 		Connection connection = mock(Connection.class);
 		doAnswer(invocation -> mock(Channel.class)).when(connection).createChannel(anyBoolean());
 		ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
@@ -602,7 +699,7 @@ public class InboundEndpointTests {
 	}
 
 	@Test
-	public void testRetryWithinOnMessageAdapterConsumerBatch() throws Exception {
+	public void testRetryWithinOnMessageAdapterConsumerBatch() {
 		ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
 		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
 		container.setConsumerBatchEnabled(true);
@@ -641,12 +738,51 @@ public class InboundEndpointTests {
 		List<Message<?>> msgs = (List<Message<?>>) payload.getFailedMessage().getPayload();
 		assertThat(msgs).hasSize(2);
 		assertThat(msgs).extracting(msg -> StaticMessageHeaderAccessor.getDeliveryAttempt(msg).get())
-			.contains(3, 3);
+				.contains(3, 3);
 		assertThat(msgs).extracting(msg -> msg.getHeaders().get(AmqpHeaders.DELIVERY_TAG, Long.class))
-			.contains(42L, 43L);
+				.contains(42L, 43L);
 		assertThat(errors.receive(0)).isNull();
 	}
 
+	@Test
+	public void testRetryWithMessageRecovererOnMessageAdapterConsumerBatch() throws InterruptedException {
+		ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
+		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
+		container.setConsumerBatchEnabled(true);
+		AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(container);
+		adapter.setOutputChannel(new DirectChannel());
+		adapter.setRetryTemplate(new RetryTemplate());
+		AtomicReference<List<org.springframework.amqp.core.Message>> recoveredMessages = new AtomicReference<>();
+		AtomicReference<Throwable> recoveredError = new AtomicReference<>();
+		CountDownLatch recoveredLatch = new CountDownLatch(1);
+		adapter.setMessageRecoverer((MessageBatchRecoverer) (messages, cause) -> {
+			recoveredMessages.set(messages);
+			recoveredError.set(cause);
+			recoveredLatch.countDown();
+		});
+		adapter.afterPropertiesSet();
+		ChannelAwareBatchMessageListener listener = (ChannelAwareBatchMessageListener) container.getMessageListener();
+		MessageProperties messageProperties = new MessageProperties();
+		messageProperties.setContentType("text/plain");
+		messageProperties.setDeliveryTag(42L);
+		List<org.springframework.amqp.core.Message> messages = new ArrayList<>();
+		messages.add(new org.springframework.amqp.core.Message("test1".getBytes(), messageProperties));
+		messageProperties = new MessageProperties();
+		messageProperties.setContentType("text/plain");
+		messageProperties.setDeliveryTag(43L);
+		messages.add(new org.springframework.amqp.core.Message("test2".getBytes(), messageProperties));
+		listener.onMessageBatch(messages, null);
+
+		assertThat(recoveredLatch.await(10, TimeUnit.SECONDS)).isTrue();
+
+		assertThat(recoveredError.get())
+				.isInstanceOf(MessagingException.class)
+				.extracting(Throwable::getMessage, InstanceOfAssertFactories.STRING)
+				.contains("Dispatcher has no");
+
+		assertThat(recoveredMessages.get()).isSameAs(messages);
+	}
+
 	public static class Foo {
 
 		private String bar;
@@ -673,7 +809,7 @@ public class InboundEndpointTests {
 
 			Foo foo = (Foo) o;
 
-			return !(bar != null ? !bar.equals(foo.bar) : foo.bar != null);
+			return Objects.equals(bar, foo.bar);
 
 		}
 

src/reference/asciidoc/amqp.adoc

@@ -199,6 +199,9 @@ The JMS inbound channel adapter is using a `JmsDestinationPollingSource` under t
 The AMQP inbound channel adapter uses an `AbstractMessageListenerContainer` and is message driven.
 In that regard, it is more similar to the JMS message-driven channel adapter.
 
+Starting with version 5.5, the `AmqpInboundChannelAdapter` can be configured with an `org.springframework.amqp.rabbit.retry.MessageRecoverer` strategy which is used in the `RecoveryCallback` when the retry operation is called internally.
+See `setMessageRecoverer()` JavaDocs for more information.
+
 ==== Configuring with Java Configuration
 
 The following Spring Boot application shows an example of configuring the inbound adapter with Java configuration:
@@ -294,6 +297,8 @@ Starting with version 5.2, if the container's `deBatchingEnabled` property is se
 
 The default `BatchingStrategy` is the `SimpleBatchingStrategy`, but this can be overridden on the adapter.
 
+NOTE: The `org.springframework.amqp.rabbit.retry.MessageBatchRecoverer` must be used with batches when recovery is required for retry operations.
+
 === Polled Inbound Channel Adapter
 
 ==== Overview
@@ -395,6 +400,9 @@ if you anticipate cases when no `replyTo` property exists in the request message
 
 See the note in <<amqp-inbound-channel-adapter>> about configuring the `listener-container` attribute.
 
+Starting with version 5.5, the `AmqpInboundChannelAdapter` can be configured with an `org.springframework.amqp.rabbit.retry.MessageRecoverer` strategy which is used in the `RecoveryCallback` when the retry operation is called internally.
+See `setMessageRecoverer()` JavaDocs for more information.
+
 ==== Configuring with Java Configuration
 
 The following Spring Boot application shows an example of how to configure the inbound gateway with Java configuration:

src/reference/asciidoc/whats-new.adoc

@@ -17,3 +17,9 @@ If you are interested in more details, see the Issue Tracker tickets that were r
 
 [[x5.5-general]]
 === General Changes
+
+[[x5.5-amqp]]
+==== AMQP Changes
+
+The `AmqpInboundChannelAdapter` and `AmqpInboundGateway` (and the respective Java DSL builders) now support an `org.springframework.amqp.rabbit.retry.MessageRecoverer` as an AMQP-specific alternative to the general purpose `RecoveryCallback`.
+See <<./amqp.adoc#amqp,AMQP Support>> for more information.