[Bug] SemaphoredDelegatingExecutor loses track of its permits on interruption and rejection #9508

Search before asking

• I searched in the 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.

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.

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 raised with nothing logged.

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?

• I'm willing to submit a PR!