diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileInboundChannelAdapterParser.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileInboundChannelAdapterParser.java
index 836998b2e6..d1e283f52c 100644
--- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileInboundChannelAdapterParser.java
+++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileInboundChannelAdapterParser.java
@@ -25,6 +25,7 @@ import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.config.ExpressionFactoryBean;
 import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
 import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.springframework.integration.file.filters.AbstractPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.remote.synchronizer.InboundFileSynchronizer;
 import org.springframework.util.StringUtils;
@@ -36,6 +37,7 @@ import org.springframework.util.StringUtils;
  * @author Mark Fisher
  * @author Gary Russell
  * @author Artem Bilan
+ *
  * @since 2.0
  */
 public abstract class AbstractRemoteFileInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser {
@@ -59,8 +61,10 @@ public abstract class AbstractRemoteFileInboundChannelAdapterParser extends Abst
 		String remoteFileSeparator = element.getAttribute("remote-file-separator");
 		synchronizerBuilder.addPropertyValue("remoteFileSeparator", remoteFileSeparator);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(synchronizerBuilder, element, "temporary-file-suffix");
+
 		FileParserUtils.configureFilter(synchronizerBuilder, element, parserContext,
-				getSimplePatternFileListFilterClass(), getRegexPatternFileListFilterClass());
+				getSimplePatternFileListFilterClass(), getRegexPatternFileListFilterClass(),
+				getPersistentAcceptOnceFileListFilterClass());
 
 		// build the MessageSource
 		BeanDefinitionBuilder messageSourceBuilder =
@@ -94,4 +98,6 @@ public abstract class AbstractRemoteFileInboundChannelAdapterParser extends Abst
 
 	protected abstract Class<? extends FileListFilter<?>> getRegexPatternFileListFilterClass();
 
+	protected abstract Class<? extends AbstractPersistentAcceptOnceFileListFilter<?>> getPersistentAcceptOnceFileListFilterClass();
+
 }
diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileStreamingInboundChannelAdapterParser.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileStreamingInboundChannelAdapterParser.java
index 6b6b338c1f..5727e93433 100644
--- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileStreamingInboundChannelAdapterParser.java
+++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileStreamingInboundChannelAdapterParser.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2017 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.
@@ -25,6 +25,7 @@ import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
 import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
 import org.springframework.integration.core.MessageSource;
+import org.springframework.integration.file.filters.AbstractPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.remote.RemoteFileOperations;
 import org.springframework.util.StringUtils;
@@ -33,6 +34,8 @@ import org.springframework.util.StringUtils;
  * Abstract base class for parsing remote file streaming inbound channel adapters.
  *
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 4.3
  */
 public abstract class AbstractRemoteFileStreamingInboundChannelAdapterParser
@@ -56,7 +59,7 @@ public abstract class AbstractRemoteFileStreamingInboundChannelAdapterParser
 		String remoteFileSeparator = element.getAttribute("remote-file-separator");
 		messageSourceBuilder.addPropertyValue("remoteFileSeparator", remoteFileSeparator);
 		FileParserUtils.configureFilter(messageSourceBuilder, element, parserContext,
-				getSimplePatternFileListFilterClass(), getRegexPatternFileListFilterClass());
+				getSimplePatternFileListFilterClass(), getRegexPatternFileListFilterClass(), getPersistentAcceptOnceFileListFilterClass());
 
 		String comparator = element.getAttribute("comparator");
 		if (StringUtils.hasText(comparator)) {
@@ -74,4 +77,6 @@ public abstract class AbstractRemoteFileStreamingInboundChannelAdapterParser
 
 	protected abstract Class<? extends FileListFilter<?>> getRegexPatternFileListFilterClass();
 
+	protected abstract Class<? extends AbstractPersistentAcceptOnceFileListFilter<?>> getPersistentAcceptOnceFileListFilterClass();
+
 }
diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileParserUtils.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileParserUtils.java
index 1e622457a7..ef605b5c5e 100644
--- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileParserUtils.java
+++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileParserUtils.java
@@ -20,12 +20,16 @@ import org.w3c.dom.Element;
 
 import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.support.ManagedList;
 import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
 import org.springframework.integration.file.DefaultFileNameGenerator;
+import org.springframework.integration.file.filters.AbstractPersistentAcceptOnceFileListFilter;
+import org.springframework.integration.file.filters.CompositeFileListFilter;
 import org.springframework.integration.file.filters.ExpressionFileListFilter;
 import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.remote.RemoteFileOperations;
+import org.springframework.integration.metadata.SimpleMetadataStore;
 import org.springframework.util.StringUtils;
 
 /**
@@ -34,6 +38,7 @@ import org.springframework.util.StringUtils;
  * @author David Turanski
  * @author Gary Russell
  * @author Artem Bilan
+ *
  * @since 3.0
  *
  */
@@ -93,7 +98,8 @@ public final class FileParserUtils {
 	}
 
 	static void configureFilter(BeanDefinitionBuilder synchronizerBuilder, Element element, ParserContext parserContext,
-			Class<? extends FileListFilter<?>> patternClass, Class<? extends FileListFilter<?>> regexClass) {
+			Class<? extends FileListFilter<?>> patternClass, Class<? extends FileListFilter<?>> regexClass,
+			Class<? extends AbstractPersistentAcceptOnceFileListFilter<?>> persistentAcceptOnceFileListFilterClass) {
 		String filter = element.getAttribute("filter");
 		String filterExpression = element.getAttribute("filter-expression");
 		String fileNamePattern = element.getAttribute("filename-pattern");
@@ -132,16 +138,44 @@ public final class FileParserUtils {
 				synchronizerBuilder.addPropertyValue("filter", expressionFilterBeanDefinition);
 			}
 			else if (hasFileNamePattern) {
-				BeanDefinitionBuilder filterBuilder = BeanDefinitionBuilder.genericBeanDefinition(patternClass);
-				filterBuilder.addConstructorArgValue(fileNamePattern);
-				synchronizerBuilder.addPropertyValue("filter", filterBuilder.getBeanDefinition());
+				BeanDefinition patternFilter =
+						BeanDefinitionBuilder.genericBeanDefinition(patternClass)
+								.addConstructorArgValue(fileNamePattern)
+								.getBeanDefinition();
+
+				composeFilters(synchronizerBuilder, persistentAcceptOnceFileListFilterClass, patternFilter);
 			}
-			else if (hasFileNameRegex) {
-				BeanDefinitionBuilder filterBuilder = BeanDefinitionBuilder.genericBeanDefinition(regexClass);
-				filterBuilder.addConstructorArgValue(fileNameRegex);
-				synchronizerBuilder.addPropertyValue("filter", filterBuilder.getBeanDefinition());
+			else {
+				BeanDefinition regexFilter = BeanDefinitionBuilder.genericBeanDefinition(regexClass)
+						.addConstructorArgValue(fileNameRegex)
+						.getBeanDefinition();
+
+				composeFilters(synchronizerBuilder, persistentAcceptOnceFileListFilterClass, regexFilter);
 			}
 		}
 	}
 
