GH-3488: Fix Persistent Filters with Recursion

Resolves https://github.com/spring-projects/spring-integration/issues/3488

Resolves two problems:

- When changes are made deep in the directory tree, they were not detected because
  the directory is in the metadata store and only passes the filter if a file
  immediately under it is changed, changing the directory's timestamp.

This is solved by subclassing `AbstractDirectoryAwareFileListFilter`, allowing its
`alwaysAcceptDirectories` property to be set.

- Only the filename was used as a metadata key; causing problems if a file with the
  same name appears multiple times in the tree.

This is solved with a new property on `AbstractDirectoryAwareFileListFilter` used by
the gateways to determine whether to filter the raw file names returned by the session
(previous behavior) or the full path relative to the root directory.

**cherry-pick to 5.4.x, 5.3.x**

* Some code style clean up

# Conflicts:
#	spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java
#	src/reference/asciidoc/whats-new.adoc

spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractDirectoryAwareFileListFilter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2021 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.
@@ -21,7 +21,10 @@ package org.springframework.integration.file.filters;
  * This permits, for example, pattern matching on just files when using recursion
  * to examine a directory tree.
  *
+ * @param <F> the file type.
+ *
  * @author Gary Russell
+ *
  * @since 5.0
  *
  */
@@ -29,6 +32,8 @@ public abstract class AbstractDirectoryAwareFileListFilter<F> extends AbstractFi
 
 	private boolean alwaysAcceptDirectories;
 
+	private boolean forRecursion;
+
 	/**
 	 * Set to true so that filters that support this feature can unconditionally pass
 	 * directories; default false.
@@ -38,6 +43,22 @@ public abstract class AbstractDirectoryAwareFileListFilter<F> extends AbstractFi
 		this.alwaysAcceptDirectories = alwaysAcceptDirectories;
 	}
 
+	@Override
+	public boolean isForRecursion() {
+		return this.forRecursion;
+	}
+
+	/**
+	 * Set to true to inform a recursive gateway operation to use the full file path as
+	 * the metadata key. Also sets {@link #alwaysAcceptDirectories}.
+	 * @param forRecursion true to use the full path.
+	 * @since 5.3.6
+	 */
+	public void setForRecursion(boolean forRecursion) {
+		this.forRecursion = forRecursion;
+		this.alwaysAcceptDirectories = forRecursion;
+	}
+
 	protected boolean alwaysAccept(F file) {
 		return file != null && this.alwaysAcceptDirectories && isDirectory(file);
 	}

spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractPersistentAcceptOnceFileListFilter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2019 the original author or authors.
+ * Copyright 2013-2021 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.
@@ -31,13 +31,15 @@ import org.springframework.util.Assert;
  * Files are deemed as already 'seen' if they exist in the store and have the
  * same modified time as the current file.
  *
+ * @param <F> the file type.
+ *
  * @author Gary Russell
  * @author Artem Bilan
  *
  * @since 3.0
  *
  */
