GH-9430: Fix leak for the TCP/IP caching

Related to: https://github.com/spring-projects/spring-integration/issues/9430

(cherry picked from commit 8242ca6a91)

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java

@@ -78,9 +78,9 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
 
 	protected final Lock lifecycleMonitor = new ReentrantLock(); // NOSONAR final
 
-	private final Map<String, TcpConnectionSupport> connections = new ConcurrentHashMap<>();
+	protected final Map<String, TcpConnectionSupport> connections = new ConcurrentHashMap<>(); // NOSONAR final
 
-	private final Lock connectionsMonitor = new ReentrantLock();
+	protected final Lock connectionsMonitor = new ReentrantLock(); // NOSONAR final
 
 	private final BlockingQueue<PendingIO> delayedReads = new LinkedBlockingQueue<>();
 
@@ -977,7 +977,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
 		this.connectionsMonitor.lock();
 		try {
 			boolean closed = false;
-			TcpConnectionSupport connection = removeConnection(connectionId);
+			TcpConnectionSupport connection = this.connections.remove(connectionId);
 			if (connection != null) {
 				try {
 					connection.close();
@@ -996,11 +996,6 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
 		}
 	}
 
-	@Nullable
-	protected TcpConnectionSupport removeConnection(String connectionId) {
-		return this.connections.remove(connectionId);
-	}
-
 	@Override
 	public String toString() {
 		return super.toString()

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactory.java

@@ -33,6 +33,7 @@ import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.support.ErrorMessage;
+import org.springframework.util.Assert;
 
 /**
  * Connection factory that caches connections from the underlying target factory. The underlying
@@ -359,8 +360,31 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
 	}
 
 	@Override
-	protected TcpConnectionSupport removeConnection(String connectionId) {
-		return this.targetConnectionFactory.removeConnection(connectionId.replaceFirst("Cached:", ""));
+	public boolean closeConnection(String connectionId) {
+		Assert.notNull(connectionId, "'connectionId' to close must not be null");
+		String targetConnectionId = connectionId.replaceFirst("Cached:", "");
+		this.connectionsMonitor.lock();
+		try {
+			TcpConnectionSupport targetConnection = this.targetConnectionFactory.connections.get(targetConnectionId);
+			if (targetConnection != null) {
+				/*
+				 * If the delegate is stopped, actually close the connection, but still release
+				 * it to the pool, it will be discarded/renewed the next time it is retrieved.
+				 */
+				if (!isRunning()) {
+					logger.debug(() -> "Factory not running - closing " + connectionId);
+					super.closeConnection(targetConnectionId);
+				}
+				CachingClientConnectionFactory.this.pool.releaseItem(targetConnection);
+				return true;
+			}
+			else {
+				return false;
+			}
+		}
+		finally {
+			this.connectionsMonitor.unlock();
+		}
 	}
 
 	@Override

spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/IpIntegrationTests.java

@@ -35,6 +35,7 @@ import org.springframework.context.ApplicationListener;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.context.event.EventListener;
 import org.springframework.integration.MessageTimeoutException;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.EnableIntegration;
@@ -51,6 +52,7 @@ import org.springframework.integration.ip.tcp.TcpSendingMessageHandler;
 import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory;
 import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory;
 import org.springframework.integration.ip.tcp.connection.CachingClientConnectionFactory;
+import org.springframework.integration.ip.tcp.connection.TcpConnectionOpenEvent;
 import org.springframework.integration.ip.tcp.connection.TcpConnectionServerListeningEvent;
 import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionFactory;
 import org.springframework.integration.ip.tcp.connection.TcpNetServerConnectionFactory;
@@ -261,6 +263,7 @@ public class IpIntegrationTests {
 				IntStream.rangeClosed('a', 'z')
 						.mapToObj((characterCode) -> (char) characterCode)
 						.map((character) -> "" + character)
+						.parallel()
 						.peek((character) -> this.outboundFlowInput.send(new GenericMessage<>(character)))
 						.map(String::toUpperCase)
 						.toList();
@@ -274,6 +277,8 @@ public class IpIntegrationTests {
 		}
 
 		assertThat(replies).containsAll(expected);
+
+		assertThat(config.openEvents).hasSizeLessThanOrEqualTo(5);
 	}
 
 	@Configuration
@@ -431,9 +436,18 @@ public class IpIntegrationTests {
 			return Tcp.netClient("localhost", 0);
 		}
 
+		final List<TcpConnectionOpenEvent> openEvents = new ArrayList<>();
+
+		@EventListener
+		void connectionOpened(TcpConnectionOpenEvent tcpConnectionOpenEvent) {
+			if ("client3".equals(tcpConnectionOpenEvent.getConnectionFactoryName())) {
+				this.openEvents.add(tcpConnectionOpenEvent);
+			}
+		}
+
 		@Bean
 		CachingClientConnectionFactory cachingClient(TcpNetClientConnectionFactory client3) {
-			var cachingClientConnectionFactory = new CachingClientConnectionFactory(client3, 10);
+			var cachingClientConnectionFactory = new CachingClientConnectionFactory(client3, 5);
 			cachingClientConnectionFactory.setConnectionWaitTimeout(10_000);
 			return cachingClientConnectionFactory;
 		}

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/CachingClientConnectionFactoryTests.java

@@ -389,12 +389,7 @@ public class CachingClientConnectionFactoryTests {
 
 	private TcpConnectionSupport mockedTcpNioConnection() throws Exception {
 		SocketChannel socketChannel = mock(SocketChannel.class);
-		if (System.getProperty("java.version").startsWith("1.8")) {
-			new DirectFieldAccessor(socketChannel).setPropertyValue("open", false);
-		}
-		else {
-			new DirectFieldAccessor(socketChannel).setPropertyValue("closed", true);
-		}
+		new DirectFieldAccessor(socketChannel).setPropertyValue("closed", true);
 		doThrow(new IOException("Foo")).when(socketChannel).write(Mockito.any(ByteBuffer.class));
 		when(socketChannel.socket()).thenReturn(mock(Socket.class));
 		TcpNioConnection conn = new TcpNioConnection(socketChannel, false, false, event -> {