Add initial_delay parameter to the transform _start API

docs/changelog/150358.yaml

@@ -0,0 +1,6 @@
+area: Transform
+issues:
+ - 150357
+pr: 150358
+summary: Add an `initial_delay` parameter to the transform `_start` API that applies a reduced sync delay until the transform first processes data, then reverts to the steady-state `sync.time.delay`
+type: enhancement

rest-api-spec/src/main/resources/rest-api-spec/api/transform.start_transform.json

@@ -32,6 +32,10 @@
         "type": "string",
         "description": "Restricts the set of transformed entities to those changed after this time"
       },
+      "initial_delay": {
+        "type": "time",
+        "description": "One-time reduced sync delay applied until the transform has processed its first document, then it reverts to the steady-state sync.time.delay"
+      },
       "timeout": {
         "type": "time",
         "default": "30s",

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

@@ -0,0 +1 @@
+9494000

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

@@ -1 +1 @@
-project_routing_usage_stats,9493000
+transform_start_initial_delay,9494000

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

@@ -52,6 +52,7 @@ public final class TransformField {
     public static final ParseField SYNC = new ParseField("sync");
     public static final ParseField TIME = new ParseField("time");
     public static final ParseField DELAY = new ParseField("delay");
+    public static final ParseField INITIAL_DELAY = new ParseField("initial_delay");
     public static final ParseField DEFER = new ParseField("defer");
     // TODO: Rename to "defer_data_validation" or similar to emphasize that not all validation is deferred
     public static final ParseField DEFER_VALIDATION = new ParseField("defer_validation");

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

@@ -7,6 +7,8 @@
 
 package org.elasticsearch.xpack.core.transform.action;
 
+import org.elasticsearch.ElasticsearchStatusException;
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.support.master.AcknowledgedRequest;
 import org.elasticsearch.action.support.tasks.BaseTasksResponse;
@@ -16,13 +18,15 @@
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.Releasable;
 import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.tasks.CancellableTask;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.tasks.TaskId;
 import org.elasticsearch.xcontent.ToXContentObject;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.xpack.core.security.cloud.CloudCredential;
 import org.elasticsearch.xpack.core.transform.TransformField;
+import org.elasticsearch.xpack.core.transform.transforms.TransformTaskParams;
 import org.elasticsearch.xpack.core.transform.utils.ExceptionsHelper;
 
 import java.io.IOException;
@@ -44,18 +48,26 @@ private StartTransformAction() {
 
     public static class Request extends AcknowledgedRequest<Request> implements Releasable {
 
+        private static final TransportVersion TRANSFORM_START_INITIAL_DELAY = TransportVersion.fromName("transform_start_initial_delay");
+
         private final String id;
         private final Instant from;
+        private final TimeValue initialDelay;
 
         // Caller's UIAM cloud credential carried on the request so it survives coordinator -> master
         // transport, where the AUTHENTICATING_CLOUD_TOKEN_THREAD_CONTEXT transient is no longer present.
         @Nullable
         private CloudCredential cloudCredential;
 
         public Request(String id, Instant from, TimeValue timeout) {
+            this(id, from, null, timeout);
+        }
+
+        public Request(String id, Instant from, TimeValue initialDelay, TimeValue timeout) {
             super(TRAPPY_IMPLICIT_DEFAULT_MASTER_NODE_TIMEOUT, timeout);
             this.id = ExceptionsHelper.requireNonNull(id, TransformField.ID.getPreferredName());
             this.from = from;
+            this.initialDelay = initialDelay;
         }
 
         public Request(StreamInput in) throws IOException {
@@ -67,6 +79,11 @@ public Request(StreamInput in) throws IOException {
             } else {
                 cloudCredential = null;
             }
+            if (in.getTransportVersion().supports(TRANSFORM_START_INITIAL_DELAY)) {
+                initialDelay = in.readOptionalTimeValue();
+            } else {
+                initialDelay = null;
+            }
         }
 
         public String getId() {
@@ -94,6 +111,10 @@ public CloudCredential setCloudCredential(@Nullable CloudCredential cloudCredent
             return previous;
         }
 
+        public TimeValue getInitialDelay() {
+            return initialDelay;
+        }
+
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             super.writeTo(out);
@@ -102,6 +123,16 @@ public void writeTo(StreamOutput out) throws IOException {
             if (out.getTransportVersion().supports(TRANSFORM_CLOUD_CREDENTIAL_ON_REQUEST)) {
                 out.writeOptionalWriteable(cloudCredential);
             }
+            if (out.getTransportVersion().supports(TRANSFORM_START_INITIAL_DELAY)) {
+                out.writeOptionalTimeValue(initialDelay);
+            } else if (initialDelay != null) {
+                throw new ElasticsearchStatusException(
+                    "Cannot send a _start request with "
+                        + TransformTaskParams.INITIAL_DELAY.getPreferredName()
+                        + " to an outdated node. Please upgrade the node to 9.6.0+ and try again.",
+                    RestStatus.BAD_REQUEST
+                );
+            }
         }
 
         @Override
@@ -113,7 +144,7 @@ public void close() {
         public int hashCode() {
             // the base class does not implement hashCode, therefore we need to hash timeout ourselves
             // cloudCredential is intentionally excluded: request-scoped secret carrier, not logical identity.
-            return Objects.hash(ackTimeout(), id, from);
+            return Objects.hash(ackTimeout(), id, from, initialDelay);
         }
 
         @Override
@@ -127,7 +158,10 @@ public boolean equals(Object obj) {
             Request other = (Request) obj;
             // the base class does not implement equals, therefore we need to check timeout ourselves
             // cloudCredential is intentionally excluded: request-scoped secret carrier, not logical identity.
-            return Objects.equals(id, other.id) && Objects.equals(from, other.from) && ackTimeout().equals(other.ackTimeout());
+            return Objects.equals(id, other.id)
+                && Objects.equals(from, other.from)
+                && Objects.equals(initialDelay, other.initialDelay)
+                && ackTimeout().equals(other.ackTimeout());
         }
 
         @Override

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

@@ -29,18 +29,22 @@ public class TransformTaskParams implements SimpleDiffable<TransformTaskParams>,
     public static final String NAME = TransformField.TASK_NAME;
     public static final ParseField FROM = TransformField.FROM;
     public static final ParseField FREQUENCY = TransformField.FREQUENCY;
+    public static final ParseField INITIAL_DELAY = TransformField.INITIAL_DELAY;
     public static final ParseField REQUIRES_REMOTE = new ParseField("requires_remote");
 
+    private static final TransportVersion TRANSFORM_START_INITIAL_DELAY = TransportVersion.fromName("transform_start_initial_delay");
+
     private final String transformId;
     private final TransformConfigVersion version;
     private final Instant from;
     private final TimeValue frequency;
     private final Boolean requiresRemote;
+    private final TimeValue initialDelay;
 
     public static final ConstructingObjectParser<TransformTaskParams, Void> PARSER = new ConstructingObjectParser<>(
         NAME,
         true,
-        a -> new TransformTaskParams((String) a[0], (String) a[1], (Long) a[2], (String) a[3], (Boolean) a[4])
+        a -> new TransformTaskParams((String) a[0], (String) a[1], (Long) a[2], (String) a[3], (Boolean) a[4], (String) a[5])
     );
 
     static {
@@ -49,28 +53,42 @@ public class TransformTaskParams implements SimpleDiffable<TransformTaskParams>,
         PARSER.declareLong(ConstructingObjectParser.optionalConstructorArg(), FROM);
         PARSER.declareString(ConstructingObjectParser.optionalConstructorArg(), FREQUENCY);
         PARSER.declareBoolean(ConstructingObjectParser.optionalConstructorArg(), REQUIRES_REMOTE);
+        PARSER.declareString(ConstructingObjectParser.optionalConstructorArg(), INITIAL_DELAY);
     }
 
-    private TransformTaskParams(String transformId, String version, Long from, String frequency, Boolean remote) {
+    private TransformTaskParams(String transformId, String version, Long from, String frequency, Boolean remote, String initialDelay) {
         this(
             transformId,
             version == null ? null : TransformConfigVersion.fromString(version),
             from == null ? null : Instant.ofEpochMilli(from),
             frequency == null ? null : TimeValue.parseTimeValue(frequency, FREQUENCY.getPreferredName()),
-            remote == null ? false : remote.booleanValue()
+            remote == null ? false : remote.booleanValue(),
+            initialDelay == null ? null : TimeValue.parseTimeValue(initialDelay, INITIAL_DELAY.getPreferredName())
         );
     }
 
     public TransformTaskParams(String transformId, TransformConfigVersion version, TimeValue frequency, boolean remote) {
-        this(transformId, version, null, frequency, remote);
+        this(transformId, version, null, frequency, remote, null);
     }
 
     public TransformTaskParams(String transformId, TransformConfigVersion version, Instant from, TimeValue frequency, boolean remote) {
+        this(transformId, version, from, frequency, remote, null);
+    }
+
+    public TransformTaskParams(
+        String transformId,
+        TransformConfigVersion version,
+        Instant from,
+        TimeValue frequency,
+        boolean remote,
+        TimeValue initialDelay
+    ) {
         this.transformId = transformId;
         this.version = version == null ? TransformConfigVersion.V_7_2_0 : version;
         this.from = from;
         this.frequency = frequency;
         this.requiresRemote = remote;
+        this.initialDelay = initialDelay;
     }
 
     public TransformTaskParams(StreamInput in) throws IOException {
@@ -79,6 +97,11 @@ public TransformTaskParams(StreamInput in) throws IOException {
         this.from = in.readOptionalInstant();
         this.frequency = in.readOptionalTimeValue();
         this.requiresRemote = in.readBoolean();
+        if (in.getTransportVersion().supports(TRANSFORM_START_INITIAL_DELAY)) {
+            this.initialDelay = in.readOptionalTimeValue();
+        } else {
+            this.initialDelay = null;
+        }
     }
 
     @Override
@@ -98,6 +121,9 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeOptionalInstant(from);
         out.writeOptionalTimeValue(frequency);
         out.writeBoolean(requiresRemote);
+        if (out.getTransportVersion().supports(TRANSFORM_START_INITIAL_DELAY)) {
+            out.writeOptionalTimeValue(initialDelay);
+        }
     }
 
     @Override
@@ -112,6 +138,9 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
             builder.field(FREQUENCY.getPreferredName(), frequency.getStringRep());
         }
         builder.field(REQUIRES_REMOTE.getPreferredName(), requiresRemote);
+        if (initialDelay != null) {
+            builder.field(INITIAL_DELAY.getPreferredName(), initialDelay.getStringRep());
+        }
         builder.endObject();
         return builder;
     }
@@ -136,6 +165,14 @@ public boolean requiresRemote() {
         return requiresRemote;
     }
 
+    /**
+     * @return the one-time reduced sync delay supplied at {@code _start}, applied only until the transform has processed
+     *         its first document, or {@code null} to always use the steady-state {@code sync.time.delay}.
+     */
+    public TimeValue getInitialDelay() {
+        return initialDelay;
+    }
+
     public static TransformTaskParams fromXContent(XContentParser parser) throws IOException {
         return PARSER.parse(parser, null);
     }
@@ -156,11 +193,12 @@ public boolean equals(Object other) {
             && Objects.equals(this.version, that.version)
             && Objects.equals(this.from, that.from)
             && Objects.equals(this.frequency, that.frequency)
-            && this.requiresRemote == that.requiresRemote;
+            && this.requiresRemote == that.requiresRemote
+            && Objects.equals(this.initialDelay, that.initialDelay);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(transformId, version, from, frequency, requiresRemote);
+        return Objects.hash(transformId, version, from, frequency, requiresRemote, initialDelay);
     }
 }

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

@@ -14,10 +14,12 @@
 import org.elasticsearch.test.TransportVersionUtils;
 import org.elasticsearch.xpack.core.security.cloud.CloudCredential;
 import org.elasticsearch.xpack.core.transform.action.StartTransformAction.Request;
+import org.elasticsearch.xpack.core.transform.transforms.TransformTaskParams;
 
 import java.io.IOException;
 import java.time.Duration;
 import java.time.Instant;
+import java.util.Collection;
 
 import static java.time.Instant.ofEpochMilli;
 import static org.elasticsearch.xpack.core.transform.transforms.TransformConfig.TRANSFORM_CLOUD_CREDENTIAL_ON_REQUEST;
@@ -26,11 +28,15 @@
 import static org.hamcrest.Matchers.nullValue;
 
 public class StartTransformActionRequestTests extends AbstractWireSerializingTransformTestCase<Request> {
+
+    private static final TransportVersion TRANSFORM_START_INITIAL_DELAY = TransportVersion.fromName("transform_start_initial_delay");
+
     @Override
     protected Request createTestInstance() {
         Request request = new Request(
             randomAlphaOfLengthBetween(1, 20),
             randomBoolean() ? ofEpochMilli(randomNonNegativeLong()) : null,
+            randomBoolean() ? randomTimeValue() : null,
             randomTimeValue()
         );
         // Randomly include a cloud credential so the wire path with the optional field is exercised
@@ -48,29 +54,51 @@ protected Writeable.Reader<Request> instanceReader() {
     protected Request mutateInstance(Request instance) {
         String id = instance.getId();
         Instant from = instance.from();
+        TimeValue initialDelay = instance.getInitialDelay();
         TimeValue timeout = instance.ackTimeout();
 
-        switch (between(0, 2)) {
+        switch (between(0, 3)) {
             case 0 -> id += randomAlphaOfLengthBetween(1, 5);
             case 1 -> from = from != null ? from.plus(Duration.ofDays(1)) : Instant.ofEpochMilli(randomNonNegativeLong());
             case 2 -> timeout = new TimeValue(timeout.duration() + randomLongBetween(1, 5), timeout.timeUnit());
+            case 3 -> initialDelay = initialDelay != null
+                ? new TimeValue(initialDelay.duration() + randomLongBetween(1, 5), initialDelay.timeUnit())
+                : randomTimeValue();
             default -> throw new AssertionError("Illegal randomization branch");
         }
 
-        Request mutated = new Request(id, from, timeout);
+        Request mutated = new Request(id, from, initialDelay, timeout);
         mutated.setCloudCredential(instance.getCloudCredential());
         return mutated;
     }
 
     @Override
     protected Request mutateInstanceForVersion(Request instance, TransportVersion version) {
         // cloudCredential is excluded from Request.equals so it passes through unchanged here; the explicit
-        // drop semantics are asserted by testCloudCredentialDroppedWhenWireVersionTooOld.
-        Request mutated = new Request(instance.getId(), instance.from(), instance.ackTimeout());
+        // drop semantics are asserted by testCloudCredentialDroppedWhenWireVersionTooOld. initialDelay is part
+        // of equals, so it must round-trip as null for versions that predate the initial_delay parameter.
+        TimeValue initialDelay = version.supports(TRANSFORM_START_INITIAL_DELAY) ? instance.getInitialDelay() : null;
+        Request mutated = new Request(instance.getId(), instance.from(), initialDelay, instance.ackTimeout());
         mutated.setCloudCredential(instance.getCloudCredential());
         return mutated;
     }
 
+    @Override
+    protected Collection<TransportVersion> bwcVersions() {
+        // Requests carrying initial_delay are rejected instead of silently dropping the value on older nodes.
+        return super.bwcVersions().stream().filter(version -> version.supports(TRANSFORM_START_INITIAL_DELAY)).toList();
+    }
+
+    public void testInitialDelayCannotSerializeToOlderNode() throws IOException {
+        testSerializationIsNotBackwardsCompatible(
+            TRANSFORM_START_INITIAL_DELAY,
+            request -> request.getInitialDelay() != null,
+            "Cannot send a _start request with "
+                + TransformTaskParams.INITIAL_DELAY.getPreferredName()
+                + " to an outdated node. Please upgrade the node to 9.6.0+ and try again."
+        );
+    }
+
     public void testCloudCredentialRoundTripPreservesValue() throws IOException {
         String secret = randomAlphaOfLengthBetween(8, 32);
         Request original = new Request(randomAlphaOfLengthBetween(1, 20), null, randomTimeValue());