david-streamlio opened a new pull request, #81:
URL: https://github.com/apache/pulsar-connectors/pull/81
Fixes #80
### Motivation
`FileListingThread` records a file as offered **before** checking that the
work queue accepted it:
```java
for (File f : listing) {
if (alreadyOffered.add(f)) {
workQueue.offer(f); // return value ignored
}
}
```
Ignoring `offer()`'s result is safe only while `workQueue` is unbounded,
which it is today. The moment it is bounded, `offer()` returns `false` when the
queue is full — the file is already in `alreadyOffered`, and the prune step
only forgets files that have **left the disk**. The file is never enqueued,
never retried, and never processed.
This is not theoretical: #28 ("Prevent OOM in FileSource by bounding
internal queues") does exactly that. It is green and conflicts with nothing,
because it touches different files — so the loss would have merged silently.
Worse, the failure triggers precisely under the backlog #28 exists to survive.
Reproduced against current master with a bounded `workQueue` of capacity 1
and 3 files on disk:
```
WORKQUEUE_SIZE=1
FILES_ON_DISK=3
AFTER_DRAIN_REOFFERED=0 <-- the other 2 files are never re-offered
```
The assumption is mine, introduced in #41. This PR removes it so #28 can
land safely.
### Modifications
Track a file only once the queue has accepted it:
```java
for (File f : listing) {
if (!alreadyOffered.contains(f) && workQueue.offer(f)) {
alreadyOffered.add(f);
}
}
```
A rejected file is simply retried on the next listing pass — the lister
slows to the consumer's rate rather than discarding work, which is the
back-pressure behavior a bounded queue is meant to produce. The `keepFile=true`
branch re-checks `workQueue.contains(f)` each pass and is unaffected.
### Tests
`FileListingThreadBoundedQueueTest` bounds the queue itself, so it holds
regardless of what `FileSource` constructs today and guards the invariant ahead
of #28:
- `rejectedFilesAreRetriedOnceTheQueueDrains` — 3 files, queue capacity 1;
every file must reach the queue as it drains.
- `filesAreNotOfferedTwiceWhileTheyRemainOnDisk` — the exactly-once property
from #41 must survive the new retry path.
### Verifying this change
The regression test genuinely catches the bug. Reverting `FileListingThread`
to master's version:
```
rejectedFilesAreRetriedOnceTheQueueDrains FAILED
org.awaitility.core.ConditionTimeoutException: Condition ... was not
fulfilled within 30 seconds.
filesAreNotOfferedTwiceWhileTheyRemainOnDisk PASSED
```
The second test passes either way, as it should — it pins the property the
bug never broke.
With the fix applied, the full `:file:test` suite is green (`BUILD
SUCCESSFUL in 1m 58s`), including the previously flaky
`ProcessedFileThreadTest`.
### Suggested merge order
1. this PR
2. #28 — rebase; its bounded queues then cannot drop files
--
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]
