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

Fixes: #9796
Issue link: #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 91f4fe4)

Author: artembilan

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.
@@ -46,6 +46,7 @@
 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;
@@ -458,7 +459,7 @@ public void destroy() {
 	 */
 	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 {
@@ -467,13 +468,15 @@ protected ConcurrentSftpClient(ClientSession clientSession, SftpVersionSelector
 		}
 
 		@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();
 			}
 		}