Skip to content

Adding a cancelled field to tell if a cancellable task is cancelled #1682

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
meghasaik wants to merge 4 commits into from
Closed

Adding a cancelled field to tell if a cancellable task is cancelled #1682

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
c60caef
Adding the cancelled field for telling if the task is cancelled.
meghasaik Dec 8, 2021
c0f8384
Changes made to add the cancelled field when the task is completed.
meghasaik Dec 10, 2021
9121895
Merge branch 'task-cancellation-indicator' of https://github.com/megh…
meghasaik Dec 13, 2021
ebb2ab1
Merge branch 'opensearch-project:main' into task-cancellation-indicator
meghasaik Dec 13, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 14 additions & 0 deletions client/rest-high-level/src/main/java/org/opensearch/client/tasks/TaskInfo.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,7 @@ public class TaskInfo {
private long startTime;
private long runningTimeNanos;
private boolean cancellable;
private boolean cancelled;
private TaskId parentTaskId;
private final Map<String, Object> status = new HashMap<>();
private final Map<String, String> headers = new HashMap<>();
Expand Down Expand Up @@ -117,6 +118,14 @@ void setCancellable(boolean cancellable) {
this.cancellable = cancellable;
}

public boolean isCancelled() {
return cancelled;
}

void setCancelled(boolean cancelled) {
this.cancelled = cancelled;
}

public TaskId getParentTaskId() {
return parentTaskId;
}
Expand Down Expand Up @@ -158,6 +167,7 @@ private void noOpParse(Object s) {}
parser.declareLong(TaskInfo::setStartTime, new ParseField("start_time_in_millis"));
parser.declareLong(TaskInfo::setRunningTimeNanos, new ParseField("running_time_in_nanos"));
parser.declareBoolean(TaskInfo::setCancellable, new ParseField("cancellable"));
parser.declareBoolean(TaskInfo::setCancelled, new ParseField("cancelled"));
parser.declareString(TaskInfo::setParentTaskId, new ParseField("parent_task_id"));
parser.declareObject(TaskInfo::setHeaders, (p, c) -> p.mapStrings(), new ParseField("headers"));
PARSER = (XContentParser p, Void v, String name) -> parser.parse(p, new TaskInfo(new TaskId(name)), null);
Expand All @@ -171,6 +181,7 @@ public boolean equals(Object o) {
return getStartTime() == taskInfo.getStartTime()
&& getRunningTimeNanos() == taskInfo.getRunningTimeNanos()
&& isCancellable() == taskInfo.isCancellable()
&& isCancelled() == taskInfo.isCancelled()
&& Objects.equals(getTaskId(), taskInfo.getTaskId())
&& Objects.equals(getType(), taskInfo.getType())
&& Objects.equals(getAction(), taskInfo.getAction())
Expand All @@ -190,6 +201,7 @@ public int hashCode() {
getStartTime(),
getRunningTimeNanos(),
isCancellable(),
isCancelled(),
getParentTaskId(),
status,
getHeaders()
Expand All @@ -216,6 +228,8 @@ public String toString() {
+ runningTimeNanos
+ ", cancellable="
+ cancellable
+ ", cancelled="
+ cancelled
+ ", parentTaskId="
+ parentTaskId
+ ", status="
Expand Down
15 changes: 14 additions & 1 deletion .../rest-high-level/src/test/java/org/opensearch/client/core/tasks/GetTaskResponseTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -90,11 +90,24 @@ static TaskInfo randomTaskInfo() {
long startTime = randomLong();
long runningTimeNanos = randomLong();
boolean cancellable = randomBoolean();
boolean cancelled = cancellable == true ? randomBoolean() : false;
TaskId parentTaskId = randomBoolean() ? TaskId.EMPTY_TASK_ID : randomTaskId();
Map<String, String> headers = randomBoolean()
? Collections.emptyMap()
: Collections.singletonMap(randomAlphaOfLength(5), randomAlphaOfLength(5));
return new TaskInfo(taskId, type, action, description, status, startTime, runningTimeNanos, cancellable, parentTaskId, headers);
return new TaskInfo(
taskId,
type,
action,
description,
status,
startTime,
runningTimeNanos,
cancellable,
cancelled,
parentTaskId,
headers
);
}

private static TaskId randomTaskId() {
Expand Down
6 changes: 5 additions & 1 deletion ...t/rest-high-level/src/test/java/org/opensearch/client/tasks/CancelTasksResponseTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,8 @@ protected CancelTasksResponseTests.ByNodeCancelTasksResponse createServerTestIns
}

for (int i = 0; i < 4; i++) {
boolean cancellable = randomBoolean();
boolean cancelled = cancellable == true ? randomBoolean() : false;
tasks.add(
new org.opensearch.tasks.TaskInfo(
new TaskId(NODE_ID, (long) i),
Expand All @@ -91,7 +93,8 @@ protected CancelTasksResponseTests.ByNodeCancelTasksResponse createServerTestIns
new FakeTaskStatus(randomAlphaOfLength(4), randomInt()),
randomLongBetween(1, 3),
randomIntBetween(5, 10),
false,
cancellable,
cancelled,
new TaskId("node1", randomLong()),
Collections.singletonMap("x-header-of", "some-value")
)
Expand Down Expand Up @@ -128,6 +131,7 @@ protected void assertInstances(
assertEquals(ti.getStartTime(), taskInfo.getStartTime());
assertEquals(ti.getRunningTimeNanos(), taskInfo.getRunningTimeNanos());
assertEquals(ti.isCancellable(), taskInfo.isCancellable());
assertEquals(ti.isCancelled(), taskInfo.isCancelled());
assertEquals(ti.getParentTaskId().getNodeId(), taskInfo.getParentTaskId().getNodeId());
assertEquals(ti.getParentTaskId().getId(), taskInfo.getParentTaskId().getId());
FakeTaskStatus status = (FakeTaskStatus) ti.getStatus();
Expand Down
2 changes: 2 additions & 0 deletions ...es/reindex/src/test/java/org/opensearch/index/reindex/TransportRethrottleActionTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -129,6 +129,7 @@ public void testRethrottleSuccessfulResponse() {
0,
0,
true,
false,
new TaskId("test", task.getId()),
Collections.emptyMap()
)
Expand Down Expand Up @@ -164,6 +165,7 @@ public void testRethrottleWithSomeSucceeded() {
0,
0,
true,
false,
new TaskId("test", task.getId()),
Collections.emptyMap()
)
Expand Down
14 changes: 13 additions & 1 deletion .../src/internalClusterTest/java/org/opensearch/action/admin/cluster/node/tasks/TasksIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -901,7 +901,19 @@ public void testNodeNotFoundButTaskFound() throws Exception {
TaskResultsService resultsService = internalCluster().getInstance(TaskResultsService.class);
resultsService.storeResult(
new TaskResult(
new TaskInfo(new TaskId("fake", 1), "test", "test", "", null, 0, 0, false, TaskId.EMPTY_TASK_ID, Collections.emptyMap()),
new TaskInfo(
new TaskId("fake", 1),
"test",
"test",
"",
null,
0,
0,
false,
false,
TaskId.EMPTY_TASK_ID,
Collections.emptyMap()
),
new RuntimeException("test")
),
new ActionListener<Void>() {
Expand Down
1 change: 1 addition & 0 deletions server/src/main/java/org/opensearch/tasks/Task.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -130,6 +130,7 @@ protected final TaskInfo taskInfo(String localNodeId, String description, Status
startTime,
System.nanoTime() - startTimeNanos,
this instanceof CancellableTask,
this instanceof CancellableTask && ((CancellableTask) this).isCancelled(),

@reta reta Dec 9, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't know to be honest why this method is final, but it certainly looks like it should not be and should be overridden by CancellableTask, without the need to use instanceof and type casting.

@andrross andrross Dec 10, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agreed. It seems to defeat the purpose of inheritance to have a parent class check at runtime whether it is an instance of a child class.

parentTask,
headers
);
Expand Down
58 changes: 56 additions & 2 deletions server/src/main/java/org/opensearch/tasks/TaskInfo.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,11 @@

package org.opensearch.tasks;

<<<<<<< HEAD
=======
Comment on lines +35 to +36

@andrross andrross Dec 13, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Merge problem?

import org.opensearch.Version;
import org.opensearch.LegacyESVersion;
>>>>>>> c0f838439041b18b511ec6c5b42bac033ff1753f
import org.opensearch.common.ParseField;
import org.opensearch.common.Strings;
import org.opensearch.common.bytes.BytesReference;
Expand Down Expand Up @@ -79,6 +84,8 @@ public final class TaskInfo implements Writeable, ToXContentFragment {

private final boolean cancellable;

private final boolean cancelled;

private final TaskId parentTaskId;

private final Map<String, String> headers;
Expand All @@ -92,9 +99,11 @@ public TaskInfo(
long startTime,
long runningTimeNanos,
boolean cancellable,
boolean cancelled,
TaskId parentTaskId,
Map<String, String> headers
) {
assert cancellable || cancelled == false : "uncancellable task cannot be cancelled";
this.taskId = taskId;
this.type = type;
this.action = action;
Expand All @@ -103,6 +112,7 @@ public TaskInfo(
this.startTime = startTime;
this.runningTimeNanos = runningTimeNanos;
this.cancellable = cancellable;
this.cancelled = cancelled;
this.parentTaskId = parentTaskId;
this.headers = headers;
}
Expand All @@ -119,6 +129,12 @@ public TaskInfo(StreamInput in) throws IOException {
startTime = in.readLong();
runningTimeNanos = in.readLong();
cancellable = in.readBoolean();
if (in.getVersion().onOrAfter(Version.V_2_0_0)) {
cancelled = in.readBoolean();
} else {
cancelled = false;
}
assert cancellable || cancelled == false : "uncancellable task cannot be cancelled";

@andrross andrross Dec 13, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm not sure assert is the right thing to do here. An AssertionError (because it is a child of error) "indicates serious problems that a reasonable application should not try to catch". This is constructing an object from parsing a stream, so are there contexts where external input could trigger this path? If so, we definitely don't want the application to throw an Error. IllegalArgumentException seems more appropriate to me.

@meghasaik meghasaik Dec 13, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This might trigger in those cases where we are trying to cancel a task which cannot be cancelled when cancelled is true and cancellable is false. But as you mentioned we can throw a IllegalArgumentException instead of the assert error.

parentTaskId = TaskId.readFromStream(in);
headers = in.readMap(StreamInput::readString, StreamInput::readString);
}
Expand All @@ -133,6 +149,9 @@ public void writeTo(StreamOutput out) throws IOException {
out.writeLong(startTime);
out.writeLong(runningTimeNanos);
out.writeBoolean(cancellable);
if (out.getVersion().onOrAfter(Version.V_2_0_0)) {
out.writeBoolean(cancelled);
}
parentTaskId.writeTo(out);
out.writeMap(headers, StreamOutput::writeString, StreamOutput::writeString);
}
Expand Down Expand Up @@ -186,6 +205,13 @@ public boolean isCancellable() {
return cancellable;
}

/**
* Returns true if the task has been cancelled
*/
public boolean isCancelled() {
return cancelled;
}

/**
* Returns the parent task id
*/
Expand Down Expand Up @@ -218,6 +244,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
}
builder.field("running_time_in_nanos", runningTimeNanos);
builder.field("cancellable", cancellable);
builder.field("cancelled", cancelled);
if (parentTaskId.isSet()) {
builder.field("parent_task_id", parentTaskId.toString());
}
Expand All @@ -243,6 +270,7 @@ public static TaskInfo fromXContent(XContentParser parser) {
long startTime = (Long) a[i++];
long runningTimeNanos = (Long) a[i++];
boolean cancellable = (Boolean) a[i++];
boolean cancelled = a[i++] == Boolean.TRUE;
String parentTaskIdString = (String) a[i++];
@SuppressWarnings("unchecked")
Map<String, String> headers = (Map<String, String>) a[i++];
Expand All @@ -252,7 +280,19 @@ public static TaskInfo fromXContent(XContentParser parser) {
}
RawTaskStatus status = statusBytes == null ? null : new RawTaskStatus(statusBytes);
TaskId parentTaskId = parentTaskIdString == null ? TaskId.EMPTY_TASK_ID : new TaskId(parentTaskIdString);
return new TaskInfo(id, type, action, description, status, startTime, runningTimeNanos, cancellable, parentTaskId, headers);
return new TaskInfo(
id,
type,
action,
description,
status,
startTime,
runningTimeNanos,
cancellable,
cancelled,
parentTaskId,
headers
);
});
static {
// Note for the future: this has to be backwards and forwards compatible with all changes to the task storage format
Expand All @@ -266,6 +306,7 @@ public static TaskInfo fromXContent(XContentParser parser) {
PARSER.declareLong(constructorArg(), new ParseField("start_time_in_millis"));
PARSER.declareLong(constructorArg(), new ParseField("running_time_in_nanos"));
PARSER.declareBoolean(constructorArg(), new ParseField("cancellable"));
PARSER.declareBoolean(optionalConstructorArg(), new ParseField("cancelled"));
PARSER.declareString(optionalConstructorArg(), new ParseField("parent_task_id"));
PARSER.declareObject(optionalConstructorArg(), (p, c) -> p.mapStrings(), new ParseField("headers"));
}
Expand All @@ -290,12 +331,25 @@ public boolean equals(Object obj) {
&& Objects.equals(runningTimeNanos, other.runningTimeNanos)
&& Objects.equals(parentTaskId, other.parentTaskId)
&& Objects.equals(cancellable, other.cancellable)
&& Objects.equals(cancelled, other.cancelled)
&& Objects.equals(status, other.status)
&& Objects.equals(headers, other.headers);
}

@Override
public int hashCode() {
return Objects.hash(taskId, type, action, description, startTime, runningTimeNanos, parentTaskId, cancellable, status, headers);
return Objects.hash(
taskId,
type,
action,
description,
startTime,
runningTimeNanos,
parentTaskId,
cancellable,
cancelled,
status,
headers
);
}
}
3 changes: 3 additions & 0 deletions server/src/main/resources/org/opensearch/tasks/task-index-mapping.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,9 @@
"cancellable": {
"type": "boolean"
},
"cancelled": {
"type": "boolean"
},
"id": {
"type": "long"
},
Expand Down
Loading