+	private static void composeFilters(BeanDefinitionBuilder synchronizerBuilder,
+			Class<? extends AbstractPersistentAcceptOnceFileListFilter<?>> persistentAcceptOnceFileListFilterClass,
+			BeanDefinition filter) {
+		BeanDefinition persistentFilter =
+				BeanDefinitionBuilder.genericBeanDefinition(persistentAcceptOnceFileListFilterClass)
+						.addConstructorArgValue(
+								BeanDefinitionBuilder
+										.genericBeanDefinition(SimpleMetadataStore.class)
+										.getBeanDefinition())
+						.addConstructorArgValue("remoteFileMessageSource")
+						.getBeanDefinition();
+
+		ManagedList<BeanDefinition> filters = new ManagedList<>();
+		filters.add(filter);
+		filters.add(persistentFilter);
+
+		BeanDefinition compositeFilterDefinition =
+				BeanDefinitionBuilder.genericBeanDefinition(CompositeFileListFilter.class)
+						.addConstructorArgValue(filters)
+						.getBeanDefinition();
+		synchronizerBuilder.addPropertyValue("filter", compositeFilterDefinition);
+	}
+
 }
diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/RemoteFileInboundChannelAdapterSpec.java b/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/RemoteFileInboundChannelAdapterSpec.java
index 5a8811cfc3..4b3dd2dab5 100644
--- a/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/RemoteFileInboundChannelAdapterSpec.java
+++ b/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/RemoteFileInboundChannelAdapterSpec.java
@@ -27,7 +27,6 @@ import org.springframework.integration.dsl.ComponentsRegistration;
 import org.springframework.integration.dsl.MessageSourceSpec;
 import org.springframework.integration.expression.FunctionExpression;
 import org.springframework.integration.file.FileReadingMessageSource;
-import org.springframework.integration.file.filters.CompositeFileListFilter;
 import org.springframework.integration.file.filters.ExpressionFileListFilter;
 import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer;
@@ -51,8 +50,6 @@ public abstract class RemoteFileInboundChannelAdapterSpec<F, S extends RemoteFil
 
 	protected final AbstractInboundFileSynchronizer<F> synchronizer;
 
-	private CompositeFileListFilter<F> filter;
-
 	private ExpressionFileListFilter<F> expressionFileListFilter;
 
 	protected RemoteFileInboundChannelAdapterSpec(AbstractInboundFileSynchronizer<F> synchronizer) {
@@ -172,19 +169,7 @@ public abstract class RemoteFileInboundChannelAdapterSpec<F, S extends RemoteFil
 	 * @return the spec.
 	 */
 	public S filter(FileListFilter<F> filter) {
-		if (this.filter == null) {
-			if (filter instanceof CompositeFileListFilter) {
-				this.filter = (CompositeFileListFilter<F>) filter;
-			}
-			else {
-				this.filter = new CompositeFileListFilter<>();
-				this.filter.addFilter(filter);
-			}
-			this.synchronizer.setFilter(this.filter);
-		}
-		else {
-			this.filter.addFilter(filter);
-		}
+		this.synchronizer.setFilter(filter);
 		return _this();
 	}
 
diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/RemoteFileStreamingInboundChannelAdapterSpec.java b/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/RemoteFileStreamingInboundChannelAdapterSpec.java
index be120550f3..64b5214888 100644
--- a/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/RemoteFileStreamingInboundChannelAdapterSpec.java
+++ b/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/RemoteFileStreamingInboundChannelAdapterSpec.java
@@ -25,7 +25,6 @@ import org.springframework.integration.dsl.ComponentsRegistration;
 import org.springframework.integration.dsl.MessageSourceSpec;
 import org.springframework.integration.expression.FunctionExpression;
 import org.springframework.integration.file.FileReadingMessageSource;
-import org.springframework.integration.file.filters.CompositeFileListFilter;
 import org.springframework.integration.file.filters.ExpressionFileListFilter;
 import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.remote.AbstractRemoteFileStreamingMessageSource;
@@ -49,8 +48,6 @@ public abstract class RemoteFileStreamingInboundChannelAdapterSpec<F,
 		extends MessageSourceSpec<S, MS>
 		implements ComponentsRegistration {
 
-	private CompositeFileListFilter<F> filter;
-
 	private ExpressionFileListFilter<F> expressionFileListFilter;
 
 	/**
@@ -101,19 +98,7 @@ public abstract class RemoteFileStreamingInboundChannelAdapterSpec<F,
 	 * @return the spec.
 	 */
 	public S filter(FileListFilter<F> filter) {
-		if (this.filter == null) {
-			if (filter instanceof CompositeFileListFilter) {
-				this.filter = (CompositeFileListFilter<F>) filter;
-			}
-			else {
-				this.filter = new CompositeFileListFilter<F>();
-				this.filter.addFilter(filter);
-			}
-			this.target.setFilter(this.filter);
-		}
-		else {
-			this.filter.addFilter(filter);
-		}
+		this.target.setFilter(filter);
 		return _this();
 	}
 
diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java
index db2474f14f..4214b518c6 100644
--- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java
+++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java
@@ -46,6 +46,8 @@ import org.springframework.util.Assert;
  * referencing a remote file.
  *
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 4.3
  *
  */
@@ -111,6 +113,10 @@ public abstract class AbstractRemoteFileStreamingMessageSource<F>
 	 * @param filter the file list filter.
 	 */
 	public void setFilter(FileListFilter<F> filter) {
+		doSetFilter(filter);
+	}
+
+	protected final void doSetFilter(FileListFilter<F> filter) {
 		this.filter = filter;
 	}
 
diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java
index fe467897bf..efbe25de10 100644
--- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java
+++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java
@@ -152,11 +152,14 @@ public abstract class AbstractInboundFileSynchronizer<F>
 
 	/**
 	 * Specify an expression that evaluates to the full path to the remote directory.
-	 *
 	 * @param remoteDirectoryExpression The remote directory expression.
 	 * @since 4.2
 	 */
 	public void setRemoteDirectoryExpression(Expression remoteDirectoryExpression) {
+		doSetRemoteDirectoryExpression(remoteDirectoryExpression);
+	}
+
+	protected final void doSetRemoteDirectoryExpression(Expression remoteDirectoryExpression) {
 		Assert.notNull(remoteDirectoryExpression, "'remoteDirectoryExpression' must not be null");
 		this.remoteDirectoryExpression = remoteDirectoryExpression;
 	}
@@ -166,6 +169,10 @@ public abstract class AbstractInboundFileSynchronizer<F>
 	 * @param filter the file list filter.
 	 */
 	public void setFilter(FileListFilter<F> filter) {
+		doSetFilter(filter);
+	}
+
+	protected final void doSetFilter(FileListFilter<F> filter) {
 		this.filter = filter;
 	}
 
diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParser.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParser.java
index b463bd4771..006f172dfe 100644
--- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParser.java
+++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParser.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2017 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,8 +17,10 @@
 package org.springframework.integration.ftp.config;
 
 import org.springframework.integration.file.config.AbstractRemoteFileInboundChannelAdapterParser;
+import org.springframework.integration.file.filters.AbstractPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.remote.synchronizer.InboundFileSynchronizer;
+import org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.ftp.filters.FtpRegexPatternFileListFilter;
 import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter;
 import org.springframework.integration.ftp.inbound.FtpInboundFileSynchronizer;
@@ -29,6 +31,8 @@ import org.springframework.integration.ftp.inbound.FtpInboundFileSynchronizingMe
  *
  * @author Mark Fisher
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 2.0
  */
 public class FtpInboundChannelAdapterParser extends AbstractRemoteFileInboundChannelAdapterParser {
@@ -53,4 +57,9 @@ public class FtpInboundChannelAdapterParser extends AbstractRemoteFileInboundCha
 		return FtpRegexPatternFileListFilter.class;
 	}
 
+	@Override
+	protected Class<? extends AbstractPersistentAcceptOnceFileListFilter<?>> getPersistentAcceptOnceFileListFilterClass() {
+		return FtpPersistentAcceptOnceFileListFilter.class;
+	}
+
 }
diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParser.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParser.java
index 06f17f2bb9..51c1e4869d 100644
--- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParser.java
+++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParser.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2017 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.
@@ -18,8 +18,10 @@ package org.springframework.integration.ftp.config;
 
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.file.config.AbstractRemoteFileStreamingInboundChannelAdapterParser;
+import org.springframework.integration.file.filters.AbstractPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.remote.RemoteFileOperations;
+import org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.ftp.filters.FtpRegexPatternFileListFilter;
 import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter;
 import org.springframework.integration.ftp.inbound.FtpStreamingMessageSource;
@@ -27,6 +29,8 @@ import org.springframework.integration.ftp.session.FtpRemoteFileTemplate;
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 4.3
  *
  */
@@ -52,4 +56,9 @@ public class FtpStreamingInboundChannelAdapterParser extends AbstractRemoteFileS
 		return FtpRegexPatternFileListFilter.class;
 	}
 
