From c003a473799a87fb910541e78c3fb6f46eaa1515 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 19 Jan 2021 17:22:05 -0500 Subject: [PATCH] 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 Co-authored-by: Gary Russell --- .../AmqpBaseInboundChannelAdapterSpec.java | 13 ++ .../amqp/dsl/AmqpBaseInboundGatewaySpec.java | 14 ++ .../inbound/AmqpInboundChannelAdapter.java | 50 +++++- .../amqp/inbound/AmqpInboundGateway.java | 34 +++- .../amqp/inbound/InboundEndpointTests.java | 162 ++++++++++++++++-- src/reference/asciidoc/amqp.adoc | 8 + src/reference/asciidoc/whats-new.adoc | 6 + 7 files changed, 271 insertions(+), 16 deletions(-) diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpBaseInboundChannelAdapterSpec.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpBaseInboundChannelAdapterSpec.java index b9f3bdd028..1bd612546f 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpBaseInboundChannelAdapterSpec.java +++ b/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> 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(); + } + } 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 d6952fe1a4..00f41ff7e7 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 @@ -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 messagesToRecover = + (List) 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(); diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java index ae8b6091cc..2a34203410 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundGateway.java +++ b/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 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(); 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 de02836854..7ea904edac 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 @@ -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 recoveredMessage = new AtomicReference<>(); + AtomicReference 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 recoveredMessage = new AtomicReference<>(); + AtomicReference 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) 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>) 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> msgs = (List>) 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> recoveredMessages = new AtomicReference<>(); + AtomicReference 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 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); } diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index 61acebf530..adcb56a451 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/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 <> 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: diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 56bca13378..7ccd2896ad 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/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.