[Transform] Ensure transform `_schedule_now` API only triggers the expected transform task (#102958)

docs/changelog/102958.yaml

@@ -0,0 +1,7 @@
+pr: 102958
+summary: Ensure transform `_schedule_now` API only triggers the expected transform
+  task
+area: Transform
+type: bug
+issues:
+ - 102956

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/transform/action/ScheduleNowTransformAction.java

@@ -18,6 +18,7 @@ import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.tasks.Task;
 import org.elasticsearch.xcontent.ToXContentObject;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xpack.core.transform.TransformField;
@@ -94,6 +95,15 @@ public class ScheduleNowTransformAction extends ActionType<ScheduleNowTransformA
             // the base class does not implement equals, therefore we need to check timeout ourselves
             return this.id.equals(other.id) && getTimeout().equals(other.getTimeout());
         }
+
+        @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;
+        }
     }
 
     public static class Response extends BaseTasksResponse implements Writeable, ToXContentObject {

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/transform/action/ScheduleNowTransformActionRequestTests.java

@@ -10,6 +10,7 @@ package org.elasticsearch.xpack.core.transform.action;
 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.test.AbstractWireSerializingTestCase;
 import org.elasticsearch.xpack.core.transform.action.ScheduleNowTransformAction.Request;
 
@@ -55,4 +56,12 @@ public class ScheduleNowTransformActionRequestTests extends AbstractWireSerializ
         assertThat(e, is(notNullValue()));
         assertThat(e.validationErrors(), contains("_schedule_now API does not support _all wildcard"));
     }
+
+    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)));
+    }
 }