diff --git a/file/src/main/java/org/apache/pulsar/io/file/FileListingThread.java b/file/src/main/java/org/apache/pulsar/io/file/FileListingThread.java
index 5958fc3748..10a3c862b9 100644
--- a/file/src/main/java/org/apache/pulsar/io/file/FileListingThread.java
+++ b/file/src/main/java/org/apache/pulsar/io/file/FileListingThread.java
@@ -44,6 +44,15 @@ public class FileListingThread extends Thread {
     private final AtomicLong queueLastUpdated = new AtomicLong(0L);
     private final Lock listingLock = new ReentrantLock();
     private final AtomicReference<FileFilter> fileFilterRef = new AtomicReference<>();
+
+    /**
+     * Files that have already been offered to the work queue and still exist on disk.
+     * Only accessed by this thread. Consulting the downstream queues instead is racy:
+     * a file is briefly in none of them while the consumer moves it between queues, and
+     * again while the cleanup thread renames/deletes it, so a listing pass in one of
+     * those windows would offer the same file twice.
+     */
+    private final Set<File> alreadyOffered = new HashSet<>();
     private final BlockingQueue<File> workQueue;
     private final BlockingQueue<File> inProcess;
     private final BlockingQueue<File> recentlyProcessed;
@@ -72,20 +81,37 @@ public void run() {
         while (true) {
             if ((queueLastUpdated.get() < System.currentTimeMillis() - pollingInterval) && listingLock.tryLock()) {
                 try {
+                    // Prune tracked files that are gone from disk (processed and then renamed or
+                    // deleted). This must happen before the listing snapshot: a file that
+                    // disappears between the prune and the listing cannot be in the listing, so
+                    // it can never be re-offered through a stale tracking entry.
+                    if (!keepOriginal) {
+                        alreadyOffered.removeIf(f -> !f.exists());
+                    }
+
                     final File directory = new File(inputDir);
                     final Set<File> listing = performListing(directory, fileFilterRef.get(), recurseDirs);
 
                     if (listing != null && !listing.isEmpty()) {
 
-                        // Remove any files that have been or are currently being processed.
-                        listing.removeAll(inProcess);
-                        if (!keepOriginal) {
-                            listing.removeAll(recentlyProcessed);
-                        }
-
-                        for (File f: listing) {
-                            if (!workQueue.contains(f)) {
-                                workQueue.offer(f);
+                        if (keepOriginal) {
+                            // Re-processing the same file is expected in keepFile mode: only
+                            // skip files that are currently queued or being processed.
+                            listing.removeAll(inProcess);
+                            for (File f: listing) {
+                                if (!workQueue.contains(f)) {
+                                    workQueue.offer(f);
+                                }
+                            }
+                        } else {
+                            // Offer each file exactly once for as long as it remains on disk.
+                            // The file stays tracked until the cleanup thread renames or
+                            // deletes it, which closes the windows where a file is on disk
+                            // but in none of the downstream queues.
+                            for (File f: listing) {
+                                if (alreadyOffered.add(f)) {
+                                    workQueue.offer(f);
+                                }
                             }
                         }
                         queueLastUpdated.set(System.currentTimeMillis());
diff --git a/file/src/test/java/org/apache/pulsar/io/file/ProcessedFileThreadTest.java b/file/src/test/java/org/apache/pulsar/io/file/ProcessedFileThreadTest.java
index 0b1f8b1584..6e4f015dfd 100644
--- a/file/src/test/java/org/apache/pulsar/io/file/ProcessedFileThreadTest.java
+++ b/file/src/test/java/org/apache/pulsar/io/file/ProcessedFileThreadTest.java
@@ -41,6 +41,32 @@ public class ProcessedFileThreadTest extends AbstractFileTest {
     private ProcessedFileThread cleanupThread;
     private FileSourceConfig fileConfig;
 
+    /**
+     * Waits (bounded) until every produced file has made it through the entire pipeline,
+     * including the final rename/delete performed by the cleanup thread. Checking only the
+     * queues is not enough: a file is briefly in none of them while it is handed from one
+     * thread to the next, and the last hop (disk rename/delete) happens after the file has
+     * already left the queues.
+     */
+    private void awaitProcessingComplete() throws InterruptedException {
+        long deadline = System.currentTimeMillis() + 60_000;
+        while (!processingComplete()) {
+            if (System.currentTimeMillis() >= deadline) {
+                fail("Pipeline did not drain within 60s: workQueue=" + workQueue.size()
+                        + ", inProcess=" + inProcess.size()
+                        + ", recentlyProcessed=" + recentlyProcessed.size()
+                        + ", files still on disk="
+                        + producedFiles.stream().filter(File::exists).count());
+            }
+            Thread.sleep(200);
+        }
+    }
+
+    private boolean processingComplete() {
+        return workQueue.isEmpty() && inProcess.isEmpty() && recentlyProcessed.isEmpty()
+                && producedFiles.stream().noneMatch(File::exists);
+    }
+
     @Test
     public final void singleFileTest() throws IOException {
 
@@ -186,10 +212,8 @@ public final void continuousRunTest() throws IOException {
             // Stop producing files
             generatorThread.halt();
 
-            // Let the consumer catch up
-            while (!workQueue.isEmpty() && !inProcess.isEmpty() && !recentlyProcessed.isEmpty()) {
-                Thread.sleep(2000);
-            }
+            // Let the pipeline finish processing every produced file
+            awaitProcessingComplete();
 
             // Make sure every single file was processed.
             for (File produced : producedFiles) {
@@ -241,10 +265,8 @@ public final void multipleConsumerTest() throws IOException {
             // Stop producing files
             generatorThread.halt();
 
-            // Let the consumer catch up
-            while (!workQueue.isEmpty() && !inProcess.isEmpty() && !recentlyProcessed.isEmpty()) {
-                Thread.sleep(2000);
-            }
+            // Let the pipeline finish processing every produced file
+            awaitProcessingComplete();
 
             // Make sure every single file was processed exactly once.
             for (File produced : producedFiles) {
@@ -293,10 +315,8 @@ public final void renameFileTest() throws IOException {
             // Stop producing files
             generatorThread.halt();
 
-            // Let the consumer catch up
-            while (!workQueue.isEmpty() && !inProcess.isEmpty() && !recentlyProcessed.isEmpty()) {
-                Thread.sleep(2000);
-            }
+            // Let the pipeline finish processing every produced file
+            awaitProcessingComplete();
 
 
             // Make sure every single file was processed.