Fix concurrency permit leak causing permanent deadlock in SimpleAsyncTaskExecutor

When concurrency limiting is enabled via setConcurrencyLimit() and
thread creation fails in doExecute() (e.g., OutOfMemoryError from
Thread.start()), the concurrency permit acquired by beforeAccess()
is never released because TaskTrackingRunnable.run() never executes.

This causes the concurrency count to permanently remain at the limit,
causing all subsequent task submissions to block forever in
ConcurrencyThrottleSupport.onLimitReached().

Root cause:
- beforeAccess() increments concurrencyCount
- doExecute() throws Error before thread starts
- TaskTrackingRunnable.run() never executes
- afterAccess() in finally block never called
- Concurrency permit permanently leaked

Solution:
Wrap doExecute() in try-catch block in the concurrency throttle path
and call afterAccess() in catch block to ensure permit is always
released, even when thread creation fails.

The fix only applies to the concurrency throttle path. The
activeThreads-only path does not need fixing because it never calls
beforeAccess(), so there is no permit to leak.

Test approach:
The test simulates thread creation failure and verifies that a
subsequent execution does not deadlock. The first execution should
fail with some exception (type doesn't matter), and the second
execution should complete within timeout if the permit was properly
released.

Signed-off-by: Park Juhyeong <wngud5957@naver.com>

spring-core/src/main/java/org/springframework/core/task/SimpleAsyncTaskExecutor.java

@@ -313,7 +313,15 @@ public class SimpleAsyncTaskExecutor extends CustomizableThreadCreator
 		Runnable taskToUse = (this.taskDecorator != null ? this.taskDecorator.decorate(task) : task);
 		if (isThrottleActive() && startTimeout > TIMEOUT_IMMEDIATE) {
 			this.concurrencyThrottle.beforeAccess();
-			doExecute(new TaskTrackingRunnable(taskToUse));
+			try {
+				doExecute(new TaskTrackingRunnable(taskToUse));
+			}
+			catch (Throwable ex) {
+				// Release concurrency permit if thread creation fails
+				this.concurrencyThrottle.afterAccess();
+				throw new TaskRejectedException(
+						"Failed to start execution thread for task: " + task, ex);
+			}
 		}
 		else if (this.activeThreads != null) {
 			doExecute(new TaskTrackingRunnable(taskToUse));

spring-core/src/test/java/org/springframework/core/task/SimpleAsyncTaskExecutorTests.java

@@ -16,6 +16,9 @@
 
 package org.springframework.core.task;
 
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
 import org.junit.jupiter.api.Test;
 
 import org.springframework.util.ConcurrencyThrottleSupport;
@@ -24,6 +27,12 @@ import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
 import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.BDDMockito.willCallRealMethod;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.spy;
+
 
 /**
  * @author Rick Evans
@@ -69,6 +78,59 @@ class SimpleAsyncTaskExecutorTests {
 		}
 	}
 
+	/**
+	 * Verify that when thread creation fails in doExecute() while concurrency
+	 * limiting is active, the concurrency permit is properly released to
+	 * prevent permanent deadlock.
+	 *
+	 * <p>This test reproduces a critical bug where OutOfMemoryError from
+	 * Thread.start() causes the executor to permanently deadlock:
+	 * <ol>
+	 *   <li>beforeAccess() increments concurrencyCount
+	 *   <li>doExecute() throws Error before thread starts
+	 *   <li>TaskTrackingRunnable.run() never executes
+	 *   <li>afterAccess() in finally block never called
+	 *   <li>Subsequent tasks block forever in onLimitReached()
+	 * </ol>
+	 *
+	 * <p>Test approach: The first execute() should fail with some exception
+	 * (type doesn't matter - could be Error or TaskRejectedException).
+	 * The second execute() is the real test: it should complete without
+	 * deadlock if the permit was properly released.
+	 */
+	@Test
+	void executeFailsToStartThreadReleasesConcurrencyPermit() throws InterruptedException {
+		// Arrange
+		SimpleAsyncTaskExecutor executor = spy(new SimpleAsyncTaskExecutor());
+		executor.setConcurrencyLimit(1);  // Enable concurrency limiting
+
+		Runnable task = () -> {};
+		Error failure = new OutOfMemoryError("TEST: Cannot start thread");
+
+		// Simulate thread creation failure
+		doThrow(failure).when(executor).doExecute(any(Runnable.class));
+
+		// Act - First execution fails
+		// Both "before fix" (throws Error) and "after fix" (throws TaskRejectedException)
+		// should throw some exception here - that's expected and correct
+		assertThatThrownBy(() -> executor.execute(task))
+				.isInstanceOf(Throwable.class);
+
+		// Arrange - Reset mock to allow second execution to succeed
+		willCallRealMethod().given(executor).doExecute(any(Runnable.class));
+
+		// Assert - Second execution should NOT deadlock
+		// This is the real test: if permit was leaked, this will timeout
+		CountDownLatch latch = new CountDownLatch(1);
+		executor.execute(() -> latch.countDown());
+
+		boolean completed = latch.await(1, TimeUnit.SECONDS);
+
+		assertThat(completed)
+				.withFailMessage("Executor should not deadlock if concurrency permit was properly released after first failure")
+				.isTrue();
+	}
+
 	@Test
 	void threadNameGetsSetCorrectly() {
 		String customPrefix = "chankPop#";