Fix SftpStreamingMessageSourceTests for race

It turns out there is some race condition between tests
against the same source dir.

* Add `DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD` to clean
up the application context state after each test method

Author: artembilan

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

@@ -26,6 +26,8 @@
 import java.io.InputStream;
 import java.util.Arrays;
 import java.util.Comparator;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -43,8 +45,10 @@
 import org.springframework.integration.file.FileHeaders;
 import org.springframework.integration.file.filters.AcceptAllFileListFilter;
 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.session.SftpFileInfo;
 import org.springframework.integration.sftp.session.SftpRemoteFileTemplate;
 import org.springframework.integration.transformer.StreamTransformer;
@@ -63,7 +67,7 @@
  *
  */
 @RunWith(SpringRunner.class)
-@DirtiesContext
+@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD)
 public class SftpStreamingMessageSourceTests extends SftpTestSupport {
 
 	@Autowired
@@ -81,6 +85,9 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport {
 	@Autowired
 	private ApplicationContext context;
 
+	@Autowired
+	private ConcurrentMap<String, String> metadataMap;
+
 	@SuppressWarnings("unchecked")
 	@Test
 	public void testAllContents() {
@@ -111,6 +118,7 @@ public void testAllContents() {
 		this.adapter.stop();
 		this.source.setFileInfoJson(false);
 		this.data.purge(null);
+		this.metadataMap.clear();
 		this.adapter.start();
 		received = (Message<byte[]>) this.data.receive(10000);
 		assertNotNull(received);
@@ -186,12 +194,20 @@ public PollerMetadata defaultPoller() {
 			return pollerMetadata;
 		}
 
+		@Bean
+		public ConcurrentMap<String, String> metadataMap() {
+			return new ConcurrentHashMap<>();
+		}
+
+
 		@Bean
 		@InboundChannelAdapter(channel = "stream", autoStartup = "false")
 		public MessageSource<InputStream> sftpMessageSource() {
 			SftpStreamingMessageSource messageSource = new SftpStreamingMessageSource(template(),
 					Comparator.comparing(LsEntry::getFilename));
-			messageSource.setFilter(new AcceptAllFileListFilter<>());
+			messageSource.setFilter(
+					new SftpPersistentAcceptOnceFileListFilter(
+							new SimpleMetadataStore(metadataMap()), "testStreaming"));
 			messageSource.setRemoteDirectory("sftpSource/");
 			return messageSource;
 		}