-public abstract class AbstractPersistentAcceptOnceFileListFilter<F> extends AbstractFileListFilter<F>
+public abstract class AbstractPersistentAcceptOnceFileListFilter<F> extends AbstractDirectoryAwareFileListFilter<F>
 		implements ReversibleFileListFilter<F>, ResettableFileListFilter<F>, Closeable {
 
 	protected final ConcurrentMetadataStore store; // NOSONAR
@@ -73,6 +75,9 @@ public abstract class AbstractPersistentAcceptOnceFileListFilter<F> extends Abst
 
 	@Override
 	public boolean accept(F file) {
+		if (alwaysAccept(file)) {
+			return true;
+		}
 		String key = buildKey(file);
 		String newValue = value(file);
 		String oldValue = this.store.putIfAbsent(key, newValue);

spring-integration-file/src/main/java/org/springframework/integration/file/filters/CompositeFileListFilter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -56,6 +56,8 @@ public class CompositeFileListFilter<F>
 
 	private boolean allSupportAccept = true;
 
+	private boolean oneIsForRecursion;
+
 
 	public CompositeFileListFilter() {
 		this.fileFilters = new LinkedHashSet<>();
@@ -63,9 +65,14 @@ public class CompositeFileListFilter<F>
 
 	public CompositeFileListFilter(Collection<? extends FileListFilter<F>> fileFilters) {
 		this.fileFilters = new LinkedHashSet<>(fileFilters);
-		this.allSupportAccept = fileFilters.stream().allMatch(FileListFilter<F>::supportsSingleFileFiltering);
+		this.allSupportAccept = fileFilters.stream().allMatch(FileListFilter::supportsSingleFileFiltering);
+		this.oneIsForRecursion = fileFilters.stream().anyMatch(FileListFilter::isForRecursion);
 	}
 
+	@Override
+	public boolean isForRecursion() {
+		return this.oneIsForRecursion;
+	}
 
 	@Override
 	public void close() throws IOException {
@@ -77,7 +84,6 @@ public class CompositeFileListFilter<F>
 	}
 
 	public CompositeFileListFilter<F> addFilter(FileListFilter<F> filter) {
-		this.allSupportAccept &= filter.supportsSingleFileFiltering();
 		return addFilters(Collections.singletonList(filter));
 	}
 
@@ -114,11 +120,10 @@ public class CompositeFileListFilter<F>
 					throw new IllegalStateException(e);
 				}
 			}
+			this.allSupportAccept &= elf.supportsSingleFileFiltering();
+			this.oneIsForRecursion |= elf.isForRecursion();
 		}
 		this.fileFilters.addAll(filtersToAdd);
-		if (this.allSupportAccept) {
-			this.allSupportAccept = filtersToAdd.stream().allMatch(FileListFilter<F>::supportsSingleFileFiltering);
-		}
 		return this;
 	}
 

spring-integration-file/src/main/java/org/springframework/integration/file/filters/FileListFilter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -64,4 +64,13 @@ public interface FileListFilter<F> {
 		return false;
 	}
 
+	/**
+	 * Return true if this filter is being used for recursion.
+	 * @return whether or not to filter based on the full path.
+	 * @since 5.3.6
+	 */
+	default boolean isForRecursion() {
+		return false;
+	}
+
 }

spring-integration-file/src/main/java/org/springframework/integration/file/filters/FileSystemPersistentAcceptOnceFileListFilter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2019 the original author or authors.
+ * Copyright 2013-2021 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.
@@ -23,6 +23,7 @@ import org.springframework.integration.metadata.ConcurrentMetadataStore;
 
 /**
  * @author Gary Russell
+ *
  * @since 3.0
  *
  */
@@ -52,5 +53,9 @@ public class FileSystemPersistentAcceptOnceFileListFilter extends AbstractPersis
 		return file.exists();
 	}
 
+	@Override
+	protected boolean isDirectory(File file) {
+		return file.isDirectory();
+	}
 
 }

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

@@ -97,6 +97,8 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 	 */
 	private FileListFilter<F> filter;
 
