Making reindex data streams actions cancellable (elastic#122438)

jdconrad · Feb 12, 2025 · 8c0ab4e · 8c0ab4e
1 parent 88550f6
commit 8c0ab4e
Show file tree

Hide file tree

Showing 3 changed files with 41 additions and 0 deletions.
diff --git a/...ate/src/main/java/org/elasticsearch/xpack/migrate/action/CreateIndexFromSourceAction.java b/...ate/src/main/java/org/elasticsearch/xpack/migrate/action/CreateIndexFromSourceAction.java
@@ -15,6 +15,9 @@
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.tasks.CancellableTask;
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.tasks.TaskId;
 import org.elasticsearch.xcontent.ObjectParser;
 import org.elasticsearch.xcontent.ParseField;
 import org.elasticsearch.xcontent.ToXContent;
@@ -191,5 +194,15 @@ public String[] indices() {
         public IndicesOptions indicesOptions() {
             return IndicesOptions.strictSingleIndexNoExpandForbidClosed();
         }
+
+        @Override
+        public Task createTask(long id, String type, String action, TaskId parentTaskId, Map<String, String> headers) {
+            return new CancellableTask(id, type, action, getDescription(), parentTaskId, headers);
+        }
+
+        @Override
+        public String getDescription() {
+            return "creating index " + destIndex + " from " + sourceIndex;
+        }
     }
 }
diff --git a/...migrate/src/main/java/org/elasticsearch/xpack/migrate/action/ReindexDataStreamAction.java b/...migrate/src/main/java/org/elasticsearch/xpack/migrate/action/ReindexDataStreamAction.java
@@ -16,6 +16,9 @@
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.features.NodeFeature;
+import org.elasticsearch.tasks.CancellableTask;
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.tasks.TaskId;
 import org.elasticsearch.xcontent.ConstructingObjectParser;
 import org.elasticsearch.xcontent.ParseField;
 import org.elasticsearch.xcontent.ToXContent;
@@ -24,6 +27,7 @@
 
 import java.io.IOException;
 import java.util.Locale;
+import java.util.Map;
 import java.util.Objects;
 import java.util.function.Predicate;
 
@@ -144,5 +148,15 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
             builder.endObject();
             return builder;
         }
+
+        @Override
+        public Task createTask(long id, String type, String action, TaskId parentTaskId, Map<String, String> headers) {
+            return new CancellableTask(id, type, action, getDescription(), parentTaskId, headers);
+        }
+
+        @Override
+        public String getDescription() {
+            return "reindexing data stream " + sourceDataStream;
+        }
     }
 }
diff --git a/...te/src/main/java/org/elasticsearch/xpack/migrate/action/ReindexDataStreamIndexAction.java b/...te/src/main/java/org/elasticsearch/xpack/migrate/action/ReindexDataStreamIndexAction.java
@@ -14,8 +14,12 @@
 import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.tasks.CancellableTask;
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.tasks.TaskId;
 
 import java.io.IOException;
+import java.util.Map;
 import java.util.Objects;
 
 public class ReindexDataStreamIndexAction extends ActionType<ReindexDataStreamIndexAction.Response> {
@@ -78,6 +82,16 @@ public String[] indices() {
         public IndicesOptions indicesOptions() {
             return IndicesOptions.strictSingleIndexNoExpandForbidClosed();
         }
+
+        @Override
+        public Task createTask(long id, String type, String action, TaskId parentTaskId, Map<String, String> headers) {
+            return new CancellableTask(id, type, action, getDescription(), parentTaskId, headers);
+        }
+
+        @Override
+        public String getDescription() {
+            return "reindexing data stream index " + sourceIndex;
+        }
     }
 
     public static class Response extends ActionResponse {