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

GH-2607: Add SMLC.enforceImmediateAckForManual option

Browse Source 
Fixes: #2607

There are use-cases when `ImmediateAcknowledgeAmqpException` can be thrown outside
the listener method, therefore there is no way to reach `Channel.basicAck()`.
For example, for `AbstractMessageListenerContainer.afterReceivePostProcessors`

* Make force ack for `ImmediateAcknowledgeAmqpException` even if `AcknowledgeMode.MANUAL`.
This is controlled with newly introduced `enforceImmediateAckForManual` flag on the `SimpleMessageListenerContainer`.
Such an option might be as tentative solution to not break behavior for existing applications using the current point release.
We may consider to make this unconditional in future versions
This commit is contained in:
Artem Bilan
2024-02-02 13:50:59 -05:00
parent 5c3e739c9c
commit c8c06eacc7
4 changed files with 108 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/SimpleRabbitListenerContainerFactory.java
View File

@@ -16,6 +16,9 @@
package org.springframework.amqp.rabbit.config;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.ImmediateAcknowledgeAmqpException;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.utils.JavaUtils;
@@ -59,6 +62,8 @@ public class SimpleRabbitListenerContainerFactory
private Boolean consumerBatchEnabled;
private Boolean enforceImmediateAckForManual;
/**
* @param batchSize the batch size.
* @since 2.2
@@ -153,6 +158,17 @@ public class SimpleRabbitListenerContainerFactory
}
}
/**
* Set to {@code true} to enforce {@link Channel#basicAck(long, boolean)}
* for {@link org.springframework.amqp.core.AcknowledgeMode#MANUAL}
* when {@link ImmediateAcknowledgeAmqpException} is thrown.
* This might be a tentative solution to not break behavior for current minor version.
* @param enforceImmediateAckForManual the flag to ack message for MANUAL mode on ImmediateAcknowledgeAmqpException
* @since 3.1.2
*/
public void setEnforceImmediateAckForManual(Boolean enforceImmediateAckForManual) {
this.enforceImmediateAckForManual = enforceImmediateAckForManual;
}
@Override
protected SimpleMessageListenerContainer createContainerInstance() {
return new SimpleMessageListenerContainer();
@@ -180,7 +196,8 @@ public class SimpleRabbitListenerContainerFactory
.acceptIfNotNull(this.consecutiveActiveTrigger, instance::setConsecutiveActiveTrigger)
.acceptIfNotNull(this.consecutiveIdleTrigger, instance::setConsecutiveIdleTrigger)
.acceptIfNotNull(this.receiveTimeout, instance::setReceiveTimeout)
.acceptIfNotNull(this.batchReceiveTimeout, instance::setBatchReceiveTimeout);
.acceptIfNotNull(this.batchReceiveTimeout, instance::setBatchReceiveTimeout)
.acceptIfNotNull(this.enforceImmediateAckForManual, instance::setEnforceImmediateAckForManual);
if (Boolean.TRUE.equals(this.consumerBatchEnabled)) {
instance.setConsumerBatchEnabled(true);
/*

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

@@ -875,10 +875,25 @@ public class BlockingQueueConsumer {
/**
* Perform a commit or message acknowledgement, as appropriate.
* NOTE: This method was never been intended tobe public.
* @param localTx Whether the channel is locally transacted.
* @return true if at least one delivery tag exists.
* @deprecated in favor of {@link #commitIfNecessary(boolean, boolean)}
*/
@Deprecated(forRemoval = true, since = "3.1.2")
public boolean commitIfNecessary(boolean localTx) {
return commitIfNecessary(localTx, false);
}
/**
* Perform a commit or message acknowledgement, as appropriate.
* NOTE: This method was never been intended tobe 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.
* @since 3.1.2
*/
boolean commitIfNecessary(boolean localTx, boolean forceAck) {
if (this.deliveryTags.isEmpty()) {
return false;
}
@@ -890,11 +905,10 @@ public class BlockingQueueConsumer {
|| (this.transactional
&& TransactionSynchronizationManager.getResource(this.connectionFactory) == null);
try {
boolean ackRequired = !this.acknowledgeMode.isAutoAck() && !this.acknowledgeMode.isManual();
boolean ackRequired = forceAck || (!this.acknowledgeMode.isAutoAck() && !this.acknowledgeMode.isManual());
if (ackRequired && (!this.transactional || isLocallyTransacted)) {
long deliveryTag = new ArrayList<Long>(this.deliveryTags).get(this.deliveryTags.size() - 1);
long deliveryTag = new ArrayList<>(this.deliveryTags).get(this.deliveryTags.size() - 1);
try {
this.channel.basicAck(deliveryTag, true);
notifyMessageAckListener(true, deliveryTag, null);
@@ -909,14 +923,12 @@ public class BlockingQueueConsumer {
// For manual acks we still need to commit
RabbitUtils.commitIfNecessary(this.channel);
}
}
finally {
this.deliveryTags.clear();
}
return true;
}
/**
@@ -931,7 +943,7 @@ public class BlockingQueueConsumer {
this.messageAckListener.onComplete(success, deliveryTag, cause);
}
catch (Exception e) {
logger.error("An exception occured in MessageAckListener.", e);
logger.error("An exception occurred in MessageAckListener.", e);
}
}

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

@@ -41,6 +41,7 @@ import org.springframework.amqp.core.BatchMessageListener;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.batch.BatchingStrategy;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils;
import org.springframework.amqp.rabbit.connection.ConsumerChannelRegistry;
@@ -134,6 +135,8 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
private long consumerStartTimeout = DEFAULT_CONSUMER_START_TIMEOUT;
private boolean enforceImmediateAckForManual;
private volatile int concurrentConsumers = 1;
private volatile Integer maxConcurrentConsumers;
@@ -504,6 +507,18 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
this.consumerStartTimeout = consumerStartTimeout;
}
/**
* Set to {@code true} to enforce {@link Channel#basicAck(long, boolean)}
* for {@link org.springframework.amqp.core.AcknowledgeMode#MANUAL}
* when {@link ImmediateAcknowledgeAmqpException} is thrown.
* This might be a tentative solution to not break behavior for current minor version.
* @param enforceImmediateAckForManual the flag to ack message for MANUAL mode on ImmediateAcknowledgeAmqpException
* @since 3.1.2
*/
public void setEnforceImmediateAckForManual(boolean enforceImmediateAckForManual) {
this.enforceImmediateAckForManual = enforceImmediateAckForManual;
}
/**
* Avoid the possibility of not configuring the CachingConnectionFactory in sync with the number of concurrent
* consumers.
@@ -1012,6 +1027,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
List<Message> messages = null;
long deliveryTag = 0;
boolean immediateAck = false;
boolean isBatchReceiveTimeoutEnabled = this.batchReceiveTimeout > 0;
long startTime = isBatchReceiveTimeoutEnabled ? System.currentTimeMillis() : 0;
for (int i = 0; i < this.batchSize; i++) {
@@ -1050,9 +1066,9 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
if (messages == null) {
messages = new ArrayList<>(this.batchSize);
}
if (isDeBatchingEnabled() && getBatchingStrategy().canDebatch(message.getMessageProperties())) {
final List<Message> messageList = messages;
getBatchingStrategy().deBatch(message, fragment -> messageList.add(fragment));
BatchingStrategy batchingStrategy = getBatchingStrategy();
if (isDeBatchingEnabled() && batchingStrategy.canDebatch(message.getMessageProperties())) {
batchingStrategy.deBatch(message, messages::add);
}
else {
messages.add(message);
@@ -1073,6 +1089,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
+ e.getMessage() + "': "
+ message.getMessageProperties().getDeliveryTag());
}
immediateAck = this.enforceImmediateAckForManual;
break;
}
catch (Exception ex) {
@@ -1081,6 +1098,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
this.logger.debug("User requested ack for failed delivery: "
+ message.getMessageProperties().getDeliveryTag());
}
immediateAck = this.enforceImmediateAckForManual;
break;
}
long tagToRollback = isAsyncReplies()
@@ -1117,14 +1135,14 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
}
}
if (messages != null) {
executeWithList(channel, messages, deliveryTag, consumer);
immediateAck = executeWithList(channel, messages, deliveryTag, consumer);
}
return consumer.commitIfNecessary(isChannelLocallyTransacted());
return consumer.commitIfNecessary(isChannelLocallyTransacted(), immediateAck);
}
private void executeWithList(Channel channel, List<Message> messages, long deliveryTag,
private boolean executeWithList(Channel channel, List<Message> messages, long deliveryTag,
BlockingQueueConsumer consumer) {
try {
@@ -1136,7 +1154,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
+ e.getMessage() + "' (last in batch): "
+ deliveryTag);
}
return;
return this.enforceImmediateAckForManual;
}
catch (Exception ex) {
if (causeChainHasImmediateAcknowledgeAmqpException(ex)) {
@@ -1144,7 +1162,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
this.logger.debug("User requested ack for failed delivery (last in batch): "
+ deliveryTag);
}
return;
return this.enforceImmediateAckForManual;
}
if (getTransactionManager() != null) {
if (getTransactionAttribute().rollbackOn(ex)) {
@@ -1173,6 +1191,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
throw ex;
}
}
return false;
}
protected void handleStartupFailure(BackOffExecution backOffExecution) {

47
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerManualAckIntegrationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2024 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.
@@ -17,6 +17,7 @@
package org.springframework.amqp.rabbit.listener;
import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -27,8 +28,10 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.ImmediateAcknowledgeAmqpException;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
@@ -45,6 +48,7 @@ import com.rabbitmq.client.Channel;
* @author Dave Syer
* @author Gunnar Hillert
* @author Gary Russell
* @author Artem Bilan
*
* @since 1.0
*
@@ -56,7 +60,7 @@ public class MessageListenerManualAckIntegrationTests {
public static final String TEST_QUEUE = "test.queue.MessageListenerManualAckIntegrationTests";
private static Log logger = LogFactory.getLog(MessageListenerManualAckIntegrationTests.class);
private static final Log logger = LogFactory.getLog(MessageListenerManualAckIntegrationTests.class);
private final Queue queue = new Queue(TEST_QUEUE);
@@ -121,6 +125,26 @@ public class MessageListenerManualAckIntegrationTests {
assertThat(template.receiveAndConvert(queue.getName())).isNull();
}
@Test
public void immediateIsAckedForManual() throws Exception {
CountDownLatch latch = new CountDownLatch(1);
container = createContainer(new ImmediateTestListener(latch));
container.setEnforceImmediateAckForManual(true);
template.convertAndSend(queue.getName(), "test data");
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
container.stop();
Channel channel = template.getConnectionFactory().createConnection().createChannel(false);
await().untilAsserted(() -> assertThat(channel.consumerCount(queue.getName())).isEqualTo(0));
assertThat(channel.messageCount(queue.getName())).isEqualTo(0);
channel.close();
}
private SimpleMessageListenerContainer createContainer(Object listener) {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(template.getConnectionFactory());
container.setMessageListener(new MessageListenerAdapter(listener));
@@ -159,4 +183,23 @@ public class MessageListenerManualAckIntegrationTests {
}
}
static class ImmediateTestListener implements MessageListener {
private final CountDownLatch latch;
ImmediateTestListener(CountDownLatch latch) {
this.latch = latch;
}
@Override
public void onMessage(Message message) {
try {
throw new ImmediateAcknowledgeAmqpException("intentional");
}
finally {
this.latch.countDown();
}
}
}
}