+	private boolean filterAfterEnhancement;
+
 	/**
 	 * A {@link FileListFilter} that runs against the <em>local</em> file system view when
 	 * using MPUT.
@@ -402,6 +404,15 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 	 */
 	public void setFilter(FileListFilter<F> filter) {
 		this.filter = filter;
+		this.filterAfterEnhancement = filter != null
+				&& filter.isForRecursion()
+				&& filter.supportsSingleFileFiltering()
+				&& this.options.contains(Option.RECURSIVE);
+		if (filter != null && !filter.isForRecursion()) {
+			this.logger.warn("When using recursion, you will normally want to set the filter's "
+					+ "'forRecursion' property; otherwise files added deep into the "
+					+ "directory tree may not be detected");
+		}
 	}
 
 	/**
@@ -947,12 +958,19 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 		List<F> lsFiles = new ArrayList<>();
 		String remoteDirectory = buildRemotePath(directory, subDirectory);
 
-		F[] files = session.list(remoteDirectory);
-		boolean recursion = this.options.contains(Option.RECURSIVE);
+		F[] list = session.list(remoteDirectory);
+		List<F> files;
+		if (!this.filterAfterEnhancement) {
+			files = filterFiles(list);
+		}
+		else {
+			files = Arrays.asList(list);
+		}
 		if (!ObjectUtils.isEmpty(files)) {
-			for (F file : filterFiles(files)) {
+			for (F file : files) {
 				if (file != null) {
-					processFile(session, directory, subDirectory, lsFiles, recursion, file);
+					processFile(session, directory, subDirectory, lsFiles, this.options.contains(Option.RECURSIVE),
+							file);
 				}
 			}
 		}
@@ -974,29 +992,51 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 		return (this.filter != null) ? this.filter.filterFiles(files) : Arrays.asList(files);
 	}
 
+	protected final F filterFile(F file) {
+		if (this.filter.accept(file)) {
+			return file;
+		}
+		else {
+			return null;
+		}
+	}
+
 	private void processFile(Session<F> session, String directory, String subDirectory, List<F> lsFiles,
 			boolean recursion, F file) throws IOException {
 
-		String fileName = getFilename(file);
-		String fileSep = this.remoteFileTemplate.getRemoteFileSeparator();
-		boolean isDots = ".".equals(fileName)
-				|| "..".equals(fileName)
-				|| fileName.endsWith(fileSep + ".")
-				|| fileName.endsWith(fileSep + "..");
-		if (this.options.contains(Option.SUBDIRS) || !isDirectory(file)) {
-			if (recursion && StringUtils.hasText(subDirectory) && (!isDots || this.options.contains(Option.ALL))) {
-				lsFiles.add(enhanceNameWithSubDirectory(file, subDirectory));
+		F fileToAdd = file;
+		if (recursion && StringUtils.hasText(subDirectory)) {
+			fileToAdd = enhanceNameWithSubDirectory(file, subDirectory);
+		}
+		if (this.filterAfterEnhancement) {
+			if (!this.filter.accept(fileToAdd)) {
+				return;
 			}
+		}
+		String fileName = getFilename(fileToAdd);
+		final boolean isDirectory = isDirectory(file);
+		boolean isDots = hasDots(fileName);
+		if ((this.options.contains(Option.SUBDIRS) || !isDirectory)
+				&& (!isDots || this.options.contains(Option.ALL))) {
+
 			else if (this.options.contains(Option.ALL) || !isDots) {
 				lsFiles.add(file);
 			}
 		}
-		if (recursion && isDirectory(file) && !isDots) {
-			lsFiles.addAll(listFilesInRemoteDir(session, directory,
-					subDirectory + fileName + fileSep));
+
+		if (recursion && isDirectory && !isDots) {
+			lsFiles.addAll(listFilesInRemoteDir(session, directory, fileName +
+					this.remoteFileTemplate.getRemoteFileSeparator()));
 		}
 	}
 
+	private boolean hasDots(String fileName) {
+		String fileSeparator = this.remoteFileTemplate.getRemoteFileSeparator();
+		return ".".equals(fileName)
+				|| "..".equals(fileName)
+				|| fileName.endsWith(fileSeparator + ".")
+				|| fileName.endsWith(fileSeparator + "..");
+	}
 	protected final List<File> filterMputFiles(File[] files) {
 		if (files == null) {
 			return Collections.emptyList();

spring-integration-file/src/test/java/org/springframework/integration/file/filters/CompositeFileListFilterTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -30,7 +30,7 @@ import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
 
 /**
  * @author Iwein Fuld
@@ -56,7 +56,7 @@ public class CompositeFileListFilterTests {
 		List<File> returnedFiles = Collections.singletonList(fileMock);
 		when(fileFilterMock1.filterFiles(isA(File[].class))).thenReturn(returnedFiles);
 		when(fileFilterMock2.filterFiles(isA(File[].class))).thenReturn(returnedFiles);
-		assertThat(compositeFileFilter.filterFiles(new File[] { fileMock })).isEqualTo(returnedFiles);
+		assertThat(compositeFileFilter.filterFiles(new File[]{ fileMock })).isEqualTo(returnedFiles);
 		verify(fileFilterMock1).filterFiles(isA(File[].class));
 		verify(fileFilterMock2).filterFiles(isA(File[].class));
 		compositeFileFilter.close();
@@ -70,7 +70,7 @@ public class CompositeFileListFilterTests {
 		List<File> returnedFiles = Collections.singletonList(fileMock);
 		when(fileFilterMock1.filterFiles(isA(File[].class))).thenReturn(returnedFiles);
 		when(fileFilterMock2.filterFiles(isA(File[].class))).thenReturn(returnedFiles);
-		assertThat(compositeFileFilter.filterFiles(new File[] { fileMock })).isEqualTo(returnedFiles);
+		assertThat(compositeFileFilter.filterFiles(new File[]{ fileMock })).isEqualTo(returnedFiles);
 		verify(fileFilterMock1).filterFiles(isA(File[].class));
 		verify(fileFilterMock2).filterFiles(isA(File[].class));
 		compositeFileFilter.close();
@@ -84,7 +84,7 @@ public class CompositeFileListFilterTests {
 
 		when(fileFilterMock2.filterFiles(isA(File[].class))).thenReturn(new ArrayList<>());
 		when(fileFilterMock1.filterFiles(isA(File[].class))).thenReturn(new ArrayList<>());
-		assertThat(compositeFileFilter.filterFiles(new File[] { fileMock }).isEmpty()).isTrue();
+		assertThat(compositeFileFilter.filterFiles(new File[]{ fileMock }).isEmpty()).isTrue();
 		compositeFileFilter.close();
 	}
 
@@ -95,7 +95,7 @@ public class CompositeFileListFilterTests {
 		compositeFileFilter.addFilter(this.fileFilterMock2);
 		List<File> noFiles = new ArrayList<>();
 		when(this.fileFilterMock1.filterFiles(isA(File[].class))).thenReturn(noFiles);
-		assertThat(compositeFileFilter.filterFiles(new File[] { this.fileMock })).isEqualTo(noFiles);
+		assertThat(compositeFileFilter.filterFiles(new File[]{ this.fileMock })).isEqualTo(noFiles);
 
 		verify(fileFilterMock1).filterFiles(isA(File[].class));
 		verify(fileFilterMock2, never()).filterFiles(isA(File[].class));
@@ -108,17 +108,17 @@ public class CompositeFileListFilterTests {
 		CompositeFileListFilter<String> compo =
 				new CompositeFileListFilter<>(Collections.singletonList(new FileListFilter<String>() {
 
-			@Override
-			public List<String> filterFiles(String[] files) {
-				return Collections.emptyList();
-			}
+					@Override
+					public List<String> filterFiles(String[] files) {
+						return Collections.emptyList();
+					}
 
-			@Override
-			public boolean supportsSingleFileFiltering() {
-				return true;
-			}
+					@Override
+					public boolean supportsSingleFileFiltering() {
+						return true;
+					}
 
-		}));
+				}));
 		assertThatExceptionOfType(UnsupportedOperationException.class).isThrownBy(() -> compo.accept("foo"));
 		compo.close();
 	}
@@ -128,25 +128,32 @@ public class CompositeFileListFilterTests {
 		CompositeFileListFilter<String> compo =
 				new CompositeFileListFilter<>(Collections.singletonList(new FileListFilter<String>() {
 
-			@Override
-			public List<String> filterFiles(String[] files) {
-				return Collections.emptyList();
-			}
+					@Override
+					public List<String> filterFiles(String[] files) {
+						return Collections.emptyList();
+					}
 
-			@Override
-			public boolean supportsSingleFileFiltering() {
-				return true;
-			}
+					@Override
+					public boolean supportsSingleFileFiltering() {
+						return true;
+					}
 
-			@Override
-			public boolean accept(String file) {
-				return true;
-			}
 
-		}));
+					@Override
+					public boolean isForRecursion() {
+						return true;
+					}
+
+					@Override
+					public boolean accept(String file) {
+						return true;
+					}
+
+				}));
 		assertThat(compo.accept("foo")).isTrue();
 		compo.addFilter(s -> null);
 		assertThat(compo.supportsSingleFileFiltering()).isFalse();
+		assertThat(compo.isForRecursion()).isTrue();
 		compo.close();
 	}
 
@@ -155,6 +162,7 @@ public class CompositeFileListFilterTests {
 		CompositeFileListFilter<String> compo =
 				new CompositeFileListFilter<>(Collections.singletonList(s -> null));
 		assertThat(compo.supportsSingleFileFiltering()).isFalse();
+		assertThat(compo.isForRecursion()).isFalse();
 		compo.close();
 	}
 

spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2019 the original author or authors.
+ * Copyright 2013-2021 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.
@@ -24,7 +24,6 @@ import java.io.Flushable;
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.List;
-import java.util.concurrent.Callable;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
@@ -32,13 +31,15 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
 
 import org.springframework.integration.metadata.ConcurrentMetadataStore;
 import org.springframework.integration.metadata.SimpleMetadataStore;
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 3.0
  *
  */
