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**

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java

@@ -137,6 +137,7 @@
  * @author Artem Bilan
  * @author Ernest Sadykov
  * @author Mark Norkin
+ *
  * @since 1.0
  */
 public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, RabbitOperations, MessageListener,
@@ -1269,7 +1270,7 @@ private Delivery consumeDelivery(Channel channel, String queueName, long timeout
 			Thread.currentThread().interrupt();
 		}
 		catch (TimeoutException e) {
-			// no result in time
+			RabbitUtils.setPhysicalCloseRequired(channel, true);
 		}
 		finally {
 			if (!(exception instanceof ConsumerCancelledException) && channel.isOpen()) {

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java

@@ -48,6 +48,7 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 import java.util.UUID;
@@ -397,6 +398,14 @@ public void testReceiveBlockingNoTimeout() throws Exception {
 		}
 	}
 
+	@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");