honeymoose/OpenSearch
1
0
Fork 0
mirror of https://github.com/honeymoose/OpenSearch.git synced 2025-03-27 10:28:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

Persistent Tasks: force writeable name of params and status to be the same as their task (#1072)

Browse Source 
Changes persistent task serialization and forces params and status to have the same writeable name as the task itself.
This commit is contained in:
Igor Motov 2017-04-13 09:49:55 -04:00 committed by Martijn van Groningen
parent 4771965931
commit fc524bc9b5
No known key found for this signature in database
GPG Key ID: AB236F4FCF2AF12A
10 changed files with 130 additions and 75 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

90
server/src/main/java/org/elasticsearch/persistent/PersistentTasksCustomMetaData.java
View File

@ -82,16 +82,21 @@ public final class PersistentTasksCustomMetaData extends AbstractNamedDiffable<M
public static final ConstructingObjectParser<Assignment, Void> ASSIGNMENT_PARSER =
new ConstructingObjectParser<>("assignment", objects -> new Assignment((String) objects[0], (String) objects[1]));
private static final NamedObjectParser<PersistentTaskParams, Void> PARAMS_PARSER =
(XContentParser p, Void c, String name) -> p.namedObject(PersistentTaskParams.class, name, null);
private static final NamedObjectParser<Status, Void> STATUS_PARSER =
(XContentParser p, Void c, String name) -> p.namedObject(Status.class, name, null);
private static final NamedObjectParser<TaskDescriptionBuilder<PersistentTaskParams>, Void> TASK_DESCRIPTION_PARSER;
static {
// Tasks parser initialization
PERSISTENT_TASKS_PARSER.declareLong(Builder::setLastAllocationId, new ParseField("last_allocation_id"));
PERSISTENT_TASKS_PARSER.declareObjectArray(Builder::setTasks, PERSISTENT_TASK_PARSER, new ParseField("tasks"));
// Task description parser initialization
ObjectParser<TaskDescriptionBuilder<PersistentTaskParams>, String> parser = new ObjectParser<>("named");
parser.declareObject(TaskDescriptionBuilder::setParams,
(p, c) -> p.namedObject(PersistentTaskParams.class, c, null), new ParseField("params"));
parser.declareObject(TaskDescriptionBuilder::setStatus,
(p, c) -> p.namedObject(Status.class, c, null), new ParseField("status"));
TASK_DESCRIPTION_PARSER = (XContentParser p, Void c, String name) -> parser.parse(p, new TaskDescriptionBuilder<>(name), name);
// Assignment parser
ASSIGNMENT_PARSER.declareStringOrNull(constructorArg(), new ParseField("executor_node"));
ASSIGNMENT_PARSER.declareStringOrNull(constructorArg(), new ParseField("explanation"));
@ -100,28 +105,46 @@ public final class PersistentTasksCustomMetaData extends AbstractNamedDiffable<M
PERSISTENT_TASK_PARSER.declareString(TaskBuilder::setId, new ParseField("id"));
PERSISTENT_TASK_PARSER.declareString(TaskBuilder::setTaskName, new ParseField("name"));
PERSISTENT_TASK_PARSER.declareLong(TaskBuilder::setAllocationId, new ParseField("allocation_id"));
PERSISTENT_TASK_PARSER.declareNamedObjects(
(TaskBuilder<PersistentTaskParams> taskBuilder, List<PersistentTaskParams> objects) -> {
if (objects.size() != 1) {
throw new IllegalArgumentException("only one params per task is allowed");
}
taskBuilder.setParams(objects.get(0));
}, PARAMS_PARSER, new ParseField("params"));
PERSISTENT_TASK_PARSER.declareNamedObjects(
(TaskBuilder<PersistentTaskParams> taskBuilder, List<Status> objects) -> {
(TaskBuilder<PersistentTaskParams> taskBuilder, List<TaskDescriptionBuilder<PersistentTaskParams>> objects) -> {
if (objects.size() != 1) {
throw new IllegalArgumentException("only one status per task is allowed");
throw new IllegalArgumentException("only one task description per task is allowed");
}
taskBuilder.setStatus(objects.get(0));
}, STATUS_PARSER, new ParseField("status"));
TaskDescriptionBuilder<PersistentTaskParams> builder = objects.get(0);
taskBuilder.setTaskName(builder.taskName);
taskBuilder.setParams(builder.params);
taskBuilder.setStatus(builder.status);
}, TASK_DESCRIPTION_PARSER, new ParseField("task"));
PERSISTENT_TASK_PARSER.declareObject(TaskBuilder::setAssignment, ASSIGNMENT_PARSER, new ParseField("assignment"));
PERSISTENT_TASK_PARSER.declareLong(TaskBuilder::setAllocationIdOnLastStatusUpdate,
new ParseField("allocation_id_on_last_status_update"));
}
/**
* Private builder used in XContent parser to build task-specific portion (params and status)
*/
private static class TaskDescriptionBuilder<Params extends PersistentTaskParams> {
private final String taskName;
private Params params;
private Status status;
private TaskDescriptionBuilder(String taskName) {
this.taskName = taskName;
}
private TaskDescriptionBuilder setParams(Params params) {
this.params = params;
return this;
}
private TaskDescriptionBuilder setStatus(Status status) {
this.status = status;
return this;
}
}
public Collection<PersistentTask<?>> tasks() {
return this.tasks.values();
}
@ -279,6 +302,18 @@ public final class PersistentTasksCustomMetaData extends AbstractNamedDiffable<M
this.status = status;
this.assignment = assignment;
this.allocationIdOnLastStatusUpdate = allocationIdOnLastStatusUpdate;
if (params != null) {
if (params.getWriteableName().equals(taskName) == false) {
throw new IllegalArgumentException("params have to have the same writeable name as task. params: " +
params.getWriteableName() + " task: " + taskName);
}
}
if (status != null) {
if (status.getWriteableName().equals(taskName) == false) {
throw new IllegalArgumentException("status has to have the same writeable name as task. status: " +
status.getWriteableName() + " task: " + taskName);
}
}
}
@SuppressWarnings("unchecked")
@ -384,21 +419,20 @@ public final class PersistentTasksCustomMetaData extends AbstractNamedDiffable<M
builder.startObject();
{
builder.field("id", id);
builder.field("name", taskName);
if (params != null) {
builder.startObject("params");
builder.startObject("task");
{
builder.startObject(taskName);
{
builder.field(params.getWriteableName(), params, xParams);
}
builder.endObject();
}
if (status != null) {
builder.startObject("status");
{
builder.field(status.getWriteableName(), status, xParams);
if (params != null) {
builder.field("params", params, xParams);
}
if (status != null) {
builder.field("status", status, xParams);
}
}
builder.endObject();
}
builder.endObject();
if (API_CONTEXT.equals(xParams.param(MetaData.CONTEXT_MODE_PARAM, API_CONTEXT))) {
// These are transient values that shouldn't be persisted to gateway cluster state or snapshot

34
server/src/main/java/org/elasticsearch/persistent/StartPersistentTaskAction.java
View File

@ -74,7 +74,7 @@ public class StartPersistentTaskAction extends Action<StartPersistentTaskAction.
private String taskId;
@Nullable
private String action;
private String taskName;
private PersistentTaskParams params;
@ -82,9 +82,9 @@ public class StartPersistentTaskAction extends Action<StartPersistentTaskAction.
}
public Request(String taskId, String action, PersistentTaskParams params) {
public Request(String taskId, String taskName, PersistentTaskParams params) {
this.taskId = taskId;
this.action = action;
this.taskName = taskName;
this.params = params;
}
@ -92,7 +92,7 @@ public class StartPersistentTaskAction extends Action<StartPersistentTaskAction.
public void readFrom(StreamInput in) throws IOException {
super.readFrom(in);
taskId = in.readString();
action = in.readString();
taskName = in.readString();
params = in.readOptionalNamedWriteable(PersistentTaskParams.class);
}
@ -100,7 +100,7 @@ public class StartPersistentTaskAction extends Action<StartPersistentTaskAction.
public void writeTo(StreamOutput out) throws IOException {
super.writeTo(out);
out.writeString(taskId);
out.writeString(action);
out.writeString(taskName);
out.writeOptionalNamedWriteable(params);
}
@ -110,9 +110,15 @@ public class StartPersistentTaskAction extends Action<StartPersistentTaskAction.
if (this.taskId == null) {
validationException = addValidationError("task id must be specified", validationException);
}
if (this.action == null) {
if (this.taskName == null) {
validationException = addValidationError("action must be specified", validationException);
}
if (params != null) {
if (params.getWriteableName().equals(taskName) == false) {
validationException = addValidationError("params have to have the same writeable name as task. params: " +
params.getWriteableName() + " task: " + taskName, validationException);
}
}
return validationException;
}
@ -121,21 +127,21 @@ public class StartPersistentTaskAction extends Action<StartPersistentTaskAction.
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
Request request1 = (Request) o;
return Objects.equals(taskId, request1.taskId) && Objects.equals(action, request1.action) &&
return Objects.equals(taskId, request1.taskId) && Objects.equals(taskName, request1.taskName) &&
Objects.equals(params, request1.params);
}
@Override
public int hashCode() {
return Objects.hash(taskId, action, params);
return Objects.hash(taskId, taskName, params);
}
public String getAction() {
return action;
public String getTaskName() {
return taskName;
}
public void setAction(String action) {
this.action = action;
public void setTaskName(String taskName) {
this.taskName = taskName;
}
public String getTaskId() {
@ -170,7 +176,7 @@ public class StartPersistentTaskAction extends Action<StartPersistentTaskAction.
}
public RequestBuilder setAction(String action) {
request.setAction(action);
request.setTaskName(action);
return this;
}
@ -219,7 +225,7 @@ public class StartPersistentTaskAction extends Action<StartPersistentTaskAction.
@Override
protected final void masterOperation(final Request request, ClusterState state,
final ActionListener<PersistentTaskResponse> listener) {
persistentTasksClusterService.createPersistentTask(request.taskId, request.action, request.params,
persistentTasksClusterService.createPersistentTask(request.taskId, request.taskName, request.params,
new ActionListener<PersistentTask<?>>() {
@Override

17
server/src/main/java/org/elasticsearch/persistent/UpdatePersistentTaskStatusAction.java
View File

@ -22,7 +22,6 @@ import org.elasticsearch.action.Action;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.ActionRequestValidationException;
import org.elasticsearch.action.support.ActionFilters;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.support.master.MasterNodeOperationRequestBuilder;
import org.elasticsearch.action.support.master.MasterNodeRequest;
import org.elasticsearch.action.support.master.TransportMasterNodeAction;
@ -38,13 +37,14 @@ import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.tasks.Task;
import org.elasticsearch.threadpool.ThreadPool;
import org.elasticsearch.transport.TransportResponse.Empty;
import org.elasticsearch.transport.TransportService;
import org.elasticsearch.persistent.PersistentTasksCustomMetaData.PersistentTask;
import java.io.IOException;
import java.util.Objects;
import static org.elasticsearch.action.ValidateActions.addValidationError;
public class UpdatePersistentTaskStatusAction extends Action<UpdatePersistentTaskStatusAction.Request,
PersistentTaskResponse,
UpdatePersistentTaskStatusAction.RequestBuilder> {
@ -69,7 +69,7 @@ public class UpdatePersistentTaskStatusAction extends Action<UpdatePersistentTas
public static class Request extends MasterNodeRequest<Request> {
private String taskId;
private long allocationId;
private long allocationId = -1L;
private Task.Status status;
public Request() {
@ -112,7 +112,16 @@ public class UpdatePersistentTaskStatusAction extends Action<UpdatePersistentTas
@Override
public ActionRequestValidationException validate() {
return null;
ActionRequestValidationException validationException = null;
if (this.taskId == null) {
validationException = addValidationError("task id must be specified", validationException);
}
if (this.allocationId == -1L) {
validationException = addValidationError("allocationId must be specified", validationException);
}
// We cannot really check if status has the same type as task because we don't have access
// to the task here. We will check it when we try to update the task
return validationException;
}
@Override

26
server/src/test/java/org/elasticsearch/persistent/PersistentTasksClusterServiceTests.java
View File

@ -36,6 +36,7 @@ import org.elasticsearch.test.VersionUtils;
import org.elasticsearch.persistent.PersistentTasksCustomMetaData.Assignment;
import org.elasticsearch.persistent.PersistentTasksCustomMetaData.PersistentTask;
import org.elasticsearch.persistent.TestPersistentTasksPlugin.TestParams;
import org.elasticsearch.persistent.TestPersistentTasksPlugin.TestPersistentTasksExecutor;
import java.util.ArrayList;
import java.util.Arrays;
@ -93,7 +94,7 @@ public class PersistentTasksClusterServiceTests extends ESTestCase {
addTestNodes(nodes, randomIntBetween(1, 10));
int numberOfTasks = randomIntBetween(2, 40);
for (int i = 0; i < numberOfTasks; i++) {
addTask(tasks, "should_assign", "assign_one", randomBoolean() ? null : "no_longer_exits");
addTask(tasks, "assign_one", randomBoolean() ? null : "no_longer_exits");
}
MetaData.Builder metaData = MetaData.builder(clusterState.metaData()).putCustom(PersistentTasksCustomMetaData.TYPE, tasks.build());
@ -117,14 +118,14 @@ public class PersistentTasksClusterServiceTests extends ESTestCase {
switch (randomInt(2)) {
case 0:
// add an unassigned task that should get assigned because it's assigned to a non-existing node or unassigned
addTask(tasks, "should_assign", "assign_me", randomBoolean() ? null : "no_longer_exits");
addTask(tasks, "assign_me", randomBoolean() ? null : "no_longer_exits");
break;
case 1:
// add a task assigned to non-existing node that should not get assigned
addTask(tasks, "should_not_assign", "dont_assign_me", randomBoolean() ? null : "no_longer_exits");
addTask(tasks, "dont_assign_me", randomBoolean() ? null : "no_longer_exits");
break;
case 2:
addTask(tasks, "assign_one", "assign_one", randomBoolean() ? null : "no_longer_exits");
addTask(tasks, "assign_one", randomBoolean() ? null : "no_longer_exits");
break;
}
@ -143,8 +144,8 @@ public class PersistentTasksClusterServiceTests extends ESTestCase {
for (PersistentTask<?> task : tasksInProgress.tasks()) {
// explanation should correspond to the action name
switch (task.getTaskName()) {
case "should_assign":
switch (((TestParams) task.getParams()).getTestParam()) {
case "assign_me":
assertThat(task.getExecutorNode(), notNullValue());
assertThat(task.isAssigned(), equalTo(true));
if (clusterState.nodes().nodeExists(task.getExecutorNode()) == false) {
@ -154,7 +155,7 @@ public class PersistentTasksClusterServiceTests extends ESTestCase {
clusterState.nodes().nodeExists(task.getExecutorNode()), equalTo(true));
assertThat(task.getAssignment().getExplanation(), equalTo("test assignment"));
break;
case "should_not_assign":
case "dont_assign_me":
assertThat(task.getExecutorNode(), nullValue());
assertThat(task.isAssigned(), equalTo(false));
assertThat(task.getAssignment().getExplanation(), equalTo("no appropriate nodes found for the assignment"));
@ -210,7 +211,9 @@ public class PersistentTasksClusterServiceTests extends ESTestCase {
private Assignment assignOnlyOneTaskAtATime(ClusterState clusterState) {
DiscoveryNodes nodes = clusterState.nodes();
PersistentTasksCustomMetaData tasksInProgress = clusterState.getMetaData().custom(PersistentTasksCustomMetaData.TYPE);
if (tasksInProgress.findTasks("assign_one", task -> nodes.nodeExists(task.getExecutorNode())).isEmpty()) {
if (tasksInProgress.findTasks(TestPersistentTasksExecutor.NAME, task ->
"assign_one".equals(((TestParams) task.getParams()).getTestParam()) &&
nodes.nodeExists(task.getExecutorNode())).isEmpty()) {
return randomNodeAssignment(clusterState.nodes());
} else {
return new Assignment(null, "only one task can be assigned at a time");
@ -404,11 +407,12 @@ public class PersistentTasksClusterServiceTests extends ESTestCase {
MetaData.Builder metaData, PersistentTasksCustomMetaData.Builder tasks,
Assignment assignment, String param) {
return clusterStateBuilder.metaData(metaData.putCustom(PersistentTasksCustomMetaData.TYPE,
tasks.addTask(UUIDs.base64UUID(), randomAlphaOfLength(10), new TestParams(param), assignment).build()));
tasks.addTask(UUIDs.base64UUID(), TestPersistentTasksExecutor.NAME, new TestParams(param), assignment).build()));
}
private void addTask(PersistentTasksCustomMetaData.Builder tasks, String action, String param, String node) {
tasks.addTask(UUIDs.base64UUID(), action, new TestParams(param), new Assignment(node, "explanation: " + action));
private void addTask(PersistentTasksCustomMetaData.Builder tasks, String param, String node) {
tasks.addTask(UUIDs.base64UUID(), TestPersistentTasksExecutor.NAME, new TestParams(param),
new Assignment(node, "explanation: " + param));
}
private DiscoveryNode newNode(String nodeId) {

4
server/src/test/java/org/elasticsearch/persistent/PersistentTasksCustomMetaDataTests.java
View File

@ -81,7 +81,7 @@ public class PersistentTasksCustomMetaDataTests extends AbstractDiffableSerializ
new Entry(MetaData.Custom.class, PersistentTasksCustomMetaData.TYPE, PersistentTasksCustomMetaData::new),
new Entry(NamedDiff.class, PersistentTasksCustomMetaData.TYPE, PersistentTasksCustomMetaData::readDiffFrom),
new Entry(PersistentTaskParams.class, TestPersistentTasksExecutor.NAME, TestParams::new),
new Entry(Task.Status.class, Status.NAME, Status::new)
new Entry(Task.Status.class, TestPersistentTasksExecutor.NAME, Status::new)
));
}
@ -164,7 +164,7 @@ public class PersistentTasksCustomMetaDataTests extends AbstractDiffableSerializ
return new NamedXContentRegistry(Arrays.asList(
new NamedXContentRegistry.Entry(PersistentTaskParams.class, new ParseField(TestPersistentTasksExecutor.NAME),
TestParams::fromXContent),
new NamedXContentRegistry.Entry(Task.Status.class, new ParseField(Status.NAME), Status::fromXContent)
new NamedXContentRegistry.Entry(Task.Status.class, new ParseField(TestPersistentTasksExecutor.NAME), Status::fromXContent)
));
}

4
server/src/test/java/org/elasticsearch/persistent/PersistentTasksExecutorIT.java
View File

@ -203,9 +203,9 @@ public class PersistentTasksExecutorIT extends ESIntegTestCase {
assertThrows(future1, IllegalStateException.class, "timed out after 10ms");
PlainActionFuture<PersistentTask<?>> failedUpdateFuture = new PlainActionFuture<>();
persistentTasksService.updateStatus(taskId, -1, new Status("should fail"), failedUpdateFuture);
persistentTasksService.updateStatus(taskId, -2, new Status("should fail"), failedUpdateFuture);
assertThrows(failedUpdateFuture, ResourceNotFoundException.class, "the task with id " + taskId +
" and allocation id -1 doesn't exist");
" and allocation id -2 doesn't exist");
// Wait for the task to disappear
WaitForPersistentTaskStatusFuture<?> future2 = new WaitForPersistentTaskStatusFuture<>();

7
server/src/test/java/org/elasticsearch/persistent/PersistentTasksExecutorResponseTests.java
View File

@ -22,6 +22,7 @@ import org.elasticsearch.common.UUIDs;
import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
import org.elasticsearch.test.AbstractStreamableTestCase;
import org.elasticsearch.persistent.PersistentTasksCustomMetaData.PersistentTask;
import org.elasticsearch.persistent.TestPersistentTasksPlugin.TestPersistentTasksExecutor;
import java.util.Collections;
@ -33,7 +34,7 @@ public class PersistentTasksExecutorResponseTests extends AbstractStreamableTest
protected PersistentTaskResponse createTestInstance() {
if (randomBoolean()) {
return new PersistentTaskResponse(
new PersistentTask<PersistentTaskParams>(UUIDs.base64UUID(), randomAsciiOfLength(10),
new PersistentTask<PersistentTaskParams>(UUIDs.base64UUID(), TestPersistentTasksExecutor.NAME,
new TestPersistentTasksPlugin.TestParams("test"),
randomLong(), PersistentTasksCustomMetaData.INITIAL_ASSIGNMENT));
} else {
@ -50,7 +51,7 @@ public class PersistentTasksExecutorResponseTests extends AbstractStreamableTest
protected NamedWriteableRegistry getNamedWriteableRegistry() {
return new NamedWriteableRegistry(Collections.singletonList(
new NamedWriteableRegistry.Entry(PersistentTaskParams.class,
TestPersistentTasksPlugin.TestPersistentTasksExecutor.NAME, TestPersistentTasksPlugin.TestParams::new)
TestPersistentTasksExecutor.NAME, TestPersistentTasksPlugin.TestParams::new)
));
}
}
}

13
server/src/test/java/org/elasticsearch/persistent/PersistentTasksNodeServiceTests.java
View File

@ -39,6 +39,7 @@ import org.elasticsearch.threadpool.ThreadPool;
import org.elasticsearch.persistent.PersistentTasksCustomMetaData.Assignment;
import org.elasticsearch.persistent.PersistentTasksCustomMetaData.PersistentTask;
import org.elasticsearch.persistent.TestPersistentTasksPlugin.TestParams;
import org.elasticsearch.persistent.TestPersistentTasksPlugin.TestPersistentTasksExecutor;
import java.io.IOException;
import java.util.ArrayList;
@ -80,7 +81,7 @@ public class PersistentTasksNodeServiceTests extends ESTestCase {
PersistentTasksService persistentTasksService = mock(PersistentTasksService.class);
@SuppressWarnings("unchecked") PersistentTasksExecutor<TestParams> action = mock(PersistentTasksExecutor.class);
when(action.getExecutor()).thenReturn(ThreadPool.Names.SAME);
when(action.getTaskName()).thenReturn("test");
when(action.getTaskName()).thenReturn(TestPersistentTasksExecutor.NAME);
int nonLocalNodesCount = randomInt(10);
// need to account for 5 original tasks on each node and their relocations
for (int i = 0; i < (nonLocalNodesCount + 1) * 10; i++) {
@ -101,11 +102,11 @@ public class PersistentTasksNodeServiceTests extends ESTestCase {
boolean added = false;
if (nonLocalNodesCount > 0) {
for (int i = 0; i < randomInt(5); i++) {
tasks.addTask(UUIDs.base64UUID(), "test_action", new TestParams("other_" + i),
tasks.addTask(UUIDs.base64UUID(), TestPersistentTasksExecutor.NAME, new TestParams("other_" + i),
new Assignment("other_node_" + randomInt(nonLocalNodesCount), "test assignment on other node"));
if (added == false && randomBoolean()) {
added = true;
tasks.addTask(UUIDs.base64UUID(), "test", new TestParams("this_param"),
tasks.addTask(UUIDs.base64UUID(), TestPersistentTasksExecutor.NAME, new TestParams("this_param"),
new Assignment("this_node", "test assignment on this node"));
}
}
@ -126,7 +127,7 @@ public class PersistentTasksNodeServiceTests extends ESTestCase {
// Add task on some other node
state = newClusterState;
newClusterState = addTask(state, "test", null, "some_other_node");
newClusterState = addTask(state, TestPersistentTasksExecutor.NAME, null, "some_other_node");
coordinator.clusterChanged(new ClusterChangedEvent("test", newClusterState, state));
// Make sure action wasn't called again
@ -134,7 +135,7 @@ public class PersistentTasksNodeServiceTests extends ESTestCase {
// Start another task on this node
state = newClusterState;
newClusterState = addTask(state, "test", new TestParams("this_param"), "this_node");
newClusterState = addTask(state, TestPersistentTasksExecutor.NAME, new TestParams("this_param"), "this_node");
coordinator.clusterChanged(new ClusterChangedEvent("test", newClusterState, state));
// Make sure action was called this time
@ -149,7 +150,7 @@ public class PersistentTasksNodeServiceTests extends ESTestCase {
// Add task on some other node
state = newClusterState;
newClusterState = addTask(state, "test", null, "some_other_node");
newClusterState = addTask(state, TestPersistentTasksExecutor.NAME, null, "some_other_node");
coordinator.clusterChanged(new ClusterChangedEvent("test", newClusterState, state));
// Make sure action wasn't called again

7
server/src/test/java/org/elasticsearch/persistent/TestPersistentTasksPlugin.java
View File

@ -124,7 +124,7 @@ public class TestPersistentTasksPlugin extends Plugin implements ActionPlugin {
PersistentTasksCustomMetaData::new),
new NamedWriteableRegistry.Entry(NamedDiff.class, PersistentTasksCustomMetaData.TYPE,
PersistentTasksCustomMetaData::readDiffFrom),
new NamedWriteableRegistry.Entry(Task.Status.class, Status.NAME, Status::new)
new NamedWriteableRegistry.Entry(Task.Status.class, TestPersistentTasksExecutor.NAME, Status::new)
);
}
@ -135,7 +135,7 @@ public class TestPersistentTasksPlugin extends Plugin implements ActionPlugin {
PersistentTasksCustomMetaData::fromXContent),
new NamedXContentRegistry.Entry(PersistentTaskParams.class, new ParseField(TestPersistentTasksExecutor.NAME),
TestParams::fromXContent),
new NamedXContentRegistry.Entry(Task.Status.class, new ParseField(Status.NAME), Status::fromXContent)
new NamedXContentRegistry.Entry(Task.Status.class, new ParseField(TestPersistentTasksExecutor.NAME), Status::fromXContent)
);
}
@ -225,7 +225,6 @@ public class TestPersistentTasksPlugin extends Plugin implements ActionPlugin {
}
public static class Status implements Task.Status {
public static final String NAME = "test";
private final String phase;
@ -246,7 +245,7 @@ public class TestPersistentTasksPlugin extends Plugin implements ActionPlugin {
@Override
public String getWriteableName() {
return NAME;
return TestPersistentTasksExecutor.NAME;
}
@Override

3
server/src/test/java/org/elasticsearch/persistent/UpdatePersistentTaskRequestTests.java
View File

@ -23,6 +23,7 @@ import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
import org.elasticsearch.tasks.Task;
import org.elasticsearch.test.AbstractStreamableTestCase;
import org.elasticsearch.persistent.TestPersistentTasksPlugin.Status;
import org.elasticsearch.persistent.TestPersistentTasksPlugin.TestPersistentTasksExecutor;
import org.elasticsearch.persistent.UpdatePersistentTaskStatusAction.Request;
import java.util.Collections;
@ -42,7 +43,7 @@ public class UpdatePersistentTaskRequestTests extends AbstractStreamableTestCase
@Override
protected NamedWriteableRegistry getNamedWriteableRegistry() {
return new NamedWriteableRegistry(Collections.singletonList(
new NamedWriteableRegistry.Entry(Task.Status.class, Status.NAME, Status::new)
new NamedWriteableRegistry.Entry(Task.Status.class, TestPersistentTasksExecutor.NAME, Status::new)
));
}
}