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.
@@ -103,7 +103,7 @@ public Boolean doWithClient(FTPClient client) {
 							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

@@ -33,6 +33,7 @@
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
+import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -43,7 +44,10 @@
 import org.springframework.integration.file.remote.SessionCallbackWithoutResult;
 import org.springframework.integration.file.remote.session.Session;
 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;
@@ -62,7 +66,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.setExpression("'foobar.txt'");
@@ -105,7 +109,7 @@ public void doInSessionWithoutResult(Session<FTPFile> session) throws IOExceptio
 				assertTrue(session.rmdir("foo/"));
 			}
 		});
-		assertFalse(template.getSession().exists("foo"));
+		assertFalse(template.exists("foo"));
 	}
 
 	@Test
@@ -133,6 +137,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());
+		assertTrue(file.renameTo(newFile));
+		file.delete();
+		newFile.delete();
+
+		SimplePool<?> pool = TestUtils.getPropertyValue(this.sessionFactory, "pool", SimplePool.class);
+		assertEquals(0, pool.getActiveCount());
+	}
+
 	@Configuration
 	public static class Config {