GH-2820: Close session in FtpRemFileTempl.exists

Fixes: #2820

The current `FtpRemoteFileTemplate.exists()` implementation obtains a
`Session` for the `NLST_AND_DIRS` mode, but doesn't close it.

* Fix the `FtpRemoteFileTemplate.exists()` method to delegate to `super`
for the proper interaction with a `Session`

**Cherry-pick to 5.1.x, 5.0.x & 4.3.x**

Author: artembilan

spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpRemoteFileTemplate.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2016 the original author or authors.
+ * Copyright 2014-2019 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.
@@ -92,7 +92,7 @@ public boolean exists(final String path) {
 						return !ObjectUtils.isEmpty(names);
 
 					case NLST_AND_DIRS:
-						return getSession().exists(path);
+						return FtpRemoteFileTemplate.super.exists(path);
 
 					default:
 						throw new IllegalStateException("Unsupported 'existsMode': " +

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/session/FtpRemoteFileTemplateTests.java

@@ -40,7 +40,10 @@
 import org.springframework.integration.file.remote.ClientCallbackWithoutResult;
 import org.springframework.integration.file.remote.SessionCallbackWithoutResult;
 import org.springframework.integration.file.remote.session.SessionFactory;
+import org.springframework.integration.file.support.FileExistsMode;
 import org.springframework.integration.ftp.FtpTestSupport;
+import org.springframework.integration.test.util.TestUtils;
+import org.springframework.integration.util.SimplePool;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.test.context.ContextConfiguration;
@@ -61,7 +64,7 @@ public class FtpRemoteFileTemplateTests extends FtpTestSupport {
 	private SessionFactory<FTPFile> sessionFactory;
 
 	@Test
-	public void testINT3412AppendStatRmdir() throws IOException {
+	public void testINT3412AppendStatRmdir() {
 		FtpRemoteFileTemplate template = new FtpRemoteFileTemplate(sessionFactory);
 		DefaultFileNameGenerator fileNameGenerator = new DefaultFileNameGenerator();
 		fileNameGenerator.setBeanFactory(mock(BeanFactory.class));
@@ -94,7 +97,7 @@ public void testINT3412AppendStatRmdir() throws IOException {
 			assertThat(files.length).isEqualTo(0);
 			assertThat(session.rmdir("foo/")).isTrue();
 		});
-		assertThat(template.getSession().exists("foo")).isFalse();
+		assertThat(template.exists("foo")).isFalse();
 	}
 
 	@Test
@@ -122,6 +125,27 @@ public void testFileCloseOnBadConnect() throws Exception {
 		newFile.delete();
 	}
 
+	@Test
+	public void testConnectionClosedAfterExists() throws Exception {
+		FtpRemoteFileTemplate template = new FtpRemoteFileTemplate(this.sessionFactory);
+		template.setRemoteDirectoryExpression(new LiteralExpression("/"));
+		template.setExistsMode(FtpRemoteFileTemplate.ExistsMode.NLST_AND_DIRS);
+		template.setBeanFactory(mock(BeanFactory.class));
+		template.afterPropertiesSet();
+		File file = new File(System.getProperty("java.io.tmpdir"), UUID.randomUUID().toString());
+		FileOutputStream fileOutputStream = new FileOutputStream(file);
+		fileOutputStream.write("foo".getBytes());
+		fileOutputStream.close();
+		template.send(new GenericMessage<>(file), FileExistsMode.IGNORE);
+		File newFile = new File(System.getProperty("java.io.tmpdir"), UUID.randomUUID().toString());
+		assertThat(file.renameTo(newFile)).isTrue();
+		file.delete();
+		newFile.delete();
+
+		SimplePool<?> pool = TestUtils.getPropertyValue(this.sessionFactory, "pool", SimplePool.class);
+		assertThat(pool.getActiveCount()).isEqualTo(0);
+	}
+
 	@Configuration
 	public static class Config {