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

GH-1409: Fix Nacks for Async Replies

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

Normally, when message has a fatal exception (such as message conversion)
`basicNack` with `multiple` true is used, to nack any previously unacked
messages (e.g. when using batch size to limit the ack traffic).
Even when using manual acks, fatal exceptions are nacked by the container
because the user does not have access to the message.

However, when using async replies, this has the side effect of nacking
unprocessed messages.

Detect whether async replies are being used and only nack individual
records that cause fatal exceptions.

Also, coerce the `AcknowledgeMode` to `MANUAL` for such listners.

Add a test for both containers; send a good message followed by a
bad one without actually completing the reply future.
After the exception occurs and the container is stopped, there should
be one messag in the queue.

* Remove warning, deprecation; add docs.

* Docs.

**Cherry-pick to `2.3.x` & `2.2.x`**
This commit is contained in:
Gary Russell
2021-12-20 11:44:08 -05:00
committed by GitHub
parent 687b515a07
commit a4f014dc35
13 changed files with 328 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
spring-amqp/src/main/java/org/springframework/amqp/core/MessageListener.java
View File

@@ -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.
@@ -43,6 +43,16 @@ public interface MessageListener {
// NOSONAR - empty
}
/**
* Return true if this listener is request/reply and the replies are
* async.
* @return true for async replies.
* @since 2.2.21
*/
default boolean isAsyncReplies() {
return false;
}
/**
* Delivers a batch of messages.
* @param messages the messages.

10
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java
View File

@@ -255,6 +255,8 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
private volatile boolean lazyLoad;
private boolean asyncReplies;
@Override
public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
this.applicationEventPublisher = applicationEventPublisher;
@@ -439,6 +441,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
this.messageListener = messageListener;
this.isBatchListener = messageListener instanceof BatchMessageListener
|| messageListener instanceof ChannelAwareBatchMessageListener;
this.asyncReplies = messageListener.isAsyncReplies();
}
/**
@@ -1016,10 +1019,12 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
return this.possibleAuthenticationFailureFatal;
}
protected boolean isPossibleAuthenticationFailureFatalSet() {
return this.possibleAuthenticationFailureFatalSet;
}
protected boolean isAsyncReplies() {
return this.asyncReplies;
}
/**
* Set to true to automatically declare elements (queues, exchanges, bindings)
@@ -1220,6 +1225,9 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
catch (IllegalStateException e) {
this.logger.debug("Could not enable micrometer timers", e);
}
if (this.isAsyncReplies() && !AcknowledgeMode.MANUAL.equals(this.acknowledgeMode)) {
this.acknowledgeMode = AcknowledgeMode.MANUAL;
}
}
@Override

38
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java
View File

@@ -779,6 +779,17 @@ public class BlockingQueueConsumer {
* @param ex the thrown application exception or error
*/
public void rollbackOnExceptionIfNecessary(Throwable ex) {
rollbackOnExceptionIfNecessary(ex, -1);
}
/**
* Perform a rollback, handling rollback exceptions properly.
* @param ex the thrown application exception or error
* @param tag delivery tag; when specified (greater than or equal to 0) only that
* message is nacked.
* @since 2.2.21.
*/
public void rollbackOnExceptionIfNecessary(Throwable ex, long tag) {
boolean ackRequired = !this.acknowledgeMode.isAutoAck()
&& (!this.acknowledgeMode.isManual() || ContainerUtils.isRejectManual(ex));
@@ -790,14 +801,20 @@ public class BlockingQueueConsumer {
RabbitUtils.rollbackIfNecessary(this.channel);
}
if (ackRequired) {
OptionalLong deliveryTag = this.deliveryTags.stream().mapToLong(l -> l).max();
if (deliveryTag.isPresent()) {
this.channel.basicNack(deliveryTag.getAsLong(), true,
ContainerUtils.shouldRequeue(this.defaultRequeueRejected, ex, logger));
if (tag < 0) {
OptionalLong deliveryTag = this.deliveryTags.stream().mapToLong(l -> l).max();
if (deliveryTag.isPresent()) {
this.channel.basicNack(deliveryTag.getAsLong(), true,
ContainerUtils.shouldRequeue(this.defaultRequeueRejected, ex, logger));
}
if (this.transactional) {
// Need to commit the reject (=nack)
RabbitUtils.commitIfNecessary(this.channel);
}
}
if (this.transactional) {
// Need to commit the reject (=nack)
RabbitUtils.commitIfNecessary(this.channel);
else {
this.channel.basicNack(tag, false,
ContainerUtils.shouldRequeue(this.defaultRequeueRejected, ex, logger));
}
}
}
@@ -806,7 +823,12 @@ public class BlockingQueueConsumer {
throw RabbitExceptionTranslator.convertRabbitAccessException(e); // NOSONAR stack trace loss
}
finally {
this.deliveryTags.clear();
if (tag < 0) {
this.deliveryTags.clear();
}
else {
this.deliveryTags.remove(tag);
}
}
}

2
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java
View File

@@ -1214,7 +1214,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
}
}
}
getChannel().basicNack(deliveryTag, true,
getChannel().basicNack(deliveryTag, !isAsyncReplies(),
ContainerUtils.shouldRequeue(isDefaultRequeueRejected(), e, this.logger));
}
catch (IOException e1) {

7
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java
View File

@@ -982,6 +982,9 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
}
break;
}
long tagToRollback = isAsyncReplies()
? message.getMessageProperties().getDeliveryTag()
: -1;
if (getTransactionManager() != null) {
if (getTransactionAttribute().rollbackOn(ex)) {
RabbitResourceHolder resourceHolder = (RabbitResourceHolder) TransactionSynchronizationManager
@@ -994,7 +997,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
* If we don't actually have a transaction, we have to roll back
* manually. See prepareHolderForRollback().
*/
consumer.rollbackOnExceptionIfNecessary(ex);
consumer.rollbackOnExceptionIfNecessary(ex, tagToRollback);
}
throw ex; // encompassing transaction will handle the rollback.
}
@@ -1006,7 +1009,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
}
}
else {
consumer.rollbackOnExceptionIfNecessary(ex);
consumer.rollbackOnExceptionIfNecessary(ex, tagToRollback);
throw ex;
}
}

