Add disabled POST _reindex/{taskId}/_cancel API

modules/reindex-management/build.gradle

@@ -7,6 +7,7 @@
  * License v 3.0 only", or the "Server Side Public License, v 1".
  */
 
+apply plugin: 'elasticsearch.internal-cluster-test'
 apply plugin: 'elasticsearch.internal-yaml-rest-test'
 apply plugin: 'elasticsearch.yaml-rest-compat-test'
 
@@ -19,6 +20,7 @@ esplugin {
 
 dependencies {
   compileOnly project(path: ':modules:reindex')
+  internalClusterTestImplementation project(path: ':modules:reindex')
 
   testImplementation project(':modules:rest-root')
 

modules/reindex-management/src/main/java/org/elasticsearch/reindex/management/CancelReindexRequest.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.reindex.management;
+
+import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.action.support.tasks.BaseTasksRequest;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.index.reindex.ReindexAction;
+import org.elasticsearch.tasks.Task;
+
+import java.io.IOException;
+
+import static org.elasticsearch.action.ValidateActions.addValidationError;
+
+/** A request to cancel an ongoing reindex task. */
+public class CancelReindexRequest extends BaseTasksRequest<CancelReindexRequest> {
+
+    private final boolean waitForCompletion;
+
+    public CancelReindexRequest(StreamInput in) throws IOException {
+        super(in);
+        waitForCompletion = in.readBoolean();
+    }
+
+    public CancelReindexRequest(boolean waitForCompletion) {
+        this.waitForCompletion = waitForCompletion;
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+        out.writeBoolean(waitForCompletion);
+    }
+
+    @Override
+    public String getDescription() {
+        return "waitForCompletion[" + waitForCompletion + "], targetTaskId[" + getTargetTaskId() + "]";
+    }
+
+    @Override
+    public ActionRequestValidationException validate() {
+        var validationException = super.validate();
+        if (getTargetTaskId().isSet() == false) {
+            validationException = addValidationError("task id must be provided", validationException);
+        }
+        return validationException;
+    }
+
+    public boolean waitForCompletion() {
+        return waitForCompletion;
+    }
+
+    @Override
+    public boolean match(Task task) {
+        return ReindexAction.NAME.equals(task.getAction()) && task.getParentTaskId().isSet() == false;
+    }
+}

modules/reindex-management/src/main/java/org/elasticsearch/reindex/management/CancelReindexResponse.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.reindex.management;
+
+import org.elasticsearch.action.FailedNodeException;
+import org.elasticsearch.action.TaskOperationFailure;
+import org.elasticsearch.action.support.tasks.BaseTasksResponse;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.xcontent.ToXContentObject;
+import org.elasticsearch.xcontent.XContentBuilder;
+
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * Response returned from {@code POST /_reindex/{taskId}/_cancel}.
+ */
+public class CancelReindexResponse extends BaseTasksResponse implements ToXContentObject {
+
+    public CancelReindexResponse(List<TaskOperationFailure> taskFailures, List<FailedNodeException> failedNodes) {
+        super(taskFailures, failedNodes);
+    }
+
+    public CancelReindexResponse(final StreamInput in) throws IOException {
+        super(in);
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject();
+        builder.field("acknowledged", true);
+        builder.endObject();
+        return builder;
+    }
+}

modules/reindex-management/src/main/java/org/elasticsearch/reindex/management/CancelReindexTaskResponse.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.reindex.management;
+
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Writeable;
+
+/** Response to a single reindex task cancel action. */
+public class CancelReindexTaskResponse implements Writeable {
+
+    public CancelReindexTaskResponse() {}
+
+    public CancelReindexTaskResponse(final StreamInput in) {}
+
+    @Override
+    public void writeTo(final StreamOutput out) {}
+}

modules/reindex-management/src/main/java/org/elasticsearch/reindex/management/ReindexManagementPlugin.java

@@ -37,7 +37,8 @@ public List<ActionHandler> getActions() {
         if (REINDEX_RESILIENCE_ENABLED) {
             return List.of(
                 new ActionHandler(TransportGetReindexAction.TYPE, TransportGetReindexAction.class),
-                new ActionHandler(TransportListReindexAction.TYPE, TransportListReindexAction.class)
+                new ActionHandler(TransportListReindexAction.TYPE, TransportListReindexAction.class),
+                new ActionHandler(TransportCancelReindexAction.TYPE, TransportCancelReindexAction.class)
             );
         } else {
             return List.of();
@@ -57,7 +58,11 @@ public List<RestHandler> getRestHandlers(
         Predicate<NodeFeature> clusterSupportsFeature
     ) {
         if (REINDEX_RESILIENCE_ENABLED) {
-            return List.of(new RestGetReindexAction(clusterSupportsFeature), new RestListReindexAction(clusterSupportsFeature));
+            return List.of(
+                new RestGetReindexAction(clusterSupportsFeature),
+                new RestListReindexAction(clusterSupportsFeature),
+                new RestCancelReindexAction(clusterSupportsFeature)
+            );
         } else {
             return List.of();
         }

modules/reindex-management/src/main/java/org/elasticsearch/reindex/management/RestCancelReindexAction.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.reindex.management;
+
+import org.elasticsearch.client.internal.node.NodeClient;
+import org.elasticsearch.features.NodeFeature;
+import org.elasticsearch.rest.BaseRestHandler;
+import org.elasticsearch.rest.RestRequest;
+import org.elasticsearch.rest.Scope;
+import org.elasticsearch.rest.ServerlessScope;
+import org.elasticsearch.rest.action.RestToXContentListener;
+import org.elasticsearch.tasks.TaskId;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.function.Predicate;
+
+import static org.elasticsearch.rest.RestRequest.Method.POST;
+
+/** REST handler for cancelling an ongoing reindex task. */
+@ServerlessScope(Scope.PUBLIC)
+public class RestCancelReindexAction extends BaseRestHandler {
+
+    private final Predicate<NodeFeature> clusterSupportsFeature;
+
+    public RestCancelReindexAction(final Predicate<NodeFeature> clusterSupportsFeature) {
+        this.clusterSupportsFeature = Objects.requireNonNull(clusterSupportsFeature);
+    }
+
+    @Override
+    public List<Route> routes() {
+        return List.of(new Route(POST, "/_reindex/{task_id}/_cancel"));
+    }
+
+    @Override
+    public String getName() {
+        return "cancel_reindex_action";
+    }
+
+    @Override
+    protected RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) {
+        if (clusterSupportsFeature.test(ReindexManagementFeatures.NEW_ENDPOINTS) == false) {
+            throw new IllegalArgumentException("endpoint not supported on all nodes in the cluster");
+        }
+        final String taskIdParam = request.param("task_id");
+        final TaskId taskId = new TaskId(taskIdParam);
+        if (taskId.isSet() == false) {
+            throw new IllegalArgumentException("invalid taskId provided: " + taskIdParam);
+        }
+
+        final boolean waitForCompletion = request.paramAsBoolean("wait_for_completion", true);
+        final CancelReindexRequest cancelRequest = new CancelReindexRequest(waitForCompletion);
+        cancelRequest.setTargetTaskId(taskId);
+
+        return channel -> client.execute(TransportCancelReindexAction.TYPE, cancelRequest, new RestToXContentListener<>(channel));
+    }
+}

modules/reindex-management/src/main/java/org/elasticsearch/reindex/management/TransportCancelReindexAction.java

@@ -0,0 +1,113 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.reindex.management;
+
+import org.elasticsearch.ExceptionsHelper;
+import org.elasticsearch.ResourceNotFoundException;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.ActionType;
+import org.elasticsearch.action.FailedNodeException;
+import org.elasticsearch.action.NoSuchNodeException;
+import org.elasticsearch.action.TaskOperationFailure;
+import org.elasticsearch.action.admin.cluster.node.tasks.cancel.CancelTasksRequest;
+import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.action.support.tasks.TransportTasksProjectAction;
+import org.elasticsearch.cluster.project.ProjectResolver;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.index.reindex.BulkByScrollTask;
+import org.elasticsearch.injection.guice.Inject;
+import org.elasticsearch.tasks.CancellableTask;
+import org.elasticsearch.tasks.TaskId;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.TransportService;
+
+import java.util.List;
+
+/** Transport action that cancels an in-flight reindex task and its descendants. */
+public class TransportCancelReindexAction extends TransportTasksProjectAction<
+    CancellableTask,
+    CancelReindexRequest,
+    CancelReindexResponse,
+    CancelReindexTaskResponse> {
+
+    public static final ActionType<CancelReindexResponse> TYPE = new ActionType<>("cluster:admin/reindex/cancel");
+
+    @Inject
+    public TransportCancelReindexAction(
+        final ClusterService clusterService,
+        final TransportService transportService,
+        final ActionFilters actionFilters,
+        final ProjectResolver projectResolver
+    ) {
+        super(
+            TYPE.name(),
+            clusterService,
+            transportService,
+            actionFilters,
+            CancelReindexRequest::new,
+            CancelReindexTaskResponse::new,
+            transportService.getThreadPool().executor(ThreadPool.Names.GENERIC),
+            projectResolver
+        );
+    }
+
+    @Override
+    protected List<CancellableTask> processTasks(final CancelReindexRequest request) {
+        final CancellableTask requestedTask = taskManager.getCancellableTask(request.getTargetTaskId().getId());
+        if (requestedTask != null && super.match(requestedTask) && request.match(requestedTask)) {
+            return List.of(requestedTask);
+        }
+        return List.of();
+    }
+
+    @Override
+    protected void taskOperation(
+        final CancellableTask actionTask,
+        final CancelReindexRequest request,
+        final CancellableTask task,
+        final ActionListener<CancelReindexTaskResponse> listener
+    ) {
+        assert task instanceof BulkByScrollTask : "Task should be a BulkByScrollTask";
+
+        taskManager.cancelTaskAndDescendants(
+            task,
+            CancelTasksRequest.DEFAULT_REASON,
+            request.waitForCompletion(),
+            ActionListener.wrap(ignored -> listener.onResponse(new CancelReindexTaskResponse()), listener::onFailure)
+        );
+    }
+
+    @Override
+    protected CancelReindexResponse newResponse(
+        final CancelReindexRequest request,
+        final List<CancelReindexTaskResponse> tasks,
+        final List<TaskOperationFailure> taskFailures,
+        final List<FailedNodeException> nodeExceptions
+    ) {
+        assert tasks.size() + taskFailures.size() + nodeExceptions.size() <= 1 : "currently only supports cancelling one task max";
+        // check whether node in requested TaskId doesn't exist and throw 404
+        for (final FailedNodeException e : nodeExceptions) {
+            if (ExceptionsHelper.unwrap(e, NoSuchNodeException.class) != null) {
+                throw reindexWithTaskIdNotFoundException(request.getTargetTaskId());
+            }
+        }
+
+        final var response = new CancelReindexResponse(taskFailures, nodeExceptions);
+        response.rethrowFailures("cancel_reindex"); // if we haven't handled any exception already, throw here
+        if (tasks.isEmpty()) {
+            throw reindexWithTaskIdNotFoundException(request.getTargetTaskId());
+        }
+        return response;
+    }
+
+    private static ResourceNotFoundException reindexWithTaskIdNotFoundException(final TaskId requestedTaskId) {
+        return new ResourceNotFoundException("reindex task [{}] either not found or completed", requestedTaskId);
+    }
+}