@@ -69,26 +70,21 @@ public class PersistentAcceptOnceFileListFilterTests extends AcceptOnceFileListF
 		final FileSystemPersistentAcceptOnceFileListFilter filter =
 				new FileSystemPersistentAcceptOnceFileListFilter(store, "foo:");
 		final File file = File.createTempFile("foo", ".txt");
-		assertThat(filter.filterFiles(new File[] { file }).size()).isEqualTo(1);
+		assertThat(filter.filterFiles(new File[]{ file }).size()).isEqualTo(1);
 		String ts = store.get("foo:" + file.getAbsolutePath());
 		assertThat(ts).isEqualTo(String.valueOf(file.lastModified()));
-		assertThat(filter.filterFiles(new File[] { file }).size()).isEqualTo(0);
+		assertThat(filter.filterFiles(new File[]{ file }).size()).isEqualTo(0);
 		file.setLastModified(file.lastModified() + 5000L);
-		assertThat(filter.filterFiles(new File[] { file }).size()).isEqualTo(1);
+		assertThat(filter.filterFiles(new File[]{ file }).size()).isEqualTo(1);
 		ts = store.get("foo:" + file.getAbsolutePath());
 		assertThat(ts).isEqualTo(String.valueOf(file.lastModified()));
-		assertThat(filter.filterFiles(new File[] { file }).size()).isEqualTo(0);
+		assertThat(filter.filterFiles(new File[]{ file }).size()).isEqualTo(0);
 
 		suspend.set(true);
 		file.setLastModified(file.lastModified() + 5000L);
 
