[I] [Bug] Bounding the file connector's work queue (#28) would silently drop files against current master [pulsar-connectors]

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

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: commits-unsubscribe@pulsar.apache.org.apache.org

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

Re: [I] [Bug] Bounding the file connector's work queue (#28) would silently drop files against current master [pulsar-connectors]

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

david-streamlio closed issue #80: [Bug] Bounding the file connector's work queue (#28) would silently drop files against current master
URL: https://github.com/apache/pulsar-connectors/issues/80


-- 
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