[AMORO-4373] Fix concurrency race condition in RollingFileCleaner causing silent orphan file leaks

[aakashofficial-k01]

Why are the changes needed?

Close #4373.

In IcebergTableMaintainer.expireSnapshots, table snapshots are planned and expired concurrently using IcebergThreadPools.getMaintenanceExecutor(). In RollingFileCleaner, doCleanFiles() was unsynchronized and directly operated on collectedFiles. When worker threads concurrently called addFile() while remote bulk deletion was in flight, newly submitted files were added to collectedFiles and subsequently wiped by finally { collectedFiles.clear(); } before being deleted, causing silent orphan storage leaks.

Brief change log

• In RollingFileCleaner, replace collectedFiles with BlockingQueue<String>.
• In doCleanFiles(), synchronize execution and atomically drain pending files into a local batch via collectedFiles.drainTo(toClean) before deletion.
• Synchronize clear().
• Add testConcurrentCleanFiles in TestRollingFileCleaner to verify concurrent additions during bulk deletion.

How was this patch tested?

• Add some test cases that check the changes thoroughly including negative and positive cases if possible
• Add screenshots for manual tests if appropriate
• Run test locally before making a pull request

Documentation

• Does this pull request introduce a new feature? no
• If yes, how is the feature documented? not applicable

[j1wonpark]

Thanks for the PR, but the premise of #4373 does not hold on master. planWith(...) only parallelizes manifest reading; the deleteWith callback runs on a direct executor by default (Iceberg 1.8.1 RemoveSnapshots.java:65), and Amoro never calls executeDeleteWith(...). So addFile runs sequentially on the commit thread. I confirmed this with a 10-thread maintenance pool over ~2.5k expired files: every callback ran on the calling thread.

The "empirical test on unpatched master" in the issue is this PR's own testConcurrentCleanFiles calling addFile from 10 threads directly. Did you actually observe this leak on a real AMS deployment, @aakashofficial-k01?

[aakashofficial-k01]

No, I haven't observed this on a live deployment, this came out of an offline code inspection of RollingFileCleaner.

Seeing Sets.newConcurrentHashSet() in RollingFileCleaner alongside planWith(maintenanceExecutor) in IcebergTableMaintainer, I assumed deletion callbacks were also dispatched across the pool and wrote an isolated multi-threaded test around that assumption. I checked RemoveSnapshots.java:65 and you're right: since Amoro doesn't call executeDeleteWith, it defaults to the direct executor on the commit thread, so the test was simulating a scenario that doesn't occur in AMS.

Closing this PR and #4373 to avoid unnecessary churn. Thanks for the review and learnings, @j1wonpark.