Drop useless AckedRequest interface (#113255)

DaveCTurner · DaveCTurner · commit 422c26b65921 · 2024-09-20T13:04:49.000+01:00
Almost every implementation of `AckedRequest` is an
`AcknowledgedRequest` too, and the distinction is rather confusing.
Moreover the other implementations of `AckedRequest` are a potential
source of `null` timeouts that we'd like to get rid of. This commit
simplifies the situation by dropping the unnecessary `AckedRequest`
interface entirely.
diff --git a/qa/smoke-test-http/src/javaRestTest/java/org/elasticsearch/http/RestClusterInfoActionCancellationIT.java b/qa/smoke-test-http/src/javaRestTest/java/org/elasticsearch/http/RestClusterInfoActionCancellationIT.java
@@ -19,7 +19,6 @@
 import org.elasticsearch.client.Response;
 import org.elasticsearch.cluster.AckedClusterStateUpdateTask;
 import org.elasticsearch.cluster.ClusterState;
-import org.elasticsearch.cluster.ack.AckedRequest;
 import org.elasticsearch.cluster.block.ClusterBlock;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
 import org.elasticsearch.cluster.block.ClusterBlocks;
@@ -102,22 +101,9 @@ private void runTest(String actionName, String endpoint) throws Exception {
 
     private void updateClusterState(Function<ClusterState, ClusterState> transformationFn) {
         final TimeValue timeout = TimeValue.timeValueSeconds(10);
-
-        final AckedRequest ackedRequest = new AckedRequest() {
-            @Override
-            public TimeValue ackTimeout() {
-                return timeout;
-            }
-
-            @Override
-            public TimeValue masterNodeTimeout() {
-                return timeout;
-            }
-        };
-
         PlainActionFuture<AcknowledgedResponse> future = new PlainActionFuture<>();
         internalCluster().getAnyMasterNodeInstance(ClusterService.class)
-            .submitUnbatchedStateUpdateTask("get_mappings_cancellation_test", new AckedClusterStateUpdateTask(ackedRequest, future) {
+            .submitUnbatchedStateUpdateTask("get_mappings_cancellation_test", new AckedClusterStateUpdateTask(timeout, timeout, future) {
                 @Override
                 public ClusterState execute(ClusterState currentState) throws Exception {
                     return transformationFn.apply(currentState);
diff --git a/server/src/main/java/org/elasticsearch/action/support/master/AcknowledgedRequest.java b/server/src/main/java/org/elasticsearch/action/support/master/AcknowledgedRequest.java
@@ -9,7 +9,6 @@
 package org.elasticsearch.action.support.master;
 
 import org.elasticsearch.action.ActionRequestValidationException;
-import org.elasticsearch.cluster.ack.AckedRequest;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.core.TimeValue;
@@ -23,9 +22,7 @@
  * Abstract base class for action requests that track acknowledgements of cluster state updates: such a request is acknowledged only once
  * the cluster state update is committed and all relevant nodes have applied it and acknowledged its application to the elected master..
  */
-public abstract class AcknowledgedRequest<Request extends MasterNodeRequest<Request>> extends MasterNodeRequest<Request>
-    implements
-        AckedRequest {
+public abstract class AcknowledgedRequest<Request extends MasterNodeRequest<Request>> extends MasterNodeRequest<Request> {
 
     public static final TimeValue DEFAULT_ACK_TIMEOUT = timeValueSeconds(30);
 
@@ -74,7 +71,6 @@ public final Request ackTimeout(TimeValue ackTimeout) {
     /**
      * @return the current ack timeout as a {@link TimeValue}
      */
-    @Override
     public final TimeValue ackTimeout() {
         return ackTimeout;
     }
diff --git a/server/src/main/java/org/elasticsearch/cluster/AckedClusterStateUpdateTask.java b/server/src/main/java/org/elasticsearch/cluster/AckedClusterStateUpdateTask.java
@@ -9,8 +9,8 @@
 package org.elasticsearch.cluster;
 
 import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.support.master.AcknowledgedRequest;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
-import org.elasticsearch.cluster.ack.AckedRequest;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.core.TimeValue;
@@ -23,21 +23,38 @@
 public abstract class AckedClusterStateUpdateTask extends ClusterStateUpdateTask implements ClusterStateAckListener {
 
     private final ActionListener<AcknowledgedResponse> listener;
-    private final AckedRequest request;
+    private final TimeValue ackTimeout;
 
-    protected AckedClusterStateUpdateTask(AckedRequest request, ActionListener<? extends AcknowledgedResponse> listener) {
-        this(Priority.NORMAL, request, listener);
+    protected AckedClusterStateUpdateTask(AcknowledgedRequest<?> request, ActionListener<? extends AcknowledgedResponse> listener) {
+        this(Priority.NORMAL, request.masterNodeTimeout(), request.ackTimeout(), listener);
+    }
+
+    protected AckedClusterStateUpdateTask(
+        TimeValue masterNodeTimeout,
+        TimeValue ackTimeout,
+        ActionListener<? extends AcknowledgedResponse> listener
+    ) {
+        this(Priority.NORMAL, masterNodeTimeout, ackTimeout, listener);
+    }
+
+    protected AckedClusterStateUpdateTask(
+        Priority priority,
+        AcknowledgedRequest<?> request,
+        ActionListener<? extends AcknowledgedResponse> listener
+    ) {
+        this(priority, request.masterNodeTimeout(), request.ackTimeout(), listener);
     }
 
     @SuppressWarnings("unchecked")
     protected AckedClusterStateUpdateTask(
         Priority priority,
-        AckedRequest request,
+        TimeValue masterNodeTimeout,
+        TimeValue ackTimeout,
         ActionListener<? extends AcknowledgedResponse> listener
     ) {
-        super(priority, request.masterNodeTimeout());
+        super(priority, masterNodeTimeout);
         this.listener = (ActionListener<AcknowledgedResponse>) listener;
-        this.request = request;
+        this.ackTimeout = ackTimeout;
     }
 
     /**
@@ -81,6 +98,6 @@ public void onFailure(Exception e) {
      * Acknowledgement timeout, maximum time interval to wait for acknowledgements
      */
     public final TimeValue ackTimeout() {
-        return request.ackTimeout();
+        return ackTimeout;
     }
 }
diff --git a/server/src/main/java/org/elasticsearch/cluster/ack/AckedRequest.java b/server/src/main/java/org/elasticsearch/cluster/ack/AckedRequest.java
diff --git a/server/src/main/java/org/elasticsearch/cluster/ack/ClusterStateUpdateRequest.java b/server/src/main/java/org/elasticsearch/cluster/ack/ClusterStateUpdateRequest.java
@@ -15,15 +15,14 @@
  * Base class to be used when needing to update the cluster state
  * Contains the basic fields that are always needed
  */
-public abstract class ClusterStateUpdateRequest<T extends ClusterStateUpdateRequest<T>> implements AckedRequest {
+public abstract class ClusterStateUpdateRequest<T extends ClusterStateUpdateRequest<T>> {
 
     private TimeValue ackTimeout;
     private TimeValue masterNodeTimeout;
 
     /**
      * Returns the maximum time interval to wait for acknowledgements
      */
-    @Override
     public TimeValue ackTimeout() {
         return ackTimeout;
     }
@@ -41,7 +40,6 @@ public T ackTimeout(TimeValue ackTimeout) {
      * Returns the maximum time interval to wait for the request to
      * be completed on the master node
      */
-    @Override
     public TimeValue masterNodeTimeout() {
         return masterNodeTimeout;
     }
diff --git a/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataCreateDataStreamService.java b/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataCreateDataStreamService.java
@@ -99,7 +99,12 @@ public void createDataStream(CreateDataStreamClusterStateUpdateRequest request,
         var delegate = new AllocationActionListener<>(listener, threadPool.getThreadContext());
         submitUnbatchedTask(
             "create-data-stream [" + request.name + "]",
-            new AckedClusterStateUpdateTask(Priority.HIGH, request, delegate.clusterStateUpdate()) {
+            new AckedClusterStateUpdateTask(
+                Priority.HIGH,
+                request.masterNodeTimeout(),
+                request.ackTimeout(),
+                delegate.clusterStateUpdate()
+            ) {
                 @Override
                 public ClusterState execute(ClusterState currentState) throws Exception {
                     // When we're manually creating a data stream (i.e. not an auto creation), we don't need to initialize the failure store
diff --git a/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataCreateIndexService.java b/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataCreateIndexService.java
@@ -296,7 +296,12 @@ private void onlyCreateIndex(final CreateIndexClusterStateUpdateRequest request,
         var delegate = new AllocationActionListener<>(listener, threadPool.getThreadContext());
         submitUnbatchedTask(
             "create-index [" + request.index() + "], cause [" + request.cause() + "]",
-            new AckedClusterStateUpdateTask(Priority.URGENT, request, delegate.clusterStateUpdate()) {
+            new AckedClusterStateUpdateTask(
+                Priority.URGENT,
+                request.masterNodeTimeout(),
+                request.ackTimeout(),
+                delegate.clusterStateUpdate()
+            ) {
 
                 @Override
                 public ClusterState execute(ClusterState currentState) throws Exception {
diff --git a/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataMigrateToDataStreamService.java b/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataMigrateToDataStreamService.java
@@ -105,7 +105,12 @@ public void migrateToDataStream(
         var delegate = new AllocationActionListener<>(listener, threadContext);
         submitUnbatchedTask(
             "migrate-to-data-stream [" + request.aliasName + "]",
-            new AckedClusterStateUpdateTask(Priority.HIGH, request, delegate.clusterStateUpdate()) {
+            new AckedClusterStateUpdateTask(
+                Priority.HIGH,
+                request.masterNodeTimeout(),
+                request.ackTimeout(),
+                delegate.clusterStateUpdate()
+            ) {
 
                 @Override
                 public ClusterState execute(ClusterState currentState) throws Exception {
diff --git a/server/src/test/java/org/elasticsearch/cluster/service/MasterServiceTests.java b/server/src/test/java/org/elasticsearch/cluster/service/MasterServiceTests.java
@@ -15,6 +15,7 @@
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.PlainActionFuture;
+import org.elasticsearch.action.support.master.AcknowledgedRequest;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.cluster.AckedClusterStateUpdateTask;
 import org.elasticsearch.cluster.ClusterName;
@@ -26,7 +27,6 @@
 import org.elasticsearch.cluster.LocalMasterServiceTask;
 import org.elasticsearch.cluster.NotMasterException;
 import org.elasticsearch.cluster.SimpleBatchedExecutor;
-import org.elasticsearch.cluster.ack.AckedRequest;
 import org.elasticsearch.cluster.block.ClusterBlocks;
 import org.elasticsearch.cluster.coordination.ClusterStatePublisher;
 import org.elasticsearch.cluster.coordination.FailedToCommitClusterStateException;
@@ -1570,7 +1570,7 @@ public void onAckFailure(Exception e) {
 
                 masterService.submitUnbatchedStateUpdateTask(
                     "test2",
-                    new AckedClusterStateUpdateTask(ackedRequest(TimeValue.ZERO, null), null) {
+                    new AckedClusterStateUpdateTask(ackedRequest(TimeValue.ZERO, TimeValue.MINUS_ONE), null) {
                         @Override
                         public ClusterState execute(ClusterState currentState) {
                             return ClusterState.builder(currentState).build();
@@ -1622,7 +1622,7 @@ public void onAckTimeout() {
 
                 masterService.submitUnbatchedStateUpdateTask(
                     "test2",
-                    new AckedClusterStateUpdateTask(ackedRequest(ackTimeout, null), null) {
+                    new AckedClusterStateUpdateTask(ackedRequest(ackTimeout, TimeValue.MINUS_ONE), null) {
                         @Override
                         public ClusterState execute(ClusterState currentState) {
                             threadPool.getThreadContext().addResponseHeader(responseHeaderName, responseHeaderValue);
@@ -1677,7 +1677,7 @@ public void onAckTimeout() {
 
                 masterService.submitUnbatchedStateUpdateTask(
                     "test2",
-                    new AckedClusterStateUpdateTask(ackedRequest(TimeValue.MINUS_ONE, null), null) {
+                    new AckedClusterStateUpdateTask(ackedRequest(TimeValue.MINUS_ONE, TimeValue.MINUS_ONE), null) {
                         @Override
                         public ClusterState execute(ClusterState currentState) {
                             return ClusterState.builder(currentState).build();
@@ -2656,20 +2656,15 @@ public static ClusterState discoveryState(MasterService masterService) {
     }
 
     /**
-     * Returns a plain {@link AckedRequest} that does not implement any functionality outside of the timeout getters.
+     * Returns a plain {@link AcknowledgedRequest} that does not implement any functionality outside of the timeout getters.
      */
-    public static AckedRequest ackedRequest(TimeValue ackTimeout, TimeValue masterNodeTimeout) {
-        return new AckedRequest() {
-            @Override
-            public TimeValue ackTimeout() {
-                return ackTimeout;
-            }
-
-            @Override
-            public TimeValue masterNodeTimeout() {
-                return masterNodeTimeout;
+    public static AcknowledgedRequest<?> ackedRequest(TimeValue ackTimeout, TimeValue masterNodeTimeout) {
+        class BareAcknowledgedRequest extends AcknowledgedRequest<BareAcknowledgedRequest> {
+            BareAcknowledgedRequest() {
+                super(masterNodeTimeout, ackTimeout);
             }
-        };
+        }
+        return new BareAcknowledgedRequest();
     }
 
     /**
diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/OperationModeUpdateTask.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ilm/OperationModeUpdateTask.java
@@ -9,11 +9,11 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.support.master.AcknowledgedRequest;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.cluster.AckedClusterStateUpdateTask;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStateUpdateTask;
-import org.elasticsearch.cluster.ack.AckedRequest;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.core.Nullable;
@@ -36,7 +36,7 @@ public class OperationModeUpdateTask extends ClusterStateUpdateTask {
 
     public static AckedClusterStateUpdateTask wrap(
         OperationModeUpdateTask task,
-        AckedRequest request,
+        AcknowledgedRequest<?> request,
         ActionListener<AcknowledgedResponse> listener
     ) {
         return new AckedClusterStateUpdateTask(task.priority(), request, listener) {
diff --git a/x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/transport/actions/TransportWatcherServiceAction.java b/x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/transport/actions/TransportWatcherServiceAction.java
@@ -16,14 +16,12 @@
 import org.elasticsearch.cluster.AckedClusterStateUpdateTask;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStateUpdateTask;
-import org.elasticsearch.cluster.ack.AckedRequest;
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.core.SuppressForbidden;
-import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.injection.guice.Inject;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
@@ -69,42 +67,35 @@ protected void masterOperation(
         final boolean manuallyStopped = request.getCommand() == WatcherServiceRequest.Command.STOP;
         final String source = manuallyStopped ? "update_watcher_manually_stopped" : "update_watcher_manually_started";
 
-        // TODO: make WatcherServiceRequest a real AckedRequest so that we have both a configurable timeout and master node timeout like
-        // we do elsewhere
-        submitUnbatchedTask(source, new AckedClusterStateUpdateTask(new AckedRequest() {
-            @Override
-            public TimeValue ackTimeout() {
-                return AcknowledgedRequest.DEFAULT_ACK_TIMEOUT;
-            }
-
-            @Override
-            public TimeValue masterNodeTimeout() {
-                return request.masterNodeTimeout();
-            }
-        }, listener) {
-            @Override
-            public ClusterState execute(ClusterState clusterState) {
-                XPackPlugin.checkReadyForXPackCustomMetadata(clusterState);
+        // TODO: make WatcherServiceRequest a real AcknowledgedRequest so that we have both a configurable timeout and master node timeout
+        // like we do elsewhere
+        submitUnbatchedTask(
+            source,
+            new AckedClusterStateUpdateTask(request.masterNodeTimeout(), AcknowledgedRequest.DEFAULT_ACK_TIMEOUT, listener) {
+                @Override
+                public ClusterState execute(ClusterState clusterState) {
+                    XPackPlugin.checkReadyForXPackCustomMetadata(clusterState);
 
-                WatcherMetadata newWatcherMetadata = new WatcherMetadata(manuallyStopped);
-                WatcherMetadata currentMetadata = clusterState.metadata().custom(WatcherMetadata.TYPE);
+                    WatcherMetadata newWatcherMetadata = new WatcherMetadata(manuallyStopped);
+                    WatcherMetadata currentMetadata = clusterState.metadata().custom(WatcherMetadata.TYPE);
 
-                // adhere to the contract of returning the original state if nothing has changed
-                if (newWatcherMetadata.equals(currentMetadata)) {
-                    return clusterState;
-                } else {
-                    ClusterState.Builder builder = new ClusterState.Builder(clusterState);
-                    builder.metadata(Metadata.builder(clusterState.getMetadata()).putCustom(WatcherMetadata.TYPE, newWatcherMetadata));
-                    return builder.build();
+                    // adhere to the contract of returning the original state if nothing has changed
+                    if (newWatcherMetadata.equals(currentMetadata)) {
+                        return clusterState;
+                    } else {
+                        ClusterState.Builder builder = new ClusterState.Builder(clusterState);
+                        builder.metadata(Metadata.builder(clusterState.getMetadata()).putCustom(WatcherMetadata.TYPE, newWatcherMetadata));
+                        return builder.build();
+                    }
                 }
-            }
 
-            @Override
-            public void onFailure(Exception e) {
-                logger.error(() -> format("could not update watcher stopped status to [%s], source [%s]", manuallyStopped, source), e);
-                listener.onFailure(e);
+                @Override
+                public void onFailure(Exception e) {
+                    logger.error(() -> format("could not update watcher stopped status to [%s], source [%s]", manuallyStopped, source), e);
+                    listener.onFailure(e);
+                }
             }
-        });
+        );
     }
 
     @Override
