Improve master service batching queues

docs/changelog/92021.yaml

@@ -0,0 +1,6 @@
+pr: 92021
+summary: Improve master service batching queues
+area: Cluster Coordination
+type: enhancement
+issues:
+ - 81626

modules/data-streams/src/main/java/org/elasticsearch/datastreams/UpdateTimeSeriesRangeService.java

@@ -10,14 +10,14 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.cluster.ClusterState;
-import org.elasticsearch.cluster.ClusterStateTaskConfig;
 import org.elasticsearch.cluster.ClusterStateTaskExecutor;
 import org.elasticsearch.cluster.ClusterStateTaskListener;
 import org.elasticsearch.cluster.LocalNodeMasterListener;
 import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.cluster.service.MasterServiceTaskQueue;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.component.AbstractLifecycleComponent;
 import org.elasticsearch.common.settings.Settings;
@@ -50,13 +50,14 @@ public class UpdateTimeSeriesRangeService extends AbstractLifecycleComponent imp
     volatile TimeValue pollInterval;
     volatile Scheduler.Cancellable job;
     private final AtomicBoolean running = new AtomicBoolean(false);
-    private final ClusterStateTaskExecutor<UpdateTimeSeriesTask> taskExecutor = new UpdateTimeSeriesExecutor();
+    private final MasterServiceTaskQueue<UpdateTimeSeriesTask> taskQueue;
 
     UpdateTimeSeriesRangeService(Settings settings, ThreadPool threadPool, ClusterService clusterService) {
         this.pollInterval = DataStreamsPlugin.TIME_SERIES_POLL_INTERVAL.get(settings);
         this.threadPool = threadPool;
         this.clusterService = clusterService;
         clusterService.getClusterSettings().addSettingsUpdateConsumer(DataStreamsPlugin.TIME_SERIES_POLL_INTERVAL, this::setPollInterval);
+        this.taskQueue = clusterService.getTaskQueue("update-time-series-range", Priority.URGENT, new UpdateTimeSeriesExecutor());
     }
 
     void perform(Runnable onComplete) {
@@ -69,8 +70,7 @@ void perform(Runnable onComplete) {
                 running.set(false);
                 onComplete.run();
             });
-            var config = ClusterStateTaskConfig.build(Priority.URGENT);
-            clusterService.submitStateUpdateTask("update_tsdb_data_stream_end_times", task, config, taskExecutor);
+            taskQueue.submitTask("update_tsdb_data_stream_end_times", task, null);
         } else {
             LOGGER.debug("not starting tsdb update task, because another execution is still running");
         }

qa/evil-tests/src/test/java/org/elasticsearch/threadpool/EvilThreadPoolTests.java

@@ -98,8 +98,7 @@ public void testExecutionErrorOnSinglePrioritizingThreadPoolExecutor() throws In
             "test",
             EsExecutors.daemonThreadFactory("test"),
             threadPool.getThreadContext(),
