GH-8745: Add RFT.shouldMarkSessionAsDirty() (#8759)

* GH-8745: Add RFT.shouldMarkSessionAsDirty()

Fixes https://github.com/spring-projects/spring-integration/issues/8745

Not all errors caught in the `RemoteFileTemplate.execute()`
are fatal to mark session as dirty and physically close the target session
in the cache

* Introduce a `RemoteFileTemplate.shouldMarkSessionAsDirty()`
to consult with an exception if it is really a fatal error to close
the session in the end.
* Override `shouldMarkSessionAsDirty()` in the `RemoteFileTemplate`
implementations to check statuses of respective protocol errors

**Cherry-pick to `6.1.x` & `6.0.x`**

* * Fix tests for pool interaction

* * Fix language in Javadocs
* Add more `not dirty` statuses to `SftpRemoteFileTemplate` & `SmbRemoteFileTemplate`

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

@@ -16,11 +16,16 @@
 
 package org.springframework.integration.sftp.session;
 
+import java.util.List;
+
 import org.apache.sshd.sftp.client.SftpClient;
+import org.apache.sshd.sftp.common.SftpConstants;
+import org.apache.sshd.sftp.common.SftpException;
 
 import org.springframework.integration.file.remote.ClientCallback;
 import org.springframework.integration.file.remote.RemoteFileTemplate;
 import org.springframework.integration.file.remote.session.SessionFactory;
+import org.springframework.lang.Nullable;
 
 /**
  * SFTP version of {@code RemoteFileTemplate} providing type-safe access to
@@ -34,6 +39,21 @@ import org.springframework.integration.file.remote.session.SessionFactory;
  */
 public class SftpRemoteFileTemplate extends RemoteFileTemplate<SftpClient.DirEntry> {
 
+	protected static final List<Integer> NOT_DIRTY_STATUSES = // NOSONAR
+			List.of(
+					SftpConstants.SSH_FX_NO_SUCH_FILE,
+					SftpConstants.SSH_FX_NO_SUCH_PATH,
+					SftpConstants.SSH_FX_INVALID_FILENAME,
+					SftpConstants.SSH_FX_INVALID_HANDLE,
+					SftpConstants.SSH_FX_FILE_ALREADY_EXISTS,
+					SftpConstants.SSH_FX_DIR_NOT_EMPTY,
+					SftpConstants.SSH_FX_NOT_A_DIRECTORY,
+					SftpConstants.SSH_FX_EOF,
+					SftpConstants.SSH_FX_CANNOT_DELETE,
+					SftpConstants.SSH_FX_FILE_IS_A_DIRECTORY,
+					SftpConstants.SSH_FX_FILE_CORRUPT
+			);
+
 	public SftpRemoteFileTemplate(SessionFactory<SftpClient.DirEntry> sessionFactory) {
 		super(sessionFactory);
 	}
@@ -48,4 +68,35 @@ public class SftpRemoteFileTemplate extends RemoteFileTemplate<SftpClient.DirEnt
 		return execute(session -> callback.doWithClient((SftpClient) session.getClientInstance()));
 	}
 
+	@Override
+	protected boolean shouldMarkSessionAsDirty(Exception ex) {
+		SftpException sftpException = findSftpException(ex);
+		if (sftpException != null) {
+			return isStatusDirty(sftpException.getStatus());
+		}
+		else {
+			return super.shouldMarkSessionAsDirty(ex);
+		}
+	}
+
+	/**
+	 * Check if {@link SftpException#getStatus()} is treated as fatal.
+	 * @param status the value from {@link SftpException#getStatus()}.
+	 * @return true if {@link SftpException#getStatus()} is treated as fatal.
+	 * @since 6.0.8
+	 */
+	protected boolean isStatusDirty(int status) {
+		return !NOT_DIRTY_STATUSES.contains(status);
+	}
+
+	@Nullable
+	private static SftpException findSftpException(Throwable ex) {
+		if (ex == null || ex instanceof SftpException) {
+			return (SftpException) ex;
+		}
+		else {
+			return findSftpException(ex.getCause());
+		}
+	}
+
 }

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpRemoteFileTemplateTests.java

@@ -22,6 +22,7 @@ import java.util.List;
 
 import org.apache.sshd.sftp.client.SftpClient;
 import org.apache.sshd.sftp.client.SftpVersionSelector;
+import org.apache.sshd.sftp.common.SftpException;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.BeanFactory;
@@ -38,6 +39,7 @@ import org.springframework.integration.file.remote.session.SessionFactory;
 import org.springframework.integration.file.support.FileExistsMode;
 import org.springframework.integration.sftp.SftpTestSupport;
 import org.springframework.integration.test.condition.LogLevels;
+import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.MessageDeliveryException;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.support.GenericMessage;
@@ -52,9 +54,7 @@ import static org.mockito.Mockito.mock;
 /**
  * @author Gary Russell
  * @author Artem Bilan
- *
  * @since 4.1
- *
  */
 @SpringJUnitConfig
 @DirtiesContext
@@ -63,7 +63,7 @@ public class SftpRemoteFileTemplateTests extends SftpTestSupport {
 	@Autowired
 	private CachingSessionFactory<SftpClient.DirEntry> sessionFactory;
 
-	@LogLevels(level = "trace", categories = { "org.apache.sshd", "org.springframework.integration.sftp" })
+	@LogLevels(level = "trace", categories = {"org.apache.sshd", "org.springframework.integration.sftp"})
 	@Test
 	public void testINT3412AppendStatRmdir() {
 		SftpRemoteFileTemplate template = new SftpRemoteFileTemplate(sessionFactory);
@@ -164,6 +164,25 @@ public class SftpRemoteFileTemplateTests extends SftpTestSupport {
 		oldVersionSession.close();
 	}
 
+	@Test
+	public void sessionIsNotDirtyOnNoSuchFileError() {
+		Session<SftpClient.DirEntry> session = this.sessionFactory.getSession();
+		session.close();
+
+		SftpRemoteFileTemplate template = new SftpRemoteFileTemplate(this.sessionFactory);
+
+		assertThatExceptionOfType(MessagingException.class)
+				.isThrownBy(() -> template.list("No_such_dir"))
+				.withRootCauseInstanceOf(SftpException.class)
+				.withStackTraceContaining("(SSH_FX_NO_SUCH_FILE): No such file or directory");
+
+		Session<SftpClient.DirEntry> newSession = this.sessionFactory.getSession();
+		assertThat(TestUtils.getPropertyValue(newSession, "targetSession"))
+				.isSameAs(TestUtils.getPropertyValue(session, "targetSession"));
+
+		newSession.close();
+	}
+
 	@Configuration
 	public static class Config {