+	@Override
+	protected Class<? extends AbstractPersistentAcceptOnceFileListFilter<?>> getPersistentAcceptOnceFileListFilterClass() {
+		return FtpPersistentAcceptOnceFileListFilter.class;
+	}
+
 }
diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/dsl/FtpInboundChannelAdapterSpec.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/dsl/FtpInboundChannelAdapterSpec.java
index d7d8c787ec..1c739edece 100644
--- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/dsl/FtpInboundChannelAdapterSpec.java
+++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/dsl/FtpInboundChannelAdapterSpec.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2016 the original author or authors.
+ * Copyright 2014-2017 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.
@@ -22,17 +22,22 @@ import java.util.Comparator;
 import org.apache.commons.net.ftp.FTPFile;
 
 import org.springframework.integration.file.dsl.RemoteFileInboundChannelAdapterSpec;
+import org.springframework.integration.file.filters.CompositeFileListFilter;
+import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.remote.session.SessionFactory;
+import org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.ftp.filters.FtpRegexPatternFileListFilter;
 import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter;
 import org.springframework.integration.ftp.inbound.FtpInboundFileSynchronizer;
 import org.springframework.integration.ftp.inbound.FtpInboundFileSynchronizingMessageSource;
+import org.springframework.integration.metadata.SimpleMetadataStore;
 
 /**
  * A {@link RemoteFileInboundChannelAdapterSpec} for a
  * {@link FtpInboundFileSynchronizingMessageSource}.
  *
  * @author Artem Bilan
+ *
  * @since 5.0
  */
 public class FtpInboundChannelAdapterSpec
@@ -52,7 +57,7 @@ public class FtpInboundChannelAdapterSpec
 	 */
 	@Override
 	public FtpInboundChannelAdapterSpec patternFilter(String pattern) {
-		return filter(new FtpSimplePatternFileListFilter(pattern));
+		return filter(composeFilters(new FtpSimplePatternFileListFilter(pattern)));
 	}
 
 	/**
@@ -63,7 +68,15 @@ public class FtpInboundChannelAdapterSpec
 	 */
 	@Override
 	public FtpInboundChannelAdapterSpec regexFilter(String regex) {
-		return filter(new FtpRegexPatternFileListFilter(regex));
+		return filter(composeFilters(new FtpRegexPatternFileListFilter(regex)));
+	}
+
+	@SuppressWarnings("unchecked")
+	private CompositeFileListFilter<FTPFile> composeFilters(FileListFilter<FTPFile> fileListFilter) {
+		CompositeFileListFilter<FTPFile> compositeFileListFilter = new CompositeFileListFilter<>();
+		compositeFileListFilter.addFilters(fileListFilter,
+				new FtpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "ftpMessageSource"));
+		return compositeFileListFilter;
 	}
 
 }
diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/dsl/FtpStreamingInboundChannelAdapterSpec.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/dsl/FtpStreamingInboundChannelAdapterSpec.java
index c8b6dde1d5..1e4ff2ccc3 100644
--- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/dsl/FtpStreamingInboundChannelAdapterSpec.java
+++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/dsl/FtpStreamingInboundChannelAdapterSpec.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2016 the original author or authors.
+ * Copyright 2014-2017 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,11 +21,15 @@ import java.util.Comparator;
 import org.apache.commons.net.ftp.FTPFile;
 
 import org.springframework.integration.file.dsl.RemoteFileStreamingInboundChannelAdapterSpec;
+import org.springframework.integration.file.filters.CompositeFileListFilter;
+import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.remote.AbstractFileInfo;
 import org.springframework.integration.file.remote.RemoteFileTemplate;
+import org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.ftp.filters.FtpRegexPatternFileListFilter;
 import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter;
 import org.springframework.integration.ftp.inbound.FtpStreamingMessageSource;
+import org.springframework.integration.metadata.SimpleMetadataStore;
 
 /**
  * A {@link RemoteFileStreamingInboundChannelAdapterSpec} for a
@@ -51,7 +55,7 @@ public class FtpStreamingInboundChannelAdapterSpec
 	 */
 	@Override
 	public FtpStreamingInboundChannelAdapterSpec patternFilter(String pattern) {
-		return filter(new FtpSimplePatternFileListFilter(pattern));
+		return filter(composeFilters(new FtpSimplePatternFileListFilter(pattern)));
 	}
 
 	/**
@@ -62,7 +66,15 @@ public class FtpStreamingInboundChannelAdapterSpec
 	 */
 	@Override
 	public FtpStreamingInboundChannelAdapterSpec regexFilter(String regex) {
-		return filter(new FtpRegexPatternFileListFilter(regex));
+		return filter(composeFilters(new FtpRegexPatternFileListFilter(regex)));
+	}
+
+	@SuppressWarnings("unchecked")
+	private CompositeFileListFilter<FTPFile> composeFilters(FileListFilter<FTPFile> fileListFilter) {
+		CompositeFileListFilter<FTPFile> compositeFileListFilter = new CompositeFileListFilter<>();
+		compositeFileListFilter.addFilters(fileListFilter,
+				new FtpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "ftpStreamingMessageSource"));
+		return compositeFileListFilter;
 	}
 
 }
diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundFileSynchronizer.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundFileSynchronizer.java
index 0df12793ae..0d55caf66f 100644
--- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundFileSynchronizer.java
+++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundFileSynchronizer.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -22,6 +22,8 @@ import org.springframework.expression.common.LiteralExpression;
 import org.springframework.integration.file.remote.session.Session;
 import org.springframework.integration.file.remote.session.SessionFactory;
 import org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer;
+import org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter;
+import org.springframework.integration.metadata.SimpleMetadataStore;
 
 /**
  * An implementation of {@link AbstractInboundFileSynchronizer} for FTP.
@@ -36,12 +38,12 @@ public class FtpInboundFileSynchronizer extends AbstractInboundFileSynchronizer<
 
 	/**
 	 * Create a synchronizer with the {@link SessionFactory} used to acquire {@link Session} instances.
-	 *
 	 * @param sessionFactory The session factory.
 	 */
 	public FtpInboundFileSynchronizer(SessionFactory<FTPFile> sessionFactory) {
 		super(sessionFactory);
-		setRemoteDirectoryExpression(new LiteralExpression(null));
+		doSetRemoteDirectoryExpression(new LiteralExpression(null));
+		doSetFilter(new FtpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "ftpMessageSource"));
 	}
 
 	@Override
diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSource.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSource.java
index f03051c1d5..e77d35d458 100644
--- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSource.java
+++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSource.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2017 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.
@@ -26,12 +26,16 @@ import org.apache.commons.net.ftp.FTPFile;
 import org.springframework.integration.file.remote.AbstractFileInfo;
 import org.springframework.integration.file.remote.AbstractRemoteFileStreamingMessageSource;
 import org.springframework.integration.file.remote.RemoteFileTemplate;
