From 498d4f98ac6d66b2bfa2997ae795c40666265f1b Mon Sep 17 00:00:00 2001 From: Marco Cozzi Date: Thu, 12 Jun 2025 15:53:42 +0200 Subject: [PATCH] GH-3079: Allow ACKs within the ShutdownTimeout period Fixes: #3079 Issue link: https://github.com/spring-projects/spring-amqp/issues/3079 Remove `cancelled()` condition from the `BlockingQueueConsumer.commitIfNecessary()` to allow initiated acks to be committed. The cancelled consumer means no new deliveries, but that does not mean we cannot ack outstanding messages. Actually, this is a leftover after previous fix with assumption that messages have to be returned to the queue with `Recovery` command when consumer is cancelled. Signed-off-by: Marco Cozzi (cherry picked from commit c3cc65dad9b752504f5c9de2e631f98bf0594c42) --- .../amqp/rabbit/listener/BlockingQueueConsumer.java | 4 ++-- 1 file changed, 2 insertions(+), 2 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 820bd8e4..4dc345cb 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 @@ -883,7 +883,7 @@ public class BlockingQueueConsumer { /** * Perform a commit or message acknowledgement, as appropriate. - * NOTE: This method was never been intended tobe public. + * NOTE: This method was never intended to be public. * @param localTx Whether the channel is locally transacted. * @param forceAck perform {@link Channel#basicAck(long, boolean)} independently of {@link #acknowledgeMode}. * @return true if at least one delivery tag exists. @@ -904,7 +904,7 @@ public class BlockingQueueConsumer { try { boolean ackRequired = forceAck || (!this.acknowledgeMode.isAutoAck() && !this.acknowledgeMode.isManual()); - if (ackRequired && (!this.transactional || (isLocallyTransacted && !cancelled()))) { + if (ackRequired && (!this.transactional || isLocallyTransacted)) { OptionalLong deliveryTag = this.deliveryTags.stream().mapToLong(l -> l).max(); deliveryTag.ifPresent((tag) -> { try {