Dedicated API for unbatched master tasks

Today when submitting a cluster state update task the caller can
indicate that it is not to be included in a batch by using an executor
obtained from `ClusterStateTaskExecutor#unbatched()`. This is kind of
weird: the caller must not re-use the executor, and also the master
service has no practical way to handle this special case any
differently.

This commit introduces a dedicated API for unbatched tasks, opening the
door to some future simplifications.

Extracted from elastic#85751

Author: DaveCTurner

build-tools-internal/src/main/resources/forbidden/es-server-signatures.txt

@@ -139,4 +139,5 @@ org.apache.logging.log4j.LogManager#getLogger()
 java.lang.String#formatted(java.lang.Object[]) @ Uses default locale - use String#format(Locale, String, Object...) instead
 
 @defaultMessage Unbatched cluster state tasks are a source of performance and stability bugs. Implement the update logic in a executor which is reused across tasks instead.
-org.elasticsearch.cluster.ClusterStateTaskExecutor#unbatched()
+org.elasticsearch.cluster.service.MasterService#submitUnbatchedStateUpdateTask(java.lang.String, org.elasticsearch.cluster.ClusterStateUpdateTask)
+org.elasticsearch.cluster.service.ClusterService#submitUnbatchedStateUpdateTask(java.lang.String, org.elasticsearch.cluster.ClusterStateUpdateTask)

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

@@ -11,7 +11,6 @@
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.elasticsearch.cluster.ClusterState;
-import org.elasticsearch.cluster.ClusterStateTaskExecutor;
 import org.elasticsearch.cluster.ClusterStateUpdateTask;
 import org.elasticsearch.cluster.LocalNodeMasterListener;
 import org.elasticsearch.cluster.metadata.DataStream;