+import org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.ftp.session.FtpFileInfo;
+import org.springframework.integration.metadata.SimpleMetadataStore;
 
 /**
  * Message source for streaming FTP remote file contents.
  *
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 4.3
  *
  */
@@ -42,7 +46,7 @@ public class FtpStreamingMessageSource extends AbstractRemoteFileStreamingMessag
 	 * @param template the template.
 	 */
 	public FtpStreamingMessageSource(RemoteFileTemplate<FTPFile> template) {
-		super(template, null);
+		this(template, null);
 	}
 
 	/**
@@ -55,6 +59,7 @@ public class FtpStreamingMessageSource extends AbstractRemoteFileStreamingMessag
 	public FtpStreamingMessageSource(RemoteFileTemplate<FTPFile> template,
 			Comparator<AbstractFileInfo<FTPFile>> comparator) {
 		super(template, comparator);
+		doSetFilter(new FtpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "ftpStreamingMessageSource"));
 	}
 
 	@Override
diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/FtpParserInboundTests-context.xml b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/FtpParserInboundTests-context.xml
index eb05818f3b..11e05cab65 100644
--- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/FtpParserInboundTests-context.xml
+++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/FtpParserInboundTests-context.xml
@@ -3,12 +3,9 @@
 	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 	   xmlns:ftp="http://www.springframework.org/schema/integration/ftp"
 	   xmlns:int="http://www.springframework.org/schema/integration"
-	   xmlns:context="http://www.springframework.org/schema/context"
-	   xmlns:ftps="http://www.springframework.org/schema/integration/ftps"
 	   xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
-	    http://www.springframework.org/schema/integration/ftps http://www.springframework.org/schema/integration/ftp/spring-integration-ftps.xsd
         http://www.springframework.org/schema/integration  http://www.springframework.org/schema/integration/spring-integration.xsd
-       http://www.springframework.org/schema/integration/ftp http://www.springframework.org/schema/integration/ftp/spring-integration-ftp.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
+       http://www.springframework.org/schema/integration/ftp http://www.springframework.org/schema/integration/ftp/spring-integration-ftp.xsd">
 
 	<bean id="ftpSessionFactory" class="org.springframework.integration.ftp.session.DefaultFtpSessionFactory">
 		<property name="host" value="localhost"/>
diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java
index 8db146114f..926afb88a1 100644
--- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java
+++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -16,10 +16,12 @@
 
 package org.springframework.integration.ftp.config;
 
+import static org.hamcrest.Matchers.instanceOf;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
@@ -27,6 +29,8 @@ import static org.mockito.Mockito.when;
 import java.lang.reflect.Method;
 import java.util.Collection;
 import java.util.Comparator;
+import java.util.Iterator;
+import java.util.Set;
 import java.util.concurrent.PriorityBlockingQueue;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -39,9 +43,11 @@ import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.ApplicationContext;
 import org.springframework.expression.Expression;
 import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
+import org.springframework.integration.file.filters.CompositeFileListFilter;
 import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.remote.session.CachingSessionFactory;
 import org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer;
+import org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter;
 import org.springframework.integration.ftp.inbound.FtpInboundFileSynchronizer;
 import org.springframework.integration.ftp.inbound.FtpInboundFileSynchronizingMessageSource;
@@ -59,6 +65,7 @@ import org.springframework.util.ReflectionUtils;
  * @author Mark Fisher
  * @author Gary Russell
  * @author Gunnar Hillert
+ * @author Artem Bilan
  */
 @ContextConfiguration
 @RunWith(SpringJUnit4ClassRunner.class)
