This is an automated email from the ASF dual-hosted git repository.
david-streamlio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar-connectors.git
The following commit(s) were added to refs/heads/master by this push:
new 6377b961 [fix][io] Fix duplicate file offers and flaky tests in file
connector (#41)
6377b961 is described below
commit 6377b9610bb668ac6a110265bb94955743db69da
Author: David Kjerrumgaard <[email protected]>
AuthorDate: Thu Jul 9 13:22:53 2026 -0700
[fix][io] Fix duplicate file offers and flaky tests in file connector (#41)
* [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.
* [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.
---
.../apache/pulsar/io/file/FileListingThread.java | 44 +++++++++++++++++-----
.../pulsar/io/file/ProcessedFileThreadTest.java | 44 ++++++++++++++++------
2 files changed, 67 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 5958fc37..10a3c862 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 class FileListingThread extends Thread {
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 0b1f8b15..6e4f015d 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 class ProcessedFileThreadTest extends
AbstractFileTest {
// 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 class ProcessedFileThreadTest extends
AbstractFileTest {
// 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 class ProcessedFileThreadTest extends
AbstractFileTest {
// 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.
