Skip to content
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

Removing unnecessary state from DataStreamReindexTask #117942

Merged
merged 1 commit into from
Dec 4, 2024
+24 −20
Merged

Removing unnecessary state from DataStreamReindexTask #117942

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
9 changes: 5 additions & 4 deletions ...in/java/org/elasticsearch/xpack/migrate/task/ReindexDataStreamPersistentTaskExecutor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,6 @@ protected ReindexDataStreamTask createTask(
params.startTime(),
params.totalIndices(),
params.totalIndicesToBeUpgraded(),
threadPool,
id,
type,
action,
Expand All @@ -74,9 +73,11 @@ protected void nodeOperation(AllocatedPersistentTask task, ReindexDataStreamTask
List<Index> indicesToBeReindexed = indices.stream()
.filter(index -> clusterService.state().getMetadata().index(index).getCreationVersion().isLegacyIndexVersion())
.toList();
reindexDataStreamTask.setPendingIndices(indicesToBeReindexed.stream().map(Index::getName).toList());
reindexDataStreamTask.setPendingIndicesCount(indicesToBeReindexed.size());
for (Index index : indicesToBeReindexed) {
reindexDataStreamTask.incrementInProgressIndicesCount();
// TODO This is just a placeholder. This is where the real data stream reindex logic will go
reindexDataStreamTask.reindexSucceeded();
}

completeSuccessfulPersistentTask(reindexDataStreamTask);
Expand All @@ -87,12 +88,12 @@ protected void nodeOperation(AllocatedPersistentTask task, ReindexDataStreamTask
}

private void completeSuccessfulPersistentTask(ReindexDataStreamTask persistentTask) {
persistentTask.reindexSucceeded();
persistentTask.allReindexesCompleted();
threadPool.schedule(persistentTask::markAsCompleted, getTimeToLive(persistentTask), threadPool.generic());
}

private void completeFailedPersistentTask(ReindexDataStreamTask persistentTask, Exception e) {
persistentTask.reindexFailed(e);
persistentTask.taskFailed(e);
threadPool.schedule(() -> persistentTask.markAsFailed(e), getTimeToLive(persistentTask), threadPool.generic());
}

Expand Down
35 changes: 19 additions & 16 deletions ...gin/migrate/src/main/java/org/elasticsearch/xpack/migrate/task/ReindexDataStreamTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,29 +10,27 @@
import org.elasticsearch.core.Tuple;
import org.elasticsearch.persistent.AllocatedPersistentTask;
import org.elasticsearch.tasks.TaskId;
import org.elasticsearch.threadpool.ThreadPool;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

public class ReindexDataStreamTask extends AllocatedPersistentTask {
public static final String TASK_NAME = "reindex-data-stream";
private final long persistentTaskStartTime;
private final int totalIndices;
private final int totalIndicesToBeUpgraded;
private final ThreadPool threadPool;
private boolean complete = false;
private Exception exception;
private List<String> inProgress = new ArrayList<>();
private List<String> pending = List.of();
private AtomicInteger inProgress = new AtomicInteger(0);
private AtomicInteger pending = new AtomicInteger();
private List<Tuple<String, Exception>> errors = new ArrayList<>();

public ReindexDataStreamTask(
long persistentTaskStartTime,
int totalIndices,
int totalIndicesToBeUpgraded,
ThreadPool threadPool,
long id,
String type,
String action,
Expand All @@ -44,7 +42,6 @@ public ReindexDataStreamTask(
this.persistentTaskStartTime = persistentTaskStartTime;
this.totalIndices = totalIndices;
this.totalIndicesToBeUpgraded = totalIndicesToBeUpgraded;
this.threadPool = threadPool;
}

@Override
Expand All @@ -55,30 +52,36 @@ public ReindexDataStreamStatus getStatus() {
totalIndicesToBeUpgraded,
complete,
exception,
inProgress.size(),
pending.size(),
inProgress.get(),
pending.get(),
errors
);
}

public void reindexSucceeded() {
public void allReindexesCompleted() {
this.complete = true;
}

public void reindexFailed(Exception e) {
public void taskFailed(Exception e) {
this.complete = true;
this.exception = e;
}

public void setInProgressIndices(List<String> inProgressIndices) {
this.inProgress = inProgressIndices;
public void reindexSucceeded() {
inProgress.decrementAndGet();
Copy link
Member

Choose a reason for hiding this comment

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

We should add asserts (in the future) that these decrementing actions never go negative

}

public void reindexFailed(String index, Exception error) {
this.errors.add(Tuple.tuple(index, error));
inProgress.decrementAndGet();
}

public void setPendingIndices(List<String> pendingIndices) {
this.pending = pendingIndices;
public void incrementInProgressIndicesCount() {
inProgress.incrementAndGet();
pending.decrementAndGet();
}

public void addErrorIndex(String index, Exception error) {
this.errors.add(Tuple.tuple(index, error));
public void setPendingIndicesCount(int size) {
pending.set(size);
}
}