INT-4175: Fix SftpSessionFactory Race Condition

JIRA: https://jira.spring.io/browse/INT-4175
Fixes GH-1980 (#1980)

When the `isSharedSession` is used for the `DefaultSftpSessionFactory`,
there is some race condition window when we can call the target `this.jschSession.connect()` several times and end up with the session is already connected.

Wrap `sftpSession.connect()` to the `this.sharedSessionLock.readLock().lock()` when `isSharedSession` to protect from that race condition.

Note: there is no test for this change because it is pretty tricky to build barriers between threads for this kind of race conditions.
Especially after introduction `this.sharedSessionLock.readLock().lock()` around guilty code

**Cherry-pick to 4.3.x & 4.2.x**

Move `sharedJschSession` connect logic to the existing locking block

Fix mock test to reflect the current ConnectionFactory state

Conflicts:
	spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpOutboundTests.java
Resolved.

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

@@ -34,6 +34,7 @@
 
 import com.jcraft.jsch.ChannelSftp.LsEntry;
 import com.jcraft.jsch.JSch;
+import com.jcraft.jsch.JSchException;
 import com.jcraft.jsch.Proxy;
 import com.jcraft.jsch.SocketFactory;
 import com.jcraft.jsch.UIKeyboardInteractive;
@@ -357,6 +358,12 @@ public SftpSession getSession() {
 						try {
 							if (this.sharedJschSession == null || !this.sharedJschSession.isConnected()) {
 								this.sharedJschSession = new JSchSessionWrapper(initJschSession());
+								try {
+									this.sharedJschSession.getSession().connect();
+								}
+								catch (JSchException e) {
+									throw new IllegalStateException("failed to connect", e);
+								}
 							}
 						}
 						finally {

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpOutboundTests.java

@@ -20,6 +20,7 @@
 import static org.junit.Assert.assertNotSame;
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
+import static org.mockito.BDDMockito.willAnswer;
 import static org.mockito.Matchers.anyString;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.doReturn;
@@ -76,6 +77,7 @@
  * @author Oleg Zhurakousky
  * @author Gary Russell
  * @author Gunnar Hillert
+ * @author Artem Bilan
  */
 public class SftpOutboundTests {
 
@@ -102,7 +104,7 @@ public void testHandleFileMessage() throws Exception {
 		File destFile = new File(targetDir, srcFile.getName() + ".test");
 		destFile.deleteOnExit();
 
-		handler.handleMessage(new GenericMessage<File>(srcFile));
+		handler.handleMessage(new GenericMessage<>(srcFile));
 		assertTrue("destination file was not created", destFile.exists());
 	}
 
@@ -113,7 +115,7 @@ public void testHandleStringMessage() throws Exception {
 			file.delete();
 		}
 		SessionFactory<LsEntry> sessionFactory = new TestSftpSessionFactory();
-		FileTransferringMessageHandler<LsEntry> handler = new FileTransferringMessageHandler<LsEntry>(sessionFactory);
+		FileTransferringMessageHandler<LsEntry> handler = new FileTransferringMessageHandler<>(sessionFactory);
 		DefaultFileNameGenerator fGenerator = new DefaultFileNameGenerator();
 		fGenerator.setBeanFactory(mock(BeanFactory.class));
 		fGenerator.setExpression("'foo.txt'");
@@ -228,8 +230,21 @@ public void testSharedSession() throws Exception {
 		ctor.setAccessible(true);
 		com.jcraft.jsch.Session jschSession1 = spy(ctor.newInstance(jsch, "foo", "host", 22));
 		com.jcraft.jsch.Session jschSession2 = spy(ctor.newInstance(jsch, "foo", "host", 22));
-		new DirectFieldAccessor(jschSession1).setPropertyValue("isConnected", true);
-		new DirectFieldAccessor(jschSession2).setPropertyValue("isConnected", true);
+
+		willAnswer(invocation -> {
+			new DirectFieldAccessor(jschSession1).setPropertyValue("isConnected", true);
+			return null;
+		})
+				.given(jschSession1)
+				.connect();
+
+		willAnswer(invocation -> {
+			new DirectFieldAccessor(jschSession2).setPropertyValue("isConnected", true);
+			return null;
+		})
+				.given(jschSession2)
+				.connect();
+
 		when(jsch.getSession("foo", "host", 22)).thenReturn(jschSession1, jschSession2);
 		final ChannelSftp channel1 = spy(new ChannelSftp());
 		doReturn("channel1").when(channel1).toString();
@@ -262,7 +277,8 @@ public ChannelSftp answer(InvocationOnMock invocation) throws Throwable {
 	@Test
 	public void testNotSharedSession() throws Exception {
 		JSch jsch = spy(new JSch());
-		Constructor<com.jcraft.jsch.Session> ctor = com.jcraft.jsch.Session.class.getDeclaredConstructor(JSch.class, String.class, String.class, int.class);
+		Constructor<com.jcraft.jsch.Session> ctor =
+				com.jcraft.jsch.Session.class.getDeclaredConstructor(JSch.class, String.class, String.class, int.class);
 		ctor.setAccessible(true);
 		com.jcraft.jsch.Session jschSession1 = spy(ctor.newInstance(jsch, "foo", "host", 22));
 		com.jcraft.jsch.Session jschSession2 = spy(ctor.newInstance(jsch, "foo", "host", 22));
@@ -291,12 +307,26 @@ public void testNotSharedSession() throws Exception {
 	@Test
 	public void testSharedSessionCachedReset() throws Exception {
 		JSch jsch = spy(new JSch());
-		Constructor<com.jcraft.jsch.Session> ctor = com.jcraft.jsch.Session.class.getDeclaredConstructor(JSch.class, String.class, String.class, int.class);
+		Constructor<com.jcraft.jsch.Session> ctor =
+				com.jcraft.jsch.Session.class.getDeclaredConstructor(JSch.class, String.class, String.class, int.class);
 		ctor.setAccessible(true);
 		com.jcraft.jsch.Session jschSession1 = spy(ctor.newInstance(jsch, "foo", "host", 22));
 		com.jcraft.jsch.Session jschSession2 = spy(ctor.newInstance(jsch, "foo", "host", 22));
-		new DirectFieldAccessor(jschSession1).setPropertyValue("isConnected", true);
-		new DirectFieldAccessor(jschSession2).setPropertyValue("isConnected", true);
+
+		willAnswer(invocation -> {
+			new DirectFieldAccessor(jschSession1).setPropertyValue("isConnected", true);
+			return null;
+		})
+				.given(jschSession1)
+				.connect();
+
+		willAnswer(invocation -> {
+			new DirectFieldAccessor(jschSession2).setPropertyValue("isConnected", true);
+			return null;
+		})
+				.given(jschSession2)
+				.connect();
+
 		when(jsch.getSession("foo", "host", 22)).thenReturn(jschSession1, jschSession2);
 		final ChannelSftp channel1 = spy(new ChannelSftp());
 		doReturn("channel1").when(channel1).toString();