Fix CachedSessionFactory Race

Close the pool so that any sessions returned after the factory is
`destroy()`ed are closed.

* Call `removeAllIdleItems()` in `close()`.

* Close sessions in `SftpStreamingMessageSourceTests`.

**cherry-pick to all supported branches**

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java

@@ -30,6 +30,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.StaticMessageHeaderAccessor;
 import org.springframework.integration.annotation.InboundChannelAdapter;
 import org.springframework.integration.annotation.Transformer;
 import org.springframework.integration.channel.QueueChannel;
@@ -134,6 +135,7 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport {
 				.isIn(" sftpSource1.txt", "sftpSource2.txt");
 
 		received.getPayload().close();
+		StaticMessageHeaderAccessor.getCloseableResource(received).close();
 	}
 
 	@Test
@@ -148,6 +150,7 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport {
 				.isIn(" sftpSource1.txt", "sftpSource2.txt");
 
 		received.getPayload().close();
+		StaticMessageHeaderAccessor.getCloseableResource(received).close();
 	}
 
 	@Test
@@ -162,6 +165,7 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport {
 				.isIn(" sftpSource1.txt", "sftpSource2.txt");
 
 		received.getPayload().close();
+		StaticMessageHeaderAccessor.getCloseableResource(received).close();
 	}
 
 	private SftpStreamingMessageSource buildSource() {