[Transform] Reduce task match load

prwhelan · prwhelan · commit 37bccfb1a12a · 2025-12-19T17:21:17.000-05:00
Move `TransportGetTransformStatsAction` to the generic threadpool, and match `Task` based on java type first before comparing the description to prevent iterating over a large abount of transform ids for tasks that are not `TransformTask`. Resolve elastic#139252
diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/action/GetTransformStatsAction.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/action/GetTransformStatsAction.java
@@ -82,8 +82,7 @@ public Request(StreamInput in) throws IOException {
         @Override
         public boolean match(Task task) {
             // Only get tasks that we have expanded to
-            return expandedIds.stream()
-                .anyMatch(transformId -> task.getDescription().equals(TransformField.PERSISTENT_TASK_DESCRIPTION_PREFIX + transformId));
+            return PutTransformAction.TransformTaskMatcher.match(task, expandedIds);
         }
 
         public String getId() {
diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/action/PutTransformAction.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/action/PutTransformAction.java
@@ -25,6 +25,7 @@
 import org.elasticsearch.xpack.core.transform.utils.TransformStrings;
 
 import java.io.IOException;
+import java.util.Collection;
 import java.util.Map;
 import java.util.Objects;
 
@@ -165,4 +166,17 @@ public Task createTask(long id, String type, String action, TaskId parentTaskId,
         }
     }
 
+    public interface TransformTaskMatcher {
+        static boolean match(Task task, String expectedTransformId) {
+            return task instanceof TransformTaskMatcher
+                && task.getDescription().equals(TransformField.PERSISTENT_TASK_DESCRIPTION_PREFIX + expectedTransformId);
+        }
+
+        static boolean match(Task task, Collection<String> expectedTransformIds) {
+            return task instanceof PutTransformAction.TransformTaskMatcher
+                && expectedTransformIds.stream()
+                    .anyMatch(transformId -> task.getDescription().equals(TransformField.PERSISTENT_TASK_DESCRIPTION_PREFIX + transformId));
+        }
+    }
+
 }
diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/action/ScheduleNowTransformAction.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/action/ScheduleNowTransformAction.java
@@ -98,11 +98,7 @@ public boolean equals(Object obj) {
 
         @Override
         public boolean match(Task task) {
-            if (task.getDescription().startsWith(TransformField.PERSISTENT_TASK_DESCRIPTION_PREFIX)) {
-                String taskId = task.getDescription().substring(TransformField.PERSISTENT_TASK_DESCRIPTION_PREFIX.length());
-                return taskId.equals(this.id);
-            }
-            return false;
+            return PutTransformAction.TransformTaskMatcher.match(task, id);
         }
     }
 
diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/action/StopTransformAction.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/action/StopTransformAction.java
@@ -170,14 +170,7 @@ public boolean equals(Object obj) {
 
         @Override
         public boolean match(Task task) {
-            if (task.getDescription().startsWith(TransformField.PERSISTENT_TASK_DESCRIPTION_PREFIX)) {
-                String taskId = task.getDescription().substring(TransformField.PERSISTENT_TASK_DESCRIPTION_PREFIX.length());
-                if (expandedIds != null) {
-                    return expandedIds.contains(taskId);
-                }
-            }
-
-            return false;
+            return PutTransformAction.TransformTaskMatcher.match(task, expandedIds);
         }
     }
 
diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/action/UpdateTransformAction.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/action/UpdateTransformAction.java
@@ -185,11 +185,7 @@ public boolean equals(Object obj) {
 
         @Override
         public boolean match(Task task) {
-            if (task.getDescription().startsWith(TransformField.PERSISTENT_TASK_DESCRIPTION_PREFIX)) {
-                String taskId = task.getDescription().substring(TransformField.PERSISTENT_TASK_DESCRIPTION_PREFIX.length());
-                return taskId.equals(this.id);
-            }
-            return false;
+            return PutTransformAction.TransformTaskMatcher.match(task, id);
         }
     }
 
