spring-projectsGH-8898: Add AbstractRemoteFileStreamingMessageSource.clearFetchedCache

Fixes: spring-projects#8898

In some cases not all retched remote files are processed, and after changing the `SessionFactory` (e.g. `RotatingServerAdvice`)
thy might not be processed on the next polling cycle

Author: kkosiacki

spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java

@@ -183,16 +183,27 @@ public void start() {
 	@Override
 	public void stop() {
 		if (this.running.compareAndSet(true, false)) {
-			if (this.filter == null || this.filter.supportsSingleFileFiltering()) {
-				this.toBeReceived.clear();
-			}
-			else {
-				// remove unprocessed files from the queue (and filter)
-				AbstractFileInfo<F> file = this.toBeReceived.poll();
-				while (file != null) {
-					resetFilterIfNecessary(file);
-					file = this.toBeReceived.poll();
-				}
+			clearFetchedCache();
+		}
+	}
+
+	/**
+	 * Clear internal queue of fetched remote files.
+	 * This functionality might be useful in combination with a
+	 * {@link org.springframework.integration.file.remote.aop.RotatingServerAdvice},
+	 * when not all fetched files are processed in between rotations.
+	 * @since 6.4
+	 */
+	public void clearFetchedCache() {
+		if (this.filter == null || this.filter.supportsSingleFileFiltering()) {
+			this.toBeReceived.clear();
+		}
+		else {
+			// remove unprocessed files from the queue (and filter)
+			AbstractFileInfo<F> file = this.toBeReceived.poll();
+			while (file != null) {
+				resetFilterIfNecessary(file);
+				file = this.toBeReceived.poll();
 			}
 		}
 	}

spring-integration-file/src/test/java/org/springframework/integration/file/remote/RemoteFileStreamingMessageSourceTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2022 the original author or authors.
+ * Copyright 2015-2024 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.
@@ -17,7 +17,6 @@
 package org.springframework.integration.file.remote;
 
 import java.io.IOException;
-import java.io.InputStream;
 import java.io.UncheckedIOException;
 import java.util.Collection;
 import java.util.Comparator;
@@ -26,7 +25,6 @@
 
 import org.junit.jupiter.api.Test;
 
-import org.springframework.beans.factory.BeanFactory;
 import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.remote.session.CachingSessionFactory;
 import org.springframework.integration.file.remote.session.Session;
@@ -37,6 +35,8 @@
 import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 /**
@@ -49,39 +49,65 @@
 public class RemoteFileStreamingMessageSourceTests {
 
 	@Test
-	@SuppressWarnings("unchecked")
+	public void fetchFilesFromRemoteAfterClearingFetchedCache() throws IOException {
+		RemoteFileTemplate<String> remoteFileTemplate = mock();
+		when(remoteFileTemplate.list("remoteDirectory")).thenReturn(new String[] {"file1", "file2"});
+		Session<String> session = mock();
+		when(session.readRaw(anyString())).thenReturn(mock());
+		when(remoteFileTemplate.getSession()).thenReturn(session);
+
+		Comparator<String> comparator = mock();
+		TestRemoteFileStreamingMessageSource testRemoteFileStreamingMessageSource =
+				new TestRemoteFileStreamingMessageSource(remoteFileTemplate, comparator);
+
+		testRemoteFileStreamingMessageSource.setRemoteDirectory("remoteDirectory");
+		testRemoteFileStreamingMessageSource.setBeanFactory(mock());
+		testRemoteFileStreamingMessageSource.start();
+
+		assertThat(testRemoteFileStreamingMessageSource.doReceive(2))
+				.isNotNull();
+
+		testRemoteFileStreamingMessageSource.clearFetchedCache();
+
+		assertThat(testRemoteFileStreamingMessageSource.doReceive(2))
+				.isNotNull();
+
+		verify(remoteFileTemplate, times(2)).list("remoteDirectory");
+
+	}
+
+	@Test
 	public void filterOutFilesNotAcceptedByFilter() throws IOException {
-		RemoteFileTemplate<String> remoteFileTemplate = mock(RemoteFileTemplate.class);
+		RemoteFileTemplate<String> remoteFileTemplate = mock();
 		when(remoteFileTemplate.list("remoteDirectory")).thenReturn(new String[] {"file1", "file2"});
-		Session<String> session = mock(Session.class);
-		when(session.readRaw(anyString())).thenReturn(mock(InputStream.class));
+		Session<String> session = mock();
+		when(session.readRaw(anyString())).thenReturn(mock());
 		when(remoteFileTemplate.getSession()).thenReturn(session);
 
-		FileListFilter<String> fileListFilter = mock(FileListFilter.class);
+		FileListFilter<String> fileListFilter = mock();
 		when(fileListFilter.supportsSingleFileFiltering()).thenReturn(true);
 		when(fileListFilter.accept("file1")).thenReturn(false);
 		when(fileListFilter.accept("file2")).thenReturn(false);
 
-		Comparator<String> comparator = mock(Comparator.class);
+		Comparator<String> comparator = mock();
 		TestRemoteFileStreamingMessageSource testRemoteFileStreamingMessageSource =
 				new TestRemoteFileStreamingMessageSource(remoteFileTemplate, comparator);
 
 		testRemoteFileStreamingMessageSource.setFilter(fileListFilter);
 		testRemoteFileStreamingMessageSource.setRemoteDirectory("remoteDirectory");
-		testRemoteFileStreamingMessageSource.setBeanFactory(mock(BeanFactory.class));
+		testRemoteFileStreamingMessageSource.setBeanFactory(mock());
 		testRemoteFileStreamingMessageSource.start();
 
 		assertThat(testRemoteFileStreamingMessageSource.doReceive(-1)).isNull();
 	}
 
 	@Test
-	@SuppressWarnings("unchecked")
 	public void sessionReturnedToCacheProperlyOnDoReceive() throws IOException {
-		Session<String> session = mock(Session.class);
+		Session<String> session = mock();
 		when(session.readRaw(anyString())).thenThrow(IOException.class);
 		when(session.list("remoteDirectory")).thenReturn(new String[] {"file1"});
 
-		SessionFactory<String> sessionFactory = mock(SessionFactory.class);
+		SessionFactory<String> sessionFactory = mock();
 		when(sessionFactory.getSession()).thenReturn(session);
 
 		CachingSessionFactory<String> cachingSessionFactory = new CachingSessionFactory<>(sessionFactory, 1);
@@ -91,7 +117,7 @@ public void sessionReturnedToCacheProperlyOnDoReceive() throws IOException {
 				new TestRemoteFileStreamingMessageSource(remoteFileTemplate, null);
 
 		testRemoteFileStreamingMessageSource.setRemoteDirectory("remoteDirectory");
-		testRemoteFileStreamingMessageSource.setBeanFactory(mock(BeanFactory.class));
+		testRemoteFileStreamingMessageSource.setBeanFactory(mock());
 		testRemoteFileStreamingMessageSource.start();
 
 		assertThatExceptionOfType(UncheckedIOException.class)

src/reference/antora/modules/ROOT/pages/sftp/max-fetch.adoc

@@ -25,5 +25,7 @@ If the poller is active when the property is changed, the change takes effect on
 Starting with version 5.1, the synchronizer can be provided with a `Comparator<?>`.
 This is useful when restricting the number of files fetched with `maxFetchSize`.
 
-Also see general xref:sftp/inbound.adoc[SFTP Inbound Channel Adapter] chapter for information about `FileListFilter` configuration.
+Starting with version 6.4, the `AbstractRemoteFileStreamingMessageSource` has now a convenient `clearFetchedCache()` API to remove references from cache for not processed remote files.
+The references stay in cache because polling configuration does not allow to process all of them in one cycle, and the target `SessionFactory` might be changed between polling cycles, e.g. via `RotatingServerAdvice`.
 
+Also see general xref:sftp/inbound.adoc[SFTP Inbound Channel Adapter] chapter for information about `FileListFilter` configuration.

src/reference/antora/modules/ROOT/pages/sftp/rotating-server-advice.adoc

@@ -83,3 +83,4 @@ public IntegrationFlow flow() {
 
 IMPORTANT: Do not configure a `TaskExecutor` on the poller when using this advice; see xref:changes-4.1-4.2.adoc#x4.2-conditional-pollers[Conditional Pollers for Message Sources] for more information.
 
+Also see a convenient `AbstractRemoteFileStreamingMessageSource.clearFetchedCache()` API when not all fetched files are processed withing a single polling cycle, but `SessionFactory` might be rotated to different one.

src/reference/antora/modules/ROOT/pages/whats-new.adoc

@@ -17,4 +17,10 @@ In general the project has been moved to the latest dependency versions.
 === New Components
 
 [[x6.4-general]]
-=== General Changes
+=== General Changes
+
+[[x6.4-remote-files-changes]]
+=== Remote File Adapters Changes
+
+The `AbstractRemoteFileStreamingMessageSource` has now a convenient `clearFetchedCache()` API to remove references from cache for not processed remote files.
+The references stay in cache because polling configuration does not allow to process all the fetched in one cycle, and the target `SessionFactory` might be changed between polling cycles, e.g. via `RotatingServerAdvice`.