19
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java
View File

@@ -21,7 +21,6 @@ import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.lang.reflect.WildcardType;
import java.util.Arrays;
import java.util.function.Consumer;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -56,7 +55,6 @@ import org.springframework.util.ClassUtils;
import org.springframework.util.concurrent.ListenableFuture;
import com.rabbitmq.client.Channel;
import reactor.core.publisher.Mono;
/**
* An abstract {@link org.springframework.amqp.core.MessageListener} adapter providing the
@@ -81,7 +79,7 @@ public abstract class AbstractAdaptableMessageListener implements ChannelAwareMe
private static final ParserContext PARSER_CONTEXT = new TemplateParserContext("!{", "}");
private static final boolean monoPresent = // NOSONAR - lower case
static final boolean monoPresent = // NOSONAR - lower case, protected
ClassUtils.isPresent("reactor.core.publisher.Mono", ChannelAwareMessageListener.class.getClassLoader());
/**
@@ -695,19 +693,4 @@ public abstract class AbstractAdaptableMessageListener implements ChannelAwareMe
}
private static class MonoHandler { // NOSONAR - pointless to name it ..Utils|Helper
static boolean isMono(Object result) {
return result instanceof Mono;
}
@SuppressWarnings("unchecked")
static void subscribe(Object returnValue, Consumer<? super Object> success,
Consumer<? super Throwable> failure, Runnable completeConsumer) {
((Mono<? super Object>) returnValue).subscribe(success, failure, completeConsumer);
}
}
}

25
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/DelegatingInvocableHandler.java
View File

@@ -20,6 +20,7 @@ import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@@ -44,6 +45,7 @@ import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.messaging.handler.annotation.support.PayloadMethodArgumentResolver;
import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
import org.springframework.util.Assert;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.validation.Validator;
@@ -84,6 +86,8 @@ public class DelegatingInvocableHandler {
private final PayloadValidator validator;
private final boolean asyncReplies;
/**
* Construct an instance with the supplied handlers for the bean.
* @param handlers the handlers.
@@ -132,9 +136,19 @@ public class DelegatingInvocableHandler {
this.resolver = beanExpressionResolver;
this.beanExpressionContext = beanExpressionContext;
this.validator = validator == null ? null : new PayloadValidator(validator);
boolean asyncReplies;
asyncReplies = defaultHandler != null && isAsyncReply(defaultHandler);
Iterator<InvocableHandlerMethod> iterator = handlers.iterator();
while (iterator.hasNext()) {
asyncReplies |= isAsyncReply(iterator.next());
}
this.asyncReplies = asyncReplies;
}
private boolean isAsyncReply(InvocableHandlerMethod method) {
return (AbstractAdaptableMessageListener.monoPresent && MonoHandler.isMono(method.getMethod().getReturnType()))
|| ListenableFuture.class.isAssignableFrom(method.getMethod().getReturnType());
}
/**
* @return the bean
@@ -143,6 +157,15 @@ public class DelegatingInvocableHandler {
return this.bean;
}
/**
* Return true if any handler method has an async reply type.
* @return the asyncReply.
* @since 2.2.21
*/
public boolean isAsyncReplies() {
return this.asyncReplies;
}
/**
* Invoke the method with the given message.
* @param message the message.

16
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/HandlerAdapter.java
View File

@@ -22,6 +22,7 @@ import java.lang.reflect.Type;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
import org.springframework.util.concurrent.ListenableFuture;
/**
* A wrapper for either an {@link InvocableHandlerMethod} or
@@ -38,6 +39,8 @@ public class HandlerAdapter {
private final DelegatingInvocableHandler delegatingHandler;
private final boolean asyncReplies;
/**
* Construct an instance with the provided method.
* @param invokerHandlerMethod the method.
@@ -45,6 +48,9 @@ public class HandlerAdapter {
public HandlerAdapter(InvocableHandlerMethod invokerHandlerMethod) {
this.invokerHandlerMethod = invokerHandlerMethod;
this.delegatingHandler = null;
this.asyncReplies = (AbstractAdaptableMessageListener.monoPresent
&& MonoHandler.isMono(invokerHandlerMethod.getMethod().getReturnType()))
|| ListenableFuture.class.isAssignableFrom(invokerHandlerMethod.getMethod().getReturnType());
}
/**
@@ -54,6 +60,7 @@ public class HandlerAdapter {
public HandlerAdapter(DelegatingInvocableHandler delegatingHandler) {
this.invokerHandlerMethod = null;
this.delegatingHandler = delegatingHandler;
this.asyncReplies = delegatingHandler.isAsyncReplies();
}
/**
@@ -139,6 +146,15 @@ public class HandlerAdapter {
}
}
/**
* Return true if any handler method has an async reply type.
* @return the asyncReply.
* @since 2.2.21
*/
public boolean isAsyncReplies() {
return this.asyncReplies;
}
/**
* Build an {@link InvocationResult} for the result and inbound payload.
* @param result the result.

5
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java
View File

@@ -107,6 +107,11 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis
return this.handlerAdapter;
}
@Override
public boolean isAsyncReplies() {
return this.handlerAdapter.isAsyncReplies();
}
/**
* Set the {@link AmqpHeaderMapper} implementation to use to map the standard
* AMQP headers. By default, a {@link org.springframework.amqp.support.SimpleAmqpHeaderMapper

44
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MonoHandler.java Normal file
View File

@@ -0,0 +1,44 @@
/*
* 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.listener.adapter;
import java.util.function.Consumer;
import reactor.core.publisher.Mono;
/**
* Class to prevent direct links to {@link Mono}.
* @author Gary Russell
* @since 2.2.21
*/
final class MonoHandler { // NOSONAR - pointless to name it ..Utils|Helper
private MonoHandler() {
}
static boolean isMono(Object result) {
return result instanceof Mono;
}
@SuppressWarnings("unchecked")
static void subscribe(Object returnValue, Consumer<? super Object> success,
Consumer<? super Throwable> failure, Runnable completeConsumer) {
((Mono<? super Object>) returnValue).subscribe(success, failure, completeConsumer);
}
}

