Preventing exceptions on node shutdown in integration tests (#88827)

Author: masseyke

server/src/main/java/org/elasticsearch/cluster/coordination/CoordinationDiagnosticsService.java

@@ -24,6 +24,7 @@
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.settings.Setting;
+import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.Releasable;
 import org.elasticsearch.core.Releasables;
@@ -650,12 +651,20 @@ void beginPollingClusterFormationInfo(
     ) {
         masterEligibleNodes.forEach(masterEligibleNode -> {
             Consumer<ClusterFormationStateOrException> responseConsumer = result -> nodeResponseConsumer.accept(masterEligibleNode, result);
-            cancellableConsumer.accept(
-                fetchClusterFormationInfo(
-                    masterEligibleNode,
-                    responseConsumer.andThen(rescheduleFetchConsumer(masterEligibleNode, responseConsumer, cancellableConsumer))
-                )
-            );
+            try {
+                cancellableConsumer.accept(
+                    fetchClusterFormationInfo(
+                        masterEligibleNode,
+                        responseConsumer.andThen(rescheduleFetchConsumer(masterEligibleNode, responseConsumer, cancellableConsumer))
+                    )
+                );
+            } catch (EsRejectedExecutionException e) {
+                if (e.isExecutorShutdown()) {
+                    logger.trace("Not rescheduling request for cluster coordination info because this node is being shutdown", e);
+                } else {
+                    throw e;
+                }
+            }
         });
     }
 
@@ -673,12 +682,20 @@ private Consumer<CoordinationDiagnosticsService.ClusterFormationStateOrException
         Consumer<Scheduler.Cancellable> cancellableConsumer
     ) {
         return response -> {
-            cancellableConsumer.accept(
-                fetchClusterFormationInfo(
-                    masterEligibleNode,
-                    responseConsumer.andThen(rescheduleFetchConsumer(masterEligibleNode, responseConsumer, cancellableConsumer))
-                )
-            );
+            try {
+                cancellableConsumer.accept(
+                    fetchClusterFormationInfo(
+                        masterEligibleNode,
+                        responseConsumer.andThen(rescheduleFetchConsumer(masterEligibleNode, responseConsumer, cancellableConsumer))
+                    )
+                );
+            } catch (EsRejectedExecutionException e) {
+                if (e.isExecutorShutdown()) {
+                    logger.trace("Not rescheduling request for cluster coordination info because this node is being shutdown", e);
+                } else {
+                    throw e;
+                }
+            }
         };
     }
 
@@ -697,6 +714,7 @@ private void cancelPollingClusterFormationInfo() {
      * @param node The node to poll for cluster formation information
      * @param responseConsumer The consumer of the cluster formation info for the node, or the exception encountered while contacting it
      * @return A Cancellable for the task that is scheduled to fetch cluster formation information
+     * @throws EsRejectedExecutionException If the task cannot be scheduled, possibly because the node is shutting down.
      */
     private Scheduler.Cancellable fetchClusterFormationInfo(
         DiscoveryNode node,