@@ -103,8 +110,16 @@ public class FtpInboundChannelAdapterParserTests {
 		String remoteFileSeparator = (String) TestUtils.getPropertyValue(fisync, "remoteFileSeparator");
 		assertNotNull(remoteFileSeparator);
 		assertEquals("", remoteFileSeparator);
-		FtpSimplePatternFileListFilter filter = (FtpSimplePatternFileListFilter) TestUtils.getPropertyValue(fisync, "filter");
+
+		FileListFilter<?> filter = TestUtils.getPropertyValue(fisync, "filter", FileListFilter.class);
 		assertNotNull(filter);
+		assertThat(filter, instanceOf(CompositeFileListFilter.class));
+		Set<?> fileFilters = TestUtils.getPropertyValue(filter, "fileFilters", Set.class);
+
+		Iterator<?> filtersIterator = fileFilters.iterator();
+		assertThat(filtersIterator.next(), instanceOf(FtpSimplePatternFileListFilter.class));
+		assertThat(filtersIterator.next(), instanceOf(FtpPersistentAcceptOnceFileListFilter.class));
+
 		Object sessionFactory = TestUtils.getPropertyValue(fisync, "remoteFileTemplate.sessionFactory");
 		assertTrue(DefaultFtpSessionFactory.class.isAssignableFrom(sessionFactory.getClass()));
 		FileListFilter<?> acceptAllFilter = context.getBean("acceptAllFilter", FileListFilter.class);
diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParserTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParserTests.java
index 5866ff43eb..1e85d55d44 100644
--- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParserTests.java
+++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParserTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2017 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.
@@ -26,13 +26,19 @@ import static org.junit.Assert.assertThat;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
+import java.util.Iterator;
+import java.util.Set;
+
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
 import org.springframework.beans.factory.FactoryBean;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
+import org.springframework.integration.file.filters.CompositeFileListFilter;
+import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.remote.session.CachingSessionFactory;
+import org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter;
 import org.springframework.integration.ftp.inbound.FtpStreamingMessageSource;
 import org.springframework.integration.ftp.session.DefaultFtpSessionFactory;
@@ -45,6 +51,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
  */
 @ContextConfiguration
 @RunWith(SpringJUnit4ClassRunner.class)
@@ -71,7 +78,16 @@ public class FtpStreamingInboundChannelAdapterParserTests {
 
 		assertNotNull(TestUtils.getPropertyValue(source, "comparator"));
 		assertThat(TestUtils.getPropertyValue(source, "remoteFileSeparator", String.class), equalTo("X"));
-		assertThat(TestUtils.getPropertyValue(source, "filter"), instanceOf(FtpSimplePatternFileListFilter.class));
+
+		FileListFilter<?> filter = TestUtils.getPropertyValue(source, "filter", FileListFilter.class);
+		assertNotNull(filter);
+		assertThat(filter, instanceOf(CompositeFileListFilter.class));
+		Set<?> fileFilters = TestUtils.getPropertyValue(filter, "fileFilters", Set.class);
+
+		Iterator<?> filtersIterator = fileFilters.iterator();
+		assertThat(filtersIterator.next(), instanceOf(FtpSimplePatternFileListFilter.class));
+		assertThat(filtersIterator.next(), instanceOf(FtpPersistentAcceptOnceFileListFilter.class));
+
 		assertSame(this.csf, TestUtils.getPropertyValue(source, "remoteFileTemplate.sessionFactory"));
 		assertEquals(31, TestUtils.getPropertyValue(source, "maxFetchSize"));
 	}
diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizerTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizerTests.java
index 11a05ac1da..918f797182 100644
--- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizerTests.java
+++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizerTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -76,9 +76,8 @@ public class FtpInboundRemoteFileSystemSynchronizerTests {
 	public void cleanup() {
 		File file = new File("test");
 		if (file.exists()) {
-			String[] files = file.list();
-			for (String fileName : files) {
-				new File(file, fileName).delete();
+			for (File f : file.listFiles()) {
+				f.delete();
 			}
 			file.delete();
 		}
@@ -127,7 +126,7 @@ public class FtpInboundRemoteFileSystemSynchronizerTests {
 		localFileListFilter.addFilter(localAcceptOnceFilter);
 		ms.setLocalFilter(localFileListFilter);
 		ms.afterPropertiesSet();
-		Message<File> atestFile =  ms.receive();
+		Message<File> atestFile = ms.receive();
 		assertNotNull(atestFile);
 		assertEquals("A.TEST.a", atestFile.getPayload().getName());
 		// The test remote files are created with the current timestamp + 1 day.
@@ -135,13 +134,13 @@ public class FtpInboundRemoteFileSystemSynchronizerTests {
 
 		assertEquals("A.TEST.a", atestFile.getHeaders().get(FileHeaders.FILENAME));
 
-		Message<File> btestFile =  ms.receive();
+		Message<File> btestFile = ms.receive();
 		assertNotNull(btestFile);
 		assertEquals("B.TEST.a", btestFile.getPayload().getName());
 		// The test remote files are created with the current timestamp + 1 day.
 		assertThat(atestFile.getPayload().lastModified(), Matchers.greaterThan(System.currentTimeMillis()));
 
-		Message<File> nothing =  ms.receive();
+		Message<File> nothing = ms.receive();
 		assertNull(nothing);
 
 		// two times because on the third receive (above) the internal queue will be empty, so it will attempt
@@ -155,7 +154,7 @@ public class FtpInboundRemoteFileSystemSynchronizerTests {
 		new File("test/A.TEST.a").delete();
 		new File("test/B.TEST.a").delete();
 		// the remote filter should prevent a re-fetch
-		nothing =  ms.receive();
+		nothing = ms.receive();
 		assertNull(nothing);
 
 		ms.stop();
@@ -164,6 +163,35 @@ public class FtpInboundRemoteFileSystemSynchronizerTests {
 	}
 
 
+	@Test
+	public void testSyncRemoteFileOnlyOnceByDefault() throws Exception {
+		File localDirectory = new File("test");
+		localDirectory.mkdir();
+
+		TestFtpSessionFactory ftpSessionFactory = new TestFtpSessionFactory();
+		ftpSessionFactory.setUsername("kermit");
+		ftpSessionFactory.setPassword("frog");
+		FtpInboundFileSynchronizer synchronizer = spy(new FtpInboundFileSynchronizer(ftpSessionFactory));
+		synchronizer.setRemoteDirectory("remote-test-dir");
+
+		synchronizer.setBeanFactory(mock(BeanFactory.class));
+		synchronizer.afterPropertiesSet();
+
+		synchronizer.synchronizeToLocalDirectory(localDirectory);
+
+
+		File[] files = localDirectory.listFiles();
+		assertEquals(3, files.length);
+
+		for (File f : files) {
+			f.delete();
+		}
+
+		synchronizer.synchronizeToLocalDirectory(localDirectory);
+
+		assertEquals(0, localDirectory.list().length);
+	}
+
 	public static class TestFtpSessionFactory extends AbstractFtpSessionFactory<FTPClient> {
 
 		private final Collection<FTPFile> ftpFiles = new ArrayList<FTPFile>();
@@ -198,7 +226,8 @@ public class FtpInboundRemoteFileSystemSynchronizerTests {
 					when(ftpClient.retrieveFile(Mockito.eq("remote-test-dir/" + fileName),
 							Mockito.any(OutputStream.class))).thenReturn(true);
 				}
-				when(ftpClient.listFiles("remote-test-dir")).thenReturn(ftpFiles.toArray(new FTPFile[ftpFiles.size()]));
+				when(ftpClient.listFiles("remote-test-dir"))
+						.thenReturn(ftpFiles.toArray(new FTPFile[ftpFiles.size()]));
 				when(ftpClient.deleteFile(Mockito.anyString())).thenReturn(true);
 				return ftpClient;
 			}
@@ -206,6 +235,7 @@ public class FtpInboundRemoteFileSystemSynchronizerTests {
 				throw new RuntimeException("Failed to create mock client", e);
 			}
 		}
+
 	}
 
 }
diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java
index 913978825f..5908898da1 100644
--- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java
+++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java
@@ -39,6 +39,7 @@ import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
 import org.springframework.integration.file.FileHeaders;
+import org.springframework.integration.file.filters.AcceptAllFileListFilter;
 import org.springframework.integration.file.remote.FileInfo;
 import org.springframework.integration.file.remote.session.SessionFactory;
 import org.springframework.integration.ftp.FtpTestSupport;
@@ -128,6 +129,7 @@ public class FtpStreamingMessageSourceTests extends FtpTestSupport {
 		public MessageSource<InputStream> ftpMessageSource() {
 			FtpStreamingMessageSource messageSource = new FtpStreamingMessageSource(template(),
 					Comparator.comparing(FileInfo::getFilename));
+			messageSource.setFilter(new AcceptAllFileListFilter<>());
 			messageSource.setRemoteDirectory("ftpSource/");
 			return messageSource;
 		}
diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpInboundChannelAdapterParser.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpInboundChannelAdapterParser.java
index ae9a6bf708..959b841205 100644
--- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpInboundChannelAdapterParser.java
+++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpInboundChannelAdapterParser.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2017 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,8 +17,10 @@
 package org.springframework.integration.sftp.config;
 
 import org.springframework.integration.file.config.AbstractRemoteFileInboundChannelAdapterParser;
+import org.springframework.integration.file.filters.AbstractPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.remote.synchronizer.InboundFileSynchronizer;
+import org.springframework.integration.sftp.filters.SftpPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.sftp.filters.SftpRegexPatternFileListFilter;
 import org.springframework.integration.sftp.filters.SftpSimplePatternFileListFilter;
 import org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizer;
@@ -29,6 +31,8 @@ import org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizing
  *
  * @author Mark Fisher
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 2.0
  */
 public class SftpInboundChannelAdapterParser extends AbstractRemoteFileInboundChannelAdapterParser {
@@ -53,4 +57,9 @@ public class SftpInboundChannelAdapterParser extends AbstractRemoteFileInboundCh
 		return SftpRegexPatternFileListFilter.class;
 	}
 
+	@Override
+	protected Class<? extends AbstractPersistentAcceptOnceFileListFilter<?>> getPersistentAcceptOnceFileListFilterClass() {
+		return SftpPersistentAcceptOnceFileListFilter.class;
+	}
+
 }
diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpStreamingInboundChannelAdapterParser.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpStreamingInboundChannelAdapterParser.java
index b58af344d6..3efe8b0585 100644
--- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpStreamingInboundChannelAdapterParser.java
+++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpStreamingInboundChannelAdapterParser.java
@@ -18,8 +18,10 @@ package org.springframework.integration.sftp.config;
 
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.file.config.AbstractRemoteFileStreamingInboundChannelAdapterParser;
+import org.springframework.integration.file.filters.AbstractPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.remote.RemoteFileOperations;
+import org.springframework.integration.sftp.filters.SftpPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.sftp.filters.SftpRegexPatternFileListFilter;
 import org.springframework.integration.sftp.filters.SftpSimplePatternFileListFilter;
 import org.springframework.integration.sftp.inbound.SftpStreamingMessageSource;
@@ -27,6 +29,8 @@ import org.springframework.integration.sftp.session.SftpRemoteFileTemplate;
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 4.3
  *
  */
@@ -52,4 +56,9 @@ public class SftpStreamingInboundChannelAdapterParser extends AbstractRemoteFile
 		return SftpRegexPatternFileListFilter.class;
 	}
 
+	@Override
+	protected Class<? extends AbstractPersistentAcceptOnceFileListFilter<?>> getPersistentAcceptOnceFileListFilterClass() {
+		return SftpPersistentAcceptOnceFileListFilter.class;
+	}
+
 }
diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/dsl/SftpInboundChannelAdapterSpec.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/dsl/SftpInboundChannelAdapterSpec.java
index 1a59deb30b..8304a68381 100644
--- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/dsl/SftpInboundChannelAdapterSpec.java
+++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/dsl/SftpInboundChannelAdapterSpec.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2014 the original author or authors.
+ * Copyright 2014-2017 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.
@@ -20,7 +20,11 @@ import java.io.File;
 import java.util.Comparator;
 
 import org.springframework.integration.file.dsl.RemoteFileInboundChannelAdapterSpec;
+import org.springframework.integration.file.filters.CompositeFileListFilter;
+import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.remote.session.SessionFactory;
+import org.springframework.integration.metadata.SimpleMetadataStore;
+import org.springframework.integration.sftp.filters.SftpPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.sftp.filters.SftpRegexPatternFileListFilter;
 import org.springframework.integration.sftp.filters.SftpSimplePatternFileListFilter;
 import org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizer;
@@ -50,7 +54,7 @@ public class SftpInboundChannelAdapterSpec
 	 */
 	@Override
 	public SftpInboundChannelAdapterSpec patternFilter(String pattern) {
-		return filter(new SftpSimplePatternFileListFilter(pattern));
+		return filter(composeFilters(new SftpSimplePatternFileListFilter(pattern)));
 	}
 
 	/**
@@ -60,7 +64,16 @@ public class SftpInboundChannelAdapterSpec
 	 */
 	@Override
 	public SftpInboundChannelAdapterSpec regexFilter(String regex) {
-		return filter(new SftpRegexPatternFileListFilter(regex));
+		return filter(composeFilters(new SftpRegexPatternFileListFilter(regex)));
+	}
+
+	@SuppressWarnings("unchecked")
+	private CompositeFileListFilter<ChannelSftp.LsEntry> composeFilters(FileListFilter<ChannelSftp.LsEntry>
+			fileListFilter) {
+		CompositeFileListFilter<ChannelSftp.LsEntry> compositeFileListFilter = new CompositeFileListFilter<>();
+		compositeFileListFilter.addFilters(fileListFilter,
+				new SftpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "sftpMessageSource"));
+		return compositeFileListFilter;
 	}
 
 }
diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/dsl/SftpStreamingInboundChannelAdapterSpec.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/dsl/SftpStreamingInboundChannelAdapterSpec.java
index b5f3452aeb..05ffc4b783 100644
--- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/dsl/SftpStreamingInboundChannelAdapterSpec.java
+++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/dsl/SftpStreamingInboundChannelAdapterSpec.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2017 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.
@@ -19,8 +19,12 @@ package org.springframework.integration.sftp.dsl;
 import java.util.Comparator;
 
 import org.springframework.integration.file.dsl.RemoteFileStreamingInboundChannelAdapterSpec;
+import org.springframework.integration.file.filters.CompositeFileListFilter;
+import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.remote.AbstractFileInfo;
 import org.springframework.integration.file.remote.RemoteFileTemplate;
+import org.springframework.integration.metadata.SimpleMetadataStore;
+import org.springframework.integration.sftp.filters.SftpPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.sftp.filters.SftpRegexPatternFileListFilter;
 import org.springframework.integration.sftp.filters.SftpSimplePatternFileListFilter;
 import org.springframework.integration.sftp.inbound.SftpStreamingMessageSource;
@@ -34,7 +38,7 @@ import com.jcraft.jsch.ChannelSftp.LsEntry;
  */
 public class SftpStreamingInboundChannelAdapterSpec
 		extends RemoteFileStreamingInboundChannelAdapterSpec<LsEntry, SftpStreamingInboundChannelAdapterSpec,
-			SftpStreamingMessageSource> {
+		SftpStreamingMessageSource> {
 
 	SftpStreamingInboundChannelAdapterSpec(RemoteFileTemplate<LsEntry> remoteFileTemplate,
 			Comparator<AbstractFileInfo<LsEntry>> comparator) {
@@ -49,7 +53,8 @@ public class SftpStreamingInboundChannelAdapterSpec
 	 */
 	@Override
 	public SftpStreamingInboundChannelAdapterSpec patternFilter(String pattern) {
-		return filter(new SftpSimplePatternFileListFilter(pattern));	}
+		return filter(composeFilters(new SftpSimplePatternFileListFilter(pattern)));
+	}
 
 	/**
 	 * Specify a regular expression to match remote files (e.g. '[0-9].*.txt').
@@ -59,7 +64,15 @@ public class SftpStreamingInboundChannelAdapterSpec
 	 */
 	@Override
 	public SftpStreamingInboundChannelAdapterSpec regexFilter(String regex) {
-		return filter(new SftpRegexPatternFileListFilter(regex));
+		return filter(composeFilters(new SftpRegexPatternFileListFilter(regex)));
+	}
+
+	@SuppressWarnings("unchecked")
+	private CompositeFileListFilter<LsEntry> composeFilters(FileListFilter<LsEntry> fileListFilter) {
+		CompositeFileListFilter<LsEntry> compositeFileListFilter = new CompositeFileListFilter<>();
+		compositeFileListFilter.addFilters(fileListFilter,
+				new SftpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "sftpStreamingMessageSource"));
+		return compositeFileListFilter;
 	}
 
 }
diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundFileSynchronizer.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundFileSynchronizer.java
index a6f177aa75..9c7a782b0a 100644
--- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundFileSynchronizer.java
+++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundFileSynchronizer.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -18,6 +18,8 @@ package org.springframework.integration.sftp.inbound;
 
 import org.springframework.integration.file.remote.session.SessionFactory;
 import org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer;
+import org.springframework.integration.metadata.SimpleMetadataStore;
+import org.springframework.integration.sftp.filters.SftpPersistentAcceptOnceFileListFilter;
 
 import com.jcraft.jsch.ChannelSftp.LsEntry;
 
@@ -27,15 +29,21 @@ import com.jcraft.jsch.ChannelSftp.LsEntry;
  * @author Josh Long
  * @author Oleg Zhurakousky
  * @author Mark Fisher
+ * @author Artem Bilan
+ *
  * @since 2.0
  */
 public class SftpInboundFileSynchronizer extends AbstractInboundFileSynchronizer<LsEntry> {
 
+	/**
+	 * Create a synchronizer with the {@code SessionFactory} used to acquire {@code Session} instances.
+	 * @param sessionFactory The session factory.
+	 */
 	public SftpInboundFileSynchronizer(SessionFactory<LsEntry> sessionFactory) {
 		super(sessionFactory);
+		doSetFilter(new SftpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "sftpMessageSource"));
 	}
 
