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

GH-3032: Fix BlockingQueueConsumer for in-flight draining

Browse Source 
Fixes: #3032
Issue link: https://github.com/spring-projects/spring-amqp/issues/3032

The fix for https://github.com/spring-projects/spring-amqp/issues/2941 has missed
"in-flight draining" for non-transactional consumers.

(cherry picked from commit 993e94a5fc)
This commit is contained in:
Artem Bilan
2025-03-28 11:27:11 -04:00
committed by Spring Builds
parent f2db09685f
commit 8bc8f07aa7
1 changed files with 22 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -535,10 +535,7 @@ public class BlockingQueueConsumer {
*/
@Nullable
public Message nextMessage() throws InterruptedException, ShutdownSignalException {
if (logger.isTraceEnabled()) {
logger.trace("Retrieving delivery for " + this);
}
return handle(this.queue.take());
return nextMessage(-1);
}
/**
@@ -557,26 +554,35 @@ public class BlockingQueueConsumer {
if (!this.missingQueues.isEmpty()) {
checkMissingQueues();
}
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;
if (this.transactional && cancelled()) {
throw consumerCancelledException(null);
}
else {
Message message = handle(timeout < 0 ? this.queue.take() : this.queue.poll(timeout, TimeUnit.MILLISECONDS));
if (cancelled() && (message == null || this.transactional)) {
Long deliveryTagToNack = null;
if (message != null) {
deliveryTagToNack = message.getMessageProperties().getDeliveryTag();
}
throw consumerCancelledException(deliveryTagToNack);
}
else {
return message;
}
}
}
private ConsumerCancelledException consumerCancelledException(@Nullable Long deliveryTagToNack) {
this.activeObjectCounter.release(this);
ConsumerCancelledException consumerCancelledException = new ConsumerCancelledException();
if (deliveryTagToNack != null) {
rollbackOnExceptionIfNecessary(consumerCancelledException, deliveryTagToNack);
}
else {
this.deliveryTags.clear();
this.activeObjectCounter.release(this);
ConsumerCancelledException consumerCancelledException = new ConsumerCancelledException();
rollbackOnExceptionIfNecessary(consumerCancelledException);
throw consumerCancelledException;
}
return consumerCancelledException;
}
/*