[I] [BUG] Fix concurrency race condition in RollingFileCleaner causing silent orphan file leaks [amoro]

["aakashofficial-k01 (via GitHub)" <gi...@apache.org>]

aakashofficial-k01 opened a new issue, #4373:
URL: https://github.com/apache/amoro/issues/4373

   ### What happened?
   
   In `IcebergTableMaintainer.expireSnapshots`, table snapshots are planned and expired concurrently using `IcebergThreadPools.getMaintenanceExecutor()`:
   ```java
   table.expireSnapshots()
       .planWith(IcebergThreadPools.getMaintenanceExecutor())
       .deleteWith(expiredFileCleaner::addFile)
       .cleanExpiredFiles(true)
       .commit();
   ```
   
   Inside `RollingFileCleaner`:
   1. Expired file paths are added to `collectedFiles` via `addFile(filePath)`.
   2. When the batch size reaches `CLEANED_FILE_GROUP_SIZE` (1,000 files), `doCleanFiles()` is called.
   3. In `upstream/master`, `doCleanFiles()` is unsynchronized and directly accesses `collectedFiles`:
   ```java
   private void doCleanFiles() {
     if (collectedFiles.isEmpty()) {
       return;
     }
     try {
       if (fileIO.supportBulkOperations()) {
         fileIO.asBulkFileIO().deleteFiles(collectedFiles);
         cleanedFileCounter.addAndGet(collectedFiles.size());
       ...
     } finally {
       collectedFiles.clear();
     }
   }
   ```
   
   When multiple worker threads concurrently call `addFile()` while a remote bulk deletion (`deleteFiles()`) is in flight, newly submitted files are added to `collectedFiles`. When `deleteFiles()` completes, the `finally { collectedFiles.clear(); }` block wipes the set, permanently purging all files added in flight before they are ever deleted from object storage.
   
   ### Empirical Reproduction
   
   A concurrent simulation with 10 worker threads submitting 3,000 files with realistic S3 network latency against unpatched master showed:
   - Files submitted: 3,000
   - Files deleted: 2,972
   - Files permanently leaked in storage: 28
   
   ### Proposed Solution
   
   1. Replace `collectedFiles` with a `BlockingQueue<String> collectedFiles = new LinkedBlockingQueue<>()`.
   2. Synchronize `doCleanFiles()` and atomically drain files into a local batch via `collectedFiles.drainTo(toClean)`.
   3. Worker threads calling `addFile()` can continue enqueueing to the tail without blocking or risking in-flight eviction.
   4. Synchronize `clear()`.
   


-- 
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@amoro.apache.org.apache.org

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

Re: [I] [Bug]: Fix concurrency race condition in RollingFileCleaner causing silent orphan file leaks [amoro]

["aakashofficial-k01 (via GitHub)" <gi...@apache.org>]

aakashofficial-k01 closed issue #4373: [Bug]: Fix concurrency race condition in RollingFileCleaner causing silent orphan file leaks
URL: https://github.com/apache/amoro/issues/4373


-- 
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@amoro.apache.org

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