From 2bd60a2f6aeb70d7d28de6b96be872960171f949 Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Thu, 9 Jul 2026 11:52:06 -0700 Subject: [PATCH 1/2] [fix][io] Fix duplicate file offers and flaky tests in file connector ProcessedFileThreadTest (renameFileTest, continuousRunTest) failed intermittently in CI with 'offer wanted 1 time but was 2'. The root cause is a race in FileListingThread, not the tests: a file is briefly in none of the tracking queues while FileConsumerThread moves it from workQueue to inProcess, and again while ProcessedFileThread takes it from recentlyProcessed and renames/deletes it on disk. A listing pass in either window re-offers the file, causing duplicate processing in production as well. Fix the race at the source: when keepFile=false, track offered files in a listing-thread-private set and only offer a file once for as long as it remains on disk. Entries are pruned before each listing snapshot once the cleanup thread has renamed or deleted the file, so a stale entry can never suppress a legitimate new file with the same name. keepFile=true keeps its intentional re-processing semantics. Also fix two bugs in the tests themselves: the drain loops exited when ANY queue was empty (&&) instead of when ALL were empty, and nothing waited for the final rename/delete to reach disk before asserting on file existence. Replace both with a bounded await that covers the full pipeline. --- .../pulsar/io/file/FileListingThread.java | 44 +++++++++++++++---- .../io/file/ProcessedFileThreadTest.java | 34 +++++++++----- 2 files changed, 57 insertions(+), 21 deletions(-) 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 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 alreadyOffered = new HashSet<>(); private final BlockingQueue workQueue; private final BlockingQueue inProcess; private final BlockingQueue 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 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..a403ea60f9 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,22 @@ 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 (System.currentTimeMillis() < deadline + && (!workQueue.isEmpty() || !inProcess.isEmpty() || !recentlyProcessed.isEmpty() + || producedFiles.stream().anyMatch(File::exists))) { + Thread.sleep(200); + } + } + @Test public final void singleFileTest() throws IOException { @@ -186,10 +202,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 +255,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 +305,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. From 5fafc4fb0da5488eed92bb57af92211ca1340c24 Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Thu, 9 Jul 2026 12:15:51 -0700 Subject: [PATCH 2/2] [fix][io] Fail awaitProcessingComplete on timeout instead of returning silently Returning at the deadline let a test proceed as if the pipeline had drained, which could mask a regression the strict times(1) assertions are meant to catch. Fail with the queue depths and the number of files still on disk instead. --- .../pulsar/io/file/ProcessedFileThreadTest.java | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) 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 a403ea60f9..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 @@ -50,13 +50,23 @@ public class ProcessedFileThreadTest extends AbstractFileTest { */ private void awaitProcessingComplete() throws InterruptedException { long deadline = System.currentTimeMillis() + 60_000; - while (System.currentTimeMillis() < deadline - && (!workQueue.isEmpty() || !inProcess.isEmpty() || !recentlyProcessed.isEmpty() - || producedFiles.stream().anyMatch(File::exists))) { + 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 {