diff --git a/supplier/file-supplier/pom.xml b/supplier/file-supplier/pom.xml
index b1ba628d..a2f89922 100644
--- a/supplier/file-supplier/pom.xml
+++ b/supplier/file-supplier/pom.xml
@@ -19,36 +19,29 @@
             <artifactId>spring-integration-file</artifactId>
         </dependency>
         <dependency>
-            <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-starter-integration</artifactId>
+            <groupId>org.springframework.cloud.fn</groupId>
+            <artifactId>metadata-store-common</artifactId>
+            <version>${revision}</version>
         </dependency>
         <dependency>
             <groupId>org.springframework.cloud.fn</groupId>
             <artifactId>file-common</artifactId>
             <version>${project.version}</version>
         </dependency>
+
         <dependency>
-            <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-starter-json</artifactId>
-            <optional>true</optional>
-        </dependency>
-        <dependency>
-            <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-starter-validation</artifactId>
-        </dependency>
-        <dependency>
-            <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-configuration-processor</artifactId>
-            <scope>provided</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-starter-test</artifactId>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-jdbc</artifactId>
             <scope>test</scope>
         </dependency>
         <dependency>
-            <groupId>io.projectreactor</groupId>
-            <artifactId>reactor-test</artifactId>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-jdbc</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.hsqldb</groupId>
+            <artifactId>hsqldb</artifactId>
             <scope>test</scope>
         </dependency>
     </dependencies>
diff --git a/supplier/file-supplier/src/main/java/org/springframework/cloud/fn/supplier/file/FileSupplierConfiguration.java b/supplier/file-supplier/src/main/java/org/springframework/cloud/fn/supplier/file/FileSupplierConfiguration.java
index 50b275e5..fe2b483b 100644
--- a/supplier/file-supplier/src/main/java/org/springframework/cloud/fn/supplier/file/FileSupplierConfiguration.java
+++ b/supplier/file-supplier/src/main/java/org/springframework/cloud/fn/supplier/file/FileSupplierConfiguration.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2020 the original author or authors.
+ * Copyright 2020-2022 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,6 +16,7 @@
 
 package org.springframework.cloud.fn.supplier.file;
 
+import java.io.File;
 import java.util.function.Supplier;
 
 import org.reactivestreams.Publisher;
@@ -38,6 +39,12 @@ import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.file.FileReadingMessageSource;
 import org.springframework.integration.file.dsl.FileInboundChannelAdapterSpec;
 import org.springframework.integration.file.dsl.Files;
+import org.springframework.integration.file.filters.ChainFileListFilter;
+import org.springframework.integration.file.filters.FileListFilter;
+import org.springframework.integration.file.filters.FileSystemPersistentAcceptOnceFileListFilter;
+import org.springframework.integration.file.filters.RegexPatternFileListFilter;
+import org.springframework.integration.file.filters.SimplePatternFileListFilter;
+import org.springframework.integration.metadata.ConcurrentMetadataStore;
 import org.springframework.messaging.Message;
 import org.springframework.util.StringUtils;
 
