From 19b466809d945d32923bc05278236e87e50746df Mon Sep 17 00:00:00 2001
From: Artem Bilan <abilan@vmware.com>
Date: Mon, 18 Jul 2022 12:38:15 -0400
Subject: [PATCH] GH-3827: Fix RemoteFile GET STREAM session leak

Fixes https://github.com/spring-projects/spring-integration/issues/3827

The `AbstractRemoteFileOutboundGateway.doGet()` for a `STREAM` option
does not close the `session` in case of error.
This may lead to some leaks or exhausted caches

* Close `session` in the `catch()` of the `AbstractRemoteFileOutboundGateway.doGet()`
* Adjust the `SftpServerOutboundTests` to configure a `CachingSessionFactory`
for the `testStream()` to verify there is no leaks attempting to
`GET STREAM` non-existing remote file twice

**Cherry-pick to `5.5.x`**
---
 .../AbstractRemoteFileOutboundGateway.java       |  3 ++-
 .../outbound/SftpServerOutboundTests-context.xml |  8 +++++++-
 .../sftp/outbound/SftpServerOutboundTests.java   | 16 +++++++++++++---
 3 files changed, 22 insertions(+), 5 deletions(-)

diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java
index c6c11760b6..948bfd5555 100644
--- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java
+++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2021 the original author or authors.
+ * Copyright 2002-2022 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.
@@ -673,6 +673,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 						.setHeader(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE, session);
 			}
 			catch (IOException e) {
+				session.close();
 				throw new MessageHandlingException(requestMessage,
 						"Error handling message in the [" + this
 								+ "]. Failed to get the remote file [" + remoteFilePath + "] as a stream", e);
diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests-context.xml b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests-context.xml
index 4702261fce..2e210d363b 100644
--- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests-context.xml
+++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests-context.xml
@@ -191,7 +191,13 @@
 							  auto-create-directory="true"
 							  remote-file-separator="/"	/>
 
-	<int-sftp:outbound-gateway session-factory="sftpSessionFactory"
+	<bean id="cachingSessionFactory" class="org.springframework.integration.file.remote.session.CachingSessionFactory">
+		<constructor-arg name="sessionFactory" ref="sftpSessionFactory"/>
+		<constructor-arg name="sessionCacheSize" value="1"/>
+		<property name="sessionWaitTimeout" value="100"/>
+	</bean>
+
+	<int-sftp:outbound-gateway session-factory="cachingSessionFactory"
 							  request-channel="inboundGetStream"
 							  command="get"
 							  command-options="-stream"
diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java
index ceccd021f1..a2a9a3eebe 100644
--- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java
+++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java
@@ -70,6 +70,7 @@ import org.springframework.integration.support.MessageBuilder;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHandlingException;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.support.GenericMessage;
@@ -273,7 +274,7 @@ public class SftpServerOutboundTests extends SftpTestSupport {
 
 	@Test
 	@SuppressWarnings("unchecked")
-	void testLSRecursive() throws IOException {
+	void testLSRecursive() {
 		String dir = "sftpSource/";
 		this.inboundLSRecursive.send(new GenericMessage<Object>(dir));
 		Message<?> result = this.output.receive(1000);
@@ -291,7 +292,7 @@ public class SftpServerOutboundTests extends SftpTestSupport {
 
 	@Test
 	@SuppressWarnings("unchecked")
-	void testLSRecursiveALL() throws IOException {
+	void testLSRecursiveALL() {
 		String dir = "sftpSource/";
 		this.inboundLSRecursiveALL.send(new GenericMessage<Object>(dir));
 		Message<?> result = this.output.receive(1000);
@@ -481,7 +482,7 @@ public class SftpServerOutboundTests extends SftpTestSupport {
 		while (output.receive(0) != null) {
 			// drain
 		}
-		this.inboundMPut.send(new GenericMessage<File>(getSourceLocalDirectory()));
+		this.inboundMPut.send(new GenericMessage<>(getSourceLocalDirectory()));
 		@SuppressWarnings("unchecked")
 		Message<List<String>> out = (Message<List<String>>) this.output.receive(1000);
 		assertThat(out).isNotNull();
@@ -655,6 +656,15 @@ public class SftpServerOutboundTests extends SftpTestSupport {
 				.containsEntry(FileHeaders.REMOTE_DIRECTORY, "sftpSource/")
 				.containsEntry(FileHeaders.REMOTE_FILE, " sftpSource1.txt");
 		verify(session).close();
+
+		assertThatExceptionOfType(MessageHandlingException.class)
+				.isThrownBy(() -> this.inboundGetStream.send(new GenericMessage<>(dir + "doesNotExist.txt")))
+				.withStackTraceContaining("No such file or directory");
+
+		// No leak for not closed session after the previous failure
+		assertThatExceptionOfType(MessageHandlingException.class)
+				.isThrownBy(() -> this.inboundGetStream.send(new GenericMessage<>(dir + "doesNotExist.txt")))
+				.withStackTraceContaining("No such file or directory");
 	}
 
 	@Test