Fix concurrency permit leak causing deadlock in SimpleAsyncTaskExecutor

[NYgomets]

Problem

When concurrency limiting is enabled via setConcurrencyLimit(), if thread creation fails in doExecute() (for example, OutOfMemoryError from Thread.start()), the concurrency permit is never released, causing permanent deadlock.

Root Cause

if (isThrottleActive() && startTimeout > TIMEOUT_IMMEDIATE) {
    this.concurrencyThrottle.beforeAccess();  // (1) Increments concurrencyCount
    doExecute(new TaskTrackingRunnable(taskToUse));  // (2) May throw Error
    // If (2) throws, TaskTrackingRunnable.run() never executes
    // Therefore afterAccess() in its finally block is never called
}

Execution flow when thread creation fails:

1. beforeAccess() increments concurrencyCount (e.g., 0 → 1)
2. doExecute() calls newThread(task).start()
3. Thread.start() throws OutOfMemoryError
4. TaskTrackingRunnable object was created but run() never executes
5. The finally block containing afterAccess() never executes
6. concurrencyCount remains at 1 permanently

Impact

After enough failures (equal to concurrencyLimit), all subsequent task submissions permanently block:

// ConcurrencyThrottleSupport.onLimitReached()
while (this.concurrencyCount >= this.concurrencyLimit) {
    this.concurrencyCondition.await();  // Blocks forever
}

The executor becomes permanently deadlocked.

Solution

if (isThrottleActive() && startTimeout > TIMEOUT_IMMEDIATE) {
    this.concurrencyThrottle.beforeAccess();
    try {
        doExecute(new TaskTrackingRunnable(taskToUse));
    }
    catch (Throwable ex) {
        // Release permit if thread creation fails
        this.concurrencyThrottle.afterAccess();
        throw new TaskRejectedException("Failed to start execution thread", ex);
    }
}

Why Only This Path?

The else if (this.activeThreads != null) path does not need fixing:

1. It never calls beforeAccess() (different code path)
2. In this path, isThrottleActive() returned false, so concurrencyLimit == -1
3. TaskTrackingRunnable.afterAccess() has guard: if (concurrencyLimit >= 0)
4. With concurrencyLimit == -1, afterAccess() becomes no-op
5. No double-decrement or negative count issue

Testing

The test reproduces the deadlock scenario:

1. Set concurrencyLimit to 1
2. Mock doExecute() to throw OutOfMemoryError
3. First execute() call - should fail with some exception
4. Reset mock to allow success
5. Second execute() call - should not deadlock (the real test)

Test approach: We don't care what exception the first call throws (could be Error or TaskRejectedException). The critical test is whether the second call completes without timeout, proving the permit was released.

Before fix: Second call times out (deadlock)
After fix: Second call completes successfully

Backward Compatibility

• No API changes
• Behavior only changes in error scenarios (thread creation failure)
• Normal execution flow unchanged
• Exception wrapping (Error → TaskRejectedException) provides better error context