Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-1293: Confirms/Returns with Republish Recoverer

Browse Source 
Resolves https://github.com/spring-projects/spring-amqp/issues/1293

Add a subclass that waits for confirms and checks for returns.

* Fix import.

* Fix javadocs.

* Fix copyright year.

* Fix Sonar issues.
This commit is contained in:
Gary Russell
2021-01-08 15:39:49 -05:00
committed by GitHub
parent d89f10debe
commit d5f81a62c3
12 changed files with 404 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CorrelationData.java
View File

@@ -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.
@@ -20,6 +20,7 @@ import java.util.UUID;
import org.springframework.amqp.core.Correlation;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import org.springframework.util.concurrent.SettableListenableFuture;
@@ -29,7 +30,7 @@ import org.springframework.util.concurrent.SettableListenableFuture;
* {@link org.springframework.amqp.rabbit.core.RabbitTemplate} methods that include one of
* these as a parameter; when the publisher confirm is received, the CorrelationData is
* returned with the ack/nack. When returns are also enabled, the
* {@link #setReturnedMessage(Message) returnedMessage} property will be populated when a
* {@link #setReturned(ReturnedMessage) returned} property will be populated when a
* message can't be delivered - the return always arrives before the confirmation. In this
* case the {@code #id} property must be set to a unique value. If no id is provided it
* will automatically set to a unique value.
@@ -44,7 +45,7 @@ public class CorrelationData implements Correlation {
private volatile String id;
private volatile Message returnedMessage;
private volatile ReturnedMessage returnedMessage;
/**
* Construct an instance with a null Id.
@@ -56,7 +57,7 @@ public class CorrelationData implements Correlation {
/**
* Construct an instance with the supplied id. Must be unique if returns are enabled
* to allow population of the {@link #setReturnedMessage(Message) returnedMessage}.
* to allow population of the {@link #setReturned(ReturnedMessage) returned} message.
* @param id the id.
*/
public CorrelationData(String id) {
@@ -98,21 +99,46 @@ public class CorrelationData implements Correlation {
* Return a returned message, if any; requires a unique
* {@link #CorrelationData(String) id}. Guaranteed to be populated before the future
* is set.
* @deprecated in favor of {@link #getReturned()}.
* @return the message or null.
* @since 2.1
*/
@Deprecated
@Nullable
public Message getReturnedMessage() {
return this.returnedMessage;
return this.returnedMessage.getMessage();
}
/**
* Set a returned message for this correlation data.
* @param returnedMessage the returned message.
* @deprecated in favor of {@link #setReturned(ReturnedMessage)}.
* @since 1.7.13
*/
@Deprecated
public void setReturnedMessage(Message returnedMessage) {
this.returnedMessage = returnedMessage;
this.returnedMessage = new ReturnedMessage(returnedMessage, 0, "not available", "not available",
"not available");
}
/**
* Get the returned message and metadata, if any. Guaranteed to be populated before
* the future is set.
* @return the {@link ReturnedMessage}.
* @since 2.3.3
*/
@Nullable
public ReturnedMessage getReturned() {
return this.returnedMessage;
}
/**
* Set the returned message and metadata.
* @param returned the {@link ReturnedMessage}.
* @since 2.3.3
*/
public void setReturned(ReturnedMessage returned) {
this.returnedMessage = returned;
}
@Override

7
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelImpl.java
View File

@@ -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.
@@ -40,6 +40,7 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData.Confirm;
import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter;
import org.springframework.amqp.rabbit.support.MessagePropertiesConverter;
@@ -1101,7 +1102,9 @@ public class PublisherCallbackChannelImpl
new Envelope(0L, false, returned.getExchange(), returned.getRoutingKey()),
StandardCharsets.UTF_8.name());
if (confirm.getCorrelationData() != null) {
confirm.getCorrelationData().setReturnedMessage(new Message(returned.getBody(), messageProperties)); // NOSONAR never null
confirm.getCorrelationData().setReturned(new ReturnedMessage(// NOSONAR never null
new Message(returned.getBody(), messageProperties), returned.getReplyCode(),
returned.getReplyText(), returned.getExchange(), returned.getRoutingKey()));
}
}
}

54
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/AmqpNackReceivedException.java Normal file
View File

@@ -0,0 +1,54 @@
/*
* Copyright 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.
* You may obtain a copy of the License at
*
* https://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.amqp.rabbit.core;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
/**
* An exception thrown when a negative acknowledgement received after publishing a
* message.
*
* @author Gary Russell
* @since 2.3.3
*
*/
public class AmqpNackReceivedException extends AmqpException {
private static final long serialVersionUID = 1L;
private final Message failedMessage;
/**
* Create an instance with the provided message and failed message.
* @param message the message.
* @param failedMessage the failed message.
*/
public AmqpNackReceivedException(String message, Message failedMessage) {
super(message);
this.failedMessage = failedMessage;
}
/**
* Return the failed message.
* @return the message.
*/
public Message getFailedMessage() {
return this.failedMessage;
}
}

23
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/retry/RepublishMessageRecoverer.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2019 the original author or authors.
* Copyright 2014-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,7 @@ import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.RabbitUtils;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
/**
@@ -177,7 +178,7 @@ public class RepublishMessageRecoverer implements MessageRecoverer {
if (null != this.errorExchangeName) {
String routingKey = this.errorRoutingKey != null ? this.errorRoutingKey
: this.prefixedOriginalRoutingKey(message);
this.errorTemplate.send(this.errorExchangeName, routingKey, message);
doSend(this.errorExchangeName, routingKey, message);
if (this.logger.isWarnEnabled()) {
this.logger.warn("Republishing failed message to exchange '" + this.errorExchangeName
+ "' with routing key " + routingKey);
@@ -185,7 +186,7 @@ public class RepublishMessageRecoverer implements MessageRecoverer {
}
else {
final String routingKey = this.prefixedOriginalRoutingKey(message);
this.errorTemplate.send(routingKey, message);
doSend(null, routingKey, message);
if (this.logger.isWarnEnabled()) {
this.logger.warn("Republishing failed message to the template's default exchange with routing key "
+ routingKey);
@@ -193,6 +194,22 @@ public class RepublishMessageRecoverer implements MessageRecoverer {
}
}
/**
* Send the message.
* @param exchange the exchange or null to use the template's default.
* @param routingKey the routing key.
* @param message the message.
* @since 2.3.3
*/
protected void doSend(@Nullable String exchange, String routingKey, Message message) {
if (exchange != null) {
this.errorTemplate.send(exchange, routingKey, message);
}
else {
this.errorTemplate.send(routingKey, message);
}
}
private String[] processStackTrace(Throwable cause, String exceptionMessage) {
String stackTraceAsString = getStackTraceAsString(cause);
if (this.maxStackTraceLength < 0) {

153
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/retry/RepublishMessageRecovererWithConfirms.java Normal file
View File

@@ -0,0 +1,153 @@
/*
* Copyright 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.
* You may obtain a copy of the License at
*
* https://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.amqp.rabbit.retry;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.springframework.amqp.core.AmqpMessageReturnedException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.ConfirmType;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.connection.CorrelationData.Confirm;
import org.springframework.amqp.rabbit.core.AmqpNackReceivedException;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator;
import org.springframework.lang.Nullable;
/**
* A {@link RepublishMessageRecoverer} supporting publisher confirms and returns.
*
* @author Gary Russell
* @since 2.3.3
*
*/
public class RepublishMessageRecovererWithConfirms extends RepublishMessageRecoverer {
private static final long DEFAULT_TIMEOUT = 10_000;
private final RabbitTemplate template;
private final ConfirmType confirmType;
private long confirmTimeout = DEFAULT_TIMEOUT;
/**
* Use the supplied template to publish the messsage with the provided confirm type.
* The template and its connection factory must be suitably configured to support the
* confirm type.
* @param errorTemplate the template.
* @param confirmType the confirmType.
*/
public RepublishMessageRecovererWithConfirms(RabbitTemplate errorTemplate, ConfirmType confirmType) {
this(errorTemplate, null, null, confirmType);
}
/**
* Use the supplied template to publish the messsage with the provided confirm type to
* the provided exchange with the default routing key. The template and its connection
* factory must be suitably configured to support the confirm type.
* @param errorTemplate the template.
* @param confirmType the confirmType.
* @param errorExchange the exchange.
*/
public RepublishMessageRecovererWithConfirms(RabbitTemplate errorTemplate, String errorExchange,
ConfirmType confirmType) {
this(errorTemplate, errorExchange, null, confirmType);
}
/**
* Use the supplied template to publish the messsage with the provided confirm type to
* the provided exchange with the provided routing key. The template and its
* connection factory must be suitably configured to support the confirm type.
* @param errorTemplate the template.
* @param confirmType the confirmType.
* @param errorExchange the exchange.
* @param errorRoutingKey the routing key.
*/
public RepublishMessageRecovererWithConfirms(RabbitTemplate errorTemplate, String errorExchange,
String errorRoutingKey, ConfirmType confirmType) {
super(errorTemplate, errorExchange, errorRoutingKey);
this.template = errorTemplate;
this.confirmType = confirmType;
}
/**
* Set the confirm timeout; default 10 seconds.
* @param confirmTimeout the timeout.
*/
public void setConfirmTimeout(long confirmTimeout) {
this.confirmTimeout = confirmTimeout;
}
@Override
protected void doSend(@Nullable
String exchange, String routingKey, Message message) {
if (ConfirmType.CORRELATED.equals(this.confirmType)) {
doSendCorrelated(exchange, routingKey, message);
}
else {
doSendSimple(exchange, routingKey, message);
}
}
private void doSendCorrelated(String exchange, String routingKey, Message message) {
CorrelationData cd = new CorrelationData();
if (exchange != null) {
this.template.send(exchange, routingKey, message, cd);
}
else {
this.template.send(routingKey, message, cd);
}
try {
Confirm confirm = cd.getFuture().get(this.confirmTimeout, TimeUnit.MILLISECONDS);
if (cd.getReturned() != null) {
throw new AmqpMessageReturnedException("Message returned", cd.getReturned());
}
if (!confirm.isAck()) {
throw new AmqpNackReceivedException("Negative acknowledgment received", message);
}
}
catch (InterruptedException ex) {
Thread.currentThread().interrupt();
throw RabbitExceptionTranslator.convertRabbitAccessException(ex);
}
catch (ExecutionException ex) {
throw RabbitExceptionTranslator.convertRabbitAccessException(ex.getCause()); // NOSONAR (stack trace)
}
catch (TimeoutException ex) {
throw RabbitExceptionTranslator.convertRabbitAccessException(ex);
}
}
private void doSendSimple(String exchange, String routingKey, Message message) {
this.template.invoke(sender -> {
if (exchange != null) {
sender.send(exchange, routingKey, message);
}
else {
sender.send(routingKey, message);
}
sender.waitForConfirmsOrDie(this.confirmTimeout);
return null;
});
}
}

4
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/MessagingTemplateConfirmsTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020 the original author or authors.
* Copyright 2020-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.
@@ -66,7 +66,7 @@ public class MessagingTemplateConfirmsTests {
rmt.send("messaging.confirms.unroutable",
new GenericMessage<>("foo", Collections.singletonMap(AmqpHeaders.PUBLISH_CONFIRM_CORRELATION, data)));
assertThat(data.getFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue();
assertThat(data.getReturnedMessage()).isNotNull();
assertThat(data.getReturned()).isNotNull();
ccf.destroy();
}

4
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java
View File

@@ -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.
@@ -855,7 +855,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests {
this.templateWithConfirmsAndReturnsEnabled.convertAndSend("", "NO_QUEUE_HERE", "foo", cd4);
assertThat(cd4.getFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue();
assertThat(callbackLatch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(cd4.getReturnedMessage()).isNotNull();
assertThat(cd4.getReturned()).isNotNull();
assertThat(resent.get()).isTrue();
assertThat(callbackThreadName.get()).startsWith("spring-rabbit-deferred-pool");
admin.deleteQueue(queue.getName());

4
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests2.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2020 the original author or authors.
* Copyright 2016-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.
@@ -125,7 +125,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests2 {
corr = new CorrelationData();
this.templateWithConfirmsEnabled.convertAndSend("", "bad route", "foo", corr);
assertThat(corr.getFuture().get(10, TimeUnit.SECONDS).isAck()).isTrue();
assertThat(corr.getReturnedMessage()).isNotNull();
assertThat(corr.getReturned()).isNotNull();
}
@Test

4
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests3.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2020 the original author or authors.
* Copyright 2018-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.
@@ -103,7 +103,7 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests3 {
assertThat(confirmLatch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(cacheCount.get()).isEqualTo(1);
assertThat(returnCalledFirst.get()).isTrue();
assertThat(correlationData.getReturnedMessage()).isNotNull();
assertThat(correlationData.getReturned()).isNotNull();
cf.destroy();
}

119
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/retry/RepublishMessageRecovererWithConfirmsIntegrationTests.java Normal file
View File

@@ -0,0 +1,119 @@
/*
* Copyright 2018-2019 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
*
* https://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.amqp.rabbit.retry;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.assertj.core.api.Assertions.fail;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.core.AmqpMessageReturnedException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.amqp.core.QueueBuilder.Overflow;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.ConfirmType;
import org.springframework.amqp.rabbit.core.AmqpNackReceivedException;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.junit.RabbitAvailable;
import org.springframework.amqp.rabbit.junit.RabbitAvailableCondition;
/**
* @author Gary Russell
* @since 2.0.5
*
*/
@RabbitAvailable(queues = RepublishMessageRecovererWithConfirmsIntegrationTests.QUEUE)
class RepublishMessageRecovererWithConfirmsIntegrationTests {
static final String QUEUE = "RepublishMessageRecovererWithConfirmsIntegrationTests";
@Test
void testSimple() {
CachingConnectionFactory ccf = new CachingConnectionFactory(
RabbitAvailableCondition.getBrokerRunning().getConnectionFactory());
ccf.setPublisherConfirmType(ConfirmType.SIMPLE);
RabbitTemplate template = new RabbitTemplate(ccf);
RepublishMessageRecovererWithConfirms recoverer = new RepublishMessageRecovererWithConfirms(template, "",
QUEUE, ConfirmType.SIMPLE);
recoverer.recover(MessageBuilder.withBody("foo".getBytes()).build(), new RuntimeException());
Message received = template.receive(QUEUE, 10_000);
assertThat(received.getBody()).isEqualTo("foo".getBytes());
ccf.destroy();
}
@Test
void testCorrelated() {
CachingConnectionFactory ccf = new CachingConnectionFactory(
RabbitAvailableCondition.getBrokerRunning().getConnectionFactory());
ccf.setPublisherConfirmType(ConfirmType.CORRELATED);
RabbitTemplate template = new RabbitTemplate(ccf);
RepublishMessageRecovererWithConfirms recoverer = new RepublishMessageRecovererWithConfirms(template, "",
QUEUE, ConfirmType.CORRELATED);
recoverer.recover(MessageBuilder.withBody("foo".getBytes()).build(), new RuntimeException());
Message received = template.receive(QUEUE, 10_000);
assertThat(received.getBody()).isEqualTo("foo".getBytes());
ccf.destroy();
}
@Test
void testCorrelatedNotRoutable() {
CachingConnectionFactory ccf = new CachingConnectionFactory(
RabbitAvailableCondition.getBrokerRunning().getConnectionFactory());
ccf.setPublisherConfirmType(ConfirmType.CORRELATED);
ccf.setPublisherReturns(true);
RabbitTemplate template = new RabbitTemplate(ccf);
template.setMandatory(true);
RepublishMessageRecovererWithConfirms recoverer = new RepublishMessageRecovererWithConfirms(template, "",
"bad.route", ConfirmType.CORRELATED);
try {
recoverer.recover(MessageBuilder.withBody("foo".getBytes()).build(), new RuntimeException());
fail("Expected exception");
}
catch (AmqpMessageReturnedException ex) {
assertThat(ex.getReturnedMessage().getBody()).isEqualTo("foo".getBytes());
assertThat(ex.getReplyText()).isEqualTo("NO_ROUTE");
}
ccf.destroy();
}
@Test
void testCorrelatedWithNack() {
CachingConnectionFactory ccf = new CachingConnectionFactory(
RabbitAvailableCondition.getBrokerRunning().getConnectionFactory());
ccf.setPublisherConfirmType(ConfirmType.CORRELATED);
RabbitTemplate template = new RabbitTemplate(ccf);
RabbitAdmin admin = new RabbitAdmin(ccf);
Queue queue = QueueBuilder.durable(QUEUE + ".nack")
.maxLength(1)
.overflow(Overflow.rejectPublish)
.build();
admin.declareQueue(queue);
RepublishMessageRecovererWithConfirms recoverer = new RepublishMessageRecovererWithConfirms(template, "",
queue.getName(), ConfirmType.CORRELATED);
recoverer.recover(MessageBuilder.withBody("foo".getBytes()).build(), new RuntimeException());
assertThatExceptionOfType(AmqpNackReceivedException.class).isThrownBy(() ->
recoverer.recover(MessageBuilder.withBody("foo".getBytes()).build(), new RuntimeException()));
admin.deleteQueue(queue.getName());
ccf.destroy();
}
}

8
src/reference/asciidoc/amqp.adoc
View File

@@ -6263,6 +6263,14 @@ Starting with versions 2.1.13, 2.2.3, the exception message is included in this
Whenever a truncation of any kind occurs, the original exception will be logged to retain the complete information.
Starting with version 2.3.3, a new subclass `RepublishMessageRecovererWithConfirms` is provided; this supports both styles of publisher confirms and will wait for the confirmation before returning (or throw an exception if not confirmed or the message is returned).
If the confirm type is `CORRELATED`, the subclass will also detect if a message is returned and throw an `AmqpMessageReturnedException`; if the publication is negatively acknowledged, it will throw an `AmqpNackReceivedException`.
If the confirm type is `SIMPLE`, the subclass will invoke the `waitForConfirmsOrDie` method on the channel.
See <<cf-pub-conf-ret>> for more information about confirms and returns.
Starting with version 2.1, an `ImmediateRequeueMessageRecoverer` is added to throw an `ImmediateRequeueAmqpException`, which notifies a listener container to requeue the current failed message.
===== Exception Classification for Spring Retry

5
src/reference/asciidoc/whats-new.adoc
View File

@@ -52,3 +52,8 @@ See the IMPORTANT note in <<post-processing>> for more information.
==== Multiple Broker Support Improvements
See <<multi-rabbit>> for more information.
==== RepublishMessageRecoverer Changes
A new subclass of this recoverer is not provided that supports publisher confirms.
See <<async-listeners>> for more information.