Skip to content

[fix][io] Fix duplicate file offers and flaky tests in file connector #41

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
david-streamlio merged 2 commits into from
Jul 9, 2026
Merged

[fix][io] Fix duplicate file offers and flaky tests in file connector #41

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2bd60a2
[fix][io] Fix duplicate file offers and flaky tests in file connector
david-streamlio Jul 9, 2026
5fafc4f
[fix][io] Fail awaitProcessingComplete on timeout instead of returnin…
david-streamlio Jul 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 35 additions & 9 deletions file/src/main/java/org/apache/pulsar/io/file/FileListingThread.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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());
Expand Down
44 changes: 32 additions & 12 deletions file/src/test/java/org/apache/pulsar/io/file/ProcessedFileThreadTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {

Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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.
Expand Down
Loading