From e4bacc3e92592562438d581c142e7f05fb859af9 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 22 Aug 2023 14:00:34 -0400 Subject: [PATCH] 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`** --- .../session/DefaultSftpSessionFactory.java | 35 ++++++++++++++-- .../integration/sftp/session/SftpSession.java | 35 +++++----------- .../sftp/session/SftpSessionFactoryTests.java | 40 ++++++++++++++++++- 3 files changed, 80 insertions(+), 30 deletions(-) diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java index 3853cfcabf..98a117228b 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java +++ b/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; @@ -281,8 +283,8 @@ public class DefaultSftpSessionFactory implements SessionFactory { - private final Lock lock = new ReentrantLock(); - private final SftpClient sftpClient; public SftpSession(SftpClient sftpClient) { @@ -113,7 +108,7 @@ public class SftpSession implements Session { } } 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) @@ -138,30 +133,18 @@ public class SftpSession implements Session { @Override public void write(InputStream inputStream, String destination) throws IOException { - this.lock.lock(); - try { - OutputStream outputStream = this.sftpClient.write(destination); - FileCopyUtils.copy(inputStream, outputStream); - } - finally { - this.lock.unlock(); - } + OutputStream outputStream = this.sftpClient.write(destination); + FileCopyUtils.copy(inputStream, outputStream); } @Override public void append(InputStream inputStream, String destination) throws IOException { - this.lock.lock(); - try { - OutputStream outputStream = - this.sftpClient.write(destination, - SftpClient.OpenMode.Create, - SftpClient.OpenMode.Write, - SftpClient.OpenMode.Append); - FileCopyUtils.copy(inputStream, outputStream); - } - finally { - this.lock.unlock(); - } + OutputStream outputStream = + this.sftpClient.write(destination, + SftpClient.OpenMode.Create, + SftpClient.OpenMode.Write, + SftpClient.OpenMode.Append); + FileCopyUtils.copy(inputStream, outputStream); } @Override diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpSessionFactoryTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpSessionFactoryTests.java index 187beee44b..80578d2966 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpSessionFactoryTests.java +++ b/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 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); + } + } + }