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

GH-2941: Fix BlockingQueueConsumer race condition on cancel

Browse Source 
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
This commit is contained in:
Artem Bilan
2025-02-07 15:52:36 -05:00
parent bc81ebcd11
commit 4deb40c0fa
2 changed files with 42 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -170,7 +170,8 @@ public class BlockingQueueConsumer {
private long consumeDelay;
private java.util.function.Consumer<String> missingQueuePublisher = str -> { };
private java.util.function.Consumer<String> 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<String> 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) {

1
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitReturnTypesTests.java
View File

@@ -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;
}