Replace DeleteIndexClusterStateUpdateRequest with private class (elastic#113509)

DaveCTurner · web-flow · commit 6ca4fb21c094 · 2024-09-26T04:42:56.000+10:00
No need to extend `IndicesClusterStateUpdateRequest` here, nor to attach the listener to the request. Instead we can encapsulate the cluster state update task directly instead. Backport of elastic#113288 to 8.x
diff --git a/server/src/internalClusterTest/java/org/elasticsearch/snapshots/ConcurrentSnapshotsIT.java b/server/src/internalClusterTest/java/org/elasticsearch/snapshots/ConcurrentSnapshotsIT.java
@@ -18,7 +18,6 @@
 import org.elasticsearch.action.admin.cluster.snapshots.restore.RestoreSnapshotResponse;
 import org.elasticsearch.action.admin.cluster.snapshots.status.SnapshotStatus;
 import org.elasticsearch.action.admin.cluster.snapshots.status.SnapshotsStatusResponse;
-import org.elasticsearch.action.admin.indices.delete.DeleteIndexClusterStateUpdateRequest;
 import org.elasticsearch.action.support.ActionTestUtils;
 import org.elasticsearch.action.support.GroupedActionListener;
 import org.elasticsearch.action.support.PlainActionFuture;
@@ -33,9 +32,7 @@
 import org.elasticsearch.common.util.concurrent.ListenableFuture;
 import org.elasticsearch.common.util.concurrent.UncategorizedExecutionException;
 import org.elasticsearch.core.PathUtils;
-import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.discovery.AbstractDisruptionTestCase;
-import org.elasticsearch.index.Index;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.repositories.IndexId;
 import org.elasticsearch.repositories.RepositoryConflictException;
@@ -59,6 +56,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
@@ -2214,12 +2212,17 @@ public void testDeleteIndexWithOutOfOrderFinalization() {
                     .anyMatch(e -> e.snapshot().getSnapshotId().getName().equals("snapshot-with-index-1") && e.state().completed())
             )
             // execute the index deletion _directly on the master_ so it happens before the snapshot finalization executes
-            .andThen(l -> masterDeleteIndexService.deleteIndices(new DeleteIndexClusterStateUpdateRequest(l.map(r -> {
-                assertTrue(r.isAcknowledged());
-                return null;
-            })).indices(new Index[] { internalCluster().clusterService().state().metadata().index(indexToDelete).getIndex() })
-                .ackTimeout(TimeValue.timeValueSeconds(10))
-                .masterNodeTimeout(TimeValue.timeValueSeconds(10))))
+            .andThen(
+                l -> masterDeleteIndexService.deleteIndices(
+                    TEST_REQUEST_TIMEOUT,
+                    TEST_REQUEST_TIMEOUT,
+                    Set.of(internalCluster().clusterService().state().metadata().index(indexToDelete).getIndex()),
+                    l.map(r -> {
+                        assertTrue(r.isAcknowledged());
+                        return null;
+                    })
+                )
+            )
             // ultimately create the index again so that taking a full snapshot will pick up any missing shard gen blob, and deleting that
             // full snapshot will clean up all dangling shard-level blobs
             .andThen((l, ignored) -> prepareCreate(indexToDelete, indexSettingsNoReplicas(1)).execute(l.map(r -> {
diff --git a/server/src/main/java/org/elasticsearch/action/admin/indices/delete/DeleteIndexClusterStateUpdateRequest.java b/server/src/main/java/org/elasticsearch/action/admin/indices/delete/DeleteIndexClusterStateUpdateRequest.java
diff --git a/server/src/main/java/org/elasticsearch/action/admin/indices/delete/TransportDeleteIndexAction.java b/server/src/main/java/org/elasticsearch/action/admin/indices/delete/TransportDeleteIndexAction.java
@@ -92,11 +92,14 @@ protected void masterOperation(
             return;
         }
 
-        DeleteIndexClusterStateUpdateRequest deleteRequest = new DeleteIndexClusterStateUpdateRequest(listener.delegateResponse((l, e) -> {
-            logger.debug(() -> "failed to delete indices [" + concreteIndices + "]", e);
-            listener.onFailure(e);
-        })).ackTimeout(request.ackTimeout()).masterNodeTimeout(request.masterNodeTimeout()).indices(concreteIndices.toArray(new Index[0]));
-
-        deleteIndexService.deleteIndices(deleteRequest);
+        deleteIndexService.deleteIndices(
+            request.masterNodeTimeout(),
+            request.ackTimeout(),
+            concreteIndices,
+            listener.delegateResponse((l, e) -> {
+                logger.debug(() -> "failed to delete indices [" + concreteIndices + "]", e);
+                listener.onFailure(e);
+            })
+        );
     }
 }
diff --git a/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataDeleteIndexService.java b/server/src/main/java/org/elasticsearch/cluster/metadata/MetadataDeleteIndexService.java
@@ -11,32 +11,35 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import org.elasticsearch.action.admin.indices.delete.DeleteIndexClusterStateUpdateRequest;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.ClusterStateAckListener;
 import org.elasticsearch.cluster.ClusterStateTaskExecutor;
+import org.elasticsearch.cluster.ClusterStateTaskListener;
 import org.elasticsearch.cluster.RestoreInProgress;
 import org.elasticsearch.cluster.SimpleBatchedAckListenerTaskExecutor;
 import org.elasticsearch.cluster.block.ClusterBlocks;
+import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.routing.RoutingTable;
 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.collect.ImmutableOpenMap;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.util.set.Sets;
+import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.injection.guice.Inject;
 import org.elasticsearch.snapshots.RestoreService;
 import org.elasticsearch.snapshots.SnapshotInProgressException;
 import org.elasticsearch.snapshots.SnapshotsService;
 
-import java.util.Arrays;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 
 import static org.elasticsearch.cluster.routing.allocation.allocator.AllocationActionListener.rerouteCompletionIsNotRequired;
@@ -48,22 +51,19 @@ public class MetadataDeleteIndexService {
 
     private static final Logger logger = LogManager.getLogger(MetadataDeleteIndexService.class);
 
-    private final Settings settings;
-
     // package private for tests
-    final ClusterStateTaskExecutor<DeleteIndexClusterStateUpdateRequest> executor;
-    private final MasterServiceTaskQueue<DeleteIndexClusterStateUpdateRequest> taskQueue;
+    final ClusterStateTaskExecutor<DeleteIndicesClusterStateUpdateTask> executor;
+    private final MasterServiceTaskQueue<DeleteIndicesClusterStateUpdateTask> taskQueue;
 
     @Inject
     public MetadataDeleteIndexService(Settings settings, ClusterService clusterService, AllocationService allocationService) {
-        this.settings = settings;
         executor = new SimpleBatchedAckListenerTaskExecutor<>() {
             @Override
             public Tuple<ClusterState, ClusterStateAckListener> executeTask(
-                DeleteIndexClusterStateUpdateRequest task,
+                DeleteIndicesClusterStateUpdateTask task,
                 ClusterState clusterState
             ) {
-                return Tuple.tuple(MetadataDeleteIndexService.deleteIndices(clusterState, Sets.newHashSet(task.indices()), settings), task);
+                return Tuple.tuple(MetadataDeleteIndexService.deleteIndices(clusterState, task.indices, settings), task);
             }
 
             @Override
@@ -81,11 +81,64 @@ public ClusterState afterBatchExecution(ClusterState clusterState, boolean clust
         taskQueue = clusterService.createTaskQueue("delete-index", Priority.URGENT, executor);
     }
 
-    public void deleteIndices(final DeleteIndexClusterStateUpdateRequest request) {
-        if (request.indices() == null || request.indices().length == 0) {
-            throw new IllegalArgumentException("Index name is required");
+    public void deleteIndices(
+        TimeValue masterNodeTimeout,
+        TimeValue ackTimeout,
+        Set<Index> indices,
+        ActionListener<AcknowledgedResponse> listener
+    ) {
+        if (indices == null || indices.isEmpty()) {
+            throw new IllegalArgumentException("Indices are required");
+        }
+        taskQueue.submitTask(
+            "delete-index " + indices,
+            new DeleteIndicesClusterStateUpdateTask(indices, ackTimeout, listener),
+            masterNodeTimeout
+        );
+    }
+
+    // package private for tests
+    static class DeleteIndicesClusterStateUpdateTask implements ClusterStateTaskListener, ClusterStateAckListener {
+
+        private final Set<Index> indices;
+        private final TimeValue ackTimeout;
+        private final ActionListener<AcknowledgedResponse> listener;
+
+        DeleteIndicesClusterStateUpdateTask(Set<Index> indices, TimeValue ackTimeout, ActionListener<AcknowledgedResponse> listener) {
+            this.indices = Objects.requireNonNull(indices);
+            this.ackTimeout = Objects.requireNonNull(ackTimeout);
+            this.listener = Objects.requireNonNull(listener);
+        }
+
+        @Override
+        public boolean mustAck(DiscoveryNode discoveryNode) {
+            return true;
+        }
+
+        @Override
+        public void onAllNodesAcked() {
+            listener.onResponse(AcknowledgedResponse.TRUE);
+        }
+
+        @Override
+        public void onAckFailure(Exception e) {
+            listener.onResponse(AcknowledgedResponse.FALSE);
+        }
+
+        @Override
+        public void onAckTimeout() {
+            listener.onResponse(AcknowledgedResponse.FALSE);
+        }
+
+        @Override
+        public TimeValue ackTimeout() {
+            return ackTimeout;
+        }
+
+        @Override
+        public void onFailure(Exception e) {
+            listener.onFailure(e);
         }
-        taskQueue.submitTask("delete-index " + Arrays.toString(request.indices()), request, request.masterNodeTimeout());
     }
 
     /**
diff --git a/server/src/test/java/org/elasticsearch/cluster/metadata/MetadataDeleteIndexServiceTests.java b/server/src/test/java/org/elasticsearch/cluster/metadata/MetadataDeleteIndexServiceTests.java
@@ -9,7 +9,6 @@
 package org.elasticsearch.cluster.metadata;
 
 import org.elasticsearch.action.ActionListener;
-import org.elasticsearch.action.admin.indices.delete.DeleteIndexClusterStateUpdateRequest;
 import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.SnapshotsInProgress;
@@ -132,8 +131,10 @@ public void testDeleteUnassigned() throws Exception {
             before,
             service.executor,
             List.of(
-                new DeleteIndexClusterStateUpdateRequest(ActionListener.noop()).indices(
-                    new Index[] { before.metadata().getIndices().get(index).getIndex() }
+                new MetadataDeleteIndexService.DeleteIndicesClusterStateUpdateTask(
+                    Set.of(before.metadata().getIndices().get(index).getIndex()),
+                    TEST_REQUEST_TIMEOUT,
+                    ActionListener.noop()
                 )
             )
         );
