high level REST api: cancel task

client/rest-high-level/src/main/java/org/elasticsearch/client/ClusterClient.java

@@ -21,6 +21,8 @@
 
 import org.apache.http.Header;
 import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.admin.cluster.node.tasks.cancel.CancelTasksRequest;
+import org.elasticsearch.action.admin.cluster.node.tasks.cancel.CancelTasksResponse;
 import org.elasticsearch.action.admin.cluster.node.tasks.list.ListTasksRequest;
 import org.elasticsearch.action.admin.cluster.node.tasks.list.ListTasksResponse;
 import org.elasticsearch.action.admin.cluster.settings.ClusterUpdateSettingsRequest;
@@ -87,4 +89,37 @@ public void listTasksAsync(ListTasksRequest request, ActionListener<ListTasksRes
         restHighLevelClient.performRequestAsyncAndParseEntity(request, RequestConverters::listTasks, ListTasksResponse::fromXContent,
                 listener, emptySet(), headers);
     }
+
+    /**
+     * Cancel one or more cluster tasks using the Task Management API
+     * <p>
+     * See
+     * <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/tasks.html"> Task Management API on elastic.co</a>
+     * </p>
+     */
+    public CancelTasksResponse cancelTasks(CancelTasksRequest cancelTasksRequest, Header... headers) throws IOException {
+        return restHighLevelClient.performRequestAndParseEntity(
+            cancelTasksRequest,
+            RequestConverters::cancelTasks,
+            parser -> CancelTasksResponse.fromXContent(parser),
+            emptySet(),
+            headers);
+    }
+
+    /**
+     * Asynchronously cancel one or more cluster tasks using the Task Management API
+     * <p>
+     * See
+     * <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/tasks.html"> Task Management API on elastic.co</a>
+     * </p>
+     */
+    public void cancelTasksAsync(CancelTasksRequest cancelTasksRequest, ActionListener<CancelTasksResponse> listener, Header... headers) {
+        restHighLevelClient.performRequestAsyncAndParseEntity(
+            cancelTasksRequest,
+            RequestConverters::cancelTasks,
+            parser -> CancelTasksResponse.fromXContent(parser),
+            listener,
+            emptySet(),
+            headers);
+    }
 }

client/rest-high-level/src/main/java/org/elasticsearch/client/RequestConverters.java

@@ -29,6 +29,7 @@
 import org.apache.http.entity.ContentType;
 import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.action.DocWriteRequest;
+import org.elasticsearch.action.admin.cluster.node.tasks.cancel.CancelTasksRequest;
 import org.elasticsearch.action.admin.cluster.node.tasks.list.ListTasksRequest;
 import org.elasticsearch.action.admin.cluster.repositories.get.GetRepositoriesRequest;
 import org.elasticsearch.action.admin.cluster.repositories.put.PutRepositoryRequest;
@@ -101,6 +102,17 @@ private RequestConverters() {
         // Contains only status utility methods
     }
 
