Broadcast cancellation to only nodes have outstanding child tasks (#5…

…4312) Today when canceling a task we broadcast ban/unban requests to all nodes in the cluster. This strategy does not scale well for hierarchical cancellation. With this change, we will track outstanding child requests and broadcast the cancellation to only nodes that have outstanding child tasks. This change also prevents a parent task from sending child requests once it got canceled. Relates #50990 Supersedes #51157 Co-authored-by: Igor Motov <[email protected]> Co-authored-by: Yannick Welsch <[email protected]>
elastic · Apr 1, 2020 · ee3d403 · ee3d403
1 parent 7738ed4
commit ee3d403
Show file tree

Hide file tree

Showing 16 changed files with 905 additions and 222 deletions.
diff --git a/client/rest-high-level/src/main/java/org/elasticsearch/client/TasksRequestConverters.java b/client/rest-high-level/src/main/java/org/elasticsearch/client/TasksRequestConverters.java
@@ -39,6 +39,9 @@ static Request cancelTasks(CancelTasksRequest req) {
         params
             .withNodes(req.getNodes())
             .withActions(req.getActions());
+        if (req.getWaitForCompletion() != null) {
+            params.withWaitForCompletion(req.getWaitForCompletion());
+        }
         request.addParameters(params.asMap());
         return request;
     }