@@ -60,7 +59,7 @@ public class UpdateTimeSeriesRangeService extends AbstractLifecycleComponent imp
     void perform(Runnable onComplete) {
         if (running.compareAndSet(false, true)) {
             LOGGER.debug("starting tsdb update task");
-            clusterService.submitStateUpdateTask("update_tsdb_data_stream_end_times", new ClusterStateUpdateTask(Priority.URGENT) {
+            submitUnbatchedTask("update_tsdb_data_stream_end_times", new ClusterStateUpdateTask(Priority.URGENT) {
                 @Override
                 public ClusterState execute(ClusterState currentState) throws Exception {
                     return updateTimeSeriesTemporalRange(currentState, Instant.now());
@@ -79,15 +78,15 @@ public void onFailure(Exception e) {
                     onComplete.run();
                 }
 
-            }, newExecutor());
+            });
         } else {
             LOGGER.debug("not starting tsdb update task, because another execution is still running");
         }
     }
 
     @SuppressForbidden(reason = "legacy usage of unbatched task") // TODO add support for batching here
-    private static <T extends ClusterStateUpdateTask> ClusterStateTaskExecutor<T> newExecutor() {
-        return ClusterStateTaskExecutor.unbatched();
+    private void submitUnbatchedTask(@SuppressWarnings("SameParameterValue") String source, ClusterStateUpdateTask task) {
+        clusterService.submitUnbatchedStateUpdateTask(source, task);
     }
 
     void setPollInterval(TimeValue newValue) {

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

@@ -16,7 +16,6 @@
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.action.support.master.AcknowledgedTransportMasterNodeAction;
 import org.elasticsearch.cluster.ClusterState;
-import org.elasticsearch.cluster.ClusterStateTaskExecutor;
 import org.elasticsearch.cluster.ClusterStateUpdateTask;
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
@@ -89,7 +88,7 @@ protected void masterOperation(
             systemIndices.validateDataStreamAccess(name, threadPool.getThreadContext());
         }
 
-        clusterService.submitStateUpdateTask(
+        submitUnbatchedTask(
             "remove-data-stream [" + Strings.arrayToCommaDelimitedString(request.getNames()) + "]",
             new ClusterStateUpdateTask(Priority.HIGH, request.masterNodeTimeout()) {
 
@@ -113,14 +112,13 @@ public ClusterState execute(ClusterState currentState) {
                 public void clusterStateProcessed(ClusterState oldState, ClusterState newState) {
                     listener.onResponse(AcknowledgedResponse.TRUE);
                 }
-            },
-            newExecutor()
+            }
         );
     }
 
     @SuppressForbidden(reason = "legacy usage of unbatched task") // TODO add support for batching here
-    private static <T extends ClusterStateUpdateTask> ClusterStateTaskExecutor<T> newExecutor() {
-        return ClusterStateTaskExecutor.unbatched();
+    private void submitUnbatchedTask(@SuppressWarnings("SameParameterValue") String source, ClusterStateUpdateTask task) {
+        clusterService.submitUnbatchedStateUpdateTask(source, task);
     }
 
     static ClusterState removeDataStream(

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

@@ -14,7 +14,6 @@
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.action.support.master.AcknowledgedTransportMasterNodeAction;
 import org.elasticsearch.cluster.ClusterState;
-import org.elasticsearch.cluster.ClusterStateTaskExecutor;
 import org.elasticsearch.cluster.ClusterStateUpdateTask;
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
@@ -64,7 +63,7 @@ protected void masterOperation(
         ActionListener<AcknowledgedResponse> listener
     ) throws Exception {
         systemIndices.validateDataStreamAccess(request.getName(), threadPool.getThreadContext());
-        clusterService.submitStateUpdateTask(
+        submitUnbatchedTask(
             "promote-data-stream [" + request.getName() + "]",
             new ClusterStateUpdateTask(Priority.HIGH, request.masterNodeTimeout()) {
 
@@ -82,14 +81,13 @@ public ClusterState execute(ClusterState currentState) {
                 public void clusterStateProcessed(ClusterState oldState, ClusterState newState) {
                     listener.onResponse(AcknowledgedResponse.TRUE);
                 }
-            },
-            newExecutor()
+            }
         );
     }
 
     @SuppressForbidden(reason = "legacy usage of unbatched task") // TODO add support for batching here
-    private static <T extends ClusterStateUpdateTask> ClusterStateTaskExecutor<T> newExecutor() {
-        return ClusterStateTaskExecutor.unbatched();
+    private void submitUnbatchedTask(@SuppressWarnings("SameParameterValue") String source, ClusterStateUpdateTask task) {
+        clusterService.submitUnbatchedStateUpdateTask(source, task);
     }
 
     static ClusterState promoteDataStream(ClusterState currentState, PromoteDataStreamAction.Request request) {

modules/reindex/src/internalClusterTest/java/org/elasticsearch/migration/FeatureMigrationIT.java

@@ -19,7 +19,6 @@
 import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
 import org.elasticsearch.action.support.ActiveShardCount;
 import org.elasticsearch.cluster.ClusterState;
-import org.elasticsearch.cluster.ClusterStateTaskExecutor;
 import org.elasticsearch.cluster.ClusterStateUpdateTask;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.metadata.Metadata;
@@ -262,7 +261,7 @@ public void testMigrationWillRunAfterError() throws Exception {
         SetOnce<Exception> failure = new SetOnce<>();
         CountDownLatch clusterStateUpdated = new CountDownLatch(1);
         internalCluster().getCurrentMasterNodeInstance(ClusterService.class)
-            .submitStateUpdateTask(this.getTestName(), new ClusterStateUpdateTask() {
+            .submitUnbatchedStateUpdateTask(this.getTestName(), new ClusterStateUpdateTask() {
                 @Override
                 public ClusterState execute(ClusterState currentState) throws Exception {
                     FeatureMigrationResults newResults = new FeatureMigrationResults(
@@ -287,7 +286,7 @@ public void onFailure(Exception e) {
                     failure.set(e);
                     clusterStateUpdated.countDown();
                 }
-            }, ClusterStateTaskExecutor.unbatched());
+            });
 
         clusterStateUpdated.await(10, TimeUnit.SECONDS); // Should be basically instantaneous
         if (failure.get() != null) {

qa/smoke-test-http/src/test/java/org/elasticsearch/http/ClusterStateRestCancellationIT.java

@@ -16,7 +16,6 @@
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.Response;
 import org.elasticsearch.cluster.ClusterState;
-import org.elasticsearch.cluster.ClusterStateTaskExecutor;
 import org.elasticsearch.cluster.ClusterStateUpdateTask;
 import org.elasticsearch.cluster.SimpleDiffable;
 import org.elasticsearch.cluster.service.ClusterService;
@@ -45,7 +44,7 @@ protected Collection<Class<? extends Plugin>> nodePlugins() {
 
     private void updateClusterState(ClusterService clusterService, UnaryOperator<ClusterState> updateOperator) {
         final PlainActionFuture<Void> future = new PlainActionFuture<>();
-        clusterService.submitStateUpdateTask("update state", new ClusterStateUpdateTask() {
+        clusterService.submitUnbatchedStateUpdateTask("update state", new ClusterStateUpdateTask() {
             @Override
             public ClusterState execute(ClusterState currentState) {
                 return updateOperator.apply(currentState);
@@ -60,7 +59,7 @@ public void onFailure(Exception e) {
             public void clusterStateProcessed(ClusterState oldState, ClusterState newState) {
                 future.onResponse(null);
             }
-        }, ClusterStateTaskExecutor.unbatched());
+        });
         future.actionGet();
     }
 

qa/smoke-test-http/src/test/java/org/elasticsearch/http/RestGetMappingsCancellationIT.java

@@ -17,7 +17,6 @@
 import org.elasticsearch.client.Response;
 import org.elasticsearch.cluster.AckedClusterStateUpdateTask;
 import org.elasticsearch.cluster.ClusterState;
-import org.elasticsearch.cluster.ClusterStateTaskExecutor;
 import org.elasticsearch.cluster.ack.AckedRequest;
 import org.elasticsearch.cluster.block.ClusterBlock;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
@@ -98,12 +97,12 @@ public TimeValue masterNodeTimeout() {
 
         PlainActionFuture<AcknowledgedResponse> future = PlainActionFuture.newFuture();
         internalCluster().getAnyMasterNodeInstance(ClusterService.class)
-            .submitStateUpdateTask("get_mappings_cancellation_test", new AckedClusterStateUpdateTask(ackedRequest, future) {
+            .submitUnbatchedStateUpdateTask("get_mappings_cancellation_test", new AckedClusterStateUpdateTask(ackedRequest, future) {
                 @Override
                 public ClusterState execute(ClusterState currentState) throws Exception {
                     return transformationFn.apply(currentState);
                 }
-            }, ClusterStateTaskExecutor.unbatched());
+            });
 
         future.actionGet();
     }

server/src/internalClusterTest/java/org/elasticsearch/action/admin/cluster/desirednodes/TransportDesiredNodesActionsIT.java

@@ -13,7 +13,6 @@
 import org.elasticsearch.action.ActionFuture;
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.cluster.ClusterState;
-import org.elasticsearch.cluster.ClusterStateTaskExecutor;
 import org.elasticsearch.cluster.ClusterStateUpdateTask;
 import org.elasticsearch.cluster.desirednodes.VersionConflictException;
 import org.elasticsearch.cluster.metadata.DesiredNodes;
@@ -373,7 +372,7 @@ private Runnable blockClusterStateUpdateThread() throws InterruptedException {
         final CountDownLatch unblockClusterStateUpdateTask = new CountDownLatch(1);
         final CountDownLatch blockingClusterStateUpdateTaskExecuting = new CountDownLatch(1);
         final ClusterService clusterService = internalCluster().getCurrentMasterNodeInstance(ClusterService.class);
-        clusterService.submitStateUpdateTask("blocking-task", new ClusterStateUpdateTask(Priority.IMMEDIATE) {
+        clusterService.submitUnbatchedStateUpdateTask("blocking-task", new ClusterStateUpdateTask(Priority.IMMEDIATE) {
             @Override
             public ClusterState execute(ClusterState currentState) throws Exception {
                 blockingClusterStateUpdateTaskExecuting.countDown();
@@ -386,7 +385,7 @@ public void onFailure(Exception e) {
                 blockingClusterStateUpdateTaskExecuting.countDown();
                 assert false : e.getMessage();
             }
-        }, ClusterStateTaskExecutor.unbatched());
+        });
 
         assertTrue(blockingClusterStateUpdateTaskExecuting.await(10, TimeUnit.SECONDS));
         return unblockClusterStateUpdateTask::countDown;

server/src/internalClusterTest/java/org/elasticsearch/cluster/ClusterHealthIT.java

@@ -286,7 +286,7 @@ public void testWaitForEventsRetriesIfOtherConditionsNotMet() {
         final AtomicBoolean keepSubmittingTasks = new AtomicBoolean(true);
         final ClusterService clusterService = internalCluster().getInstance(ClusterService.class, internalCluster().getMasterName());
         final PlainActionFuture<Void> completionFuture = new PlainActionFuture<>();
-        clusterService.submitStateUpdateTask("looping task", new ClusterStateUpdateTask(Priority.LOW) {
+        clusterService.submitUnbatchedStateUpdateTask("looping task", new ClusterStateUpdateTask(Priority.LOW) {
             @Override
             public ClusterState execute(ClusterState currentState) {
                 return currentState;
@@ -301,12 +301,12 @@ public void onFailure(Exception e) {
             @Override
             public void clusterStateProcessed(ClusterState oldState, ClusterState newState) {
                 if (keepSubmittingTasks.get()) {
-                    clusterService.submitStateUpdateTask("looping task", this, ClusterStateTaskExecutor.unbatched());
+                    clusterService.submitUnbatchedStateUpdateTask("looping task", this);
                 } else {
                     completionFuture.onResponse(null);
                 }
             }
-        }, ClusterStateTaskExecutor.unbatched());
+        });
 
         try {
             createIndex("index");
@@ -377,7 +377,7 @@ public void testWaitForEventsTimesOutIfMasterBusy() {
         final AtomicBoolean keepSubmittingTasks = new AtomicBoolean(true);
         final ClusterService clusterService = internalCluster().getInstance(ClusterService.class, internalCluster().getMasterName());
         final PlainActionFuture<Void> completionFuture = new PlainActionFuture<>();
-        clusterService.submitStateUpdateTask("looping task", new ClusterStateUpdateTask(Priority.LOW) {
+        clusterService.submitUnbatchedStateUpdateTask("looping task", new ClusterStateUpdateTask(Priority.LOW) {
             @Override
             public ClusterState execute(ClusterState currentState) {
                 return currentState;
@@ -392,12 +392,12 @@ public void onFailure(Exception e) {
             @Override
             public void clusterStateProcessed(ClusterState oldState, ClusterState newState) {
                 if (keepSubmittingTasks.get()) {
-                    clusterService.submitStateUpdateTask("looping task", this, ClusterStateTaskExecutor.unbatched());
+                    clusterService.submitUnbatchedStateUpdateTask("looping task", this);
                 } else {
                     completionFuture.onResponse(null);
                 }
             }
-        }, ClusterStateTaskExecutor.unbatched());
+        });
 
         try {
             final ClusterHealthResponse clusterHealthResponse = client().admin()

server/src/internalClusterTest/java/org/elasticsearch/cluster/MinimumMasterNodesIT.java

@@ -333,7 +333,7 @@ public void testCannotCommitStateThreeNodes() throws Exception {
         final AtomicReference<Exception> failure = new AtomicReference<>();
         logger.debug("--> submitting for cluster state to be rejected");
         final ClusterService masterClusterService = internalCluster().clusterService(master);
-        masterClusterService.submitStateUpdateTask("test", new ClusterStateUpdateTask() {
+        masterClusterService.submitUnbatchedStateUpdateTask("test", new ClusterStateUpdateTask() {
             @Override
             public void clusterStateProcessed(ClusterState oldState, ClusterState newState) {
                 latch.countDown();
@@ -355,7 +355,7 @@ public void onFailure(Exception e) {
                 failure.set(e);
                 latch.countDown();
             }
-        }, ClusterStateTaskExecutor.unbatched());
+        });
 
         logger.debug("--> waiting for cluster state to be processed/rejected");
         latch.await();