-
 	@Override
 	protected boolean isFile(LsEntry file) {
 		return (file != null && file.getAttrs() != null && !file.getAttrs().isDir() && !file.getAttrs().isLink());
diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSource.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSource.java
index a75b70638b..5be1960c08 100644
--- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSource.java
+++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSource.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2017 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,6 +24,8 @@ import java.util.List;
 import org.springframework.integration.file.remote.AbstractFileInfo;
 import org.springframework.integration.file.remote.AbstractRemoteFileStreamingMessageSource;
 import org.springframework.integration.file.remote.RemoteFileTemplate;
+import org.springframework.integration.metadata.SimpleMetadataStore;
+import org.springframework.integration.sftp.filters.SftpPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.sftp.session.SftpFileInfo;
 
 import com.jcraft.jsch.ChannelSftp.LsEntry;
@@ -32,6 +34,8 @@ import com.jcraft.jsch.ChannelSftp.LsEntry;
  * Message source for streaming SFTP remote file contents.
  *
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 4.3
  *
  */
@@ -42,7 +46,7 @@ public class SftpStreamingMessageSource extends AbstractRemoteFileStreamingMessa
 	 * @param template the template.
 	 */
 	public SftpStreamingMessageSource(RemoteFileTemplate<LsEntry> template) {
-		super(template, null);
+		this(template, null);
 	}
 
 	/**
@@ -55,6 +59,7 @@ public class SftpStreamingMessageSource extends AbstractRemoteFileStreamingMessa
 	public SftpStreamingMessageSource(RemoteFileTemplate<LsEntry> template,
 			Comparator<AbstractFileInfo<LsEntry>> comparator) {
 		super(template, comparator);
+		doSetFilter(new SftpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "sftpStreamingMessageSource"));
 	}
 
 	@Override
diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpStreamingInboundChannelAdapterParserTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpStreamingInboundChannelAdapterParserTests.java
index 48cc6dec0a..791dd44c73 100644
--- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpStreamingInboundChannelAdapterParserTests.java
+++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpStreamingInboundChannelAdapterParserTests.java
@@ -26,14 +26,20 @@ import static org.junit.Assert.assertThat;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
+import java.util.Iterator;
+import java.util.Set;
+
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
 import org.springframework.beans.factory.FactoryBean;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
+import org.springframework.integration.file.filters.CompositeFileListFilter;
 import org.springframework.integration.file.filters.ExpressionFileListFilter;
+import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.remote.session.CachingSessionFactory;
+import org.springframework.integration.sftp.filters.SftpPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.sftp.filters.SftpSimplePatternFileListFilter;
 import org.springframework.integration.sftp.inbound.SftpStreamingMessageSource;
 import org.springframework.integration.sftp.session.DefaultSftpSessionFactory;
@@ -76,7 +82,16 @@ public class SftpStreamingInboundChannelAdapterParserTests {
 
 		assertNotNull(TestUtils.getPropertyValue(source, "comparator"));
 		assertThat(TestUtils.getPropertyValue(source, "remoteFileSeparator", String.class), equalTo("X"));
-		assertThat(TestUtils.getPropertyValue(source, "filter"), instanceOf(SftpSimplePatternFileListFilter.class));
+
+		FileListFilter<?> filter = TestUtils.getPropertyValue(source, "filter", FileListFilter.class);
+		assertNotNull(filter);
+		assertThat(filter, instanceOf(CompositeFileListFilter.class));
+		Set<?> fileFilters = TestUtils.getPropertyValue(filter, "fileFilters", Set.class);
+
+		Iterator<?> filtersIterator = fileFilters.iterator();
+		assertThat(filtersIterator.next(), instanceOf(SftpSimplePatternFileListFilter.class));
+		assertThat(filtersIterator.next(), instanceOf(SftpPersistentAcceptOnceFileListFilter.class));
+
 		assertSame(this.csf, TestUtils.getPropertyValue(source, "remoteFileTemplate.sessionFactory"));
 		assertEquals(31, TestUtils.getPropertyValue(source, "maxFetchSize"));
 
diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java
index 1efd40376a..6534cdf075 100644
--- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java
+++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java
@@ -38,6 +38,7 @@ import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
 import org.springframework.integration.file.FileHeaders;
+import org.springframework.integration.file.filters.AcceptAllFileListFilter;
 import org.springframework.integration.file.remote.FileInfo;
 import org.springframework.integration.file.remote.session.SessionFactory;
 import org.springframework.integration.scheduling.PollerMetadata;
@@ -129,6 +130,7 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport {
 		public MessageSource<InputStream> sftpMessageSource() {
 			SftpStreamingMessageSource messageSource = new SftpStreamingMessageSource(template(),
 					Comparator.comparing(FileInfo::getFilename));
+			messageSource.setFilter(new AcceptAllFileListFilter<>());
 			messageSource.setRemoteDirectory("sftpSource/");
 			return messageSource;
 		}
diff --git a/src/reference/asciidoc/ftp.adoc b/src/reference/asciidoc/ftp.adoc
index 0ae991aeea..58967df2ab 100644
--- a/src/reference/asciidoc/ftp.adoc
+++ b/src/reference/asciidoc/ftp.adoc
@@ -256,7 +256,7 @@ Unlike outbound gateways and adapters where the root object of the SpEL Evaluati
 So, the root object of the SpEL Evaluation Context is the original name of the remote file (String).
 
 The inbound channel adapter first retrieves the file to a local directory and then emits each file according to the poller configuration.
-Starting with _version 5.0_ you can now limit the number of files fetched from the FTP server when new file retrievals are needed.
+Starting with _version 5.0_, you can now limit the number of files fetched from the FTP server when new file retrievals are needed.
 This can be beneficial when the target files are very large and/or when running in a clustered system with a persistent file list filter discussed below.
 Use `max-fetch-size` for this purpose; a negative value (default) means no limit and all matching files will be retrieved.
 
@@ -283,6 +283,10 @@ This filter matches on the filename and the remote modified time.
 Since _version 4.0_, this filter requires a `ConcurrentMetadataStore`.
 When used with a shared data store (such as `Redis` with the `RedisMetadataStore`) this allows filter keys to be shared across multiple application or server instances.
 
+Starting with _version 5.0_, the `FtpPersistentAcceptOnceFileListFilter` with in-memory `SimpleMetadataStore` is applied by default for the `FtpInboundFileSynchronizer`.
+This filter is also applied together with the `regex` or `pattern` option in the XML configuration as well as via `FtpInboundChannelAdapterSpec` in Java DSL.
+Any other use-cases can be reached via `CompositeFileListFilter` (or `ChainFileListFilter`).
+
 The above discussion refers to filtering the files before retrieving them.
 Once the files have been retrieved, an additional filter is applied to the files on the file system.
 By default, this is an `AcceptOnceFileListFilter` which, as discussed, retains state in memory and does not consider the file's modified time.
@@ -291,7 +295,7 @@ Unless your application removes files after processing, the adapter will re-proc
 Also, if you configure the `filter` to use a `FtpPersistentAcceptOnceFileListFilter`, and the remote file timestamp changes (causing it to be re-fetched), the default local filter will not allow this new file to be processed.
 
 Use the `local-filter` attribute to configure the behavior of the local file system filter.
-Starting with _verion 4.3.8_, a `FileSystemPersistentAcceptOnceFileListFilter` is configured by default.
+Starting with _version 4.3.8_, a `FileSystemPersistentAcceptOnceFileListFilter` is configured by default.
 This filter stores the accepted file names and modified timestamp in an instance of the `MetadataStore` strategy (<<metadata-store>>), and will detect changes to the local file modified time.
 The default `MetadataStore` is a `SimpleMetadataStore` which stores state in memory.
 
@@ -528,12 +532,11 @@ See <<file-splitter>> and <<stream-transformer>> for more information about thes
 
 Only one of `filename-pattern`, `filename-regex`, `filter` or `filter-expression` is allowed.
 
-IMPORTANT: Unlike the non-streaming inbound channel adapter, this adapter does not prevent duplicates by default.
-If you do not delete the remote file (e.g. using an outbound gateway with an rm command) and you wish to prevent the
-file being processed again, you can configure an `FtpPersistentFileListFilter` in the `filter` attribute.
-If you don't actually want to persist the state, an in-memory `SimpleMetadataStore` can be used with the filter.
-If you wish to use a filename pattern (or regex) as well, use a `CompositeFileListFilter`.
-The java configuration below shows one technique to remove the remote file after processing.
+IMPORTANT: Starting with _version 5.0_, by default, the `FtpStreamingMessageSource` adapter prevents duplicates for remote files via `FtpPersistentAcceptOnceFileListFilter` based on the in-memory `SimpleMetadataStore`.
+This filter is also applied by default together with the filename pattern (or regex) as well.
+If there is a requirement to allow duplicates, the `AcceptAllFileListFilter` can be used.
+Any other use-cases can be reached via `CompositeFileListFilter` (or `ChainFileListFilter`).
+The java configuration below shows one technique to remove the remote file after processing, avoiding duplicates.
 
 Use the `max-fetch-size` attribute to limit the number of files fetched on each poll when a fetch is necessary; set to 1 and use a persistent filter when running in a clustered environment.
 
@@ -560,10 +563,9 @@ public class FtpJavaApplication {
     @Bean
     @InboundChannelAdapter(channel = "stream")
     public MessageSource<InputStream> ftpMessageSource() {
-        FtpStreamingMessageSource messageSource = new FtpStreamingMessageSource(template(), null);
+        FtpStreamingMessageSource messageSource = new FtpStreamingMessageSource(template());
         messageSource.setRemoteDirectory("ftpSource/");
-        messageSource.setFilter(new FtpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(),
-                           "streaming"));
+        messageSource.setFilter(new AcceptAllFileListFilter<>());
         messageSource.setMaxFetchSize(1);
         return messageSource;
     }
diff --git a/src/reference/asciidoc/sftp.adoc b/src/reference/asciidoc/sftp.adoc
index 3e180e5a53..59386e9fc6 100644
--- a/src/reference/asciidoc/sftp.adoc
+++ b/src/reference/asciidoc/sftp.adoc
@@ -345,6 +345,10 @@ This filter matches on the filename and the remote modified time.
 Since _version 4.0_, this filter requires a `ConcurrentMetadataStore`.
 When used with a shared data store (such as `Redis` with the `RedisMetadataStore`) this allows filter keys to be shared across multiple application or server instances.
 
+Starting with _version 5.0_, the `SftpPersistentAcceptOnceFileListFilter` with in-memory `SimpleMetadataStore` is applied by default for the `SftpInboundFileSynchronizer`.
+This filter is also applied together with the `regex` or `pattern` option in the XML configuration as well as via `FtpInboundChannelAdapterSpec` in Java DSL.
+Any other use-cases can be reached via `CompositeFileListFilter` (or `ChainFileListFilter`).
+
 The above discussion refers to filtering the files before retrieving them.
 Once the files have been retrieved, an additional filter is applied to the files on the file system.
 By default, this is an`AcceptOnceFileListFilter` which, as discussed, retains state in memory and does not consider the file's modified time.
@@ -353,7 +357,7 @@ Unless your application removes files after processing, the adapter will re-proc
 Also, if you configure the `filter` to use a `FtpPersistentAcceptOnceFileListFilter`, and the remote file timestamp changes (causing it to be re-fetched), the default local filter will not allow this new file to be processed.
 
 Use the `local-filter` attribute to configure the behavior of the local file system filter.
-Starting with _verion 4.3.8_, a `FileSystemPersistentAcceptOnceFileListFilter` is configured by default.
+Starting with _version 4.3.8_, a `FileSystemPersistentAcceptOnceFileListFilter` is configured by default.
 This filter stores the accepted file names and modified timestamp in an instance of the `MetadataStore` strategy (<<metadata-store>>), and will detect changes to the local file modified time.
 The default `MetadataStore` is a `SimpleMetadataStore` which stores state in memory.
 
@@ -567,12 +571,11 @@ See <<file-splitter>> and <<stream-transformer>> for more information about thes
 
 Only one of `filename-pattern`, `filename-regex`, `filter` or `filter-expression` is allowed.
 
-IMPORTANT: Unlike the non-streaming inbound channel adapter, this adapter does not prevent duplicates by default.
-If you do not delete the remote file (e.g. using an outbound gateway with an rm command) and you wish to prevent the
-file being processed again, you can configure an `SftpPersistentFileListFilter` in the `filter` attribute.
-If you don't actually want to persist the state, an in-memory `SimpleMetadataStore` can be used with the filter.
-If you wish to use a filename pattern (or regex) as well, use a `CompositeFileListFilter`.
-The java configuration below shows one technique to remove the remote file after processing.
+IMPORTANT: Starting with _version 5.0_, by default, the `SftpStreamingMessageSource` adapter prevents duplicates for remote files via `SftpPersistentAcceptOnceFileListFilter` based on the in-memory `SimpleMetadataStore`.
+This filter is also applied by default together with the filename pattern (or regex) as well.
+If there is a requirement to allow duplicates, the `AcceptAllFileListFilter` can be used.
+Any other use-cases can be reached via `CompositeFileListFilter` (or `ChainFileListFilter`).
+The java configuration below shows one technique to remove the remote file after processing, avoiding duplicates.
 
 Use the `max-fetch-size` attribute to limit the number of files fetched on each poll when a fetch is necessary; set to 1 and use a persistent filter when running in a clustered environment.
 
@@ -599,10 +602,9 @@ public class SftpJavaApplication {
     @Bean
     @InboundChannelAdapter(channel = "stream")
     public MessageSource<InputStream> ftpMessageSource() {
-        SftpStreamingMessageSource messageSource = new SftpStreamingMessageSource(template(), null);
+        SftpStreamingMessageSource messageSource = new SftpStreamingMessageSource(template());
         messageSource.setRemoteDirectory("sftpSource/");
-        messageSource.setFilter(new SftpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(),
-                           "streaming"));
+        messageSource.setFilter(new AcceptAllFileListFilter<>());
         messageSource.setMaxFetchSize(1);
         return messageSource;
     }
diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc
index 3c72f76318..b472699bf3 100644
--- a/src/reference/asciidoc/whats-new.adoc
+++ b/src/reference/asciidoc/whats-new.adoc
@@ -97,17 +97,17 @@ The file outbound channel adapter (`FileWritingMessageHandler`) now supports the
 
 The inbound channel adapters now have a property `max-fetch-size` which is used to limit the number of files fetched during a poll when there are no files currently in the local directory.
 They also are configured with a `FileSystemPersistentAcceptOnceFileListFilter` in the `local-filter` by default.
-See <<ftp-inbound>> and <<sftp-inbound>> for more information.
 
 The regex and pattern filters can now be configured to always pass directories.
 This can be useful when using recursion in the outbound gateways.
-See <<ftp-outbound-gateway>> and <<sftp-outbound-gateway>> for more information.
+
+All the Inbound Channel Adapters (streaming and synchronization-based) now use an appropriate `AbstractPersistentAcceptOnceFileListFilter` implementation by default to prevent remote files duplicate downloads.
 
 The FTP and SFTP outbound gateways now support the `REPLACE_IF_MODIFIED` `FileExistsMode` when fetching remote files.
-See <<ftp-outbound-gateway>> and <<sftp-outbound-gateway>> for more information.
 
 The (S)FTP streaming inbound channel adapters now add remote file information in a message header.
-See <<ftp-streaming>> and <<sftp-streaming>> for more information.
+
+See <<ftp>> and <<sftp>> for more information.
 
 The FTP and SFTP outbound channel adapters, as well as `PUT` command of the outbound gateways, now support `InputStream` as `payload`, too.