diff --git a/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/transform/action/FakeTransformTask.java b/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/transform/action/FakeTransformTask.java
@@ -0,0 +1,32 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.core.transform.action;
+
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.tasks.TaskId;
+
+import java.util.Map;
+
+class FakeTransformTask extends Task implements PutTransformAction.TransformTaskMatcher {
+    FakeTransformTask(long id, String type, String action, String description, TaskId parentTask, Map<String, String> headers) {
+        super(id, type, action, description, parentTask, headers);
+    }
+
+    FakeTransformTask(
+        long id,
+        String type,
+        String action,
+        String description,
+        TaskId parentTask,
+        long startTime,
+        long startTimeNanos,
+        Map<String, String> headers
+    ) {
+        super(id, type, action, description, parentTask, startTime, startTimeNanos, headers);
+    }
+}
diff --git a/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/transform/action/GetTransformStatsActionRequestTests.java b/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/transform/action/GetTransformStatsActionRequestTests.java
@@ -47,4 +47,11 @@ public void testCreateTask() {
         assertThat(task, is(instanceOf(CancellableTask.class)));
         assertThat(task.getDescription(), is(equalTo("get_transform_stats[some-transform]")));
     }
+
+    public void testMatch() {
+        Request request = new Request("some-transform", null, false);
+        assertTrue(request.match(new FakeTransformTask(123, "", "", "data_frame_some-transform", null, null)));
+        assertFalse(request.match(new FakeTransformTask(123, "", "", "some-transform", null, null)));
+        assertFalse(request.match(new Task(123, "", "", "data_frame_some-transform", null, null)));
+    }
 }
diff --git a/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/transform/action/ScheduleNowTransformActionRequestTests.java b/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/transform/action/ScheduleNowTransformActionRequestTests.java
@@ -10,7 +10,7 @@
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.core.TimeValue;
-import org.elasticsearch.persistent.AllocatedPersistentTask;
+import org.elasticsearch.tasks.Task;
 import org.elasticsearch.test.AbstractWireSerializingTestCase;
 import org.elasticsearch.xpack.core.transform.action.ScheduleNowTransformAction.Request;
 
@@ -59,9 +59,10 @@ public void testValidationFailure() {
 
     public void testMatch() {
         Request request = new Request("my-transform-7", TimeValue.timeValueSeconds(5));
-        assertTrue(request.match(new AllocatedPersistentTask(123, "", "", "data_frame_my-transform-7", null, null)));
-        assertFalse(request.match(new AllocatedPersistentTask(123, "", "", "data_frame_my-transform-", null, null)));
-        assertFalse(request.match(new AllocatedPersistentTask(123, "", "", "data_frame_my-transform-77", null, null)));
-        assertFalse(request.match(new AllocatedPersistentTask(123, "", "", "my-transform-7", null, null)));
+        assertTrue(request.match(new FakeTransformTask(123, "", "", "data_frame_my-transform-7", null, null)));
+        assertFalse(request.match(new FakeTransformTask(123, "", "", "data_frame_my-transform-", null, null)));
+        assertFalse(request.match(new FakeTransformTask(123, "", "", "data_frame_my-transform-77", null, null)));
+        assertFalse(request.match(new FakeTransformTask(123, "", "", "my-transform-7", null, null)));
+        assertFalse(request.match(new Task(123, "", "", "data_frame_my-transform-7", null, null)));
     }
 }
