[PR] [fix][io] Fix duplicate file offers and flaky tests in file connector [pulsar-connectors]

["david-streamlio (via GitHub)" <gi...@apache.org>]

david-streamlio opened a new pull request, #41:
URL: https://github.com/apache/pulsar-connectors/pull/41

   ### Motivation
   
   `ProcessedFileThreadTest.renameFileTest` and `continuousRunTest` fail intermittently in CI with `TooManyActualInvocations: offer(...) Wanted 1 time, but was 2 times` — they flaked on two unrelated PRs today alone (#31, #17), taxing every merge.
   
   The root cause is a race in `FileListingThread`, not in the tests. A file is briefly in **none** of the tracking collections during two hand-off windows:
   
   1. `FileConsumerThread`: between `workQueue.take()` and `inProcess.add(file)`
   2. `ProcessedFileThread`: between `recentlyProcessed.take()` and the rename/delete reaching disk
   
   A listing pass that lands in either window sees the file on disk, absent from every queue, and offers it again — so this is duplicate processing (duplicate records to the topic) in production too, not just test noise. The exactly-once assertions in the tests were correctly catching a real bug.
   
   ### Modifications
   
   **`FileListingThread`** — close the race at its source. For `keepFile=false`, track offered files in a listing-thread-private set: a file is offered exactly once for as long as it remains on disk, and entries are pruned (before each listing snapshot, so no stale-entry window exists) once the cleanup thread renames or deletes the file. A later new file with the same name is offered normally. `keepFile=true` keeps its intentional re-processing semantics unchanged.
   
   **`ProcessedFileThreadTest`** — fix two test-side bugs: the drain loops used `&&` (exit when *any* queue is empty) instead of waiting for *all* queues to drain, and nothing waited for the final rename/delete to reach disk before asserting file existence. Both replaced with a single bounded await covering the full pipeline.
   
   ### Verifying this change
   
   - Full `:file:test` suite passes locally, plus two additional repeat runs of `ProcessedFileThreadTest` (the three 30-second soak tests included) — all green.
   - The strict `times(1)` assertions are retained, so the tests remain a regression guard for the duplicate-offer race.
   
   Note: conflicts trivially with #14 (which renames `FileListingThread` → `FileListingTask`); whichever lands second rebases — happy to do so if #14 merges first.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [fix][io] Fix duplicate file offers and flaky tests in file connector [pulsar-connectors]

["Copilot (via GitHub)" <gi...@apache.org>]

Copilot commented on code in PR #41:
URL: https://github.com/apache/pulsar-connectors/pull/41#discussion_r3554092094


##########
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);
+        }
+    }

Review Comment:
   `awaitProcessingComplete()` returns silently after the 60s deadline even if the pipeline still hasn’t drained. In `continuousRunTest`/`multipleConsumerTest` that can allow the test to continue and potentially pass without actually waiting for the final delete/rename on disk (which this helper’s Javadoc says it is supposed to guarantee). Make the helper fail when the timeout is hit and the completion condition is still not satisfied so the tests can’t produce false positives.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [fix][io] Fix duplicate file offers and flaky tests in file connector [pulsar-connectors]

["david-streamlio (via GitHub)" <gi...@apache.org>]

david-streamlio merged PR #41:
URL: https://github.com/apache/pulsar-connectors/pull/41


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org