GH-1194: Fix cache limit with Pub Confirms channel

Resolves https://github.com/spring-projects/spring-amqp/issues/1194

Deferred channel closes were unconditionally returned to the cache.

Move the cache size check etc to `doReturnToCache`.

**cherry-pick to 2.2.x, 2.1.x**

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -1137,13 +1137,9 @@ public class CachingConnectionFactory extends AbstractConnectionFactory
 			else if (methodName.equals("close")) {
 				// Handle close method: don't pass the call on.
 				if (CachingConnectionFactory.this.active) {
-					synchronized (this.channelList) {
-						if (CachingConnectionFactory.this.active && !RabbitUtils.isPhysicalCloseRequired() &&
-								(this.channelList.size() < getChannelCacheSize()
-										|| this.channelList.contains(proxy))) {
-							logicalClose((ChannelProxy) proxy);
-							return null;
-						}
+					if (!RabbitUtils.isPhysicalCloseRequired()) {
+						logicalClose((ChannelProxy) proxy);
+						return null;
 					}
 				}
 
@@ -1291,7 +1287,18 @@ public class CachingConnectionFactory extends AbstractConnectionFactory
 				synchronized (this.channelList) {
 					// Allow for multiple close calls...
 					if (CachingConnectionFactory.this.active) {
-						if (!this.channelList.contains(proxy)) {
+						boolean alreadyCached = this.channelList.contains(proxy);
+						if (this.channelList.size() >= getChannelCacheSize() && !alreadyCached) {
+							if (logger.isTraceEnabled()) {
+								logger.trace("Cache limit reached: " + this.target);
+							}
+							try {
+								physicalClose(proxy);
+							}
+							catch (@SuppressWarnings(UNUSED) Exception e) {
+							}
+						}
+						else if (!alreadyCached) {
 							if (logger.isTraceEnabled()) {
 								logger.trace("Returning cached Channel: " + this.target);
 							}

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2019-2020 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.
@@ -22,12 +22,22 @@ import static org.mockito.BDDMockito.willAnswer;
 import static org.mockito.Mockito.mock;
 
 import java.io.IOException;
+import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.IntStream;
 
 import org.junit.jupiter.api.Test;
 
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.ConfirmType;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.rabbit.junit.RabbitAvailable;
+import org.springframework.amqp.rabbit.junit.RabbitAvailableCondition;
+import org.springframework.core.task.SimpleAsyncTaskExecutor;
+
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Method;
 import com.rabbitmq.client.ShutdownListener;
@@ -38,10 +48,11 @@ import com.rabbitmq.client.ShutdownSignalException;
  * @since 2.1.5
  *
  */
+@RabbitAvailable
 public class PublisherCallbackChannelTests {
 
 	@Test
-	public void shutdownWhileCreate() throws IOException, TimeoutException {
+	void shutdownWhileCreate() throws IOException, TimeoutException {
 		Channel delegate = mock(Channel.class);
 		AtomicBoolean npe = new AtomicBoolean();
 		willAnswer(inv -> {
@@ -59,4 +70,72 @@ public class PublisherCallbackChannelTests {
 		channel.close();
 	}
 
+	@Test
+	void testNotCached() throws Exception {
+		CachingConnectionFactory cf = new CachingConnectionFactory(
+				RabbitAvailableCondition.getBrokerRunning().getConnectionFactory());
+		cf.setPublisherConfirmType(ConfirmType.CORRELATED);
+		cf.setChannelCacheSize(2);
+		cf.afterPropertiesSet();
+		RabbitTemplate template = new RabbitTemplate(cf);
+		CountDownLatch confirmLatch = new CountDownLatch(2);
+		template.setConfirmCallback((correlationData, ack, cause) -> {
+			try {
+				Thread.sleep(50);
+				confirmLatch.countDown();
+			}
+			catch (InterruptedException e) {
+				Thread.currentThread().interrupt();
+			}
+		});
+		CountDownLatch openedLatch = new CountDownLatch(2);
+		CountDownLatch closeLatch = new CountDownLatch(1);
+		CountDownLatch closedLatch = new CountDownLatch(2);
+		CountDownLatch waitForOtherLatch = new CountDownLatch(1);
+		SimpleAsyncTaskExecutor exec = new SimpleAsyncTaskExecutor();
+		IntStream.range(0, 2).forEach(i -> {
+			// this will open 3 or 4 channels
+			exec.execute(() -> {
+				template.execute(chann -> {
+					openedLatch.countDown();
+					template.convertAndSend("", "foo", "msg", msg -> {
+						if (i == 0) {
+							try {
+								waitForOtherLatch.await();
+							}
+							catch (InterruptedException e) {
+								Thread.currentThread().interrupt();
+							}
+						}
+						else {
+							waitForOtherLatch.countDown();
+						}
+						return msg;
+					}, new CorrelationData("" + i));
+					closeLatch.await();
+					return null;
+				});
+				closedLatch.countDown();
+			});
+		});
+		assertThat(openedLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		Connection conn = cf.createConnection();
+		Channel chan1 = conn.createChannel(false);
+		Channel chan2 = conn.createChannel(false);
+		chan1.close();
+		chan2.close();
+		Properties cacheProperties = cf.getCacheProperties();
+		assertThat(cacheProperties.getProperty("idleChannelsNotTx")).isEqualTo("2");
+		closeLatch.countDown();
+		assertThat(confirmLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(closedLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		cacheProperties = cf.getCacheProperties();
+		int n = 0;
+		while (n++ < 100 && Integer.parseInt(cacheProperties.getProperty("idleChannelsNotTx")) < 2) {
+			Thread.sleep(100);
+			cacheProperties = cf.getCacheProperties();
+		}
+		assertThat(cacheProperties.getProperty("idleChannelsNotTx")).isEqualTo("2");
+	}
+
 }