Fix ExecutionSequencer queue stall when delegate executor throws RejectedExecutionException.

When a delegate `Executor` throws a `RejectedExecutionException` during task execution or queue draining, `ExecutionSequencer` fails to catch the exception, leaving `taskFuture` and `newFuture` permanently uncompleted. As a result, the `ExecutionSequencer` queue permanently stalls, causing all subsequently submitted tasks to leak in memory.

This change handles exceptions thrown by delegate executors in both non-reentrant and reentrant queue draining paths, cancelling unexecuted task futures to allow downstream tasks to execute.

RELNOTES=`util.concurrent`: Fixed `ExecutionSequencer` queue stall when delegate executor throws `RejectedExecutionException`.
PiperOrigin-RevId: 962615495

Author: java-team-github-bot

android/guava-tests/test/com/google/common/util/concurrent/ExecutionSequencerTest.java

@@ -22,6 +22,7 @@
 import static java.util.concurrent.Executors.newCachedThreadPool;
 import static java.util.concurrent.Executors.newFixedThreadPool;
 import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.junit.Assert.assertThrows;
 
 import com.google.common.annotations.GwtIncompatible;
 import com.google.common.annotations.J2ktIncompatible;
@@ -32,9 +33,12 @@
 import java.util.List;
 import java.util.concurrent.Callable;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.logging.Logger;
 import junit.framework.TestCase;
 import org.jspecify.annotations.NullUnmarked;
@@ -296,6 +300,80 @@ public void testAvoidsStackOverflow_multipleThreads() throws Exception {
     }
   }
 