diff --git a/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/transform/action/StopTransformActionRequestTests.java b/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/transform/action/StopTransformActionRequestTests.java
@@ -66,14 +66,15 @@ public void testSameButDifferentTimeout() {
     public void testMatch() {
         String transformId = "transform-id";
 
-        Task transformTask = new Task(
+        Task transformTask = new FakeTransformTask(
             1L,
             "persistent",
             "action",
             TransformField.PERSISTENT_TASK_DESCRIPTION_PREFIX + transformId,
             TaskId.EMPTY_TASK_ID,
             Collections.emptyMap()
-        );
+        ) {
+        };
 
         Request request = new Request("unrelated", false, false, null, false, false);
         request.setExpandedIds(Set.of("foo", "bar"));
diff --git a/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/transform/action/UpdateTransformActionRequestTests.java b/x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/transform/action/UpdateTransformActionRequestTests.java
@@ -9,7 +9,7 @@
 
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.core.TimeValue;
-import org.elasticsearch.persistent.AllocatedPersistentTask;
+import org.elasticsearch.tasks.Task;
 import org.elasticsearch.xpack.core.transform.action.UpdateTransformAction.Request;
 import org.elasticsearch.xpack.core.transform.transforms.AuthorizationStateTests;
 import org.elasticsearch.xpack.core.transform.transforms.TransformConfigTests;
@@ -73,9 +73,10 @@ protected Request mutateInstance(Request instance) {
 
     public void testMatch() {
         Request request = new Request(randomTransformConfigUpdate(), "my-transform-7", false, null);
-        assertTrue(request.match(new AllocatedPersistentTask(123, "", "", "data_frame_my-transform-7", null, null)));
-        assertFalse(request.match(new AllocatedPersistentTask(123, "", "", "data_frame_my-transform-", null, null)));
-        assertFalse(request.match(new AllocatedPersistentTask(123, "", "", "data_frame_my-transform-77", null, null)));
-        assertFalse(request.match(new AllocatedPersistentTask(123, "", "", "my-transform-7", null, null)));
+        assertTrue(request.match(new FakeTransformTask(123, "", "", "data_frame_my-transform-7", null, null)));
+        assertFalse(request.match(new FakeTransformTask(123, "", "", "data_frame_my-transform-", null, null)));
+        assertFalse(request.match(new FakeTransformTask(123, "", "", "data_frame_my-transform-77", null, null)));
+        assertFalse(request.match(new FakeTransformTask(123, "", "", "my-transform-7", null, null)));
+        assertFalse(request.match(new Task(123, "", "", "data_frame_my-transform-7", null, null)));
     }
 }
diff --git a/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransportGetTransformStatsAction.java b/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransportGetTransformStatsAction.java
@@ -24,7 +24,6 @@
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.index.IndexNotFoundException;
@@ -34,6 +33,7 @@
 import org.elasticsearch.tasks.CancellableTask;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.tasks.TaskId;
+import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.core.transform.action.GetTransformStatsAction;
 import org.elasticsearch.xpack.core.transform.action.GetTransformStatsAction.Request;
@@ -79,6 +79,7 @@ public TransportGetTransformStatsAction(
         ActionFilters actionFilters,
         ClusterService clusterService,
         TransformServices transformServices,
+        ThreadPool threadPool,
         Client client,
         Settings settings
     ) {
@@ -89,7 +90,7 @@ public TransportGetTransformStatsAction(
             actionFilters,
             Request::new,
             Response::new,
-            EsExecutors.DIRECT_EXECUTOR_SERVICE
+            threadPool.executor(ThreadPool.Names.GENERIC)
         );
         this.transformConfigManager = transformServices.configManager();
         this.transformCheckpointService = transformServices.checkpointService();
diff --git a/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/transforms/TransformTask.java b/x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/transforms/TransformTask.java
@@ -30,6 +30,7 @@
 import org.elasticsearch.xpack.core.indexing.IndexerState;
 import org.elasticsearch.xpack.core.transform.TransformField;
 import org.elasticsearch.xpack.core.transform.TransformMessages;
+import org.elasticsearch.xpack.core.transform.action.PutTransformAction;
 import org.elasticsearch.xpack.core.transform.action.StartTransformAction;
 import org.elasticsearch.xpack.core.transform.transforms.AuthorizationState;
 import org.elasticsearch.xpack.core.transform.transforms.SettingsConfig;
@@ -57,7 +58,11 @@
 import static org.elasticsearch.xpack.core.transform.TransformMessages.CANNOT_START_FAILED_TRANSFORM;
 import static org.elasticsearch.xpack.core.transform.TransformMessages.CANNOT_STOP_SINGLE_FAILED_TRANSFORM;
 
-public class TransformTask extends AllocatedPersistentTask implements TransformScheduler.Listener, TransformContext.Listener {
+public class TransformTask extends AllocatedPersistentTask
+    implements
+        TransformScheduler.Listener,
+        TransformContext.Listener,
+        PutTransformAction.TransformTaskMatcher {
 
     // Default interval the scheduler sends an event if the config does not specify a frequency
     private static final Logger logger = LogManager.getLogger(TransformTask.class);

Original file line number	Diff line number	Diff line change
`@@ -82,8 +82,7 @@ public Request(StreamInput in) throws IOException {`
`82`	`82`	`@Override`
`83`	`83`	`public boolean match(Task task) {`
`84`	`84`	`// Only get tasks that we have expanded to`
`85`		`- return expandedIds.stream()`
`86`		`- .anyMatch(transformId -> task.getDescription().equals(TransformField.PERSISTENT_TASK_DESCRIPTION_PREFIX + transformId));`
	`85`	`+ return PutTransformAction.TransformTaskMatcher.match(task, expandedIds);`
`87`	`86`	`}`
`88`	`87`
`89`	`88`	`public String getId() {`
Original file line number	Diff line number	Diff line change
`@@ -98,11 +98,7 @@ public boolean equals(Object obj) {`
`98`	`98`
`99`	`99`	`@Override`
`100`	`100`	`public boolean match(Task task) {`
`101`		`- if (task.getDescription().startsWith(TransformField.PERSISTENT_TASK_DESCRIPTION_PREFIX)) {`
`102`		`- String taskId = task.getDescription().substring(TransformField.PERSISTENT_TASK_DESCRIPTION_PREFIX.length());`
`103`		`- return taskId.equals(this.id);`
`104`		`- }`
`105`		`- return false;`
	`101`	`+ return PutTransformAction.TransformTaskMatcher.match(task, id);`
`106`	`102`	`}`
`107`	`103`	`}`
`108`	`104`
Original file line number	Diff line number	Diff line change
`@@ -170,14 +170,7 @@ public boolean equals(Object obj) {`
`170`	`170`
`171`	`171`	`@Override`
`172`	`172`	`public boolean match(Task task) {`
`173`		`- if (task.getDescription().startsWith(TransformField.PERSISTENT_TASK_DESCRIPTION_PREFIX)) {`
`174`		`- String taskId = task.getDescription().substring(TransformField.PERSISTENT_TASK_DESCRIPTION_PREFIX.length());`
`175`		`- if (expandedIds != null) {`
`176`		`- return expandedIds.contains(taskId);`
`177`		`- }`
`178`		`- }`
`179`		`-`
`180`		`- return false;`
	`173`	`+ return PutTransformAction.TransformTaskMatcher.match(task, expandedIds);`
`181`	`174`	`}`
`182`	`175`	`}`
`183`	`176`
Original file line number	Diff line number	Diff line change
`@@ -185,11 +185,7 @@ public boolean equals(Object obj) {`
`185`	`185`
`186`	`186`	`@Override`
`187`	`187`	`public boolean match(Task task) {`
`188`		`- if (task.getDescription().startsWith(TransformField.PERSISTENT_TASK_DESCRIPTION_PREFIX)) {`
`189`		`- String taskId = task.getDescription().substring(TransformField.PERSISTENT_TASK_DESCRIPTION_PREFIX.length());`
`190`		`- return taskId.equals(this.id);`
`191`		`- }`
`192`		`- return false;`
	`188`	`+ return PutTransformAction.TransformTaskMatcher.match(task, id);`
`193`	`189`	`}`
`194`	`190`	`}`
`195`	`191`
Original file line number	Diff line number	Diff line change
`@@ -47,4 +47,11 @@ public void testCreateTask() {`
`47`	`47`	`assertThat(task, is(instanceOf(CancellableTask.class)));`
`48`	`48`	`assertThat(task.getDescription(), is(equalTo("get_transform_stats[some-transform]")));`
`49`	`49`	`}`
	`50`	`+`
	`51`	`+ public void testMatch() {`
	`52`	`+ Request request = new Request("some-transform", null, false);`
	`53`	`+ assertTrue(request.match(new FakeTransformTask(123, "", "", "data_frame_some-transform", null, null)));`
	`54`	`+ assertFalse(request.match(new FakeTransformTask(123, "", "", "some-transform", null, null)));`
	`55`	`+ assertFalse(request.match(new Task(123, "", "", "data_frame_some-transform", null, null)));`
	`56`	`+ }`
`50`	`57`	`}`