From e73c3f22a03046412d568e368ddd15f8317a092e Mon Sep 17 00:00:00 2001
From: Artem Bilan <artem.bilan@broadcom.com>
Date: Tue, 28 Jan 2025 12:16:54 -0500
Subject: [PATCH] GH-9796: Fix `SftpSession.write()` for concurrency

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 91f4fe48b660aca7fee55996dd5457ed9417ac60)
---
 .../sftp/session/DefaultSftpSessionFactory.java   | 15 +++++++++------
 1 file changed, 9 insertions(+), 6 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 c38c1ddfc0..0082cbe9e3 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
@@ -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");
  * you may not use this file except in compliance with the License.
@@ -45,6 +45,7 @@ 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.SftpErrorDataHandler;
+import org.apache.sshd.sftp.client.SftpMessage;
 import org.apache.sshd.sftp.client.SftpVersionSelector;
 import org.apache.sshd.sftp.client.impl.AbstractSftpClient;
 import org.apache.sshd.sftp.client.impl.DefaultSftpClient;
@@ -438,7 +439,7 @@ public class DefaultSftpSessionFactory implements SessionFactory<SftpClient.DirE
 	 */
 	protected class ConcurrentSftpClient extends DefaultSftpClient {
 
-		private final Lock sendLock = new ReentrantLock();
+		private final Lock sftpWriteLock = new ReentrantLock();
 
 		protected ConcurrentSftpClient(ClientSession clientSession, SftpVersionSelector initialVersionSelector,
 				SftpErrorDataHandler errorDataHandler) throws IOException {
@@ -447,13 +448,15 @@ public class DefaultSftpSessionFactory implements SessionFactory<SftpClient.DirE
 		}
 
 		@Override
-		public int send(int cmd, Buffer buffer) throws IOException {
-			this.sendLock.lock();
+		public SftpMessage write(int cmd, Buffer buffer) throws IOException {
+			this.sftpWriteLock.lock();
 			try {
-				return super.send(cmd, buffer);
+				SftpMessage sftpMessage = super.write(cmd, buffer);
+				sftpMessage.waitUntilSent();
+				return sftpMessage;
 			}
 			finally {
-				this.sendLock.unlock();
+				this.sftpWriteLock.unlock();
 			}
 		}