[ML] refactoring start task a bit, removing unused code (elastic#40798)

benwtrent · Apr 4, 2019 · 1a95dc8 · 1a95dc8
1 parent 1af2b2b
commit 1a95dc8
Show file tree

Hide file tree

Showing 7 changed files with 67 additions and 703 deletions.
diff --git a/...java/org/elasticsearch/xpack/core/dataframe/action/StartDataFrameTransformTaskAction.java b/...java/org/elasticsearch/xpack/core/dataframe/action/StartDataFrameTransformTaskAction.java
@@ -15,6 +15,7 @@
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.xcontent.ToXContentObject;
 import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.tasks.Task;
 import org.elasticsearch.xpack.core.dataframe.DataFrameField;
 import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper;
 
@@ -62,6 +63,11 @@ public void writeTo(StreamOutput out) throws IOException {
             out.writeString(id);
         }
 
+        @Override
+        public boolean match(Task task) {
+            return task.getDescription().equals(DataFrameField.PERSISTENT_TASK_DESCRIPTION_PREFIX + id);
+        }
+
         @Override
         public ActionRequestValidationException validate() {
             return null;

diff --git a/.../org/elasticsearch/xpack/dataframe/action/TransportStartDataFrameTransformTaskAction.java b/.../org/elasticsearch/xpack/dataframe/action/TransportStartDataFrameTransformTaskAction.java
@@ -24,7 +24,6 @@
 import org.elasticsearch.xpack.dataframe.transforms.DataFrameTransformTask;
 
 import java.util.List;
-import java.util.function.Consumer;
 
 /**
  * Internal only transport class to change an allocated persistent task's state to started
@@ -44,27 +43,6 @@ public TransportStartDataFrameTransformTaskAction(TransportService transportServ
         this.licenseState = licenseState;
     }
 
-    @Override
-    protected void processTasks(StartDataFrameTransformTaskAction.Request request, Consumer<DataFrameTransformTask> operation) {
-        DataFrameTransformTask matchingTask = null;
-
-        // todo: re-factor, see rollup TransportTaskHelper
-        for (Task task : taskManager.getTasks().values()) {
-            if (task instanceof DataFrameTransformTask
-                && ((DataFrameTransformTask) task).getTransformId().equals(request.getId())) {
-                if (matchingTask != null) {
-                    throw new IllegalArgumentException("Found more than one matching task for data frame transform [" + request.getId()
-                        + "] when " + "there should only be one.");
-                }
-                matchingTask = (DataFrameTransformTask) task;
-            }
-        }
-
-        if (matchingTask != null) {
-            operation.accept(matchingTask);
-        }
-    }
-
     @Override
     protected void doExecute(Task task, StartDataFrameTransformTaskAction.Request request,
                              ActionListener<StartDataFrameTransformTaskAction.Response> listener) {

diff --git a/...main/java/org/elasticsearch/xpack/dataframe/persistence/DataFramePersistentTaskUtils.java b/...main/java/org/elasticsearch/xpack/dataframe/persistence/DataFramePersistentTaskUtils.java
diff --git a/.../data-frame/src/main/java/org/elasticsearch/xpack/dataframe/util/BatchedDataIterator.java b/.../data-frame/src/main/java/org/elasticsearch/xpack/dataframe/util/BatchedDataIterator.java