GH-8708: Fix concurrency around SFTP client (#8709)

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

According to the `org.apache.sshd.common.channel.ChannelAsyncOutputStream.writeBuffer()` JavaDocs cannot be used concurrently.

* Introduce internal `DefaultSftpSessionFactory.ConcurrentSftpClient` extension
of the `DefaultSftpClient` to set a `Lock` around `super.send(cmd, buffer);`
* Remove lock from the `SftpSession` since it now is managed by the mentioned `ConcurrentSftpClient`

**Cherry-pick to `6.1.x` & `6.0.x`**
# Conflicts:
#	spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java

spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java

@@ -36,13 +36,15 @@ import org.apache.sshd.client.session.ClientSession;
 import org.apache.sshd.common.SshConstants;
 import org.apache.sshd.common.config.keys.FilePasswordProvider;
 import org.apache.sshd.common.keyprovider.KeyIdentityProvider;
+import org.apache.sshd.common.util.buffer.Buffer;
 import org.apache.sshd.common.util.io.resource.AbstractIoResource;
 import org.apache.sshd.common.util.io.resource.IoResource;
 import org.apache.sshd.common.util.net.SshdSocketAddress;
 import org.apache.sshd.common.util.security.SecurityUtils;
 import org.apache.sshd.sftp.client.SftpClient;
-import org.apache.sshd.sftp.client.SftpClientFactory;
+import org.apache.sshd.sftp.client.SftpErrorDataHandler;
 import org.apache.sshd.sftp.client.SftpVersionSelector;
+import org.apache.sshd.sftp.client.impl.DefaultSftpClient;
 
 import org.springframework.core.io.Resource;
 import org.springframework.integration.file.remote.session.SessionFactory;
@@ -278,8 +280,8 @@ public class DefaultSftpSessionFactory implements SessionFactory<SftpClient.DirE
 			boolean freshSftpClient = false;
 			if (sftpClient == null || !sftpClient.isOpen()) {
 				sftpClient =
-						SftpClientFactory.instance()
-								.createSftpClient(initClientSession(), this.sftpVersionSelector);
+						new ConcurrentSftpClient(initClientSession(), this.sftpVersionSelector,
+								SftpErrorDataHandler.EMPTY);
 				freshSftpClient = true;
 			}
 			sftpSession = new SftpSession(sftpClient);
@@ -388,4 +390,31 @@ public class DefaultSftpSessionFactory implements SessionFactory<SftpClient.DirE
 		this.sharedSftpClient = null;
 	}
 
+	/**
+	 * The {@link DefaultSftpClient} extension to lock the {@link #send(int, Buffer)}
+	 * for concurrent interaction.
+	 */
+	private static class ConcurrentSftpClient extends DefaultSftpClient {
+
+		private final Lock sendLock = new ReentrantLock();
+
+		ConcurrentSftpClient(ClientSession clientSession, SftpVersionSelector initialVersionSelector,
+				SftpErrorDataHandler errorDataHandler) throws IOException {
+
+			super(clientSession, initialVersionSelector, errorDataHandler);
+		}
+
+		@Override
+		public int send(int cmd, Buffer buffer) throws IOException {
+			this.sendLock.lock();
+			try {
+				return super.send(cmd, buffer);
+			}
+			finally {
+				this.sendLock.unlock();
+			}
+		}
+
+	}
+
 }

spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 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.
@@ -46,7 +46,6 @@ import org.springframework.util.StringUtils;
  * @author Oleg Zhurakousky
  * @author Gary Russell
  * @author Artem Bilan
- *
  * @since 2.0
  */
 public class SftpSession implements Session<SftpClient.DirEntry> {
@@ -97,7 +96,7 @@ public class SftpSession implements Session<SftpClient.DirEntry> {
 			}
 		}
 		remoteDir =
-				remoteDir.length() > 0 && remoteDir.charAt(0) == '/'
+				!remoteDir.isEmpty() && remoteDir.charAt(0) == '/'
 						? remoteDir
 						: this.sftpClient.canonicalPath(remoteDir);
 		return StreamSupport.stream(this.sftpClient.readDir(remoteDir).spliterator(), false)
@@ -122,22 +121,18 @@ public class SftpSession implements Session<SftpClient.DirEntry> {
 
 	@Override
 	public void write(InputStream inputStream, String destination) throws IOException {
-		synchronized (this.sftpClient) {
-			OutputStream outputStream = this.sftpClient.write(destination);
-			FileCopyUtils.copy(inputStream, outputStream);
-		}
+		OutputStream outputStream = this.sftpClient.write(destination);
+		FileCopyUtils.copy(inputStream, outputStream);
 	}
 
 	@Override
 	public void append(InputStream inputStream, String destination) throws IOException {
-		synchronized (this.sftpClient) {
-			OutputStream outputStream =
-					this.sftpClient.write(destination,
-							SftpClient.OpenMode.Create,
-							SftpClient.OpenMode.Write,
-							SftpClient.OpenMode.Append);
-			FileCopyUtils.copy(inputStream, outputStream);
-		}
+		OutputStream outputStream =
+				this.sftpClient.write(destination,
+						SftpClient.OpenMode.Create,
+						SftpClient.OpenMode.Write,
+						SftpClient.OpenMode.Append);
+		FileCopyUtils.copy(inputStream, outputStream);
 	}
 
 	@Override

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

@@ -18,11 +18,13 @@ package org.springframework.integration.sftp.session;
 
 import java.io.File;
 import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.net.ConnectException;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.stream.IntStream;
 
 import org.apache.sshd.client.SshClient;
 import org.apache.sshd.client.auth.password.PasswordIdentityProvider;
@@ -30,6 +32,7 @@ import org.apache.sshd.client.keyverifier.AcceptAllServerKeyVerifier;
 import org.apache.sshd.common.SshException;
 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.server.SftpSubsystemFactory;
 import org.junit.jupiter.api.Test;
 
@@ -45,7 +48,6 @@ import static org.awaitility.Awaitility.await;
  * @author Gary Russell
  * @author Artem Bilan
  * @author Auke Zaaiman
- *
  * @since 3.0.2
  */
 public class SftpSessionFactoryTests {
@@ -154,4 +156,40 @@ public class SftpSessionFactoryTests {
 		}
 	}
 
+	@Test
+	void concurrentSessionListDoesntCauseFailure() throws IOException {
+		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();
+
+			DefaultSftpSessionFactory sftpSessionFactory = new DefaultSftpSessionFactory();
+			sftpSessionFactory.setHost("localhost");
+			sftpSessionFactory.setPort(server.getPort());
+			sftpSessionFactory.setUser("user");
+			sftpSessionFactory.setPassword("pass");
+			sftpSessionFactory.setAllowUnknownKeys(true);
+
+			SftpSession session = sftpSessionFactory.getSession();
+
+			List<SftpClient.DirEntry[]> dirEntries =
+					IntStream.range(0, 10)
+							.boxed()
+							.parallel()
+							.map(i -> {
+								try {
+									return session.list(".");
+								}
+								catch (IOException e) {
+									throw new UncheckedIOException(e);
+								}
+							})
+							.toList();
+
+			assertThat(dirEntries).hasSize(10);
+		}
+	}
+
 }