GH-8792: Fix File StreamingMS for one file filter

Fixes #8792

1. Use a `SftpStreamingMessageSource` with a `maxFetchSize = 5` and a `ChainFileListFilter` filter composed with
`SftpSystemMarkerFilePresentFileListFilter` which `supportsSingleFileFiltering == false`
2. Put 2 files in the folder and invoke `SftpStreamingMessageSource.receive()` method twice.
3. Put 5 files in the folder and invoke `SftpStreamingMessageSource.receive()`` method five times.
4. The last two files won't be received.

When you set max fetch size to a number bigger than one (for example 5) and at a certain point
it is necessary to `this.toBeReceived.clear()` inside `AbstractRemoteFileStreamingMessageSource.doReceive()`,
those removed elements from toBeReceived are not rolled back.

* Fix `AbstractRemoteFileStreamingMessageSource.listFiles()` to calculate `maxFetchSize` as `getMaxFetchSize() - this.fetched.get()`

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

(cherry picked from commit 22c4db2)

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

Author: artembilan

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

@@ -215,7 +215,7 @@ protected Object doReceive(int maxFetchSize) {
 			if (this.filter != null && this.filter.supportsSingleFileFiltering()
 					&& !this.filter.accept(file.getFileInfo())) {
 
-				if (this.toBeReceived.size() > 0) { // don't re-fetch already filtered files
+				if (!this.toBeReceived.isEmpty()) { // don't re-fetch already filtered files
 					file = poll();
 					continue;
 				}
@@ -267,7 +267,7 @@ private Object remoteFileToMessage(AbstractFileInfo<F> file) {
 	}
 
 	protected AbstractFileInfo<F> poll() {
-		if (this.toBeReceived.size() == 0) {
+		if (this.toBeReceived.isEmpty()) {
 			listFiles();
 		}
 		return this.toBeReceived.poll();
@@ -297,7 +297,7 @@ private void listFiles() {
 		if (!ObjectUtils.isEmpty(files)) {
 			List<AbstractFileInfo<F>> fileInfoList;
 			if (this.filter != null && !this.filter.supportsSingleFileFiltering()) {
-				int maxFetchSize = getMaxFetchSize();
+				int maxFetchSize = getMaxFetchSize() - this.fetched.get();
 				List<F> filteredFiles = this.filter.filterFiles(files);
 				if (maxFetchSize > 0 && filteredFiles.size() > maxFetchSize) {
 					rollbackFromFileToListEnd(filteredFiles, filteredFiles.get(maxFetchSize));

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

@@ -16,13 +16,17 @@
 
 package org.springframework.integration.sftp.inbound;
 
+import java.io.File;
+import java.io.IOException;
 import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
 import java.util.Comparator;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 
 import com.jcraft.jsch.ChannelSftp.LsEntry;
+import org.apache.commons.io.FileUtils;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.annotation.Autowired;
@@ -38,11 +42,14 @@
 import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
 import org.springframework.integration.file.FileHeaders;
 import org.springframework.integration.file.filters.AcceptAllFileListFilter;
+import org.springframework.integration.file.filters.ChainFileListFilter;
 import org.springframework.integration.file.remote.session.SessionFactory;
 import org.springframework.integration.metadata.SimpleMetadataStore;
 import org.springframework.integration.scheduling.PollerMetadata;
 import org.springframework.integration.sftp.SftpTestSupport;
 import org.springframework.integration.sftp.filters.SftpPersistentAcceptOnceFileListFilter;
+import org.springframework.integration.sftp.filters.SftpSimplePatternFileListFilter;
+import org.springframework.integration.sftp.filters.SftpSystemMarkerFilePresentFileListFilter;
 import org.springframework.integration.sftp.session.SftpFileInfo;
 import org.springframework.integration.sftp.session.SftpRemoteFileTemplate;
 import org.springframework.integration.transformer.StreamTransformer;
@@ -167,6 +174,69 @@ public void testMaxFetchLambdaFilter() throws Exception {
 		StaticMessageHeaderAccessor.getCloseableResource(received).close();
 	}
 
+
+	@Test
+	public void maxFetchIsAdjustedWhenNoSupportsSingleFileFiltering() throws Exception {
+		SftpStreamingMessageSource messageSource = buildSource();
+		ChainFileListFilter<LsEntry> chainFileListFilter = new ChainFileListFilter<>();
+		SftpSystemMarkerFilePresentFileListFilter sftpSystemMarkerFilePresentFileListFilter =
+				new SftpSystemMarkerFilePresentFileListFilter(
+						new SftpSimplePatternFileListFilter("*"), ".trg");
+		SftpPersistentAcceptOnceFileListFilter sftpPersistentAcceptOnceFileListFilter =
+				new SftpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "prefix");
+		chainFileListFilter.addFilter(sftpSystemMarkerFilePresentFileListFilter);
+		chainFileListFilter.addFilter(sftpPersistentAcceptOnceFileListFilter);
+		messageSource.setFilter(chainFileListFilter);
+		messageSource.setMaxFetchSize(5);
+		messageSource.afterPropertiesSet();
+		messageSource.start();
+
+		addFileAndTrigger("file001");
+		addFileAndTrigger("file002");
+
+		Message<InputStream> received = messageSource.receive();
+		assertThat(received).isNotNull();
+		assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("file001");
+
+		received = messageSource.receive();
+		assertThat(received).isNotNull();
+		assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("file002");
+
+		addFileAndTrigger("file003");
+		addFileAndTrigger("file004");
+		addFileAndTrigger("file005");
+		addFileAndTrigger("file006");
+		addFileAndTrigger("file007");
+
+		received = messageSource.receive();
+		assertThat(received).isNotNull();
+		assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("file003");
+
+		received = messageSource.receive();
+		assertThat(received).isNotNull();
+		assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("file004");
+
+		received = messageSource.receive();
+		assertThat(received).isNotNull();
+		assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("file005");
+
+		received = messageSource.receive();
+		assertThat(received).isNotNull();
+		assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("file006");
+
+		received = messageSource.receive();
+		assertThat(received).isNotNull();
+		assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("file007");
+	}
+
+	private void addFileAndTrigger(String filename) throws IOException {
+		File file = new File(this.sourceRemoteDirectory, filename);
+		FileUtils.writeStringToFile(file, "source1", StandardCharsets.UTF_8);
+
+		file = new File(this.sourceRemoteDirectory, filename + ".trg");
+		file.createNewFile();
+	}
+
 	private SftpStreamingMessageSource buildSource() {
 		SftpStreamingMessageSource messageSource =
 				new SftpStreamingMessageSource(this.config.template(),