Handle scheduler shutdown in MasterService (#105552)

Today if `ThreadPool#scheduler` is shut down while submitting a task for
execution by the `MasterService` then we directly throw the rejection
exception to the caller. By the looks of it most callers don't expect
this method to throw anything, so this commit adjusts the behaviour to
fail the task instead.

Closes #105549

Author: DaveCTurner

server/src/main/java/org/elasticsearch/cluster/service/MasterService.java

@@ -1470,8 +1470,13 @@ private void completeTask(Exception e) {
 
         @Override
         public String toString() {
-            return Strings.format("master service timeout handler for [%s][%s] after [%s]", source, taskHolder.get(), timeout);
+            return getTimeoutTaskDescription(source, taskHolder.get(), timeout);
         }
+
+    }
+
+    static String getTimeoutTaskDescription(String source, Object task, TimeValue timeout) {
+        return Strings.format("master service timeout handler for [%s][%s] after [%s]", source, task, timeout);
     }
 
     /**
@@ -1522,11 +1527,25 @@ public void submitTask(String source, T task, @Nullable TimeValue timeout) {
             final var taskHolder = new AtomicReference<>(task);
             final Scheduler.Cancellable timeoutCancellable;
             if (timeout != null && timeout.millis() > 0) {
-                timeoutCancellable = threadPool.schedule(
-                    new TaskTimeoutHandler<>(timeout, source, taskHolder),
-                    timeout,
-                    threadPool.generic()
-                );
+                try {
+                    timeoutCancellable = threadPool.schedule(
+                        new TaskTimeoutHandler<>(timeout, source, taskHolder),
+                        timeout,
+                        threadPool.generic()
+                    );
+                } catch (Exception e) {
+                    assert e instanceof EsRejectedExecutionException esre && esre.isExecutorShutdown() : e;
+                    task.onFailure(
+                        new FailedToCommitClusterStateException(
+                            "could not schedule timeout handler for [%s][%s] on queue [%s]",
+                            e,
+                            source,
+                            task,
+                            name
+                        )
+                    );
+                    return;
+                }
             } else {
                 timeoutCancellable = null;
             }

server/src/test/java/org/elasticsearch/cluster/service/MasterServiceTests.java

@@ -2121,6 +2121,59 @@ public void onFailure(Exception e) {
         }
     }
 
+    public void testTimeoutRejectionBehaviourAtSubmission() {
+
+        final var source = randomIdentifier();
+        final var taskDescription = randomIdentifier();
+        final var timeout = TimeValue.timeValueMillis(between(0, 100000));
+
+        final var actionCount = new AtomicInteger();
+        final var deterministicTaskQueue = new DeterministicTaskQueue();
+        final var threadPool =
+            // a threadpool which simulates the rejection of a master service timeout handler, but runs all other tasks as normal
+            deterministicTaskQueue.getThreadPool(r -> {
+                if (r.toString().equals(MasterService.getTimeoutTaskDescription(source, taskDescription, timeout))) {
+                    // assertTrue because this should happen exactly once
+                    assertTrue(actionCount.compareAndSet(0, 1));
+                    throw new EsRejectedExecutionException("simulated rejection", true);
+                } else {
+                    return r;
+                }
+            });
+
+        try (var masterService = createMasterService(true, null, threadPool, new StoppableExecutorServiceWrapper(threadPool.generic()))) {
+            masterService.createTaskQueue(
+                "queue",
+                randomFrom(Priority.values()),
+                batchExecutionContext -> fail(null, "should not execute batch")
+            ).submitTask(source, new ClusterStateTaskListener() {
+                @Override
+                public void onFailure(Exception e) {
+                    if (e instanceof FailedToCommitClusterStateException
+                        && e.getMessage().startsWith("could not schedule timeout handler")
+                        && e.getCause() instanceof EsRejectedExecutionException esre
+                        && esre.isExecutorShutdown()
+                        && esre.getMessage().equals("simulated rejection")) {
+                        // assertTrue because we must receive the exception we synthesized, exactly once, after triggering the rejection
+                        assertTrue(actionCount.compareAndSet(1, 2));
+                    } else {
+                        // fail the test if we get anything else
+                        throw new AssertionError("unexpected exception", e);
+                    }
+                }
+
+                @Override
+                public String toString() {
+                    return taskDescription;
+                }
+            }, timeout);
+
+            assertFalse(deterministicTaskQueue.hasRunnableTasks());
+            assertFalse(deterministicTaskQueue.hasDeferredTasks());
+            assertEquals(2, actionCount.get()); // ensures this test doesn't accidentally become trivial: both expected actions happened
+        }
+    }
+
     @TestLogging(reason = "verifying DEBUG logs", value = "org.elasticsearch.cluster.service.MasterService:DEBUG")
     public void testRejectionBehaviourAtCompletion() {
 

test/framework/src/main/java/org/elasticsearch/test/ESIntegTestCase.java

@@ -177,6 +177,7 @@
 import java.util.concurrent.Callable;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -971,7 +972,17 @@ private ClusterHealthStatus ensureColor(
             // been removed by the master so that the health check applies to the set of nodes we expect to be part of the cluster.
             .waitForNodes(Integer.toString(cluster().size()));
 
-        final ClusterHealthResponse clusterHealthResponse = clusterAdmin().health(healthRequest).actionGet();
+        final ClusterHealthResponse clusterHealthResponse;
+        try {
+            clusterHealthResponse = clusterAdmin().health(healthRequest).get();
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            logger.error("interrupted while waiting for health response", e);
+            throw new AssertionError("interrupted while waiting for health response", e);
+        } catch (ExecutionException e) {
+            logger.error("failed to get health response", e);
+            throw new AssertionError("failed to get health response", e);
+        }
         if (clusterHealthResponse.isTimedOut()) {
             final var allocationExplainRef = new AtomicReference<ClusterAllocationExplainResponse>();
             final var clusterStateRef = new AtomicReference<ClusterStateResponse>();