Fix concurrency permit leak causing deadlock in SimpleAsyncTaskExecutor

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

@@ -313,7 +313,15 @@ public void execute(Runnable task, long startTimeout) {
 		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.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 @@ void taskRejectedWhenConcurrencyLimitReached() {
 		}
 	}
 
+	/**
+	 * 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#";