-		Future<Integer> result = Executors.newSingleThreadExecutor().submit(new Callable<Integer>() {
-
-			@Override
-			public Integer call() throws Exception {
-				return filter.filterFiles(new File[] {file}).size();
-			}
-		});
+		Future<Integer> result = Executors.newSingleThreadExecutor()
+				.submit(() -> filter.filterFiles(new File[]{ file }).size());
 		assertThat(latch2.await(10, TimeUnit.SECONDS)).isTrue();
 		store.put("foo:" + file.getAbsolutePath(), "43");
 		latch1.countDown();
@@ -102,8 +98,8 @@ public class PersistentAcceptOnceFileListFilterTests extends AcceptOnceFileListF
 	@Override
 	@Test
 	public void testRollback() {
-		AbstractPersistentAcceptOnceFileListFilter<String> filter = new AbstractPersistentAcceptOnceFileListFilter<String>(
-				new SimpleMetadataStore(), "rollback:") {
+		AbstractPersistentAcceptOnceFileListFilter<String> filter =
+				new AbstractPersistentAcceptOnceFileListFilter<>(new SimpleMetadataStore(), "rollback:") {
 
 					@Override
 					protected long modified(String file) {
@@ -114,6 +110,12 @@ public class PersistentAcceptOnceFileListFilterTests extends AcceptOnceFileListF
 					protected String fileName(String file) {
 						return file;
 					}
+
+					@Override
+					protected boolean isDirectory(String file) {
+						return false;
+					}
+
 				};
 		doTestRollback(filter);
 	}
@@ -122,7 +124,7 @@ public class PersistentAcceptOnceFileListFilterTests extends AcceptOnceFileListF
 	public void testRollbackFileSystem() throws Exception {
 		FileSystemPersistentAcceptOnceFileListFilter filter = new FileSystemPersistentAcceptOnceFileListFilter(
 				new SimpleMetadataStore(), "rollback:");
-		File[] files = new File[] {new File("foo"), new File("bar"), new File("baz")};
+		File[] files = new File[]{ new File("foo"), new File("bar"), new File("baz") };
 		List<File> passed = filter.filterFiles(files);
 		assertThat(passed.size()).isEqualTo(0);
 		for (File file : files) {
@@ -155,7 +157,7 @@ public class PersistentAcceptOnceFileListFilterTests extends AcceptOnceFileListF
 		class MS extends SimpleMetadataStore implements Flushable, Closeable {
 
 			@Override
-			public void flush() throws IOException {
+			public void flush() {
 				flushes.incrementAndGet();
 			}
 
@@ -176,7 +178,7 @@ public class PersistentAcceptOnceFileListFilterTests extends AcceptOnceFileListF
 		FileSystemPersistentAcceptOnceFileListFilter filter = new FileSystemPersistentAcceptOnceFileListFilter(
 				store, prefix);
 		final File file = File.createTempFile("foo", ".txt");
-		File[] files = new File[] { file };
+		File[] files = new File[]{ file };
 		List<File> passed = filter.filterFiles(files);
 		assertThat(Arrays.equals(files, passed.toArray())).isTrue();
 		filter.rollback(passed.get(0), passed);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2020 the original author or authors.
+ * Copyright 2016-2021 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.
@@ -32,6 +32,7 @@ import org.junit.jupiter.api.io.TempDir;
  * Abstract base class for tests requiring remote file servers, e.g. (S)FTP.
  *
  * @author Gary Russell
+ *
  * @since 4.3
  *
  */
@@ -48,6 +49,8 @@ public abstract class RemoteFileTestSupport {
 
 	protected static File localTemporaryFolder;
 
+	protected static File scratchTemporaryFolder;
+
 	protected volatile File sourceRemoteDirectory;
 
 	protected volatile File targetRemoteDirectory;
@@ -169,6 +172,14 @@ public abstract class RemoteFileTestSupport {
 		fos.close();
 	}
 
+	public static File getScratchTempFolder() {
+		if (scratchTemporaryFolder == null) {
+			scratchTemporaryFolder = new File(temporaryFolder.toFile().getAbsolutePath() + File.separator + "scratch");
+			scratchTemporaryFolder.mkdirs();
+		}
+		return scratchTemporaryFolder;
+	}
+
 	protected static File getRemoteTempFolder() {
 		if (remoteTemporaryFolder == null) {
 			remoteTemporaryFolder = new File(temporaryFolder.toFile().getAbsolutePath() + File.separator + "source");

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2021 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.
@@ -38,7 +38,7 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
@@ -390,6 +390,11 @@ public class StreamingInboundTests {
 			return file;
 		}
 
+		@Override
+		protected boolean isDirectory(String file) {
+			return false;
+		}
+
 	}
 
 }