+    static Request cancelTasks(CancelTasksRequest cancelTasksRequest) {
+        Request request = new Request(HttpPost.METHOD_NAME, "/_tasks/_cancel");
+        Params params = new Params(request);
+        params.withTimeout(
+            cancelTasksRequest.getTimeout())
+            .withTaskId(cancelTasksRequest.getTaskId())
+            .withNodes(cancelTasksRequest.getNodes())
+            .withActions(cancelTasksRequest.getActions());
+        return request;
+    }
+
     static Request delete(DeleteRequest deleteRequest) {
         String endpoint = endpoint(deleteRequest.index(), deleteRequest.type(), deleteRequest.id());
         Request request = new Request(HttpDelete.METHOD_NAME, endpoint);
@@ -978,6 +990,13 @@ Params withActions(String[] actions) {
             return this;
         }
 
+        Params withTaskId(TaskId taskId) {
+            if (taskId != null && taskId.isSet()) {
+                return putParam("task_id", taskId.toString());
+            }
+            return this;
+        }
+
         Params withParentTaskId(TaskId parentTaskId) {
             if (parentTaskId != null && parentTaskId.isSet()) {
                 return putParam("parent_task_id", parentTaskId.toString());

client/rest-high-level/src/test/java/org/elasticsearch/client/ClusterClientIT.java

@@ -20,6 +20,8 @@
 package org.elasticsearch.client;
 
 import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.action.admin.cluster.node.tasks.cancel.CancelTasksRequest;
+import org.elasticsearch.action.admin.cluster.node.tasks.cancel.CancelTasksResponse;
 import org.elasticsearch.action.admin.cluster.node.tasks.list.ListTasksRequest;
 import org.elasticsearch.action.admin.cluster.node.tasks.list.ListTasksResponse;
 import org.elasticsearch.action.admin.cluster.node.tasks.list.TaskGroup;
@@ -32,6 +34,7 @@
 import org.elasticsearch.common.xcontent.support.XContentMapValues;
 import org.elasticsearch.indices.recovery.RecoverySettings;
 import org.elasticsearch.rest.RestStatus;
+import org.elasticsearch.tasks.TaskId;
 import org.elasticsearch.tasks.TaskInfo;
 
 import java.io.IOException;
@@ -136,4 +139,30 @@ public void testListTasks() throws IOException {
         }
         assertTrue("List tasks were not found", listTasksFound);
     }
+
+    public void testCancelTasks() throws IOException {
+        ListTasksRequest listRequest = new ListTasksRequest();
+        ListTasksResponse listResponse = execute(
+            listRequest,
+            highLevelClient().cluster()::listTasks,
+            highLevelClient().cluster()::listTasksAsync
+        );
+
+        // TODO[PCS] submit a task that is cancellable and assert it's cancelled
+        // this case is covered in TasksIT.testTasksCancellation
+        TaskInfo firstTask = listResponse.getTasks().get(0);
+        String node = listResponse.getPerNodeTasks().keySet().iterator().next();
+
+        CancelTasksRequest request = new CancelTasksRequest();
+        request.setTaskId(new TaskId(node, firstTask.getId()));
+        request.setReason("testreason");
+        CancelTasksResponse response = execute(
+            request,
+            highLevelClient().cluster()::cancelTasks,
+            highLevelClient().cluster()::cancelTasksAsync
+        );
+        // Since the task may or may not have been cancelled, assert that we received a response only
+        // The actual testing of task cancellation is covered by TasksIT.testTasksCancellation
+        assertThat(response, notNullValue());
+    }
 }

client/rest-high-level/src/test/java/org/elasticsearch/client/RequestConvertersTests.java

@@ -29,6 +29,8 @@
 import org.apache.http.util.EntityUtils;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.DocWriteRequest;
+import org.elasticsearch.action.admin.cluster.node.tasks.cancel.CancelTasksRequest;
+import org.elasticsearch.action.admin.cluster.node.tasks.cancel.CancelTasksResponse;
 import org.elasticsearch.action.admin.cluster.node.tasks.list.ListTasksRequest;
 import org.elasticsearch.action.admin.cluster.repositories.get.GetRepositoriesRequest;
 import org.elasticsearch.action.admin.cluster.repositories.put.PutRepositoryRequest;
@@ -1465,6 +1467,20 @@ public void testIndexPutSettings() throws IOException {
         assertEquals(expectedParams, request.getParameters());
     }
 
+    public void testCancelTasks() {
+        CancelTasksRequest request = new CancelTasksRequest();
+        Map<String, String> expectedParams = new HashMap<>();
+        TaskId taskId = new TaskId(randomAlphaOfLength(5), randomNonNegativeLong());
+        request.setTaskId(taskId);
+        expectedParams.put("task_id", taskId.toString());
+        Request httpRequest = RequestConverters.cancelTasks(request);
+        assertThat(httpRequest, notNullValue());
+        assertThat(httpRequest.getMethod(), equalTo(HttpPost.METHOD_NAME));
+        assertThat(httpRequest.getEntity(), nullValue());
+        assertThat(httpRequest.getEndpoint(), equalTo("/_tasks/_cancel"));
+        assertThat(httpRequest.getParameters(), equalTo(expectedParams));
+    }
+
     public void testListTasks() {
         {
             ListTasksRequest request = new ListTasksRequest();

client/rest-high-level/src/test/java/org/elasticsearch/client/documentation/ClusterClientDocumentationIT.java

@@ -21,9 +21,10 @@
 
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.action.ActionListener;
-import org.elasticsearch.action.FailedNodeException;
 import org.elasticsearch.action.LatchedActionListener;
 import org.elasticsearch.action.TaskOperationFailure;
+import org.elasticsearch.action.admin.cluster.node.tasks.cancel.CancelTasksRequest;
+import org.elasticsearch.action.admin.cluster.node.tasks.cancel.CancelTasksResponse;
 import org.elasticsearch.action.admin.cluster.node.tasks.list.ListTasksRequest;
 import org.elasticsearch.action.admin.cluster.node.tasks.list.ListTasksResponse;
 import org.elasticsearch.action.admin.cluster.node.tasks.list.TaskGroup;
@@ -80,19 +81,19 @@ public void testClusterPutSettings() throws IOException {
         // end::put-settings-request
 
         // tag::put-settings-create-settings
-        String transientSettingKey = 
+        String transientSettingKey =
                 RecoverySettings.INDICES_RECOVERY_MAX_BYTES_PER_SEC_SETTING.getKey();
         int transientSettingValue = 10;
-        Settings transientSettings = 
+        Settings transientSettings =
                 Settings.builder()
                 .put(transientSettingKey, transientSettingValue, ByteSizeUnit.BYTES)
                 .build(); // <1>
 
-        String persistentSettingKey = 
+        String persistentSettingKey =
                 EnableAllocationDecider.CLUSTER_ROUTING_ALLOCATION_ENABLE_SETTING.getKey();
-        String persistentSettingValue = 
+        String persistentSettingValue =
                 EnableAllocationDecider.Allocation.NONE.name();
-        Settings persistentSettings = 
+        Settings persistentSettings =
                 Settings.builder()
                 .put(persistentSettingKey, persistentSettingValue)
                 .build(); // <2>
@@ -105,9 +106,9 @@ public void testClusterPutSettings() throws IOException {
 
         {
             // tag::put-settings-settings-builder
-            Settings.Builder transientSettingsBuilder = 
+            Settings.Builder transientSettingsBuilder =
                     Settings.builder()
-                    .put(transientSettingKey, transientSettingValue, ByteSizeUnit.BYTES); 
+                    .put(transientSettingKey, transientSettingValue, ByteSizeUnit.BYTES);
             request.transientSettings(transientSettingsBuilder); // <1>
             // end::put-settings-settings-builder
         }
@@ -164,7 +165,7 @@ public void testClusterUpdateSettingsAsync() throws Exception {
             ClusterUpdateSettingsRequest request = new ClusterUpdateSettingsRequest();
 
             // tag::put-settings-execute-listener
-            ActionListener<ClusterUpdateSettingsResponse> listener = 
+            ActionListener<ClusterUpdateSettingsResponse> listener =
                     new ActionListener<ClusterUpdateSettingsResponse>() {
                 @Override
                 public void onResponse(ClusterUpdateSettingsResponse response) {
@@ -272,4 +273,74 @@ public void onFailure(Exception e) {
             assertTrue(latch.await(30L, TimeUnit.SECONDS));
         }
     }
+
+    public void testCancelTasks() throws IOException {
+        RestHighLevelClient client = highLevelClient();
+        {
+            // tag::cancel-tasks-request
+            CancelTasksRequest request = new CancelTasksRequest();
+            // end::cancel-tasks-request
+
+            // tag::cancel-tasks-request-filter
+            request.setTaskId(new TaskId("nodeId1", 42)); //<1>
+            request.setActions("cluster:*"); // <2>
+            request.setNodes("nodeId1", "nodeId2"); // <3>
+            // end::cancel-tasks-request-filter
+
+        }
+
+        CancelTasksRequest request = new CancelTasksRequest();
+        request.setTaskId(TaskId.EMPTY_TASK_ID);
+
+        // tag::cancel-tasks-execute
+        CancelTasksResponse response = client.cluster().cancelTasks(request);
+        // end::cancel-tasks-execute
+
+        assertThat(response, notNullValue());
+
+        // tag::cancel-tasks-response-tasks
+        List<TaskInfo> tasks = response.getTasks(); // <1>
+        // end::cancel-tasks-response-tasks
+
+
+        // tag::cancel-tasks-response-failures
+        List<ElasticsearchException> nodeFailures = response.getNodeFailures(); // <1>
+        List<TaskOperationFailure> taskFailures = response.getTaskFailures(); // <2>
+        // end::-tasks-response-failures
+
+        assertThat(response.getNodeFailures(), equalTo(emptyList()));
+        assertThat(response.getTaskFailures(), equalTo(emptyList()));
+    }
+
+    public void testAsyncCancelTasks() throws InterruptedException {
+
+        RestHighLevelClient client = highLevelClient();
+        {
+            CancelTasksRequest request = new CancelTasksRequest();
+
+            // tag::cancel-tasks-execute-listener
+            ActionListener<CancelTasksResponse> listener =
+                new ActionListener<CancelTasksResponse>() {
+                    @Override
+                    public void onResponse(CancelTasksResponse response) {
+                        // <1>
+                    }
+                    @Override
+                    public void onFailure(Exception e) {
+                        // <2>
+                    }
+                };
+            // end::cancel-tasks-execute-listener
+
+            // Replace the empty listener by a blocking listener in test
+            final CountDownLatch latch = new CountDownLatch(1);
+            listener = new LatchedActionListener<>(listener, latch);
+
+            // tag::cancel-tasks-execute-async
+            client.cluster().cancelTasksAsync(request, listener); // <1>
+            // end::cancel-tasks-execute-async
+
+            assertTrue(latch.await(30L, TimeUnit.SECONDS));
+        }
+    }
 }

docs/java-rest/high-level/cluster/cancel_tasks.asciidoc

@@ -0,0 +1,82 @@
+[[java-rest-high-cluster-cancel-tasks]]
+=== Cancel Tasks API
+
+The Cancel Tasks API allows cancellation of a currently running task.
+
+==== Cancel Tasks Request
+
+A `CancelTasksRequest`:
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests}/ClusterClientDocumentationIT.java[cancel-tasks-request]
+--------------------------------------------------
+There are no required parameters. The task cancellation command supports the same
+task selection parameters as the list tasks command.
+
+==== Parameters
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests}/ClusterClientDocumentationIT.java[list-tasks-request-filter]
+--------------------------------------------------
+<1> Cancel a task
+<2> Cancel only cluster-related tasks
+<3> Cancel all tasks running on nodes nodeId1 and nodeId2
+
+==== Synchronous Execution
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests}/ClusterClientDocumentationIT.java[list-tasks-execute]
+--------------------------------------------------
+
+==== Asynchronous Execution
+
+The asynchronous execution requires `CancelTasksRequest` instance and an
+`ActionListener` instance to be passed to the asynchronous method:
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests}/ClusterClientDocumentationIT.java[cancel-tasks-execute-async]
+--------------------------------------------------
+<1> The `CancelTasksRequest` to execute and the `ActionListener` to use
+when the execution completes
+
+The asynchronous method does not block and returns immediately. Once it is
+completed the `ActionListener` is called back using the `onResponse` method
+if the execution successfully completed or using the `onFailure` method if
+it failed.
+
+A typical listener for `CancelTasksResponse` looks like:
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests}/ClusterClientDocumentationIT.java[cancel-tasks-execute-listener]
+--------------------------------------------------
+<1> Called when the execution is successfully completed. The response is
+provided as an argument
+<2> Called in case of a failure. The raised exception is provided as an argument
+
+==== Cancel Tasks Response
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests}/ClusterClientDocumentationIT.java[list-tasks-response-tasks]
+--------------------------------------------------
+<1> List of cancelled tasks
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests}/ClusterClientDocumentationIT.java[list-tasks-response-calc]
+--------------------------------------------------
+<1> List of cancelled tasks grouped by a node
+<2> List of cancelled tasks grouped by a parent task
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests}/ClusterClientDocumentationIT.java[list-tasks-response-failures]
+--------------------------------------------------
+<1> List of node failures
+<2> List of task cancellation failures
+

docs/java-rest/high-level/supported-apis.asciidoc

@@ -105,9 +105,11 @@ The Java High Level REST Client supports the following Cluster APIs:
 
 * <<java-rest-high-cluster-put-settings>>
 * <<java-rest-high-cluster-list-tasks>>
+* <<java-rest-high-cluster-cancel-tasks>>
 
 include::cluster/put_settings.asciidoc[]
 include::cluster/list_tasks.asciidoc[]
+include::cluster/cancel_tasks.asciidoc[]
 
 == Snapshot APIs