Skip to content

Commit

Permalink
Reindex naming changes (#49482)
Browse files Browse the repository at this point in the history 
Renamed types and action names to fit that we now call it a reindex task
and not a job. Removed action and named writeable todos.

Relates #42612
  • Loading branch information
@henningandersen
henningandersen committed Nov 27, 2019
1 parent f578d58 commit 375f3d8
Show file tree
Hide file tree
Showing 16 changed files with 94 additions and 94 deletions.
2 changes: 1 addition & 1 deletion client/rest-high-level/src/test/java/org/elasticsearch/client/TasksIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -119,7 +119,7 @@ public void testGetValidTask() throws Exception {
org.elasticsearch.tasks.TaskInfo info = taskResponse.getTaskInfo();
assertTrue(info.isCancellable());
assertEquals("reindex from [source1] to [dest]", info.getDescription());
assertEquals("reindex/job[c]", info.getAction());
assertEquals("reindex[c]", info.getAction());
if (taskResponse.isCompleted() == false) {
assertBusy(ReindexIT.checkCompletionStatus(client(), taskId.toString()));
}
Expand Down
15 changes: 7 additions & 8 deletions ...search/index/reindex/ReindexJobState.java → ...x/reindex/ReindexPersistentTaskState.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -31,13 +31,12 @@

import java.io.IOException;

public class ReindexJobState implements Task.Status, PersistentTaskState {
public class ReindexPersistentTaskState implements Task.Status, PersistentTaskState {

// TODO: Name
public static final String NAME = ReindexTask.NAME;

public static final ConstructingObjectParser<ReindexJobState, Void> PARSER =
new ConstructingObjectParser<>(NAME, a -> new ReindexJobState((String) a[0], (String) a[1]));
public static final ConstructingObjectParser<ReindexPersistentTaskState, Void> PARSER =
new ConstructingObjectParser<>(NAME, a -> new ReindexPersistentTaskState((String) a[0], (String) a[1]));

private static String EPHEMERAL_TASK_ID = "ephemeral_task_id";
private static String STATUS = "status";
Expand All @@ -50,17 +49,17 @@ public class ReindexJobState implements Task.Status, PersistentTaskState {
private final TaskId ephemeralTaskId;
private final Status status;

private ReindexJobState(String ephemeralTaskId, String status) {
private ReindexPersistentTaskState(String ephemeralTaskId, String status) {
this(new TaskId(ephemeralTaskId), Status.valueOf(status));
}

ReindexJobState(TaskId ephemeralTaskId, Status status) {
ReindexPersistentTaskState(TaskId ephemeralTaskId, Status status) {
assert status != null : "Status cannot be null";
this.ephemeralTaskId = ephemeralTaskId;
this.status = status;
}

public ReindexJobState(StreamInput in) throws IOException {
public ReindexPersistentTaskState(StreamInput in) throws IOException {
ephemeralTaskId = TaskId.readFromStream(in);
status = in.readEnum(Status.class);
}
Expand Down Expand Up @@ -96,7 +95,7 @@ public TaskId getEphemeralTaskId() {
return ephemeralTaskId;
}

public static ReindexJobState fromXContent(XContentParser parser) {
public static ReindexPersistentTaskState fromXContent(XContentParser parser) {
return PARSER.apply(parser, null);
}

Expand Down
18 changes: 10 additions & 8 deletions modules/reindex/src/main/java/org/elasticsearch/index/reindex/ReindexPlugin.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -72,26 +72,28 @@ public class ReindexPlugin extends Plugin implements ActionPlugin, PersistentTas
new ActionHandler<>(UpdateByQueryAction.INSTANCE, TransportUpdateByQueryAction.class),
new ActionHandler<>(DeleteByQueryAction.INSTANCE, TransportDeleteByQueryAction.class),
new ActionHandler<>(RethrottleAction.INSTANCE, TransportRethrottleAction.class),
new ActionHandler<>(StartReindexJobAction.INSTANCE, TransportStartReindexJobAction.class)
new ActionHandler<>(StartReindexTaskAction.INSTANCE, TransportStartReindexTaskAction.class)
);
}

@Override
public List<NamedWriteableRegistry.Entry> getNamedWriteables() {
return Arrays.asList(
new NamedWriteableRegistry.Entry(Task.Status.class, BulkByScrollTask.Status.NAME, BulkByScrollTask.Status::new),
new NamedWriteableRegistry.Entry(PersistentTaskParams.class, ReindexJob.NAME, ReindexJob::new),
new NamedWriteableRegistry.Entry(Task.Status.class, ReindexJobState.NAME, ReindexJobState::new),
new NamedWriteableRegistry.Entry(PersistentTaskState.class, ReindexJobState.NAME, ReindexJobState::new));
new NamedWriteableRegistry.Entry(PersistentTaskParams.class, ReindexTaskParams.NAME, ReindexTaskParams::new),
new NamedWriteableRegistry.Entry(Task.Status.class, ReindexPersistentTaskState.NAME, ReindexPersistentTaskState::new),
new NamedWriteableRegistry.Entry(PersistentTaskState.class, ReindexPersistentTaskState.NAME, ReindexPersistentTaskState::new));
}

@Override
public List<NamedXContentRegistry.Entry> getNamedXContent() {
return Arrays.asList(
new NamedXContentRegistry.Entry(PersistentTaskParams.class, new ParseField(ReindexJob.NAME), ReindexJob::fromXContent),
new NamedXContentRegistry.Entry(Task.Status.class, new ParseField(ReindexJobState.NAME), ReindexJobState::fromXContent),
new NamedXContentRegistry.Entry(PersistentTaskState.class, new ParseField(ReindexJobState.NAME),
ReindexJobState::fromXContent));
new NamedXContentRegistry.Entry(PersistentTaskParams.class, new ParseField(ReindexTaskParams.NAME),
ReindexTaskParams::fromXContent),
new NamedXContentRegistry.Entry(Task.Status.class, new ParseField(ReindexPersistentTaskState.NAME),
ReindexPersistentTaskState::fromXContent),
new NamedXContentRegistry.Entry(PersistentTaskState.class, new ParseField(ReindexPersistentTaskState.NAME),
ReindexPersistentTaskState::fromXContent));
}

@Override
Expand Down
36 changes: 18 additions & 18 deletions modules/reindex/src/main/java/org/elasticsearch/index/reindex/ReindexTask.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -47,8 +47,7 @@ public class ReindexTask extends AllocatedPersistentTask {

private static final Logger logger = LogManager.getLogger(ReindexTask.class);

// TODO: Name
public static final String NAME = "reindex/job";
public static final String NAME = "reindex";

private final NodeClient client;
private final ReindexIndexClient reindexIndexClient;
Expand All @@ -59,7 +58,7 @@ public class ReindexTask extends AllocatedPersistentTask {
private volatile String description;
private volatile boolean assignmentConflictDetected;

public static class ReindexPersistentTasksExecutor extends PersistentTasksExecutor<ReindexJob> {
public static class ReindexPersistentTasksExecutor extends PersistentTasksExecutor<ReindexTaskParams> {

private final ClusterService clusterService;
private final Client client;
Expand All @@ -80,14 +79,14 @@ public static class ReindexPersistentTasksExecutor extends PersistentTasksExecut
}

@Override
protected void nodeOperation(AllocatedPersistentTask task, ReindexJob reindexJob, PersistentTaskState state) {
protected void nodeOperation(AllocatedPersistentTask task, ReindexTaskParams reindexTaskParams, PersistentTaskState state) {
ReindexTask reindexTask = (ReindexTask) task;
reindexTask.execute(reindexJob);
reindexTask.execute(reindexTaskParams);
}

@Override
protected AllocatedPersistentTask createTask(long id, String type, String action, TaskId parentTaskId,
PersistentTasksCustomMetaData.PersistentTask<ReindexJob> taskInProgress,
PersistentTasksCustomMetaData.PersistentTask<ReindexTaskParams> taskInProgress,
Map<String, String> headers) {
headers.putAll(taskInProgress.getParams().getHeaders());
Reindexer reindexer = new Reindexer(clusterService, client, threadPool, scriptService, reindexSslConfig);
Expand Down Expand Up @@ -139,20 +138,20 @@ BulkByScrollTask getChildTask() {
return childTask;
}

private void execute(ReindexJob reindexJob) {
private void execute(ReindexTaskParams reindexTaskParams) {
long allocationId = getAllocationId();

ReindexTaskStateUpdater taskUpdater = new ReindexTaskStateUpdater(reindexIndexClient, client.threadPool(), getPersistentTaskId(),
allocationId, new ActionListener<>() {
@Override
public void onResponse(ReindexTaskStateDoc stateDoc) {
reindexDone(stateDoc, reindexJob.shouldStoreResult());
reindexDone(stateDoc, reindexTaskParams.shouldStoreResult());
}

@Override
public void onFailure(Exception e) {
logger.info("Reindex task failed", e);
updateClusterStateToFailed(reindexJob.shouldStoreResult(), ReindexJobState.Status.DONE, e);
updateClusterStateToFailed(reindexTaskParams.shouldStoreResult(), ReindexPersistentTaskState.Status.DONE, e);
}
}, this::handleCheckpointAssignmentConflict);

Expand All @@ -166,7 +165,7 @@ public void onResponse(ReindexTaskStateDoc stateDoc) {
public void onResponse(Void aVoid) {
// TODO: need to store status in state so we can continue from it.
transientStatus = childTask.getStatus();
performReindex(reindexJob, stateDoc, taskUpdater);
performReindex(reindexTaskParams, stateDoc, taskUpdater);
}

@Override
Expand All @@ -178,7 +177,7 @@ public void onFailure(Exception e) {

@Override
public void onFailure(Exception ex) {
updateClusterStateToFailed(reindexJob.shouldStoreResult(), ReindexJobState.Status.ASSIGNMENT_FAILED, ex);
updateClusterStateToFailed(reindexTaskParams.shouldStoreResult(), ReindexPersistentTaskState.Status.ASSIGNMENT_FAILED, ex);
}
});
}
Expand All @@ -196,7 +195,7 @@ protected void onCancelled() {

private void reindexDone(ReindexTaskStateDoc stateDoc, boolean shouldStoreResult) {
TaskManager taskManager = getTaskManager();
updatePersistentTaskState(new ReindexJobState(taskId, ReindexJobState.Status.DONE), new ActionListener<>() {
updatePersistentTaskState(new ReindexPersistentTaskState(taskId, ReindexPersistentTaskState.Status.DONE), new ActionListener<>() {
@Override
public void onResponse(PersistentTasksCustomMetaData.PersistentTask<?> persistentTask) {
if (shouldStoreResult) {
Expand Down Expand Up @@ -226,7 +225,8 @@ public void onFailure(Exception ex) {
}

private void sendStartedNotification(boolean shouldStoreResult) {
updatePersistentTaskState(new ReindexJobState(taskId, ReindexJobState.Status.STARTED), new ActionListener<>() {
updatePersistentTaskState(new ReindexPersistentTaskState(taskId, ReindexPersistentTaskState.Status.STARTED),
new ActionListener<>() {
@Override
public void onResponse(PersistentTasksCustomMetaData.PersistentTask<?> persistentTask) {
}
Expand All @@ -239,14 +239,14 @@ public void onFailure(Exception e) {
});
}

private void performReindex(ReindexJob reindexJob, ReindexTaskStateDoc stateDoc, ReindexTaskStateUpdater taskUpdater) {
private void performReindex(ReindexTaskParams reindexTaskParams, ReindexTaskStateDoc stateDoc, ReindexTaskStateUpdater taskUpdater) {
ReindexRequest reindexRequest = stateDoc.getReindexRequest();
ScrollableHitSource.Checkpoint initialCheckpoint = stateDoc.getCheckpoint();
ThreadContext threadContext = client.threadPool().getThreadContext();

Supplier<ThreadContext.StoredContext> context = threadContext.newRestorableContext(false);
// TODO: Eventually we only want to retain security context
try (ThreadContext.StoredContext ignore = stashWithHeaders(threadContext, reindexJob.getHeaders())) {
try (ThreadContext.StoredContext ignore = stashWithHeaders(threadContext, reindexTaskParams.getHeaders())) {
reindexer.execute(childTask, reindexRequest, new ContextPreservingActionListener<>(context, new ActionListener<>() {
@Override
public void onResponse(BulkByScrollResponse response) {
Expand All @@ -264,11 +264,11 @@ public void onFailure(Exception e) {
}, true);
}
// send this after we started reindex to ensure sub-tasks are created.
sendStartedNotification(reindexJob.shouldStoreResult());
sendStartedNotification(reindexTaskParams.shouldStoreResult());
}

private void updateClusterStateToFailed(boolean shouldStoreResult, ReindexJobState.Status status, Exception ex) {
updatePersistentTaskState(new ReindexJobState(taskId, status), new ActionListener<>() {
private void updateClusterStateToFailed(boolean shouldStoreResult, ReindexPersistentTaskState.Status status, Exception ex) {
updatePersistentTaskState(new ReindexPersistentTaskState(taskId, status), new ActionListener<>() {
@Override
public void onResponse(PersistentTasksCustomMetaData.PersistentTask<?> persistentTask) {
markEphemeralTaskFailed(shouldStoreResult, ex);
Expand Down
13 changes: 6 additions & 7 deletions ...asticsearch/index/reindex/ReindexJob.java → ...arch/index/reindex/ReindexTaskParams.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -31,14 +31,13 @@
import java.io.IOException;
import java.util.Map;

public class ReindexJob implements PersistentTaskParams {
public class ReindexTaskParams implements PersistentTaskParams {

// TODO: Name
public static final String NAME = ReindexTask.NAME;

@SuppressWarnings("unchecked")
public static final ConstructingObjectParser<ReindexJob, Void> PARSER
= new ConstructingObjectParser<>(NAME, a -> new ReindexJob((Boolean) a[0], (Map<String, String>) a[1]));
public static final ConstructingObjectParser<ReindexTaskParams, Void> PARSER
= new ConstructingObjectParser<>(NAME, a -> new ReindexTaskParams((Boolean) a[0], (Map<String, String>) a[1]));

private static String STORE_RESULT = "store_result";
private static String HEADERS = "headers";
Expand All @@ -51,12 +50,12 @@ public class ReindexJob implements PersistentTaskParams {
private final boolean storeResult;
private final Map<String, String> headers;

public ReindexJob(boolean storeResult, Map<String, String> headers) {
public ReindexTaskParams(boolean storeResult, Map<String, String> headers) {
this.storeResult = storeResult;
this.headers = headers;
}

public ReindexJob(StreamInput in) throws IOException {
public ReindexTaskParams(StreamInput in) throws IOException {
storeResult = in.readBoolean();
headers = in.readMap(StreamInput::readString, StreamInput::readString);
}
Expand Down Expand Up @@ -94,7 +93,7 @@ public Map<String, String> getHeaders() {
return headers;
}

public static ReindexJob fromXContent(XContentParser parser) {
public static ReindexTaskParams fromXContent(XContentParser parser) {
return PARSER.apply(parser, null);
}
}
10 changes: 5 additions & 5 deletions modules/reindex/src/main/java/org/elasticsearch/index/reindex/RestReindexAction.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,7 @@ public RestChannelConsumer prepareRequest(RestRequest request, NodeClient client
boolean waitForCompletion = request.paramAsBoolean("wait_for_completion", true);

// Build the internal request
StartReindexJobAction.Request internal = new StartReindexJobAction.Request(setCommonOptions(request, buildRequest(request)),
StartReindexTaskAction.Request internal = new StartReindexTaskAction.Request(setCommonOptions(request, buildRequest(request)),
waitForCompletion);
/*
* Let's try and validate before forking so the user gets some error. The
Expand All @@ -88,12 +88,12 @@ public RestChannelConsumer prepareRequest(RestRequest request, NodeClient client
params.put(BulkByScrollTask.Status.INCLUDE_CREATED, Boolean.toString(true));
params.put(BulkByScrollTask.Status.INCLUDE_UPDATED, Boolean.toString(true));

return channel -> client.execute(StartReindexJobAction.INSTANCE, internal, new ActionListener<>() {
return channel -> client.execute(StartReindexTaskAction.INSTANCE, internal, new ActionListener<>() {

private BulkIndexByScrollResponseContentListener listener = new BulkIndexByScrollResponseContentListener(channel, params);

@Override
public void onResponse(StartReindexJobAction.Response response) {
public void onResponse(StartReindexTaskAction.Response response) {
listener.onResponse(response.getReindexResponse());
}

Expand All @@ -103,9 +103,9 @@ public void onFailure(Exception e) {
}
});
} else {
return channel -> client.execute(StartReindexJobAction.INSTANCE, internal, new RestBuilderListener<>(channel) {
return channel -> client.execute(StartReindexTaskAction.INSTANCE, internal, new RestBuilderListener<>(channel) {
@Override
public RestResponse buildResponse(StartReindexJobAction.Response response, XContentBuilder builder) throws Exception {
public RestResponse buildResponse(StartReindexTaskAction.Response response, XContentBuilder builder) throws Exception {
builder.startObject();
// This is the ephemeral task-id from the first node that is assigned the task (for BWC).
builder.field("task", response.getTaskId());
Expand Down
9 changes: 4 additions & 5 deletions .../index/reindex/StartReindexJobAction.java → ...index/reindex/StartReindexTaskAction.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -35,13 +35,12 @@

import java.io.IOException;

public class StartReindexJobAction extends ActionType<StartReindexJobAction.Response> {
public class StartReindexTaskAction extends ActionType<StartReindexTaskAction.Response> {

public static final StartReindexJobAction INSTANCE = new StartReindexJobAction();
// TODO: Name
public static final String NAME = "indices:data/write/start_reindex";
public static final StartReindexTaskAction INSTANCE = new StartReindexTaskAction();
public static final String NAME = "indices:data/write/reindex/start";

private StartReindexJobAction() {
private StartReindexTaskAction() {
super(NAME, Response::new);
}

Expand Down
Loading

0 comments on commit 375f3d8

Please sign in to comment.