Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-9796: Fix SftpSession.write() for concurrency

Browse Source 
Fixes: #9796
Issue link: https://github.com/spring-projects/spring-integration/issues/9796

The `org.apache.sshd.sftp.client.impl.SftpOutputStreamAsync` is shared object for the `DefaultSftpClient`
and it cannot be used concurrently.

The guarded `send()` operation in the `ConcurrentSftpClient` is not enough
since `DefaultSftpClient.write()` is called directly from the `SftpOutputStreamAsync.internalFlush()`.
And this in the end is called from the `SftpSession.write()`

* Fix `DefaultSftpSessionFactory.ConcurrentSftpClient` to override the `write()` method instead.
Guard it with a `Lock` and call `sftpMessage.waitUntilSent();` to ensure that no concurrent access
to the underlying `SftpOutputStreamAsync`.

(cherry picked from commit 91f4fe48b6)
This commit is contained in:
Artem Bilan
2025-01-28 12:16:54 -05:00
committed by Spring Builds
parent e744f86c11
commit 4b0d0e2d14
1 changed files with 9 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,5 @@
/* /*
* Copyright 2002-2024 the original author or authors. * Copyright 2002-2025 the original author or authors.
* *
* Licensed under the Apache License, Version 2.0 (the "License"); * Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License. * you may not use this file except in compliance with the License.
@@ -46,6 +46,7 @@ import org.apache.sshd.common.util.net.SshdSocketAddress;
import org.apache.sshd.common.util.security.SecurityUtils; import org.apache.sshd.common.util.security.SecurityUtils;
import org.apache.sshd.sftp.client.SftpClient; import org.apache.sshd.sftp.client.SftpClient;
import org.apache.sshd.sftp.client.SftpErrorDataHandler; import org.apache.sshd.sftp.client.SftpErrorDataHandler;
import org.apache.sshd.sftp.client.SftpMessage;
import org.apache.sshd.sftp.client.SftpVersionSelector; import org.apache.sshd.sftp.client.SftpVersionSelector;
import org.apache.sshd.sftp.client.impl.AbstractSftpClient; import org.apache.sshd.sftp.client.impl.AbstractSftpClient;
import org.apache.sshd.sftp.client.impl.DefaultSftpClient; import org.apache.sshd.sftp.client.impl.DefaultSftpClient;
@@ -458,7 +459,7 @@ public class DefaultSftpSessionFactory
*/ */
protected class ConcurrentSftpClient extends DefaultSftpClient { protected class ConcurrentSftpClient extends DefaultSftpClient {
private final Lock sendLock = new ReentrantLock(); private final Lock sftpWriteLock = new ReentrantLock();
protected ConcurrentSftpClient(ClientSession clientSession, SftpVersionSelector initialVersionSelector, protected ConcurrentSftpClient(ClientSession clientSession, SftpVersionSelector initialVersionSelector,
SftpErrorDataHandler errorDataHandler) throws IOException { SftpErrorDataHandler errorDataHandler) throws IOException {
@@ -467,13 +468,15 @@ public class DefaultSftpSessionFactory
} }
@Override @Override
public int send(int cmd, Buffer buffer) throws IOException { public SftpMessage write(int cmd, Buffer buffer) throws IOException {
this.sendLock.lock(); this.sftpWriteLock.lock();
try { try {
return super.send(cmd, buffer); SftpMessage sftpMessage = super.write(cmd, buffer);
sftpMessage.waitUntilSent();
return sftpMessage;
} }
finally { finally {
this.sendLock.unlock(); this.sftpWriteLock.unlock();
} }
} }