From 4deb40c0facb6fd11b4e068e3655314689616993 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 7 Feb 2025 15:52:36 -0500 Subject: [PATCH] GH-2941: Fix BlockingQueueConsumer race condition on cancel Fixes: #2941 Issue link: https://github.com/spring-projects/spring-amqp/issues/2941 Now `BlockingQueueConsumer.basicCancel()` performs `RabbitUtils.closeMessageConsumer()` to initiate `basicRecovery` on the transactional consumer to re-queue all the un-acked messages. However, there is a race condition when one in-flight message may still be delivered to the listener and then TX commit is initiated. There a `basicAck()` is initiated. However, such a tag might already be discarded because of the previous `basicRecovery`. Therefore, adjust `BlockingQueueConsumer.commitIfNecessary()` to skip `basicAck()` if locally transacted and already cancelled. Right, this may lead to the duplication delivery, but having abnormal shutdown situation we cannot guarantee that this message to commit has been processed properly. Also, adjust `BlockingQueueConsumer.nextMessage()` to rollback a message if consumer is canceled instead of going through the loop via listener * Increase `replyTimeout` in the `EnableRabbitReturnTypesTests` for resource-sensitive builds --- .../listener/BlockingQueueConsumer.java | 63 ++++++++++++------- .../EnableRabbitReturnTypesTests.java | 1 + 2 files changed, 42 insertions(+), 22 deletions(-) diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java index 88b204cc..3f3d7e53 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java @@ -170,7 +170,8 @@ public class BlockingQueueConsumer { private long consumeDelay; - private java.util.function.Consumer missingQueuePublisher = str -> { }; + private java.util.function.Consumer missingQueuePublisher = str -> { + }; private boolean globalQos; @@ -468,12 +469,13 @@ public class BlockingQueueConsumer { protected void basicCancel(boolean expected) { this.normalCancel = expected; + this.cancelled.set(true); + this.abortStarted = System.currentTimeMillis(); + Collection consumerTags = getConsumerTags(); if (!CollectionUtils.isEmpty(consumerTags)) { RabbitUtils.closeMessageConsumer(this.channel, consumerTags, this.transactional); } - this.cancelled.set(true); - this.abortStarted = System.currentTimeMillis(); } protected boolean hasDelivery() { @@ -560,12 +562,26 @@ public class BlockingQueueConsumer { if (!this.missingQueues.isEmpty()) { checkMissingQueues(); } - Message message = handle(this.queue.poll(timeout, TimeUnit.MILLISECONDS)); - if (message == null && this.cancelled.get()) { - this.activeObjectCounter.release(this); - throw new ConsumerCancelledException(); + if (!cancelled()) { + Message message = handle(this.queue.poll(timeout, TimeUnit.MILLISECONDS)); + if (message != null && cancelled()) { + this.activeObjectCounter.release(this); + ConsumerCancelledException consumerCancelledException = new ConsumerCancelledException(); + rollbackOnExceptionIfNecessary(consumerCancelledException, + message.getMessageProperties().getDeliveryTag()); + throw consumerCancelledException; + } + else { + return message; + } + } + else { + this.deliveryTags.clear(); + this.activeObjectCounter.release(this); + ConsumerCancelledException consumerCancelledException = new ConsumerCancelledException(); + rollbackOnExceptionIfNecessary(consumerCancelledException); + throw consumerCancelledException; } - return message; } /* @@ -792,7 +808,7 @@ public class BlockingQueueConsumer { if (this.abortStarted == 0) { // signal handle delivery to use offer this.abortStarted = System.currentTimeMillis(); } - if (!this.cancelled()) { + if (!cancelled()) { try { RabbitUtils.closeMessageConsumer(this.channel, getConsumerTags(), this.transactional); } @@ -894,22 +910,25 @@ public class BlockingQueueConsumer { /* * If we have a TX Manager, but no TX, act like we are locally transacted. */ - boolean isLocallyTransacted = localTx - || (this.transactional - && TransactionSynchronizationManager.getResource(this.connectionFactory) == null); + boolean isLocallyTransacted = + localTx || + (this.transactional && + TransactionSynchronizationManager.getResource(this.connectionFactory) == null); try { boolean ackRequired = forceAck || (!this.acknowledgeMode.isAutoAck() && !this.acknowledgeMode.isManual()); - if (ackRequired && (!this.transactional || isLocallyTransacted)) { - long deliveryTag = new ArrayList<>(this.deliveryTags).get(this.deliveryTags.size() - 1); - try { - this.channel.basicAck(deliveryTag, true); - notifyMessageAckListener(true, deliveryTag, null); - } - catch (Exception e) { - logger.error("Error acking.", e); - notifyMessageAckListener(false, deliveryTag, e); - } + if (ackRequired && (!this.transactional || (isLocallyTransacted && !cancelled()))) { + OptionalLong deliveryTag = this.deliveryTags.stream().mapToLong(l -> l).max(); + deliveryTag.ifPresent((tag) -> { + try { + this.channel.basicAck(tag, true); + notifyMessageAckListener(true, tag, null); + } + catch (Exception e) { + logger.error("Error acking.", e); + notifyMessageAckListener(false, tag, e); + } + }); } if (isLocallyTransacted) { diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitReturnTypesTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitReturnTypesTests.java index 8783d663..5cd07d93 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitReturnTypesTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitReturnTypesTests.java @@ -113,6 +113,7 @@ public class EnableRabbitReturnTypesTests { public RabbitTemplate template(CachingConnectionFactory cf, Jackson2JsonMessageConverter converter) { RabbitTemplate template = new RabbitTemplate(cf); template.setMessageConverter(converter); + template.setReplyTimeout(30_000); return template; }