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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
