From 33f166f47b2180ddb6150e321c09d96bb45b3976 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 8 May 2020 12:55:15 -0400 Subject: [PATCH] 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** --- .../connection/CachingConnectionFactory.java | 25 ++++-- .../PublisherCallbackChannelTests.java | 83 ++++++++++++++++++- 2 files changed, 97 insertions(+), 11 deletions(-) diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java index 0a8002d6..889a31ce 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactory.java +++ b/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); } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelTests.java index 2549ff54..e04d1b8d 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/PublisherCallbackChannelTests.java +++ b/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"); + } + }