Tidy ContinuousComputation rejection handling (#91442)

Today we implicitly assume the executor used by the
`ContinuousComputation` has an unbounded queue. This commit makes that
explicit.

Relates #91386

Author: DaveCTurner

server/src/main/java/org/elasticsearch/cluster/routing/allocation/allocator/ContinuousComputation.java

@@ -11,6 +11,8 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.common.util.concurrent.AbstractRunnable;
+import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
+import org.elasticsearch.threadpool.ThreadPool;
 
 import java.util.Objects;
 import java.util.concurrent.ExecutorService;
@@ -30,10 +32,10 @@ public abstract class ContinuousComputation<T> {
     private final Processor processor = new Processor();
 
     /**
-     * @param executorService the background executor service to use to run the computations. No more than one task is executed at once.
+     * @param threadPool Each computation runs on a {@code GENERIC} thread from this thread pool. At most one task executes at once.
      */
-    public ContinuousComputation(ExecutorService executorService) {
-        this.executorService = executorService;
+    public ContinuousComputation(ThreadPool threadPool) {
+        this.executorService = threadPool.generic();
     }
 
     /**
@@ -77,7 +79,8 @@ public void onFailure(Exception e) {
 
         @Override
         public void onRejection(Exception e) {
-            // shutting down, just give up
+            // The executor has an unbounded queue so we must be shutting down to get here.
+            assert e instanceof EsRejectedExecutionException esre && esre.isExecutorShutdown() : e;
             logger.debug("rejected", e);
         }
 

server/src/main/java/org/elasticsearch/cluster/routing/allocation/allocator/DesiredBalanceShardsAllocator.java

@@ -97,7 +97,7 @@ public DesiredBalanceShardsAllocator(
         this.clusterService = clusterService;
         this.reconciler = reconciler;
         this.desiredBalanceComputer = desiredBalanceComputer;
-        this.desiredBalanceComputation = new ContinuousComputation<>(threadPool.generic()) {
+        this.desiredBalanceComputation = new ContinuousComputation<>(threadPool) {
 
             @Override
             protected void processInput(DesiredBalanceInput desiredBalanceInput) {

server/src/test/java/org/elasticsearch/cluster/routing/allocation/allocator/ContinuousComputationTests.java

@@ -46,7 +46,7 @@ public static void terminateThreadPool() {
     public void testConcurrency() throws Exception {
 
         final var result = new AtomicReference<Integer>();
-        final var computation = new ContinuousComputation<Integer>(threadPool.generic()) {
+        final var computation = new ContinuousComputation<Integer>(threadPool) {
 
             public final Semaphore executePermit = new Semaphore(1);
 
@@ -104,7 +104,7 @@ public void testSkipsObsoleteValues() throws Exception {
         final var finalInput = new Object();
 
         final var result = new AtomicReference<Object>();
-        final var computation = new ContinuousComputation<Object>(threadPool.generic()) {
+        final var computation = new ContinuousComputation<Object>(threadPool) {
             @Override
             protected void processInput(Object input) {
                 assertNotEquals(input, skippedInput);