david-streamlio opened a new issue, #80:
URL: https://github.com/apache/pulsar-connectors/issues/80
## Summary
#28 ("Prevent OOM in FileSource by bounding internal queues") is green and
does not conflict with anything textually, but **merging it as-is against
current master would cause silent, permanent file loss.** It must not merge
without a companion change to `FileListingThread`.
## The interaction
#41 (merged) rewrote `FileListingThread` to offer each file exactly once
while it remains on disk:
```java
for (File f : listing) {
if (alreadyOffered.add(f)) {
workQueue.offer(f); // return value ignored
}
}
```
Ignoring the return value was safe because `workQueue` was an **unbounded**
`LinkedBlockingQueue`, so `offer()` always succeeded.
#28 replaces the three queues with **bounded** ones:
```java
workQueue = new LinkedBlockingQueue<>(queueCapacity);
```
Now `offer()` returns `false` when the queue is full. The file has already
been recorded in `alreadyOffered`, and the prune step only forgets files that
have **left the disk**. So the file is never enqueued, never retried, and never
processed — for as long as it exists.
The failure mode is worst precisely under the condition #28 is designed to
handle: a backlog large enough to fill the queue.
## Reproduction
Against current master, with a bounded `workQueue` of capacity 1 and 3 files
on disk, driving only `FileListingThread`:
```
WORKQUEUE_SIZE=1
FILES_ON_DISK=3
AFTER_DRAIN_REOFFERED=0 <-- the other 2 files are never re-offered
```
After draining the queue (as a consumer would), the lister does not re-offer
the two files it dropped. They are lost silently — no exception, no log.
## Fix
Only record a file as offered once it has actually been accepted:
```java
for (File f : listing) {
if (!alreadyOffered.contains(f) && workQueue.offer(f)) {
alreadyOffered.add(f);
}
}
```
A file that cannot be enqueued is simply retried on the next listing pass,
which is the natural back-pressure behavior — the lister slows to the
consumer's rate instead of discarding work.
The `keepFile=true` branch already re-checks `workQueue.contains(f)` each
pass and is unaffected.
## Ownership
The latent assumption is mine, introduced in #41; #28 merely exposes it.
Happy to prepare the `FileListingThread` fix (with a regression test that
bounds the queue and asserts nothing is dropped) so #28 can merge on top of it
— say the word and I'll open it.
cc @Praveenkumar76 (author of #28)
--
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]
