Remove non-state transforms (#110646)

Reserved cluster state is build from handlers that modify cluster state.
In the past there was one "non-state" handler that stored it's state
outside cluster state. However, with that use case now gone, non-state
transforms are no longer needed. This commit removes support for
non-state transforms.

Author: rjernst

server/src/main/java/org/elasticsearch/reservedstate/TransformState.java

@@ -8,24 +8,13 @@
 
 package org.elasticsearch.reservedstate;
 
-import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.cluster.ClusterState;
 
 import java.util.Set;
-import java.util.function.Consumer;
 
 /**
  * A {@link ClusterState} wrapper used by the ReservedClusterStateService to pass the
  * current state as well as previous keys set by an {@link ReservedClusterStateHandler} to each transform
  * step of the cluster state update.
- *
- * Each {@link ReservedClusterStateHandler} can also provide a non cluster state transform consumer that should run after
- * the cluster state is fully validated. This allows for handlers to perform extra steps, like clearing caches or saving
- * other state outside the cluster state. The consumer, if provided, must return a {@link NonStateTransformResult} with
- * the keys that will be saved as reserved in the cluster state.
  */
-public record TransformState(ClusterState state, Set<String> keys, Consumer<ActionListener<NonStateTransformResult>> nonStateTransform) {
-    public TransformState(ClusterState state, Set<String> keys) {
-        this(state, keys, null);
-    }
-}
+public record TransformState(ClusterState state, Set<String> keys) {}

server/src/main/java/org/elasticsearch/reservedstate/service/ReservedClusterStateService.java

@@ -13,7 +13,6 @@
 import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionResponse;
-import org.elasticsearch.action.support.RefCountingListener;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.metadata.ReservedStateErrorMetadata;
 import org.elasticsearch.cluster.metadata.ReservedStateMetadata;
@@ -22,16 +21,13 @@
 import org.elasticsearch.cluster.service.MasterServiceTaskQueue;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.core.Tuple;
-import org.elasticsearch.reservedstate.NonStateTransformResult;
 import org.elasticsearch.reservedstate.ReservedClusterStateHandler;
 import org.elasticsearch.reservedstate.TransformState;
 import org.elasticsearch.xcontent.ConstructingObjectParser;
 import org.elasticsearch.xcontent.ParseField;
 import org.elasticsearch.xcontent.XContentParser;
 
 import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.LinkedHashSet;
 import java.util.List;
@@ -156,7 +152,6 @@ public void initEmpty(String namespace, ActionListener<ActionResponse.Empty> lis
             new ReservedStateUpdateTask(
                 namespace,
                 emptyState,
-                List.of(),
                 Map.of(),
                 List.of(),
                 // error state should not be possible since there is no metadata being parsed or processed
@@ -210,65 +205,44 @@ public void process(String namespace, ReservedStateChunk reservedStateChunk, Con
             return;
         }
 
-        // We trial run all handler validations to ensure that we can process all of the cluster state error free. During
-        // the trial run we collect 'consumers' (functions) for any non cluster state transforms that need to run.
-        var trialRunResult = trialRun(namespace, state, reservedStateChunk, orderedHandlers);
+        // We trial run all handler validations to ensure that we can process all of the cluster state error free.
+        var trialRunErrors = trialRun(namespace, state, reservedStateChunk, orderedHandlers);
         // this is not using the modified trial state above, but that doesn't matter, we're just setting errors here
-        var error = checkAndReportError(namespace, trialRunResult.errors, reservedStateVersion);
+        var error = checkAndReportError(namespace, trialRunErrors, reservedStateVersion);
 
         if (error != null) {
             errorListener.accept(error);
             return;
         }
-
-        // Since we have validated that the cluster state update can be correctly performed in the trial run, we now
-        // execute the non cluster state transforms. These are assumed to be async and we continue with the cluster state update
-        // after all have completed. This part of reserved cluster state update is non-atomic, some or all of the non-state
-        // transformations can succeed, and we can fail to eventually write the reserved cluster state.
-        executeNonStateTransformationSteps(trialRunResult.nonStateTransforms, new ActionListener<>() {
-            @Override
-            public void onResponse(Collection<NonStateTransformResult> nonStateTransformResults) {
-                // Once all of the non-state transformation results complete, we can proceed to
-                // do the final save of the cluster state. The non-state transformation reserved keys are applied
-                // to the reserved state after all other key handlers.
-                updateTaskQueue.submitTask(
-                    "reserved cluster state [" + namespace + "]",
-                    new ReservedStateUpdateTask(
-                        namespace,
-                        reservedStateChunk,
-                        nonStateTransformResults,
-                        handlers,
-                        orderedHandlers,
-                        ReservedClusterStateService.this::updateErrorState,
-                        new ActionListener<>() {
-                            @Override
-                            public void onResponse(ActionResponse.Empty empty) {
-                                logger.info("Successfully applied new reserved cluster state for namespace [{}]", namespace);
-                                errorListener.accept(null);
-                            }
-
-                            @Override
-                            public void onFailure(Exception e) {
-                                // Don't spam the logs on repeated errors
-                                if (isNewError(existingMetadata, reservedStateVersion.version())) {
-                                    logger.debug("Failed to apply reserved cluster state", e);
-                                    errorListener.accept(e);
-                                } else {
-                                    errorListener.accept(null);
-                                }
-                            }
+        updateTaskQueue.submitTask(
+            "reserved cluster state [" + namespace + "]",
+            new ReservedStateUpdateTask(
+                namespace,
+                reservedStateChunk,
+                handlers,
+                orderedHandlers,
+                ReservedClusterStateService.this::updateErrorState,
+                new ActionListener<>() {
+                    @Override
+                    public void onResponse(ActionResponse.Empty empty) {
+                        logger.info("Successfully applied new reserved cluster state for namespace [{}]", namespace);
+                        errorListener.accept(null);
+                    }
+
+                    @Override
+                    public void onFailure(Exception e) {
+                        // Don't spam the logs on repeated errors
+                        if (isNewError(existingMetadata, reservedStateVersion.version())) {
+                            logger.debug("Failed to apply reserved cluster state", e);
+                            errorListener.accept(e);
+                        } else {
+                            errorListener.accept(null);
                         }
-                    ),
-                    null
-                );
-            }
-
-            @Override
-            public void onFailure(Exception e) {
-                // If we encounter an error while runnin the non-state transforms, we avoid saving any cluster state.
-                errorListener.accept(checkAndReportError(namespace, List.of(stackTrace(e)), reservedStateVersion));
-            }
-        });
+                    }
+                }
+            ),
+            null
+        );
     }
 
     // package private for testing
@@ -324,14 +298,13 @@ public void onFailure(Exception e) {
     /**
      * Goes through all of the handlers, runs the validation and the transform part of the cluster state.
      * <p>
-     * While running the handlers we also collect any non cluster state transformation consumer actions that
-     * need to be performed asynchronously before we attempt to save the cluster state. The trial run does not
-     * result in an update of the cluster state, it's only purpose is to verify if we can correctly perform a
-     * cluster state update with the given reserved state chunk.
+     * The trial run does not result in an update of the cluster state, it's only purpose is to verify
+     * if we can correctly perform a cluster state update with the given reserved state chunk.
      *
      * Package private for testing
+     * @return Any errors that occured
      */
-    TrialRunResult trialRun(
+    List<String> trialRun(
         String namespace,
         ClusterState currentState,
         ReservedStateChunk stateChunk,
@@ -341,7 +314,6 @@ TrialRunResult trialRun(
         Map<String, Object> reservedState = stateChunk.state();
 
         List<String> errors = new ArrayList<>();
-        List<Consumer<ActionListener<NonStateTransformResult>>> nonStateTransforms = new ArrayList<>();
 
         ClusterState state = currentState;
 
@@ -351,39 +323,12 @@ TrialRunResult trialRun(
                 Set<String> existingKeys = keysForHandler(existingMetadata, handlerName);
                 TransformState transformState = handler.transform(reservedState.get(handlerName), new TransformState(state, existingKeys));
                 state = transformState.state();
-                if (transformState.nonStateTransform() != null) {
-                    nonStateTransforms.add(transformState.nonStateTransform());
-                }
             } catch (Exception e) {
                 errors.add(format("Error processing %s state change: %s", handler.name(), stackTrace(e)));
             }
         }
 
-        return new TrialRunResult(nonStateTransforms, errors);
-    }
-
-    /**
-     * Runs the non cluster state transformations asynchronously, collecting the {@link NonStateTransformResult} objects.
-     * <p>
-     * Once all non cluster state transformations have completed, we submit the cluster state update task, which
-     * updates all of the handler state, including the keys produced by the non cluster state transforms. The new reserved
-     * state version isn't written to the cluster state until the cluster state task runs.
-     *
-     * Package private for testing
-     */
-    static void executeNonStateTransformationSteps(
-        List<Consumer<ActionListener<NonStateTransformResult>>> nonStateTransforms,
-        ActionListener<Collection<NonStateTransformResult>> listener
-    ) {
-        final List<NonStateTransformResult> result = Collections.synchronizedList(new ArrayList<>(nonStateTransforms.size()));
-        try (var listeners = new RefCountingListener(listener.map(ignored -> result))) {
-            for (var transform : nonStateTransforms) {
-                // non cluster state transforms don't modify the cluster state, they however are given a chance to return a more
-                // up-to-date version of the modified keys we should save in the reserved state. These calls are
-                // async and report back when they are done through the postTasksListener.
-                transform.accept(listeners.acquire(result::add));
-            }
-        }
+        return errors;
     }
 
     /**
@@ -449,9 +394,4 @@ private void addStateHandler(String key, Set<String> keys, LinkedHashSet<String>
     public void installStateHandler(ReservedClusterStateHandler<?> handler) {
         this.handlers.put(handler.name(), handler);
     }
-
-    /**
-     * Helper record class to combine the result of a trial run, non cluster state actions and any errors
-     */
-    record TrialRunResult(List<Consumer<ActionListener<NonStateTransformResult>>> nonStateTransforms, List<String> errors) {}
 }

server/src/main/java/org/elasticsearch/reservedstate/service/ReservedStateUpdateTask.java

@@ -20,7 +20,6 @@
 import org.elasticsearch.cluster.metadata.ReservedStateHandlerMetadata;
 import org.elasticsearch.cluster.metadata.ReservedStateMetadata;
 import org.elasticsearch.gateway.GatewayService;
-import org.elasticsearch.reservedstate.NonStateTransformResult;
 import org.elasticsearch.reservedstate.ReservedClusterStateHandler;
 import org.elasticsearch.reservedstate.TransformState;
 
@@ -51,20 +50,17 @@ public class ReservedStateUpdateTask implements ClusterStateTaskListener {
     private final Collection<String> orderedHandlers;
     private final Consumer<ErrorState> errorReporter;
     private final ActionListener<ActionResponse.Empty> listener;
-    private final Collection<NonStateTransformResult> nonStateTransformResults;
 
     public ReservedStateUpdateTask(
         String namespace,
         ReservedStateChunk stateChunk,
-        Collection<NonStateTransformResult> nonStateTransformResults,
         Map<String, ReservedClusterStateHandler<?>> handlers,
         Collection<String> orderedHandlers,
         Consumer<ErrorState> errorReporter,
         ActionListener<ActionResponse.Empty> listener
     ) {
         this.namespace = namespace;
         this.stateChunk = stateChunk;
-        this.nonStateTransformResults = nonStateTransformResults;
         this.handlers = handlers;
         this.orderedHandlers = orderedHandlers;
         this.errorReporter = errorReporter;
@@ -115,12 +111,6 @@ protected ClusterState execute(final ClusterState currentState) {
 
         checkAndThrowOnError(errors, reservedStateVersion);
 
-        // Once we have set all of the handler state from the cluster state update tasks, we add the reserved keys
-        // from the non cluster state transforms.
-        for (var transform : nonStateTransformResults) {
-            reservedMetadataBuilder.putHandler(new ReservedStateHandlerMetadata(transform.handlerName(), transform.updatedKeys()));
-        }
-
         // Remove the last error if we had previously encountered any in prior processing of reserved state
         reservedMetadataBuilder.errorMetadata(null);