diff --git a/client/rest-high-level/src/main/java/org/elasticsearch/client/tasks/CancelTasksRequest.java b/client/rest-high-level/src/main/java/org/elasticsearch/client/tasks/CancelTasksRequest.java
@@ -33,6 +33,7 @@ public class CancelTasksRequest implements Validatable {
     private Optional<TimeValue> timeout = Optional.empty();
     private Optional<TaskId> parentTaskId = Optional.empty();
     private Optional<TaskId> taskId = Optional.empty();
+    private Boolean waitForCompletion;
 
     CancelTasksRequest(){}
 
@@ -76,6 +77,14 @@ public Optional<TaskId> getTaskId() {
         return taskId;
     }
 
+    public Boolean getWaitForCompletion() {
+        return waitForCompletion;
+    }
+
+    public void setWaitForCompletion(boolean waitForCompletion) {
+        this.waitForCompletion = waitForCompletion;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;
@@ -85,12 +94,13 @@ public boolean equals(Object o) {
             Objects.equals(getActions(), that.getActions()) &&
             Objects.equals(getTimeout(), that.getTimeout()) &&
             Objects.equals(getParentTaskId(), that.getParentTaskId()) &&
-            Objects.equals(getTaskId(), that.getTaskId()) ;
+            Objects.equals(getTaskId(), that.getTaskId()) &&
+            Objects.equals(waitForCompletion, that.waitForCompletion);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(getNodes(), getActions(), getTimeout(), getParentTaskId(), getTaskId());
+        return Objects.hash(getNodes(), getActions(), getTimeout(), getParentTaskId(), getTaskId(), waitForCompletion);
     }
 
     @Override
@@ -101,6 +111,7 @@ public String toString() {
             ", timeout=" + timeout +
             ", parentTaskId=" + parentTaskId +
             ", taskId=" + taskId +
+            ", waitForCompletion=" + waitForCompletion +
             '}';
     }
 
@@ -110,6 +121,7 @@ public static class Builder {
         private Optional<TaskId> parentTaskId = Optional.empty();
         private List<String> actionsFilter = new ArrayList<>();
         private List<String> nodesFilter = new ArrayList<>();
+        private Boolean waitForCompletion;
 
         public Builder withTimeout(TimeValue timeout){
             this.timeout = Optional.of(timeout);
@@ -138,13 +150,21 @@ public Builder withNodesFiltered(List<String> nodes){
             return this;
         }
 
+        public Builder withWaitForCompletion(boolean waitForCompletion) {
+            this.waitForCompletion = waitForCompletion;
+            return this;
+        }
+
         public CancelTasksRequest build() {
             CancelTasksRequest request = new CancelTasksRequest();
             timeout.ifPresent(request::setTimeout);
             taskId.ifPresent(request::setTaskId);
             parentTaskId.ifPresent(request::setParentTaskId);
             request.setNodes(nodesFilter);
             request.setActions(actionsFilter);
+            if (waitForCompletion != null) {
+                request.setWaitForCompletion(waitForCompletion);
+            }
             return request;
         }
     }

diff --git a/...t/rest-high-level/src/test/java/org/elasticsearch/client/TasksRequestConvertersTests.java b/...t/rest-high-level/src/test/java/org/elasticsearch/client/TasksRequestConvertersTests.java
@@ -22,6 +22,7 @@
 import org.apache.http.client.methods.HttpGet;
 import org.apache.http.client.methods.HttpPost;
 import org.elasticsearch.action.admin.cluster.node.tasks.list.ListTasksRequest;
+import org.elasticsearch.client.tasks.CancelTasksRequest;
 import org.elasticsearch.tasks.TaskId;
 import org.elasticsearch.test.ESTestCase;
 
@@ -40,14 +41,15 @@ public void testCancelTasks() {
             new org.elasticsearch.client.tasks.TaskId(randomAlphaOfLength(5), randomNonNegativeLong());
         org.elasticsearch.client.tasks.TaskId parentTaskId =
             new org.elasticsearch.client.tasks.TaskId(randomAlphaOfLength(5), randomNonNegativeLong());
-        org.elasticsearch.client.tasks.CancelTasksRequest request =
-            new org.elasticsearch.client.tasks.CancelTasksRequest.Builder()
-                .withTaskId(taskId)
-                .withParentTaskId(parentTaskId)
-                .build();
+        CancelTasksRequest.Builder builder = new CancelTasksRequest.Builder().withTaskId(taskId).withParentTaskId(parentTaskId);
         expectedParams.put("task_id", taskId.toString());
         expectedParams.put("parent_task_id", parentTaskId.toString());
-        Request httpRequest = TasksRequestConverters.cancelTasks(request);
+        if (randomBoolean()) {
+            boolean waitForCompletion = randomBoolean();
+            builder.withWaitForCompletion(waitForCompletion);
+            expectedParams.put("wait_for_completion", Boolean.toString(waitForCompletion));
+        }
+        Request httpRequest = TasksRequestConverters.cancelTasks(builder.build());
         assertThat(httpRequest, notNullValue());
         assertThat(httpRequest.getMethod(), equalTo(HttpPost.METHOD_NAME));
         assertThat(httpRequest.getEntity(), nullValue());

diff --git a/...evel/src/test/java/org/elasticsearch/client/documentation/TasksClientDocumentationIT.java b/...evel/src/test/java/org/elasticsearch/client/documentation/TasksClientDocumentationIT.java
@@ -164,7 +164,8 @@ public void testCancelTasks() throws IOException {
             // tag::cancel-tasks-request-filter
             CancelTasksRequest byTaskIdRequest = new org.elasticsearch.client.tasks.CancelTasksRequest.Builder() // <1>
                 .withTaskId(new org.elasticsearch.client.tasks.TaskId("myNode",44L)) // <2>
-                .build(); // <3>
+                .withWaitForCompletion(true) // <3>
+                .build(); // <4>
             // end::cancel-tasks-request-filter
 
         }

diff --git a/docs/java-rest/high-level/tasks/cancel_tasks.asciidoc b/docs/java-rest/high-level/tasks/cancel_tasks.asciidoc
@@ -22,7 +22,9 @@ include-tagged::{doc-tests}/TasksClientDocumentationIT.java[cancel-tasks-request
 --------------------------------------------------
 <1> Cancel a task
 <2> Cancel only cluster-related tasks
-<3> Cancel all tasks running on nodes nodeId1 and nodeId2
+<3> Should the request block until the cancellation of the task and its child tasks is completed.
+Otherwise, the request can return soon after the cancellation is started. Defaults to `false`.
+<4> Cancel all tasks running on nodes nodeId1 and nodeId2
 
 ==== Synchronous Execution
 

diff --git a/docs/reference/cluster/tasks.asciidoc b/docs/reference/cluster/tasks.asciidoc
@@ -232,6 +232,11 @@ list tasks command, so multiple tasks can be cancelled at the same time. For
 example, the following command will cancel all reindex tasks running on the 
 nodes `nodeId1` and `nodeId2`.
 
+`wait_for_completion`::
+(Optional, boolean) If `true`, the request blocks until the cancellation of the
+task and its child tasks is completed. Otherwise, the request can return soon
+after the cancellation is started. Defaults to `false`.
+
 [source,console]
 --------------------------------------------------
 POST _tasks/_cancel?nodes=nodeId1,nodeId2&actions=*reindex

diff --git a/rest-api-spec/src/main/resources/rest-api-spec/api/tasks.cancel.json b/rest-api-spec/src/main/resources/rest-api-spec/api/tasks.cancel.json
@@ -39,6 +39,10 @@
       "parent_task_id":{
         "type":"string",
         "description":"Cancel tasks with specified parent task id (node_id:task_number). Set to -1 to cancel all."
+      },
+      "wait_for_completion": {
+        "type":"boolean",
+        "description":"Should the request block until the cancellation of the task and its child tasks is completed. Defaults to false"
       }
     }
   }

diff --git a/...ain/java/org/elasticsearch/action/admin/cluster/node/tasks/cancel/CancelTasksRequest.java b/...ain/java/org/elasticsearch/action/admin/cluster/node/tasks/cancel/CancelTasksRequest.java
@@ -19,6 +19,7 @@
 
 package org.elasticsearch.action.admin.cluster.node.tasks.cancel;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.action.support.tasks.BaseTasksRequest;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
@@ -33,20 +34,28 @@
 public class CancelTasksRequest extends BaseTasksRequest<CancelTasksRequest> {
 
     public static final String DEFAULT_REASON = "by user request";
+    public static final boolean DEFAULT_WAIT_FOR_COMPLETION = false;
 
     private String reason = DEFAULT_REASON;
+    private boolean waitForCompletion = DEFAULT_WAIT_FOR_COMPLETION;
 
     public CancelTasksRequest() {}
 
     public CancelTasksRequest(StreamInput in) throws IOException {
         super(in);
         this.reason = in.readString();
+        if (in.getVersion().onOrAfter(Version.V_8_0_0)) {
+            waitForCompletion = in.readBoolean();
+        }
     }
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
         out.writeString(reason);
+        if (out.getVersion().onOrAfter(Version.V_8_0_0)) {
+            out.writeBoolean(waitForCompletion);
+        }
     }
 
     @Override
@@ -68,4 +77,16 @@ public CancelTasksRequest setReason(String reason) {
     public String getReason() {
         return reason;
     }
+
+    /**
+     * If {@code true}, the request blocks until the cancellation of the task and its child tasks is completed.
+     * Otherwise, the request can return soon after the cancellation is started. Defaults to {@code false}.
+     */
+    public void setWaitForCompletion(boolean waitForCompletion) {
+        this.waitForCompletion = waitForCompletion;
+    }
+
+    public boolean waitForCompletion() {
+        return waitForCompletion;
+    }
 }
diff --git a/...a/org/elasticsearch/action/admin/cluster/node/tasks/cancel/CancelTasksRequestBuilder.java b/...a/org/elasticsearch/action/admin/cluster/node/tasks/cancel/CancelTasksRequestBuilder.java
@@ -31,4 +31,8 @@ public CancelTasksRequestBuilder(ElasticsearchClient client, CancelTasksAction a
         super(client, action, new CancelTasksRequest());
     }
 
+    public CancelTasksRequestBuilder waitForCompletion(boolean waitForCompletion) {
+        request.setWaitForCompletion(waitForCompletion);
+        return this;
+    }
 }