scheduling task for future removal

Author: masseyke

x-pack/plugin/migrate/src/main/java/org/elasticsearch/xpack/migrate/task/ReindexDataStreamPersistentTaskExecutor.java

@@ -108,8 +108,9 @@ protected void nodeOperation(
         ReindexDataStreamTaskParams params,
         PersistentTaskState persistentTaskState
     ) {
-        if (isComplete(persistentTaskState)) {
-            task.markAsCompleted();
+        Long completionTime = getCompletionTime(persistentTaskState);
+        if (completionTime != null && task instanceof ReindexDataStreamTask reindexDataStreamTask) {
+            reindexDataStreamTask.allReindexesCompleted(threadPool, getTimeToLive(completionTime));
             return;
         }
         ReindexDataStreamPersistentTaskState state = (ReindexDataStreamPersistentTaskState) persistentTaskState;
@@ -320,14 +321,14 @@ private void completeFailedPersistentTask(
         persistentTask.taskFailed(threadPool, updateCompletionTimeAndGetTimeToLive(persistentTask, state), e);
     }
 
-    private boolean isComplete(PersistentTaskState persistentTaskState) {
+    private Long getCompletionTime(PersistentTaskState persistentTaskState) {
         if (persistentTaskState == null) {
-            return false;
+            return null;
         } else {
             if (persistentTaskState instanceof ReindexDataStreamPersistentTaskState state) {
-                return state.isComplete();
+                return state.completionTime();
             } else {
-                return false;
+                return null;
             }
         }
     }
@@ -361,6 +362,10 @@ private TimeValue updateCompletionTimeAndGetTimeToLive(
                 completionTime = state.completionTime();
             }
         }
-        return TimeValue.timeValueMillis(TASK_KEEP_ALIVE_TIME.millis() - (threadPool.absoluteTimeInMillis() - completionTime));
+        return getTimeToLive(completionTime);
+    }
+
+    private TimeValue getTimeToLive(long completionTimeInMillis) {
+        return TimeValue.timeValueMillis(TASK_KEEP_ALIVE_TIME.millis() - (threadPool.absoluteTimeInMillis() - completionTimeInMillis));
     }
 }

x-pack/plugin/migrate/src/main/java/org/elasticsearch/xpack/migrate/task/ReindexDataStreamTask.java

@@ -120,7 +120,8 @@ public void incrementInProgressIndicesCount(String index) {
 
     private boolean isComplete() {
         PersistentTasksCustomMetadata persistentTasksCustomMetadata = clusterService.state()
-            .getMetadata().getProject()
+            .getMetadata()
+            .getProject()
             .custom(PersistentTasksCustomMetadata.TYPE);
         PersistentTasksCustomMetadata.PersistentTask<?> persistentTask = persistentTasksCustomMetadata.getTask(getPersistentTaskId());
         if (persistentTask != null) {