Batch ILM policy cluster state updates [elastic#122917] (elastic#126529) (elastic#126587)

* Use a task queue to ensure ILM policy change cluster state updates are batched

* Update docs/changelog/126529.yaml

* Update docs/changelog/126529.yaml

* Switch to using SimpleBatchedAckListenerTaskExecutor

* Get timeout from request

* Ditch the try-catch

(cherry picked from commit 7d7fa76)

# Conflicts:
#	x-pack/plugin/ilm/src/main/java/org/elasticsearch/xpack/ilm/action/TransportPutLifecycleAction.java

Author: lukewhiting

docs/changelog/126529.yaml

@@ -0,0 +1,6 @@
+pr: 126529
+summary: "Batch ILM policy cluster state updates [#122917]"
+area: ILM+SLM
+type: enhancement
+issues:
+ - 122917

x-pack/plugin/ilm/src/main/java/org/elasticsearch/xpack/ilm/action/TransportPutLifecycleAction.java

@@ -16,15 +16,18 @@
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.cluster.AckedClusterStateUpdateTask;
 import org.elasticsearch.cluster.ClusterState;
-import org.elasticsearch.cluster.ClusterStateUpdateTask;
+import org.elasticsearch.cluster.ClusterStateAckListener;
+import org.elasticsearch.cluster.SimpleBatchedAckListenerTaskExecutor;
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.cluster.metadata.RepositoriesMetadata;
 import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.cluster.service.MasterServiceTaskQueue;
+import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.core.Nullable;
-import org.elasticsearch.core.SuppressForbidden;
+import org.elasticsearch.core.Tuple;
 import org.elasticsearch.injection.guice.Inject;
 import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.reservedstate.ReservedClusterStateHandler;
@@ -66,6 +69,7 @@ public class TransportPutLifecycleAction extends TransportMasterNodeAction<PutLi
     private final NamedXContentRegistry xContentRegistry;
     private final Client client;
     private final XPackLicenseState licenseState;
+    private final MasterServiceTaskQueue<UpdateLifecyclePolicyTask> taskQueue;
 
     @Inject
     public TransportPutLifecycleAction(
@@ -90,6 +94,7 @@ public TransportPutLifecycleAction(
         this.xContentRegistry = namedXContentRegistry;
         this.licenseState = licenseState;
         this.client = client;
+        this.taskQueue = clusterService.createTaskQueue("ilm-put-lifecycle-queue", Priority.NORMAL, new IlmLifecycleExecutor());
     }
 
     @Override
@@ -117,10 +122,15 @@ protected void masterOperation(
             }
         }
 
-        submitUnbatchedTask(
-            "put-lifecycle-" + request.getPolicy().getName(),
-            new UpdateLifecyclePolicyTask(request, listener, licenseState, filteredHeaders, xContentRegistry, client)
+        UpdateLifecyclePolicyTask putTask = new UpdateLifecyclePolicyTask(
+            request,
+            listener,
+            licenseState,
+            filteredHeaders,
+            xContentRegistry,
+            client
         );
+        taskQueue.submitTask("put-lifecycle-" + request.getPolicy().getName(), putTask, putTask.timeout());
     }
 
     public static class UpdateLifecyclePolicyTask extends AckedClusterStateUpdateTask {
@@ -223,11 +233,6 @@ public ClusterState execute(ClusterState currentState) throws Exception {
         }
     }
 
-    @SuppressForbidden(reason = "legacy usage of unbatched task") // TODO add support for batching here
-    private void submitUnbatchedTask(@SuppressWarnings("SameParameterValue") String source, ClusterStateUpdateTask task) {
-        clusterService.submitUnbatchedStateUpdateTask(source, task);
-    }
-
     /**
      * Returns 'true' if the ILM policy is effectually the same (same policy and headers), and thus can be a no-op update.
      */
@@ -328,4 +333,15 @@ public Optional<String> reservedStateHandlerName() {
     public Set<String> modifiedKeys(PutLifecycleRequest request) {
         return Set.of(request.getPolicy().getName());
     }
+
+    private static class IlmLifecycleExecutor extends SimpleBatchedAckListenerTaskExecutor<UpdateLifecyclePolicyTask> {
+
+        @Override
+        public Tuple<ClusterState, ClusterStateAckListener> executeTask(UpdateLifecyclePolicyTask task, ClusterState clusterState)
+            throws Exception {
+            return Tuple.tuple(task.execute(clusterState), task);
+        }
+
+    }
+
 }