From b945d1cc0216fc1a8a8af8c1f8f5a7d234d2542e Mon Sep 17 00:00:00 2001
From: Artem Bilan <abilan@pivotal.io>
Date: Mon, 2 Oct 2017 14:26:27 -0400
Subject: [PATCH] Fix race condition in BogusDoubleAck test

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
---
 ...veryCachingConnectionIntegrationTests.java | 28 +++++++++----------
 src/reference/asciidoc/amqp.adoc              |  2 ++
 2 files changed, 16 insertions(+), 14 deletions(-)

diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.java
index 73d3ad96..9f40d1cc 100644
--- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.java
+++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/MessageListenerRecoveryCachingConnectionIntegrationTests.java
@@ -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());
 			}
 		}
 	}
diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc
index 91cc052b..df83f1ad 100644
--- a/src/reference/asciidoc/amqp.adoc
+++ b/src/reference/asciidoc/amqp.adoc
@@ -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>>.