GH-9909: Fix SftpSession for shared client

Fixes: #9909
Issue link: https://github.com/spring-projects/spring-integration/issues/9909

The `SftpSession.close()` closes `sftpClient` and its `clientSession` unconditionally.
At the same time the `DefaultSftpSessionFactory.isSharedSession` is expected to expose only a single shared client.
When this `DefaultSftpSessionFactory` is used concurrently, there is a chance that one thread would close
that shared client and another won't be able to interact due to `clientSession` is closed.

* Fix `SftpSession` accepting an `isSharedClient` flag and doing nothing in the `close()` if client is shared.
* Propagate `isSharedSession` state down to the `SftpSession` from the `DefaultSftpSessionFactory`
* Closed shared client and its `clientSession` in the `DefaultSftpSessionFactory.destroy()`

(cherry picked from commit 2df9611dde)

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpSessionFactoryTests.java

@@ -24,6 +24,11 @@ import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.IntStream;
 
 import org.apache.sshd.client.SshClient;
@@ -37,6 +42,8 @@ import org.apache.sshd.core.CoreModuleProperties;
 import org.apache.sshd.server.SshServer;
 import org.apache.sshd.server.keyprovider.SimpleGeneratorHostKeyProvider;
 import org.apache.sshd.sftp.client.SftpClient;
+import org.apache.sshd.sftp.client.SftpErrorDataHandler;
+import org.apache.sshd.sftp.client.SftpVersionSelector;
 import org.apache.sshd.sftp.client.impl.AbstractSftpClient;
 import org.apache.sshd.sftp.server.SftpSubsystemFactory;
 import org.junit.jupiter.api.Test;
@@ -268,4 +275,64 @@ public class SftpSessionFactoryTests {
 		}
 	}
 
+	@Test
+	void sharedSessionConcurrentAccess() throws Exception {
+		try (SshServer server = SshServer.setUpDefaultServer()) {
+			server.setPasswordAuthenticator((arg0, arg1, arg2) -> true);
+			server.setPort(0);
+			server.setKeyPairProvider(new SimpleGeneratorHostKeyProvider(new File("hostkey.ser").toPath()));
+			server.setSubsystemFactories(Collections.singletonList(new SftpSubsystemFactory()));
+			server.start();
+
+			AtomicInteger clientInstances = new AtomicInteger();
+
+			DefaultSftpSessionFactory sftpSessionFactory = new DefaultSftpSessionFactory(true) {
+
+				@Override
+				protected SftpClient createSftpClient(ClientSession clientSession,
+						SftpVersionSelector initialVersionSelector, SftpErrorDataHandler errorDataHandler)
+						throws IOException {
+
+					clientInstances.incrementAndGet();
+					return super.createSftpClient(clientSession, initialVersionSelector, errorDataHandler);
+				}
+
+			};
+			sftpSessionFactory.setHost("localhost");
+			sftpSessionFactory.setPort(server.getPort());
+			sftpSessionFactory.setUser("user");
+			sftpSessionFactory.setPassword("pass");
+			sftpSessionFactory.setAllowUnknownKeys(true);
+
+			ExecutorService executorService = Executors.newFixedThreadPool(10);
+
+			CountDownLatch executionLatch = new CountDownLatch(20);
+			List<Exception> errors = Collections.synchronizedList(new ArrayList<>());
+
+			for (int i = 0; i < 20; i++) {
+				executorService.execute(() -> {
+					try (SftpSession session = sftpSessionFactory.getSession()) {
+						session.list(".");
+					}
+					catch (Exception e) {
+						errors.add(e);
+					}
+					executionLatch.countDown();
+				});
+			}
+
+			assertThat(executionLatch.await(10, TimeUnit.SECONDS)).isTrue();
+			synchronized (errors) {
+				assertThat(errors).isEmpty();
+			}
+
+			assertThat(clientInstances).hasValue(1);
+
+			executorService.shutdown();
+			assertThat(executorService.awaitTermination(10, TimeUnit.SECONDS)).isTrue();
+
+			sftpSessionFactory.destroy();
+		}
+	}
+
 }