[Transform] Fix Missing Config Errors (#134963) (#135389)

When there are more Transforms than the current page size, GET
_transform will mistakenly display an error message `Found
task for transform [...], but no configuration for it. To
delete this transform use DELETE with force=true.`.

When the page size is smaller than the reported total count of
transforms, we will now get all Transform Ids and compare that to the
running transforms to generate the dangling tasks error message.

Resolve #134263

Author: prwhelan

docs/changelog/134963.yaml

@@ -0,0 +1,6 @@
+pr: 134963
+summary: Fix a bug in the GET _transform API that incorrectly claims some Transform configurations are missing
+area: Transform
+type: bug
+issues:
+ - 134263

server/src/main/resources/transport/definitions/referable/transform_check_for_dangling_tasks.csv

@@ -0,0 +1 @@
+9170000,9112009,9000018,8841070,8840011

server/src/main/resources/transport/upper_bounds/8.18.csv

@@ -1 +1 @@
-initial_elasticsearch_8_18_8,8840010
+transform_check_for_dangling_tasks,8840011

server/src/main/resources/transport/upper_bounds/8.19.csv

@@ -1 +1 @@
-initial_elasticsearch_8_19_5,8841069
+transform_check_for_dangling_tasks,8841070

server/src/main/resources/transport/upper_bounds/9.0.csv

@@ -1 +1 @@
-initial_elasticsearch_9_0_8,9000017
+transform_check_for_dangling_tasks,9000018

server/src/main/resources/transport/upper_bounds/9.1.csv

@@ -1 +1 @@
-initial_elasticsearch_9_1_5,9112008
+transform_check_for_dangling_tasks,9112009

server/src/main/resources/transport/upper_bounds/9.2.csv

@@ -1 +1 @@
-esql_sample_operator_status,9127000
+transform_check_for_dangling_tasks,9170000

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

@@ -59,6 +59,7 @@ public final class TransformField {
     public static final ParseField MAX_AGE = new ParseField("max_age");
 
     public static final ParseField ALLOW_NO_MATCH = new ParseField("allow_no_match");
+    public static final ParseField CHECK_FOR_DANGLING_TASKS = new ParseField("check_dangling_tasks");
     /**
      * Fields for checkpointing
      */

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

@@ -7,6 +7,7 @@
 
 package org.elasticsearch.xpack.core.transform.action;
 
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.TransportVersions;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.ActionType;
@@ -16,6 +17,7 @@
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.logging.DeprecationCategory;
 import org.elasticsearch.common.logging.DeprecationLogger;
+import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.xcontent.ParseField;
 import org.elasticsearch.xcontent.ToXContentObject;
 import org.elasticsearch.xcontent.XContentBuilder;
@@ -39,6 +41,8 @@ public class GetTransformAction extends ActionType<GetTransformAction.Response>
     public static final GetTransformAction INSTANCE = new GetTransformAction();
     public static final String NAME = "cluster:monitor/transform/get";
 
+    static final TransportVersion DANGLING_TASKS = TransportVersion.fromName("transform_check_for_dangling_tasks");
+
     private static final DeprecationLogger deprecationLogger = DeprecationLogger.getLogger(GetTransformAction.class);
 
     private GetTransformAction() {
@@ -47,24 +51,49 @@ private GetTransformAction() {
 
     public static class Request extends AbstractGetResourcesRequest {
 
+        // for legacy purposes, this transport action previously had no timeout
+        private static final TimeValue LEGACY_TIMEOUT_VALUE = TimeValue.MAX_VALUE;
         private static final int MAX_SIZE_RETURN = 1000;
+        private final boolean checkForDanglingTasks;
+        private final TimeValue timeout;
 
         public Request(String id) {
-            super(id, PageParams.defaultParams(), true);
+            this(id, false, LEGACY_TIMEOUT_VALUE);
         }
 
-        public Request() {
-            super(null, PageParams.defaultParams(), true);
+        public Request(String id, boolean checkForDanglingTasks, TimeValue timeout) {
+            super(id, PageParams.defaultParams(), true);
+            this.checkForDanglingTasks = checkForDanglingTasks;
+            this.timeout = timeout;
         }
 
         public Request(StreamInput in) throws IOException {
             super(in);
+            this.checkForDanglingTasks = in.getTransportVersion().onOrAfter(DANGLING_TASKS) ? in.readBoolean() : true;
+            this.timeout = in.getTransportVersion().onOrAfter(DANGLING_TASKS) ? in.readTimeValue() : LEGACY_TIMEOUT_VALUE;
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            super.writeTo(out);
+            if (out.getTransportVersion().onOrAfter(DANGLING_TASKS)) {
+                out.writeBoolean(checkForDanglingTasks);
+                out.writeTimeValue(timeout);
+            }
         }
 
         public String getId() {
             return getResourceId();
         }
 
+        public boolean checkForDanglingTasks() {
+            return checkForDanglingTasks;
+        }
+
+        public TimeValue timeout() {
+            return timeout;
+        }
+
         @Override
         public ActionRequestValidationException validate() {
             ActionRequestValidationException exception = null;
@@ -86,6 +115,20 @@ public String getCancelableTaskDescription() {
         public String getResourceIdField() {
             return TransformField.ID.getPreferredName();
         }
+
+        @Override
+        public boolean equals(Object obj) {
+            return this == obj
+                || (obj instanceof Request other
+                    && super.equals(obj)
+                    && (checkForDanglingTasks == other.checkForDanglingTasks)
+                    && Objects.equals(timeout, other.timeout));
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(super.hashCode(), checkForDanglingTasks, timeout);
+        }
     }
 
     public static class Response extends AbstractGetResourcesResponse<TransformConfig> implements ToXContentObject {

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

@@ -7,28 +7,37 @@
 
 package org.elasticsearch.xpack.core.transform.action;
 
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.common.io.stream.Writeable;
-import org.elasticsearch.test.AbstractWireSerializingTestCase;
+import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.xpack.core.ml.AbstractBWCWireSerializationTestCase;
 import org.elasticsearch.xpack.core.transform.action.GetTransformAction.Request;
 
-public class GetTransformActionRequestTests extends AbstractWireSerializingTestCase<Request> {
+import static org.elasticsearch.xpack.core.transform.action.GetTransformAction.DANGLING_TASKS;
+
+public class GetTransformActionRequestTests extends AbstractBWCWireSerializationTestCase<Request> {
 
     @Override
     protected Request createTestInstance() {
         if (randomBoolean()) {
             return new Request(Metadata.ALL);
         }
-        return new Request(randomAlphaOfLengthBetween(1, 20));
+        return new Request(randomAlphaOfLengthBetween(1, 20), randomBoolean(), randomPositiveTimeValue());
     }
 
     @Override
     protected Request mutateInstance(Request instance) {
-        return null;// TODO implement https://github.com/elastic/elasticsearch/issues/25929
+        return randomValueOtherThan(instance, this::createTestInstance);
     }
 
     @Override
     protected Writeable.Reader<Request> instanceReader() {
         return Request::new;
     }
+
+    @Override
+    protected Request mutateInstanceForVersion(Request instance, TransportVersion version) {
+        return version.onOrAfter(DANGLING_TASKS) ? instance : new Request(instance.getId(), true, TimeValue.MAX_VALUE);
+    }
 }