GH-8713: Add support for custom SftpClient

Fixes #8713

* Introduce `DefaultSftpSessionFactory.createSftpClient()` factory method
which can be overridden for any custom `SftpClient` use-case
* Add changes to the docs
* Some code clean up

**Cherry-pick to `6.1.x`**

# Conflicts:
#	spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java
#	src/reference/antora/modules/ROOT/pages/sftp/session-factory.adoc
#	src/reference/antora/modules/ROOT/pages/whats-new.adoc

Author: artembilan

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

@@ -53,6 +53,10 @@
 
 /**
  * Factory for creating {@link SftpSession} instances.
+ * <p>
+ * The {@link #createSftpClient(ClientSession, SftpVersionSelector, SftpErrorDataHandler)}
+ * can be overridden to provide a custom {@link SftpClient}.
+ * The {@link ConcurrentSftpClient} is used by default.
  *
  * @author Josh Long
  * @author Mario Gray
@@ -64,6 +68,7 @@
  * @author Artem Bilan
  * @author Krzysztof Debski
  * @author Auke Zaaiman
+ * @author Adama Sorho
  *
  * @since 2.0
  */
@@ -279,9 +284,7 @@ public SftpSession getSession() {
 		try {
 			boolean freshSftpClient = false;
 			if (sftpClient == null || !sftpClient.isOpen()) {
-				sftpClient =
-						new ConcurrentSftpClient(initClientSession(), this.sftpVersionSelector,
-								SftpErrorDataHandler.EMPTY);
+				sftpClient = createSftpClient(initClientSession(), this.sftpVersionSelector, SftpErrorDataHandler.EMPTY);
 				freshSftpClient = true;
 			}
 			sftpSession = new SftpSession(sftpClient);
@@ -290,8 +293,8 @@ public SftpSession getSession() {
 				this.sharedSftpClient = sftpClient;
 			}
 		}
-		catch (Exception e) {
-			throw new IllegalStateException("failed to create SFTP Session", e);
+		catch (Exception ex) {
+			throw new IllegalStateException("failed to create SFTP Session", ex);
 		}
 		finally {
 			if (this.sharedSessionLock != null) {
@@ -390,15 +393,32 @@ public void resetSharedSession() {
 		this.sharedSftpClient = null;
 	}
 
+	/**
+	 * Can be overridden to provide a custom {@link SftpClient} to {@link #getSession()}.
+	 * @param clientSession the {@link ClientSession}
+	 * @param initialVersionSelector the initial {@link SftpVersionSelector}
+	 * @param errorDataHandler the {@link SftpErrorDataHandler} to handle incoming data
+	 *                         through the error stream.
+	 * @return {@link SftpClient}
+	 * @throws IOException if failed to initialize
+	 * @since 6.1.3
+	 */
+	protected SftpClient createSftpClient(
+			ClientSession clientSession, SftpVersionSelector initialVersionSelector,
+			SftpErrorDataHandler errorDataHandler) throws IOException {
+
+		return new ConcurrentSftpClient(clientSession, initialVersionSelector, errorDataHandler);
+	}
+
 	/**
 	 * The {@link DefaultSftpClient} extension to lock the {@link #send(int, Buffer)}
 	 * for concurrent interaction.
 	 */
-	private static class ConcurrentSftpClient extends DefaultSftpClient {
+	protected static class ConcurrentSftpClient extends DefaultSftpClient {
 
 		private final Lock sendLock = new ReentrantLock();
 
-		ConcurrentSftpClient(ClientSession clientSession, SftpVersionSelector initialVersionSelector,
+		protected ConcurrentSftpClient(ClientSession clientSession, SftpVersionSelector initialVersionSelector,
 				SftpErrorDataHandler errorDataHandler) throws IOException {
 
 			super(clientSession, initialVersionSelector, errorDataHandler);