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

Fix race condition in BogusDoubleAck test

Browse Source 
https://build.spring.io/browse/AMQP-MEIGHT-1204

Since the default `prefetchCount` is `250` now, we have all the 10
messages prefetched to the consumer.
The `basicAck` from the listener method in the `MANUAL` mode is async
operation and we end up (sporadically) with processing all other
prefetched messages until receive the async `handleShutdownSignal`.

* Come back to the `prefetchCount = 1` in the test-case to restore
expected test behavior.
* Document `prefetchCount = 1` with `MANUAL` ack mode
This commit is contained in:
Artem Bilan
2017-10-02 14:26:27 -04:00
parent ffff7d7c12
commit b945d1cc02
2 changed files with 16 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.java
View File

@@ -63,6 +63,8 @@ import com.rabbitmq.client.Channel;
* @author Dave Syer
* @author Gunnar Hillert
* @author Gary Russell
* @author Artem Bilan
*
* @since 1.0
*
*/
@@ -459,6 +461,7 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
container.setMessageListener(new MessageListenerAdapter(listener));
container.setQueueNames(queueName);
container.setPrefetchCount(1);
container.setConcurrentConsumers(concurrentConsumers);
container.setChannelTransacted(transactional);
container.setAcknowledgeMode(acknowledgeMode);
@@ -483,22 +486,19 @@ public class MessageListenerRecoveryCachingConnectionIntegrationTests {
@Override
public void onMessage(Message message, Channel channel) throws Exception {
String value = new String(message.getBody());
try {
logger.debug("Acking: " + value);
logger.debug("Acking: " + value);
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
if (failed.compareAndSet(false, true)) {
// intentional error (causes exception on connection thread):
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
if (failed.compareAndSet(false, true)) {
// intentional error (causes exception on connection thread):
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}
}
finally {
if (this.received.add(value)) {
latch.countDown();
}
else {
logger.debug(value + " already received, redelivered="
+ message.getMessageProperties().isRedelivered());
}
if (this.received.add(value)) {
latch.countDown();
}
else {
logger.debug(value + " already received, redelivered="
+ message.getMessageProperties().isRedelivered());
}
}
}

2
src/reference/asciidoc/amqp.adoc
View File

@@ -1348,6 +1348,8 @@ to a large amount of memory in the client process), and if strict message orderi
(the prefetch value should be set back to 1 in this case).
Also, with low-volume messaging and multiple consumers (including concurrency within a single listener container instance), you may wish to reduce the prefetch to get a more even distribution of messages across consumers.
It is also recomended to use `prefetch = 1` with the `MANUAL` ack mode.
The `basicAck` is async operation and if something wrong happens on the Broker (double ack for the same delivery tag, for example), you end up with processed subsequent messages in the batch, but unacked on the Broker and other consumer may see them.
See <<containerAttributes>>.