From 41a61eb4de214fc6ad863c8953d1407bfcb899e9 Mon Sep 17 00:00:00 2001
From: Artem Bilan <abilan@pivotal.io>
Date: Tue, 11 Dec 2018 14:18:19 -0500
Subject: [PATCH] AMQP-847: Close channel in RabbitTemplate.receive

JIRA: https://jira.spring.io/browse/AMQP-847

To avoid unacked messages race condition when client timeouts, but at
this moment the message becomes available in queue, physically close
a receive channel on the `TimeoutException` from the `Future.get()`

**Cherry-pick to 2.0.x & 1.7.x**
---
 .../springframework/amqp/rabbit/core/RabbitTemplate.java | 4 ++++
 .../amqp/rabbit/core/RabbitTemplateIntegrationTests.java | 9 +++++++++
 2 files changed, 13 insertions(+)

diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java
index ceb78f0b..da71f64f 100644
--- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java
+++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java
@@ -890,6 +890,7 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware,
 				}
 				channel.basicCancel(consumer.getConsumerTag());
 				if (delivery == null) {
+					RabbitUtils.setPhysicalCloseRequired(channel, true);
 					return null;
 				}
 				else {
@@ -1029,6 +1030,9 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware,
 						}
 						receiveMessage = buildMessageFromDelivery(delivery);
 					}
+					else {
+						RabbitUtils.setPhysicalCloseRequired(channel, true);
+					}
 				}
 				if (receiveMessage != null) {
 					Object receive = receiveMessage;
diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java
index e88c008b..483d9ed8 100644
--- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java
+++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java
@@ -47,6 +47,7 @@ import java.io.UnsupportedEncodingException;
 import java.lang.reflect.Field;
 import java.util.Arrays;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.Callable;
@@ -348,6 +349,14 @@ public class RabbitTemplateIntegrationTests {
 		assertNull(this.template.receive(ROUTE));
 	}
 
+	@Test
+	public void testReceiveTimeoutRequeue() {
+		assertNull(this.template.receiveAndConvert(ROUTE, 1));
+		assertEquals(0,
+				TestUtils.getPropertyValue(this.connectionFactory, "cachedChannelsNonTransactional", List.class)
+						.size());
+	}
+
 	@Test
 	public void testReceiveBlockingTx() throws Exception {
 		this.template.convertAndSend(ROUTE, "blockTX");