"Failed to obtain pooled item" when concurrent file uploadings on cached shared session factory with unlimited poolsize

[metalpalo]

In what version(s) of Spring Integration are you seeing this issue?

5.3.8.RELEASE

Describe the bug
After upgrade spring-integration-sftp from 5.1.9.RELEASE to 5.3.8.RELEASE it seems that is problem to upload multiple files on shared and cached session. From logs I see that it tried to open more than one physical connection.
I tested it on two server with limited physical connection count.I got following errors like:

Caused by: org.springframework.messaging.MessagingException: Failed to execute on session; nested exception is org.springframework.integration.util.PoolItemNotAvailableException: Failed to obtain pooled item
	at org.springframework.integration.file.remote.RemoteFileTemplate.execute(RemoteFileTemplate.java:448) ~[spring-integration-file-5.3.8.RELEASE.jar:5.3.8.RELEASE]
	at org.springframework.integration.file.remote.RemoteFileTemplate.send(RemoteFileTemplate.java:301) ~[spring-integration-file-5.3.8.RELEASE.jar:5.3.8.RELEASE]
	at org.springframework.integration.file.remote.RemoteFileTemplate.send(RemoteFileTemplate.java:289) ~[spring-integration-file-5.3.8.RELEASE.jar:5.3.8.RELEASE]
	at org.springframework.integration.file.remote.RemoteFileTemplate.send(RemoteFileTemplate.java:281) ~[spring-integration-file-5.3.8.RELEASE.jar:5.3.8.RELEASE]
	at com.ferratum.collectionservices.service.impl.SftpServiceImpl.uploadFile(SftpServiceImpl.java:48) ~[classes/:na]
	... 144 common frames omitted
Caused by: org.springframework.integration.util.PoolItemNotAvailableException: Failed to obtain pooled item
	at org.springframework.integration.util.SimplePool.getItem(SimplePool.java:188) ~[spring-integration-core-5.3.8.RELEASE.jar:5.3.8.RELEASE]
	at org.springframework.integration.file.remote.session.CachingSessionFactory.getSession(CachingSessionFactory.java:132) ~[spring-integration-file-5.3.8.RELEASE.jar:5.3.8.RELEASE]
	at org.springframework.integration.file.remote.session.DelegatingSessionFactory.getSession(DelegatingSessionFactory.java:111) ~[spring-integration-file-5.3.8.RELEASE.jar:5.3.8.RELEASE]
	at org.springframework.integration.file.remote.session.DelegatingSessionFactory.getSession(DelegatingSessionFactory.java:105) ~[spring-integration-file-5.3.8.RELEASE.jar:5.3.8.RELEASE]
	at org.springframework.integration.file.remote.RemoteFileTemplate.execute(RemoteFileTemplate.java:434) ~[spring-integration-file-5.3.8.RELEASE.jar:5.3.8.RELEASE]
	... 148 common frames omitted
Caused by: java.lang.IllegalStateException: failed to create SFTP Session
	at org.springframework.integration.sftp.session.DefaultSftpSessionFactory.getSession(DefaultSftpSessionFactory.java:404) ~[spring-integration-sftp-5.3.8.RELEASE.jar:5.3.8.RELEASE]
	at org.springframework.integration.sftp.session.DefaultSftpSessionFactory.getSession(DefaultSftpSessionFactory.java:61) ~[spring-integration-sftp-5.3.8.RELEASE.jar:5.3.8.RELEASE]
	at org.springframework.integration.file.remote.session.CachingSessionFactory$1.createForPool(CachingSessionFactory.java:83) ~[spring-integration-file-5.3.8.RELEASE.jar:5.3.8.RELEASE]
	at org.springframework.integration.file.remote.session.CachingSessionFactory$1.createForPool(CachingSessionFactory.java:80) ~[spring-integration-file-5.3.8.RELEASE.jar:5.3.8.RELEASE]
	at org.springframework.integration.util.SimplePool.doGetItem(SimplePool.java:198) ~[spring-integration-core-5.3.8.RELEASE.jar:5.3.8.RELEASE]
	at org.springframework.integration.util.SimplePool.getItem(SimplePool.java:179) ~[spring-integration-core-5.3.8.RELEASE.jar:5.3.8.RELEASE]
	... 152 common frames omitted
Caused by: java.lang.IllegalStateException: failed to connect
	at org.springframework.integration.sftp.session.SftpSession.connect(SftpSession.java:303) ~[spring-integration-sftp-5.3.8.RELEASE.jar:5.3.8.RELEASE]
	at org.springframework.integration.sftp.session.DefaultSftpSessionFactory.getSession(DefaultSftpSessionFactory.java:398) ~[spring-integration-sftp-5.3.8.RELEASE.jar:5.3.8.RELEASE]
	... 157 common frames omitted
Caused by: com.jcraft.jsch.JSchException: Session.connect: java.net.SocketException: Connection reset
	at com.jcraft.jsch.Session.connect(Session.java:565) ~[jsch-0.1.54.jar:na]
	at com.jcraft.jsch.Session.connect(Session.java:183) ~[jsch-0.1.54.jar:na]
	at org.springframework.integration.sftp.session.SftpSession.connect(SftpSession.java:294) ~[spring-integration-sftp-5.3.8.RELEASE.jar:5.3.8.RELEASE]

To Reproduce

For instance, I upload 10 files from separated threads concurrently to same server.
I use following setting for DefaultSftpSessionFactory

 DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true);
            factory.setHost(host);
            factory.setPort(port);
            factory.setUser(username);
            if (privateKey != null) {
                factory.setPrivateKey(privateKey);
                factory.setPrivateKeyPassphrase(privateKeyPassphrase);
            } else {
                factory.setPassword(password);
            }
            factory.setAllowUnknownKeys(true);
            CachingSessionFactory cachingSessionFactory = new CachingSessionFactory<>(factory,0);

Expected behavior

All files should be uploaded within one physical connection with 10 channels multiplexed on them.
It works like charm on version 5.1.9.RELEASE.

Can somebody tell me what changed between versions 5.1.9 and 5.3.8.

[garyrussell]

There was some optimization done in getSession(), and it broke the shared session logic in a multi-threaded environment - the sharedJschSession field is captured before the lock is acquired.

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

Lines 384 to 413 in 50e20a9

	public SftpSession getSession() {
	JSchSessionWrapper jschSession = this.sharedJschSession;
	SftpSession sftpSession;
	if (this.sharedSessionLock != null) {
	this.sharedSessionLock.lock();
	}
	try {
	boolean freshJschSession = false;
	if (jschSession == null || !jschSession.isConnected()) {
	jschSession = new JSchSessionWrapper(initJschSession());
	freshJschSession = true;
	}
	sftpSession = new SftpSession(jschSession);
	JavaUtils.INSTANCE.acceptIfNotNull(this.channelConnectTimeout, sftpSession::setChannelConnectTimeout);
	sftpSession.connect();
	if (this.isSharedSession && freshJschSession) {
	this.sharedJschSession = jschSession;
	}
	}
	catch (Exception e) {
	throw new IllegalStateException("failed to create SFTP Session", e);
	}
	finally {
	if (this.sharedSessionLock != null) {
	this.sharedSessionLock.unlock();
	}
	}
	jschSession.addChannel();
	return sftpSession;
	}