[ML] refactoring start task a bit, removing unused code

x-pack/plugin/core/src/main/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;

x-pack/plugin/data-frame/src/main/java/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) {