Skip to content

Store reindexing result in reindex index #45260

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

Merged
merged 13 commits into from
Aug 14, 2019
Merged

Store reindexing result in reindex index #45260

Show file tree
Hide file tree
Changes from 7 commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
1f3eb16
WIP
Tim-Brooks Aug 5, 2019
a839779
Merge remote-tracking branch 'upstream/reindex_v2' into store_respons…
Tim-Brooks Aug 6, 2019
dc03c6d
WIP
Tim-Brooks Aug 6, 2019
d0501de
Tests
Tim-Brooks Aug 6, 2019
65f7965
Changes
Tim-Brooks Aug 6, 2019
7ff1847
Add license
Tim-Brooks Aug 6, 2019
6fac306
Changes
Tim-Brooks Aug 7, 2019
d4c46d5
Merge branch 'reindex_v2' into store_response_in_index
Tim-Brooks Aug 13, 2019
730e205
Merge branch 'reindex_v2' into store_response_in_index
Tim-Brooks Aug 13, 2019
e4b96de
Preserve status
Tim-Brooks Aug 13, 2019
3315b93
Fix timeout
Tim-Brooks Aug 14, 2019
2f8d3f0
Merge branch 'reindex_v2' into store_response_in_index
Tim-Brooks Aug 14, 2019
08cf6dc
Changes
Tim-Brooks Aug 14, 2019
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
61 changes: 22 additions & 39 deletions modules/reindex/src/main/java/org/elasticsearch/index/reindex/ReindexJobState.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,8 +19,6 @@

package org.elasticsearch.index.reindex;

import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.common.Nullable;
import org.elasticsearch.common.ParseField;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.io.stream.StreamOutput;
Expand All @@ -39,41 +37,31 @@ public class ReindexJobState implements Task.Status, PersistentTaskState {
public static final String NAME = ReindexTask.NAME;

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

private static String EPHEMERAL_TASK_ID = "ephemeral_task_id";
private static String REINDEX_RESPONSE = "reindex_response";
private static String REINDEX_EXCEPTION = "reindex_exception";
private static String STATUS = "status";

static {
PARSER.declareString(ConstructingObjectParser.constructorArg(), new ParseField(EPHEMERAL_TASK_ID));
PARSER.declareObject(ConstructingObjectParser.optionalConstructorArg(), (p, c) -> BulkByScrollResponse.fromXContent(p),
new ParseField(REINDEX_RESPONSE));
PARSER.declareObject(ConstructingObjectParser.optionalConstructorArg(), (p, c) -> ElasticsearchException.fromXContent(p),
new ParseField(REINDEX_EXCEPTION));
PARSER.declareString(ConstructingObjectParser.constructorArg(), new ParseField(STATUS));
}

private final TaskId ephemeralTaskId;
private final BulkByScrollResponse reindexResponse;
private final ElasticsearchException jobException;
private final Status status;

private ReindexJobState(String ephemeralTaskId, BulkByScrollResponse reindexResponse, ElasticsearchException jobException) {
this(new TaskId(ephemeralTaskId), reindexResponse, jobException);
private ReindexJobState(String ephemeralTaskId, String status) {
this(new TaskId(ephemeralTaskId), Status.valueOf(status));
}

ReindexJobState(TaskId ephemeralTaskId, @Nullable BulkByScrollResponse reindexResponse,
@Nullable ElasticsearchException jobException) {
ReindexJobState(TaskId ephemeralTaskId, Status status) {
this.ephemeralTaskId = ephemeralTaskId;
assert (reindexResponse == null) || (jobException == null) : "Either response or exception must be null";
this.reindexResponse = reindexResponse;
this.jobException = jobException;
this.status = status;
Copy link
Contributor

Choose a reason for hiding this comment

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

Should we assert that status is != null? Mostly to ensure that the isDone() method cannot return true on null.

}

public ReindexJobState(StreamInput in) throws IOException {
ephemeralTaskId = TaskId.readFromStream(in);
reindexResponse = in.readOptionalWriteable(BulkByScrollResponse::new);
jobException = in.readException();
status = in.readEnum(Status.class);
}

@Override
Expand All @@ -84,35 +72,23 @@ public String getWriteableName() {
@Override
public void writeTo(StreamOutput out) throws IOException {
ephemeralTaskId.writeTo(out);
out.writeOptionalWriteable(reindexResponse);
out.writeException(jobException);
out.writeEnum(status);
}

@Override
public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
builder.startObject();
builder.field(EPHEMERAL_TASK_ID, ephemeralTaskId.toString());
if (reindexResponse != null) {
builder.field(REINDEX_RESPONSE);
builder.startObject();
reindexResponse.toXContent(builder, params);
builder.endObject();
}
if (jobException != null) {
builder.field(REINDEX_EXCEPTION);
builder.startObject();
jobException.toXContent(builder, params);
builder.endObject();
}
builder.field(STATUS, status);
return builder.endObject();
}

public BulkByScrollResponse getReindexResponse() {
return reindexResponse;
public boolean isDone() {
return status != Status.STARTED;
}

public ElasticsearchException getJobException() {
return jobException;
public Status getStatus() {
return status;
}

public TaskId getEphemeralTaskId() {
Expand All @@ -122,4 +98,11 @@ public TaskId getEphemeralTaskId() {
public static ReindexJobState fromXContent(XContentParser parser) {
return PARSER.apply(parser, null);
}

public enum Status {
STARTED,
FAILED_TO_READ_FROM_REINDEX_INDEX,
FAILED_TO_WRITE_TO_REINDEX_INDEX,
DONE
}
}
Loading