178
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/AsyncReplyToTests.java Normal file
View File

@@ -0,0 +1,178 @@
/*
* 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.listener;
import static org.assertj.core.api.Assertions.assertThat;
import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessagePropertiesBuilder;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.config.DirectRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
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;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.SettableListenableFuture;
import com.rabbitmq.client.Channel;
/**
* @author Gary Russell
* @since 2.2.21
*
*/
@SpringJUnitConfig
@RabbitAvailable(queues = { "async1", "async2" })
public class AsyncReplyToTests {
@Test
void ackSingleWhenFatalSMLC(@Autowired Config config, @Autowired RabbitListenerEndpointRegistry registry,
@Autowired RabbitTemplate template, @Autowired RabbitAdmin admin) throws IOException, InterruptedException {
template.send("async1", MessageBuilder.withBody("\"foo\"".getBytes()).andProperties(
MessagePropertiesBuilder.newInstance()
.setContentType("application/json")
.setReplyTo("nowhere")
.build())
.build());
template.send("async1", MessageBuilder.withBody("junk".getBytes()).andProperties(
MessagePropertiesBuilder.newInstance()
.setContentType("application/json")
.setReplyTo("nowhere")
.build())
.build());
assertThat(config.smlcLatch.await(10, TimeUnit.SECONDS)).isTrue();
registry.getListenerContainer("smlc").stop();
assertThat(admin.getQueueInfo("async1").getMessageCount()).isEqualTo(1);
}
@Test
void ackSingleWhenFatalDMLC(@Autowired Config config, @Autowired RabbitListenerEndpointRegistry registry,
@Autowired RabbitTemplate template, @Autowired RabbitAdmin admin) throws IOException, InterruptedException {
template.send("async2", MessageBuilder.withBody("\"foo\"".getBytes()).andProperties(
MessagePropertiesBuilder.newInstance()
.setContentType("application/json")
.setReplyTo("nowhere")
.build())
.build());
template.send("async2", MessageBuilder.withBody("junk".getBytes()).andProperties(
MessagePropertiesBuilder.newInstance()
.setContentType("application/json")
.setReplyTo("nowhere")
.build())
.build());
assertThat(config.dmlcLatch.await(10, TimeUnit.SECONDS)).isTrue();
registry.getListenerContainer("dmlc").stop();
assertThat(admin.getQueueInfo("async2").getMessageCount()).isEqualTo(1);
}
@Configuration
@EnableRabbit
static class Config {
volatile CountDownLatch smlcLatch = new CountDownLatch(1);
volatile CountDownLatch dmlcLatch = new CountDownLatch(1);
@RabbitListener(id = "smlc", queues = "async1", containerFactory = "smlcf")
ListenableFuture<String> listen1(String in, Channel channel) {
return new SettableListenableFuture<>();
}
@RabbitListener(id = "dmlc", queues = "async2", containerFactory = "dmlcf")
ListenableFuture<String> listen2(String in, Channel channel) {
return new SettableListenableFuture<>();
}
@Bean
MessageConverter converter() {
return new Jackson2JsonMessageConverter();
}
@Bean
ConnectionFactory cf() throws IOException, TimeoutException {
return new CachingConnectionFactory(RabbitAvailableCondition.getBrokerRunning().getConnectionFactory());
}
@Bean
SimpleRabbitListenerContainerFactory smlcf(ConnectionFactory cf, MessageConverter converter) {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setConnectionFactory(cf);
factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
factory.setMessageConverter(converter);
factory.setErrorHandler(new ConditionalRejectingErrorHandler() {
@Override
public void handleError(Throwable t) {
smlcLatch.countDown();
super.handleError(t);
}
});
return factory;
}
@Bean
DirectRabbitListenerContainerFactory dmlcf(ConnectionFactory cf, MessageConverter converter) {
DirectRabbitListenerContainerFactory factory = new DirectRabbitListenerContainerFactory();
factory.setConnectionFactory(cf);
factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
factory.setMessageConverter(converter);
factory.setErrorHandler(new ConditionalRejectingErrorHandler() {
@Override
public void handleError(Throwable t) {
dmlcLatch.countDown();
super.handleError(t);
}
});
return factory;
}
@Bean
RabbitTemplate template(ConnectionFactory cf) {
return new RabbitTemplate(cf);
}
@Bean
RabbitAdmin admin(ConnectionFactory cf) {
return new RabbitAdmin(cf);
}
}
}

1
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerTests.java
View File

@@ -727,6 +727,7 @@ public class SimpleMessageListenerContainerTests {
verify(channel).basicAck(2, true);
container.stop();
verify(listener).containerAckMode(AcknowledgeMode.AUTO);
verify(listener).isAsyncReplies();
verifyNoMoreInteractions(listener);
}

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

@@ -3535,6 +3535,9 @@ If the async result is completed with an `AmqpRejectAndDontRequeueException`, th
If the container's `defaultRequeueRejected` property is `false`, you can override that by setting the future's exception to a `ImmediateRequeueException` and the message will be requeued.
If some exception occurs within the listener method that prevents creation of the async result object, you MUST catch that exception and return an appropriate return object that will cause the message to be acknowledged or requeued.
Starting with versions 2.2.21, 2.3.13, 2.4.1, the `AcknowledgeMode` will be automatically set the `MANUAL` when async return types are detected.
In addition, incoming messages with fatal exceptions will be negatively acknowledged individually, previously any prior unacknowledged message were also negatively acknowledged.
[[threading]]
===== Threading and Asynchronous Consumers