[common] Fix permit accounting of SemaphoredDelegatingExecutor under interruption and rejection

[LuciferYang]

Purpose

close #9508

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 broke that accounting, and this PR fixes only the accounting: no caller sees a behavior change.

execute() caught InterruptedException from the permit wait, restored the interrupt flag, and then submitted the task anyway. The wrapper released a permit that was never acquired, so getAvailablePermits() climbed above permitCount and the ceiling this class exists to enforce was raised with nothing logged. The task still has to run: execute() has no failed-future channel, and the callers that reach this class through CompletableFuture.supplyAsync(task, executor) would otherwise wait on a stage that never completes. So it still runs, and the wrapper now knows that no permit backs it and does not hand one back.

Worth spelling out why the fix does not simply reject the task, since that was this PR's first version. Semaphore.acquire() goes through AQS acquireSharedInterruptibly, which throws as soon as the caller carries an interrupt flag, even with every permit free. Rejecting on InterruptedException therefore does not mean "interrupted while waiting for a permit"; it means "any execute() from an already-interrupted thread", which is exactly what a Flink or Spark task looks like while it is being cancelled. A probe with permitCount 2, both permits free and the flag set showed execute() throwing while the task never ran.

All four submit/execute paths now return the acquired permit when the delegate rejects the task, which previously leaked it, because the wrapper that would have released it never ran. The release goes through a release-once flag on the wrapper rather than an unconditional release() at the call site, for two reasons: a delegate that runs the task inline (CallerRunsPolicy on a saturated bounded pool, or any direct executor) can both run the wrapper and let a RejectedExecutionException thrown by the task itself out of the same call, and the same flag records whether a permit was ever acquired for this task at all.

Scope note: every current construction site (FileOperationThreadPool, ManifestReadThreadPool, GlobalIndexReadThreadPool, CatalogSplitEnumerator) wraps a process-wide static pool with an unbounded queue and the default AbortPolicy, so the interrupt path is the one reachable in production today. The catch stays narrowed to RejectedExecutionException, the contract signal for "the delegate will not take this task"; other throwables out of the delegate are left alone, and the release-once flag makes widening that catch safe later if a bounded or custom delegate ever needs it.

Tests

SemaphoredDelegatingExecutorTest is new. The class had no test before.

• testInterruptedExecuteRunsTaskWithoutInflatingPermits: waits until the submitter is provably parked on a zero-permit semaphore, interrupts it once, then drains the delegate and asserts the task ran, the interrupt flag survived, and the permit count is unchanged rather than one higher.
• testExecuteWithInterruptFlagAlreadySetKeepsPermitCount: the same invariants for a submitter that carries the flag before it calls execute() with both permits free, which is the case that made the first version of this PR reject work during cancellation.
• testRejectedByDelegateReleasesPermit: a shut-down delegate, asserting the permit returns to 1 after each of the four entry points (execute, submit(Callable), submit(Runnable), submit(Runnable, result)).
• testInlineExecutionReleasesPermitOnlyOnce: a ThreadPoolExecutor with corePoolSize 1, a queue of 1 and CallerRunsPolicy, saturated so the wrapper runs in the calling thread while the task's own RejectedExecutionException escapes execute(); asserts the permit count is 1, not 2.
• testNormalExecutionKeepsPermitsBalanced: five tasks over two permits, asserting all five ran and the count is back to two after the delegate terminates. This one also passes without the fix; it is the first pin on the normal release path, which had none.

Three of the five fail against master: both interrupt tests on the inflated count (1 instead of 0, and 3 instead of 2) and the rejection test on the leaked permit.

mvn -pl paimon-common clean test on JDK 8: 12470 tests, 0 failures, 0 errors. SinkSavepointITCase, the test the earlier version of this PR left running until the CI job timeout, passes locally 3 for 3 with this version. checkstyle, spotless, enforcer and rat run clean.

[LuciferYang]

Pushed 28a2d67 after the first version of this PR timed out the UTCase and ITCase Flink 1.x Common job.

The problem was mine and it was wider than the commit message claimed. Semaphore.acquire() goes through AQS acquireSharedInterruptibly, which throws as soon as the calling thread carries an interrupt flag, even with every permit free. Rejecting on InterruptedException therefore did not mean "interrupted while waiting for a permit", it meant "any execute() from an already-interrupted thread", which is what a Flink task looks like while it is being cancelled. A probe with permitCount 2, both permits free and the flag set showed execute() throwing while the task never ran. The job had stalled 53 minutes inside SinkSavepointITCase, which is a loop of stop-with-savepoint and restore with FailingFileIO injecting failures.

The new commit keeps the original behavior: an interrupted submitter restores the flag and hands the task to the delegate as before, and only the accounting changes, by recording that no permit backs that task so its wrapper does not hand back a permit nobody acquired. That job is now green in 42m45s, with SinkSavepointITCase passing in 35s.

The one remaining red check, UTCase and ITCase Spark 4.x, is unrelated: Data Evolution: concurrent merge and small files compact failed with Snapshot file ... snapshot-90 does not exist. It might have been expired by other jobs operating on this table. That test races ten MERGE INTOs against ten sys.compact calls and retries only two specific conflict messages, so a snapshot expiring between the compactor's split listing and its read fails it; the sibling test in the same file needed a flakiness fix in #9475. This change only affects semaphore permit accounting and cannot expire a snapshot.

[JingsongLi]

+1

[LuciferYang]

Thank you @JingsongLi