Radix-sorted, lookup-free, typesafe, nonblocking, priority-boosting pending tasks

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/internalClusterTest/java/org/elasticsearch/cluster/service/ClusterServiceIT.java

@@ -17,6 +17,7 @@
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.ESIntegTestCase.ClusterScope;
 import org.elasticsearch.test.ESIntegTestCase.Scope;
+import org.elasticsearch.threadpool.ThreadPool;
 
 import java.util.Arrays;
 import java.util.HashSet;
@@ -25,6 +26,7 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.StreamSupport;
 
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThan;
@@ -414,11 +416,7 @@ public void onFailure(Exception e) {
                 });
             }
 
-            final var startNanoTime = System.nanoTime();
-            while (TimeUnit.MILLISECONDS.convert(System.nanoTime() - startNanoTime, TimeUnit.NANOSECONDS) <= 0) {
-                // noinspection BusyWait
-                Thread.sleep(100);
-            }
+            waitForTimeToElapse();
 
             pendingClusterTasks = clusterService.getMasterService().pendingTasks();
             assertThat(pendingClusterTasks.size(), greaterThanOrEqualTo(5));
@@ -441,4 +439,28 @@ public void onFailure(Exception e) {
             block2.countDown();
         }
     }
+
+    private static void waitForTimeToElapse() throws InterruptedException {
+        final ThreadPool[] threadPools = StreamSupport.stream(internalCluster().getInstances(ClusterService.class).spliterator(), false)
+            .map(ClusterService::threadPool)
+            .toArray(ThreadPool[]::new);
+        final long[] startTimes = Arrays.stream(threadPools).mapToLong(ThreadPool::relativeTimeInMillis).toArray();
+
+        final var startNanoTime = System.nanoTime();
+        while (TimeUnit.MILLISECONDS.convert(System.nanoTime() - startNanoTime, TimeUnit.NANOSECONDS) <= 100) {
+            // noinspection BusyWait
+            Thread.sleep(100);
+        }
+
+        outer: do {
+            for (int i = 0; i < threadPools.length; i++) {
+                if (threadPools[i].relativeTimeInMillis() <= startTimes[i]) {
+                    // noinspection BusyWait
+                    Thread.sleep(100);
+                    continue outer;
+                }
+            }
+            return;
+        } while (true);
+    }
 }

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);