+  /*
+   * Verifies that when a non-reentrant delegate executor rejects a task (e.g., throwing a
+   * RejectedExecutionException because a bounded queue is full), ExecutionSequencer catches the
+   * failure and unblocks the queue, preventing subsequent tasks from stalling or leaking.
+   */
+  public void testRecoversFromNonReentrantRejectedExecution() throws Exception {
+    // 1. Submit Task 1 (firstCallable) using directExecutor. It blocks because firstFuture is not
+    // yet set.
+    Future<?> unused = serializer.submitAsync(firstCallable, directExecutor());
+
+    Executor rejectingExecutor =
+        task -> {
+          throw new RejectedExecutionException();
+        };
+    TestCallable secondCallable = new TestCallable(immediateVoidFuture());
+
+    // 2. Submit Task 2 using the rejectingExecutor. When Task 1 completes, ExecutionSequencer will
+    // attempt to dispatch Task 2 via rejectingExecutor.execute(...), which will throw
+    // RejectedExecutionException.
+    Future<?> failingFuture = serializer.submitAsync(secondCallable, rejectingExecutor);
+
+    // 3. Submit Task 3 using directExecutor. Task 3 is queued after Task 2 in ExecutionSequencer.
+    TestCallable thirdCallable = new TestCallable(immediateVoidFuture());
+    ListenableFuture<@Nullable Void> thirdFuture =
+        serializer.submitAsync(thirdCallable, directExecutor());
+
+    // Complete Task 1's future. This triggers listener notification for Task 2.
+    // Task 2's dispatch fails with RejectedExecutionException inside rejectingExecutor.execute().
+    firstFuture.set(null);
+
+    // Verify queue recovery: Task 2's failure is handled, and Task 3 runs successfully.
+    assertThat(secondCallable.called).isFalse();
+    assertThrows(ExecutionException.class, () -> getDone(failingFuture));
+    assertThat(thirdCallable.called).isTrue();
+    assertThat(thirdFuture.isDone()).isTrue();
+  }
+
+  /*
+   * Verifies that when a reentrant delegate executor rejects a task, TaskNonReentrantExecutor.run()
+   * handles the failure without stalling the queue or leaking subsequent submitted tasks.
+   */
+  public void testRecoversFromReentrantRejectedExecution() throws Exception {
+    Executor rejectingExecutor =
+        task -> {
+          throw new RejectedExecutionException();
+        };
+
+    TestCallable secondCallable = new TestCallable(immediateVoidFuture());
+    TestCallable thirdCallable = new TestCallable(immediateVoidFuture());
+    AtomicReference<ListenableFuture<@Nullable Void>> failingFuture = new AtomicReference<>();
+    AtomicReference<ListenableFuture<@Nullable Void>> thirdFuture = new AtomicReference<>();
+
+    // Submit an outer task using directExecutor.
+    // Inside its execution, we submit Task 2 (rejectingExecutor) and Task 3 (directExecutor)
+    // reentrantly to test reentrant queue draining error handling.
+    Future<?> unused =
+        serializer.submitAsync(
+            () -> {
+              // Reentrant submission of Task 2 to rejectingExecutor.
+              failingFuture.set(serializer.submitAsync(secondCallable, rejectingExecutor));
+              // Reentrant submission of Task 3 to directExecutor.
+              thirdFuture.set(serializer.submitAsync(thirdCallable, directExecutor()));
+              return immediateVoidFuture();
+            },
+            directExecutor());
+
+    // Verify reentrant queue recovery: Task 3 runs successfully despite Task 2's executor
+    // rejection.
+    assertThat(secondCallable.called).isFalse();
+    assertThrows(ExecutionException.class, () -> getDone(failingFuture.get()));
+    assertThat(thirdCallable.called).isTrue();
+    assertThat(thirdFuture.get().isDone()).isTrue();
+  }
+
   @SuppressWarnings("ObjectToString") // Intended behavior
   public void testToString() {
     Future<?> unused = serializer.submitAsync(firstCallable, directExecutor());

android/guava/src/com/google/common/util/concurrent/ExecutionSequencer.java

@@ -16,9 +16,9 @@
 
 import static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.common.base.Preconditions.checkState;
-import static com.google.common.util.concurrent.ExecutionSequencer.RunningState.CANCELLED;
 import static com.google.common.util.concurrent.ExecutionSequencer.RunningState.NOT_RUN;
 import static com.google.common.util.concurrent.ExecutionSequencer.RunningState.STARTED;
+import static com.google.common.util.concurrent.ExecutionSequencer.RunningState.WILL_NOT_RUN;
 import static com.google.common.util.concurrent.Futures.immediateCancelledFuture;
 import static com.google.common.util.concurrent.Futures.immediateFuture;
 import static com.google.common.util.concurrent.Futures.immediateVoidFuture;
@@ -31,6 +31,7 @@
 import java.util.concurrent.Callable;
 import java.util.concurrent.Executor;
 import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.atomic.AtomicReference;
 import org.jspecify.annotations.Nullable;
 
@@ -135,7 +136,7 @@ private static final class ThreadConfinedTaskQueue {
     @LazyInit @Nullable Thread thread;
 
     /** Only used by the thread associated with this object */
-    @Nullable Runnable nextTask;
+    @Nullable TrustedListenableFutureTask<?> nextTask;
 
     /** Only used by the thread associated with this object */
     @Nullable Executor nextExecutor;
@@ -267,7 +268,7 @@ public String toString() {
 
   enum RunningState {
     NOT_RUN,
-    CANCELLED,
+    WILL_NOT_RUN,
     STARTED,
   }
 
@@ -308,7 +309,7 @@ private static final class TaskNonReentrantExecutor extends AtomicReference<Runn
      * Set before calling delegate.execute(); set to null once run, so that it can be GCed; this
      * object may live on after, if submitAsync returns an incomplete future.
      */
-    @Nullable Runnable task;
+    @Nullable TrustedListenableFutureTask<?> task;
 
     /** Thread that called execute(). Set in execute, cleared when delegate.execute() returns. */
     @LazyInit @Nullable Thread submitting;
@@ -336,12 +337,13 @@ public void execute(Runnable task) {
       // from the cancelled operation and its own. This can cause one runnable to queue two tasks,
       // breaking the invariant this method relies on to iteratively run the next task after the
       // previous one completes.
-      if (get() == RunningState.CANCELLED) {
+      if (get() == RunningState.WILL_NOT_RUN) {
         delegate = null;
         sequencer = null;
         return;
       }
       submitting = Thread.currentThread();
+      TrustedListenableFutureTask<?> trustedFutureTask = (TrustedListenableFutureTask<?>) task;
 
       try {
         /*
@@ -361,16 +363,31 @@ public void execute(Runnable task) {
           // Task must be null, since each execution on this executor can only produce one more
           // execution.
           checkState(submittingTaskQueue.nextTask == null);
-          submittingTaskQueue.nextTask = task;
+          submittingTaskQueue.nextTask = trustedFutureTask;
           // requireNonNull(delegate) is safe for reasons similar to requireNonNull(sequencer).
           submittingTaskQueue.nextExecutor = requireNonNull(delegate);
           delegate = null;
         } else {
           // requireNonNull(delegate) is safe for reasons similar to requireNonNull(sequencer).
           Executor localDelegate = requireNonNull(delegate);
           delegate = null;
-          this.task = task;
-          localDelegate.execute(this);
+          this.task = trustedFutureTask;
+          try {
+            localDelegate.execute(this);
+          } catch (RejectedExecutionException e) {
+            /*
+             * `trySetCancelled` is irrelevant when we know the task won't run and when it is
+             * harmless for our `setExceptionInternal` call here to race with a user call to
+             * `cancel`. But we do it anyway in case we want to use the `RunningState` for another
+             * purpose someday.
+             */
+            boolean unused = trySetCancelled();
+            trustedFutureTask.setExceptionInternal(e);
+            /*
+             * We don't need to clear `this.task` because no one will have a
+             * reference to `this` after we return.
+             */
+          }
         }
       } finally {
         // Important to null this out here - if we did *not* execute inline, we might still
@@ -437,14 +454,18 @@ public void run() {
         task = null;
         localTask.run();
         // Now check if our task attempted to reentrantly execute the next task.
-        Runnable queuedTask;
+        TrustedListenableFutureTask<?> queuedTask;
         Executor queuedExecutor;
         // Intentionally using non-short-circuit operator
         while ((queuedTask = executingTaskQueue.nextTask) != null
             && (queuedExecutor = executingTaskQueue.nextExecutor) != null) {
           executingTaskQueue.nextTask = null;
           executingTaskQueue.nextExecutor = null;
-          queuedExecutor.execute(queuedTask);
+          try {
+            queuedExecutor.execute(queuedTask);
+          } catch (RejectedExecutionException e) {
+            queuedTask.setExceptionInternal(e);
+          }
         }
       } finally {
         // Null out the thread field, so that we don't leak a reference to Thread, and so that
@@ -462,7 +483,7 @@ private boolean trySetStarted() {
     }
 
     private boolean trySetCancelled() {
-      return compareAndSet(NOT_RUN, CANCELLED);
+      return compareAndSet(NOT_RUN, WILL_NOT_RUN);
     }
   }
 }

cycle_suppress_list.txt

@@ -41,8 +41,14 @@ FIELD com.google.common.collect.TreeTraverser.PostOrderNode.childIterator
 FIELD com.google.common.collect.TreeTraverser.PreOrderIterator.stack
 FIELD com.google.common.util.concurrent.AbstractFuture.Listener.executor com.google.common.util.concurrent.MoreExecutors.rejectionPropagatingExecutor.$
 FIELD com.google.common.util.concurrent.AbstractService.listeners
+
 # NonReentrantExecutor is not exposed to this field
 FIELD com.google.common.util.concurrent.ExecutionSequencer.ThreadConfinedTaskQueue.nextExecutor com.google.common.util.concurrent.ExecutionSequencer.TaskNonReentrantExecutor
+
+# There is no way for a CombinedFutureInterruptibleTask to receive a TaskNonReentrantExecutor as its listenerExecutor:
+# A TaskNonReentrantExecutor instance is used as an Executor only in a direct call to addListener.
+FIELD com.google.common.util.concurrent.CombinedFuture.CombinedFutureInterruptibleTask.listenerExecutor com.google.common.util.concurrent.ExecutionSequencer.TaskNonReentrantExecutor
+
 # Real cycle, but the runningState field is null'ed on completion of the future.
 FIELD com.google.common.util.concurrent.AggregateFuture.runningState
 FIELD java.util.AbstractMap.keySet com.google.common.collect.AbstractMapBasedMultimap.NavigableKeySet

guava-tests/test/com/google/common/util/concurrent/ExecutionSequencerTest.java

@@ -22,6 +22,7 @@
 import static java.util.concurrent.Executors.newCachedThreadPool;
 import static java.util.concurrent.Executors.newFixedThreadPool;
 import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.junit.Assert.assertThrows;
 
 import com.google.common.annotations.GwtIncompatible;
 import com.google.common.annotations.J2ktIncompatible;
@@ -32,9 +33,12 @@
 import java.util.List;
 import java.util.concurrent.Callable;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.logging.Logger;
 import junit.framework.TestCase;
 import org.jspecify.annotations.NullUnmarked;
@@ -296,6 +300,80 @@ public void testAvoidsStackOverflow_multipleThreads() throws Exception {
     }
   }
 
+  /*
+   * Verifies that when a non-reentrant delegate executor rejects a task (e.g., throwing a
+   * RejectedExecutionException because a bounded queue is full), ExecutionSequencer catches the
+   * failure and unblocks the queue, preventing subsequent tasks from stalling or leaking.
+   */
+  public void testRecoversFromNonReentrantRejectedExecution() throws Exception {
+    // 1. Submit Task 1 (firstCallable) using directExecutor. It blocks because firstFuture is not
+    // yet set.
+    Future<?> unused = serializer.submitAsync(firstCallable, directExecutor());
+
+    Executor rejectingExecutor =
+        task -> {
+          throw new RejectedExecutionException();
+        };
+    TestCallable secondCallable = new TestCallable(immediateVoidFuture());
+
+    // 2. Submit Task 2 using the rejectingExecutor. When Task 1 completes, ExecutionSequencer will
+    // attempt to dispatch Task 2 via rejectingExecutor.execute(...), which will throw
+    // RejectedExecutionException.
+    Future<?> failingFuture = serializer.submitAsync(secondCallable, rejectingExecutor);
+
+    // 3. Submit Task 3 using directExecutor. Task 3 is queued after Task 2 in ExecutionSequencer.
+    TestCallable thirdCallable = new TestCallable(immediateVoidFuture());
+    ListenableFuture<@Nullable Void> thirdFuture =
+        serializer.submitAsync(thirdCallable, directExecutor());
+
+    // Complete Task 1's future. This triggers listener notification for Task 2.
+    // Task 2's dispatch fails with RejectedExecutionException inside rejectingExecutor.execute().
+    firstFuture.set(null);
+
+    // Verify queue recovery: Task 2's failure is handled, and Task 3 runs successfully.
+    assertThat(secondCallable.called).isFalse();
+    assertThrows(ExecutionException.class, () -> getDone(failingFuture));
+    assertThat(thirdCallable.called).isTrue();
+    assertThat(thirdFuture.isDone()).isTrue();
+  }
+
+  /*
+   * Verifies that when a reentrant delegate executor rejects a task, TaskNonReentrantExecutor.run()
+   * handles the failure without stalling the queue or leaking subsequent submitted tasks.
+   */
+  public void testRecoversFromReentrantRejectedExecution() throws Exception {
+    Executor rejectingExecutor =
+        task -> {
+          throw new RejectedExecutionException();
+        };
+
+    TestCallable secondCallable = new TestCallable(immediateVoidFuture());
+    TestCallable thirdCallable = new TestCallable(immediateVoidFuture());
+    AtomicReference<ListenableFuture<@Nullable Void>> failingFuture = new AtomicReference<>();
+    AtomicReference<ListenableFuture<@Nullable Void>> thirdFuture = new AtomicReference<>();
+
+    // Submit an outer task using directExecutor.
+    // Inside its execution, we submit Task 2 (rejectingExecutor) and Task 3 (directExecutor)
+    // reentrantly to test reentrant queue draining error handling.
+    Future<?> unused =
+        serializer.submitAsync(
+            () -> {
+              // Reentrant submission of Task 2 to rejectingExecutor.
+              failingFuture.set(serializer.submitAsync(secondCallable, rejectingExecutor));
+              // Reentrant submission of Task 3 to directExecutor.
+              thirdFuture.set(serializer.submitAsync(thirdCallable, directExecutor()));
+              return immediateVoidFuture();
+            },
+            directExecutor());
+
+    // Verify reentrant queue recovery: Task 3 runs successfully despite Task 2's executor
+    // rejection.
+    assertThat(secondCallable.called).isFalse();
+    assertThrows(ExecutionException.class, () -> getDone(failingFuture.get()));
+    assertThat(thirdCallable.called).isTrue();
+    assertThat(thirdFuture.get().isDone()).isTrue();
+  }
+
   @SuppressWarnings("ObjectToString") // Intended behavior
   public void testToString() {
     Future<?> unused = serializer.submitAsync(firstCallable, directExecutor());