-            threadPool.scheduler(),
-            PrioritizedEsThreadPoolExecutor.StarvationWatcher.NOOP_STARVATION_WATCHER
+            threadPool.scheduler()
         );
         try {
             checkExecutionError(getExecuteRunner(prioritizedExecutor));
@@ -208,8 +207,7 @@ public void testExecutionExceptionOnSinglePrioritizingThreadPoolExecutor() throw
             "test",
             EsExecutors.daemonThreadFactory("test"),
             threadPool.getThreadContext(),
-            threadPool.scheduler(),
-            PrioritizedEsThreadPoolExecutor.StarvationWatcher.NOOP_STARVATION_WATCHER
+            threadPool.scheduler()
         );
         try {
             checkExecutionException(getExecuteRunner(prioritizedExecutor), true);

server/src/internalClusterTest/java/org/elasticsearch/action/support/AutoCreateIndexIT.java

@@ -10,7 +10,6 @@
 
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.index.IndexResponse;
-import org.elasticsearch.cluster.ClusterStateTaskConfig;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.test.ESIntegTestCase;
@@ -28,17 +27,12 @@ public class AutoCreateIndexIT extends ESIntegTestCase {
     public void testBatchingWithDeprecationWarnings() throws Exception {
         final var masterNodeClusterService = internalCluster().getCurrentMasterNodeInstance(ClusterService.class);
         final var barrier = new CyclicBarrier(2);
-        masterNodeClusterService.submitStateUpdateTask(
-            "block",
-            e -> { assert false : e; },
-            ClusterStateTaskConfig.build(Priority.NORMAL),
-            batchExecutionContext -> {
-                barrier.await(10, TimeUnit.SECONDS);
-                barrier.await(10, TimeUnit.SECONDS);
-                batchExecutionContext.taskContexts().forEach(c -> c.success(() -> {}));
-                return batchExecutionContext.initialState();
-            }
-        );
+        masterNodeClusterService.getTaskQueue("block", Priority.NORMAL, batchExecutionContext -> {
+            barrier.await(10, TimeUnit.SECONDS);
+            barrier.await(10, TimeUnit.SECONDS);
+            batchExecutionContext.taskContexts().forEach(c -> c.success(() -> {}));
+            return batchExecutionContext.initialState();
+        }).submitTask("block", e -> { assert false : e; }, null);
 
         barrier.await(10, TimeUnit.SECONDS);
 

server/src/main/java/org/elasticsearch/action/admin/cluster/desirednodes/TransportDeleteDesiredNodesAction.java

@@ -13,15 +13,14 @@
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.master.TransportMasterNodeAction;
 import org.elasticsearch.cluster.ClusterState;
-import org.elasticsearch.cluster.ClusterStateTaskConfig;
-import org.elasticsearch.cluster.ClusterStateTaskExecutor;
 import org.elasticsearch.cluster.ClusterStateTaskListener;
 import org.elasticsearch.cluster.SimpleBatchedExecutor;
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
 import org.elasticsearch.cluster.metadata.DesiredNodesMetadata;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.cluster.service.MasterServiceTaskQueue;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.core.Tuple;
@@ -31,7 +30,7 @@
 
 public class TransportDeleteDesiredNodesAction extends TransportMasterNodeAction<DeleteDesiredNodesAction.Request, ActionResponse.Empty> {
 
-    private final ClusterStateTaskExecutor<DeleteDesiredNodesTask> taskExecutor = new DeleteDesiredNodesExecutor();
+    private final MasterServiceTaskQueue<DeleteDesiredNodesTask> taskQueue;
 
     @Inject
     public TransportDeleteDesiredNodesAction(
@@ -52,6 +51,7 @@ public TransportDeleteDesiredNodesAction(
             in -> ActionResponse.Empty.INSTANCE,
             ThreadPool.Names.SAME
         );
+        this.taskQueue = clusterService.getTaskQueue("delete-desired-nodes", Priority.HIGH, new DeleteDesiredNodesExecutor());
     }
 
     @Override
@@ -61,12 +61,7 @@ protected void masterOperation(
         ClusterState state,
         ActionListener<ActionResponse.Empty> listener
     ) throws Exception {
-        clusterService.submitStateUpdateTask(
-            "delete-desired-nodes",
-            new DeleteDesiredNodesTask(listener),
-            ClusterStateTaskConfig.build(Priority.HIGH, request.masterNodeTimeout()),
-            taskExecutor
-        );
+        taskQueue.submitTask("delete-desired-nodes", new DeleteDesiredNodesTask(listener), request.masterNodeTimeout());
     }
 
     @Override

server/src/main/java/org/elasticsearch/action/admin/cluster/desirednodes/TransportUpdateDesiredNodesAction.java

@@ -14,7 +14,6 @@
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.master.TransportMasterNodeAction;
 import org.elasticsearch.cluster.ClusterState;
-import org.elasticsearch.cluster.ClusterStateTaskConfig;
 import org.elasticsearch.cluster.ClusterStateTaskExecutor;
 import org.elasticsearch.cluster.ClusterStateTaskListener;
 import org.elasticsearch.cluster.block.ClusterBlockException;
@@ -27,6 +26,7 @@
 import org.elasticsearch.cluster.routing.RerouteService;
 import org.elasticsearch.cluster.routing.allocation.AllocationService;
 import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.cluster.service.MasterServiceTaskQueue;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.tasks.Task;
@@ -41,7 +41,7 @@ public class TransportUpdateDesiredNodesAction extends TransportMasterNodeAction
     private static final Logger logger = LogManager.getLogger(TransportUpdateDesiredNodesAction.class);
 
     private final DesiredNodesSettingsValidator settingsValidator;
-    private final ClusterStateTaskExecutor<UpdateDesiredNodesTask> taskExecutor;
+    private final MasterServiceTaskQueue<UpdateDesiredNodesTask> taskQueue;
 
     @Inject
     public TransportUpdateDesiredNodesAction(
@@ -66,7 +66,11 @@ public TransportUpdateDesiredNodesAction(
             ThreadPool.Names.SAME
         );
         this.settingsValidator = settingsValidator;
-        this.taskExecutor = new UpdateDesiredNodesExecutor(clusterService.getRerouteService(), allocationService);
+        this.taskQueue = clusterService.getTaskQueue(
+            "update-desired-nodes",
+            Priority.URGENT,
+            new UpdateDesiredNodesExecutor(clusterService.getRerouteService(), allocationService)
+        );
     }
 
     @Override
@@ -83,12 +87,7 @@ protected void masterOperation(
     ) throws Exception {
         try {
             settingsValidator.validate(request.getNodes());
-            clusterService.submitStateUpdateTask(
-                "update-desired-nodes",
-                new UpdateDesiredNodesTask(request, listener),
-                ClusterStateTaskConfig.build(Priority.URGENT, request.masterNodeTimeout()),
-                taskExecutor
-            );
+            taskQueue.submitTask("update-desired-nodes", new UpdateDesiredNodesTask(request, listener), request.masterNodeTimeout());
         } catch (Exception e) {
             listener.onFailure(e);
         }

server/src/main/java/org/elasticsearch/action/admin/cluster/health/TransportClusterHealthAction.java

@@ -22,6 +22,7 @@
 import org.elasticsearch.cluster.LocalMasterServiceTask;
 import org.elasticsearch.cluster.NotMasterException;
 import org.elasticsearch.cluster.block.ClusterBlockException;
+import org.elasticsearch.cluster.coordination.FailedToCommitClusterStateException;
 import org.elasticsearch.cluster.health.ClusterHealthStatus;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.metadata.ProcessClusterEventTimeoutException;
@@ -31,6 +32,7 @@
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.util.CollectionUtils;
+import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
 import org.elasticsearch.core.SuppressForbidden;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.IndexNotFoundException;
@@ -163,13 +165,20 @@ public void onFailure(Exception e) {
                     if (e instanceof ProcessClusterEventTimeoutException) {
                         listener.onResponse(getResponse(request, clusterService.state(), waitCount, TimeoutState.TIMED_OUT));
                     } else {
-                        final Level level = e instanceof NotMasterException ? Level.TRACE : Level.ERROR;
-                        assert e instanceof NotMasterException : e; // task cannot fail, nor will it trigger a publication which fails
+                        final Level level = isExpectedFailure(e) ? Level.TRACE : Level.ERROR;
                         logger.log(level, () -> "unexpected failure during [" + source + "]", e);
+                        assert isExpectedFailure(e) : e; // task cannot fail, nor will it trigger a publication which fails
                         // TransportMasterNodeAction implements the retry logic, which is triggered by passing a NotMasterException
                         listener.onFailure(e);
                     }
                 }
+
+                static boolean isExpectedFailure(Exception e) {
+                    return e instanceof NotMasterException
+                        || e instanceof FailedToCommitClusterStateException
+                            && e.getCause()instanceof EsRejectedExecutionException esre
+                            && esre.isExecutorShutdown();
+                }
             });
         }
     }

server/src/main/java/org/elasticsearch/action/admin/indices/create/AutoCreateAction.java

@@ -20,7 +20,6 @@
 import org.elasticsearch.action.support.master.TransportMasterNodeAction;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStateAckListener;
-import org.elasticsearch.cluster.ClusterStateTaskConfig;
 import org.elasticsearch.cluster.ClusterStateTaskExecutor;
 import org.elasticsearch.cluster.ClusterStateTaskListener;
 import org.elasticsearch.cluster.block.ClusterBlockException;
@@ -36,6 +35,7 @@
 import org.elasticsearch.cluster.routing.allocation.AllocationService;
 import org.elasticsearch.cluster.routing.allocation.allocator.AllocationActionMultiListener;
 import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.cluster.service.MasterServiceTaskQueue;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
@@ -77,7 +77,7 @@ public static final class TransportAction extends TransportMasterNodeAction<Crea
         private final AutoCreateIndex autoCreateIndex;
         private final SystemIndices systemIndices;
 
-        private final ClusterStateTaskExecutor<CreateIndexTask> executor;
+        private final MasterServiceTaskQueue<CreateIndexTask> taskQueue;
 
         @Inject
         public TransportAction(
@@ -107,7 +107,7 @@ public TransportAction(
             this.createIndexService = createIndexService;
             this.metadataCreateDataStreamService = metadataCreateDataStreamService;
             this.autoCreateIndex = autoCreateIndex;
-            this.executor = batchExecutionContext -> {
+            this.taskQueue = clusterService.getTaskQueue("auto-create", Priority.URGENT, batchExecutionContext -> {
                 final var listener = new AllocationActionMultiListener<CreateIndexResponse>(threadPool.getThreadContext());
                 final var taskContexts = batchExecutionContext.taskContexts();
                 final var successfulRequests = Maps.<CreateIndexRequest, String>newMapWithExpectedSize(taskContexts.size());
@@ -129,7 +129,7 @@ public TransportAction(
                     listener.noRerouteNeeded();
                 }
                 return state;
-            };
+            });
         }
 
         @Override
@@ -139,11 +139,10 @@ protected void masterOperation(
             ClusterState state,
             ActionListener<CreateIndexResponse> listener
         ) {
-            clusterService.submitStateUpdateTask(
+            taskQueue.submitTask(
                 "auto create [" + request.index() + "]",
                 new CreateIndexTask(request, listener),
-                ClusterStateTaskConfig.build(Priority.URGENT, request.masterNodeTimeout()),
-                executor
+                request.masterNodeTimeout()
             );
         }
 

server/src/main/java/org/elasticsearch/action/admin/indices/rollover/TransportRolloverAction.java

@@ -22,7 +22,6 @@
 import org.elasticsearch.action.support.master.TransportMasterNodeAction;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.cluster.ClusterState;
-import org.elasticsearch.cluster.ClusterStateTaskConfig;
 import org.elasticsearch.cluster.ClusterStateTaskExecutor;
 import org.elasticsearch.cluster.ClusterStateTaskListener;
 import org.elasticsearch.cluster.block.ClusterBlockException;
@@ -35,6 +34,7 @@
 import org.elasticsearch.cluster.routing.allocation.AllocationService;
 import org.elasticsearch.cluster.routing.allocation.allocator.AllocationActionMultiListener;
 import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.cluster.service.MasterServiceTaskQueue;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
@@ -64,7 +64,7 @@ public class TransportRolloverAction extends TransportMasterNodeAction<RolloverR
     private static final Logger logger = LogManager.getLogger(TransportRolloverAction.class);
 
     private final Client client;
-    private final RolloverExecutor rolloverTaskExecutor;
+    private final MasterServiceTaskQueue<RolloverTask> rolloverTaskQueue;
 
     @Inject
     public TransportRolloverAction(
@@ -89,7 +89,11 @@ public TransportRolloverAction(
             ThreadPool.Names.SAME
         );
         this.client = client;
-        this.rolloverTaskExecutor = new RolloverExecutor(clusterService, allocationService, rolloverService, threadPool);
+        this.rolloverTaskQueue = clusterService.getTaskQueue(
+            "rollover",
+            Priority.NORMAL,
+            new RolloverExecutor(clusterService, allocationService, rolloverService, threadPool)
+        );
     }
 
     @Override
@@ -173,8 +177,7 @@ protected void masterOperation(
                 if (rolloverRequest.areConditionsMet(trialConditionResults)) {
                     String source = "rollover_index source [" + trialRolloverIndexName + "] to target [" + trialRolloverIndexName + "]";
                     RolloverTask rolloverTask = new RolloverTask(rolloverRequest, statsResponse, trialRolloverResponse, listener);
-                    ClusterStateTaskConfig config = ClusterStateTaskConfig.build(Priority.NORMAL, rolloverRequest.masterNodeTimeout());
-                    clusterService.submitStateUpdateTask(source, rolloverTask, config, rolloverTaskExecutor);
+                    rolloverTaskQueue.submitTask(source, rolloverTask, rolloverRequest.masterNodeTimeout());
                 } else {
                     // conditions not met
                     listener.onResponse(trialRolloverResponse);

server/src/main/java/org/elasticsearch/cluster/ClusterState.java

@@ -73,8 +73,8 @@
  * across master elections (and therefore is preserved in a rolling restart).
  * <p>
  * Updates are triggered by submitting tasks to the {@link MasterService} on the elected master, typically using a {@link
- * TransportMasterNodeAction} to route a request to the master on which the task is submitted with {@link
- * ClusterService#submitStateUpdateTask}. Submitted tasks have an associated {@link ClusterStateTaskConfig} which defines a priority and a
+ * TransportMasterNodeAction} to route a request to the master on which the task is submitted via a queue obtained with {@link
+ * ClusterService#getTaskQueue}, which has an associated priority. Submitted tasks have an associated
  * timeout. Tasks are processed in priority order, so a flood of higher-priority tasks can starve lower-priority ones from running.
  * Therefore, avoid priorities other than {@link Priority#NORMAL} where possible. Tasks associated with client actions should typically have
  * a timeout, or otherwise be sensitive to client cancellations, to avoid surprises caused by the execution of stale tasks long after they