GH-2907: Use CF.closeTimeout for confirms wait

Fixes: #2907

Issue link: #2907

The current hard-coded `5 seconds` is not enough in real applications under heavy load

* Fix `CachingConnectionFactory` to use `getCloseTimeout()` for `publisherCallbackChannel.waitForConfirms()`
which is `30 seconds` by default, but can be modified via `CachingConnectionFactory.setCloseTimeout()`

(cherry picked from commit 562bc77)

Author: artembilan

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

@@ -478,8 +478,9 @@ protected ExecutorService getExecutorService() {
 	}
 
 	/**
-	 * How long to wait (milliseconds) for a response to a connection close operation from the broker; default 30000 (30
-	 * seconds).
+	 * How long to wait (milliseconds) for a response to a connection close operation from the broker;
+	 * default 30000 (30 seconds).
+	 * Also used for {@link com.rabbitmq.client.Channel#waitForConfirms()}.
 	 * @param closeTimeout the closeTimeout to set.
 	 */
 	public void setCloseTimeout(int closeTimeout) {

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

@@ -1087,8 +1087,6 @@ public String toString() {
 
 	private final class CachedChannelInvocationHandler implements InvocationHandler {
 
-		private static final int ASYNC_CLOSE_TIMEOUT = 5_000;
-
 		private final ChannelCachingConnectionProxy theConnection;
 
 		private final Deque<ChannelProxy> channelList;
@@ -1302,7 +1300,7 @@ private void returnToCache(ChannelProxy proxy) {
 					getChannelsExecutor()
 							.execute(() -> {
 								try {
-									publisherCallbackChannel.waitForConfirms(ASYNC_CLOSE_TIMEOUT);
+									publisherCallbackChannel.waitForConfirms(getCloseTimeout());
 								}
 								catch (InterruptedException ex) {
 									Thread.currentThread().interrupt();
@@ -1426,10 +1424,10 @@ private void asyncClose() {
 				executorService.execute(() -> {
 					try {
 						if (ConfirmType.CORRELATED.equals(CachingConnectionFactory.this.confirmType)) {
-							channel.waitForConfirmsOrDie(ASYNC_CLOSE_TIMEOUT);
+							channel.waitForConfirmsOrDie(getCloseTimeout());
 						}
 						else {
-							Thread.sleep(ASYNC_CLOSE_TIMEOUT);
+							Thread.sleep(5_000); // NOSONAR - some time to give the channel a chance to ack
 						}
 					}
 					catch (@SuppressWarnings(UNUSED) InterruptedException e1) {