LuciferYang opened a new issue, #9508: URL: https://github.com/apache/paimon/issues/9508
### Search before asking - [x] I searched in the [issues](https://github.com/apache/paimon/issues) and found nothing similar. ### Paimon version master, `9c7deebbd` (2.1-SNAPSHOT) ### Compute Engine Java API. The wrapper is reached from Flink and Spark through the shared pools it wraps: `FileOperationThreadPool`, `ManifestReadThreadPool`, `GlobalIndexReadThreadPool`, and `CatalogSplitEnumerator`. ### Minimal reproduce step `SemaphoredDelegatingExecutor` bounds concurrency on a delegate pool with one semaphore permit per submitted task, released by the per-task wrapper when the task finishes. Two paths lose track of that count. **1. `execute()` inflates the count when the permit wait is interrupted.** ```java ExecutorService delegate = Executors.newSingleThreadExecutor(); SemaphoredDelegatingExecutor executor = new SemaphoredDelegatingExecutor(delegate, 0, true); Thread submitter = new Thread(() -> executor.execute(() -> {})); submitter.start(); // wait until the submitter is parked on the zero-permit semaphore, then submitter.interrupt(); // the task runs anyway, and its wrapper releases a permit that was never acquired: // executor.getAvailablePermits() == 1 with permitCount == 0 ``` `execute()` catches `InterruptedException`, restores the interrupt flag, and then falls through to `super.execute(new RunnableWithPermitRelease(command))`. The three `submit()` overloads return a failed future at that point instead, so only `execute()` is affected. **2. Every submit/execute path leaks a permit when the delegate refuses the task.** ```java ExecutorService delegate = Executors.newSingleThreadExecutor(); delegate.shutdownNow(); SemaphoredDelegatingExecutor executor = new SemaphoredDelegatingExecutor(delegate, 1, true); try { executor.execute(() -> {}); } catch (RejectedExecutionException expected) { } // executor.getAvailablePermits() == 0: the permit was acquired, the wrapper never ran, // and nothing released it ``` ### What doesn't meet your expectations? `getAvailablePermits()` should never exceed `permitCount`, and a task whose permit was never acquired should not run — otherwise the concurrency ceiling this class exists to enforce is quietly raised, with no error anywhere. The leak is the mirror image: the four pools above are process-wide statics that are never shut down, so a permit that is never returned is a permanent loss of one unit of parallelism. Enough of them and every later caller blocks in `acquire()` forever. Of the two, the interrupt path is the one reachable today — Flink and Spark interrupt the task thread on cancellation, and `TableCommitImpl.close()` interrupts the maintain thread through `shutdownNow()`. The rejection path needs a delegate that refuses work, which no current construction site produces (all four use an unbounded queue and the default `AbortPolicy`), so it is the contract half of the same invariant. ### Anything else? There is a third case worth naming because the obvious fix walks into it: a delegate that runs the task in the calling thread (`ThreadPoolExecutor.CallerRunsPolicy` on a saturated bounded pool, or any direct executor) can both run the wrapper — which releases the permit in its `finally` — and let a `RejectedExecutionException` thrown by the task itself out of the same `execute()` call. Releasing the permit again in a `catch` block there inflates the count exactly the way case 1 does, so the release has to be idempotent per task rather than unconditional at the call site. ### Are you willing to submit a PR? - [x] I'm willing to submit a PR! -- 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]