@@ -49,6 +56,8 @@ import org.springframework.util.StringUtils;
 @EnableConfigurationProperties({FileSupplierProperties.class, FileConsumerProperties.class})
 public class FileSupplierConfiguration {
 
+	private static final String METADATA_STORE_PREFIX = "local-file-system-metadata-";
+
 	private final FileSupplierProperties fileSupplierProperties;
 
 	private final FileConsumerProperties fileConsumerProperties;
@@ -60,22 +69,32 @@ public class FileSupplierConfiguration {
 
 	public FileSupplierConfiguration(FileSupplierProperties fileSupplierProperties,
 									FileConsumerProperties fileConsumerProperties) {
+
 		this.fileSupplierProperties = fileSupplierProperties;
 		this.fileConsumerProperties = fileConsumerProperties;
 	}
 
 	@Bean
-	public FileInboundChannelAdapterSpec fileMessageSource() {
-		final FileInboundChannelAdapterSpec fileInboundChannelAdapterSpec =
-				Files.inboundAdapter(this.fileSupplierProperties.getDirectory());
+	public ChainFileListFilter<File> filter(ConcurrentMetadataStore metadataStore) {
+		ChainFileListFilter<File> chainFilter = new ChainFileListFilter<>();
 		if (StringUtils.hasText(this.fileSupplierProperties.getFilenamePattern())) {
-			fileInboundChannelAdapterSpec.patternFilter(this.fileSupplierProperties.getFilenamePattern());
+			chainFilter.addFilter(new SimplePatternFileListFilter(this.fileSupplierProperties.getFilenamePattern()));
 		}
 		else if (this.fileSupplierProperties.getFilenameRegex() != null) {
-			fileInboundChannelAdapterSpec.regexFilter(this.fileSupplierProperties.getFilenameRegex().pattern());
+			chainFilter.addFilter(new RegexPatternFileListFilter(this.fileSupplierProperties.getFilenameRegex()));
 		}
-		fileInboundChannelAdapterSpec.preventDuplicates(this.fileSupplierProperties.isPreventDuplicates());
-		return fileInboundChannelAdapterSpec;
+
+		if (this.fileSupplierProperties.isPreventDuplicates()) {
+			chainFilter.addFilter(new FileSystemPersistentAcceptOnceFileListFilter(metadataStore, METADATA_STORE_PREFIX));
+		}
+
+		return chainFilter;
+	}
+
+	@Bean
+	public FileInboundChannelAdapterSpec fileMessageSource(FileListFilter<File> fileListFilter) {
+		return Files.inboundAdapter(this.fileSupplierProperties.getDirectory())
+				.filter(fileListFilter);
 	}
 
 	@Bean
@@ -86,7 +105,7 @@ public class FileSupplierConfiguration {
 				.subscribeOn(Schedulers.boundedElastic())
 				.repeatWhenEmpty(it -> it.delayElements(this.fileSupplierProperties.getDelayWhenEmpty()))
 				.repeat()
-				.doOnSubscribe(s -> this.fileMessageSource.start());
+				.doOnRequest(r -> this.fileMessageSource.start());
 	}
 
 	@Bean
@@ -106,4 +125,5 @@ public class FileSupplierConfiguration {
 			return () -> Flux.from(fileReadingFlow());
 		}
 	}
+
 }
diff --git a/supplier/file-supplier/src/test/java/org/springframework/cloud/fn/supplier/file/DefaultFileSupplierTests.java b/supplier/file-supplier/src/test/java/org/springframework/cloud/fn/supplier/file/DefaultFileSupplierTests.java
index 234e5fd8..e1068243 100644
--- a/supplier/file-supplier/src/test/java/org/springframework/cloud/fn/supplier/file/DefaultFileSupplierTests.java
+++ b/supplier/file-supplier/src/test/java/org/springframework/cloud/fn/supplier/file/DefaultFileSupplierTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2020 the original author or authors.
+ * Copyright 2020-2022 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,16 +19,22 @@ package org.springframework.cloud.fn.supplier.file;
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.util.List;
 
 import org.junit.jupiter.api.Test;
-import reactor.core.publisher.Flux;
-import reactor.test.StepVerifier;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.integration.file.FileHeaders;
 import org.springframework.integration.file.FileReadingMessageSource;
+import org.springframework.integration.jdbc.metadata.JdbcMetadataStore;
+import org.springframework.integration.metadata.ConcurrentMetadataStore;
+import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.messaging.Message;
+import org.springframework.test.context.TestPropertySource;
+
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -37,15 +43,21 @@ import static org.assertj.core.api.Assertions.assertThat;
  * @author Artem Bilan
  * @author Soby Chacko
  */
+@TestPropertySource(properties = "metadata.store.type = jdbc")
 public class DefaultFileSupplierTests extends AbstractFileSupplierTests {
 
 	@Autowired
 	@Qualifier("fileMessageSource")
 	private FileReadingMessageSource fileMessageSource;
 
+	@Autowired
+	private ConcurrentMetadataStore metadataStore;
+
+	@Autowired
+	private JdbcTemplate jdbcTemplate;
+
 	@Test
 	public void testBasicFlow() throws IOException {
-
 		Path firstFile = tempDir.resolve("first.file");
 		Files.write(firstFile, "first.file".getBytes());
 
@@ -83,6 +95,16 @@ public class DefaultFileSupplierTests extends AbstractFileSupplierTests {
 		stepVerifier.verify();
 
 		assertThat(this.fileMessageSource.isRunning()).isTrue();
+
+		assertThat(this.metadataStore).isInstanceOf(JdbcMetadataStore.class);
+
+		List<String> metadataStoreContent =
+				jdbcTemplate.queryForList("SELECT metadata_key FROM int_metadata_store", String.class);
+
+		assertThat(metadataStoreContent).hasSize(2);
+		assertThat(metadataStoreContent.get(0)).startsWith("local-file-system-metadata-");
+		assertThat(metadataStoreContent.get(0)).endsWith("first.file");
+		assertThat(metadataStoreContent.get(1)).endsWith("test.file");
 	}
 
 }
diff --git a/supplier/s3-supplier/README.adoc b/supplier/s3-supplier/README.adoc
index c01ccc61..973a03b4 100644
--- a/supplier/s3-supplier/README.adoc
+++ b/supplier/s3-supplier/README.adoc
@@ -23,7 +23,7 @@ Once injected, you can use the `get` method of the `Supplier` to invoke it and t
 All configuration properties are prefixed with `s3.supplier`.
 There are also properties that need to be used with the prefix `s3.common` and `file.consumer`.
 
-For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierProperties.java[AwsS3upplierProperties],
+For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierProperties.java[AwsS3SupplierProperties],
 link:../../common/file-common/src/main/java/org/springframework/cloud/fn/common/file/FileConsumerProperties.java[FileConsumerProperties], and
 link:../../common/aws-s3-common/src/main/java/org/springframework/cloud/fn/common/aws/s3/AmazonS3Properties.java[AmazonS3Properties].