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: [email protected]

For queries about this service, please contact Infrastructure at:
[email protected]

Reply via email to