[ML] Add internal flag to flush api to control whether or not to flush indices (#96803)

When datafeeds send flush requests they don't require the indices to be refreshed (and in fact may be detrimentally expensive in the case of small bucket sizes).

This change adds an (internal) flag to control whether or not a refresh is required when flushing.

Author: edsavage

server/src/main/java/org/elasticsearch/TransportVersion.java

@@ -135,9 +135,10 @@ private static TransportVersion registerTransportVersion(int id, String uniqueId
     public static final TransportVersion V_8_500_009 = registerTransportVersion(8_500_009, "35091358-fd41-4106-a6e2-d2a1315494c1");
     public static final TransportVersion V_8_500_010 = registerTransportVersion(8_500_010, "9818C628-1EEC-439B-B943-468F61460675");
     public static final TransportVersion V_8_500_011 = registerTransportVersion(8_500_011, "2209F28D-B52E-4BC4-9889-E780F291C32E");
+    public static final TransportVersion V_8_500_012 = registerTransportVersion(8_500_012, "BB6F4AF1-A860-4FD4-A138-8150FFBE0ABD");
 
     private static class CurrentHolder {
-        private static final TransportVersion CURRENT = findCurrent(V_8_500_011);
+        private static final TransportVersion CURRENT = findCurrent(V_8_500_012);
 
         // finds the pluggable current version, or uses the given fallback
         private static TransportVersion findCurrent(TransportVersion fallback) {

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/action/FlushJobAction.java

@@ -6,6 +6,7 @@
  */
 package org.elasticsearch.xpack.core.ml.action;
 
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.support.tasks.BaseTasksResponse;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -62,6 +63,7 @@ public static Request parseRequest(String jobId, XContentParser parser) {
 
         private boolean calcInterim = false;
         private boolean waitForNormalization = true;
+        private boolean refreshRequired = true;
         private String start;
         private String end;
         private String advanceTime;
@@ -77,6 +79,9 @@ public Request(StreamInput in) throws IOException {
             advanceTime = in.readOptionalString();
             skipTime = in.readOptionalString();
             waitForNormalization = in.readBoolean();
+            if (in.getTransportVersion().onOrAfter(TransportVersion.V_8_500_012)) {
+                refreshRequired = in.readBoolean();
+            }
         }
 
         @Override
@@ -88,6 +93,9 @@ public void writeTo(StreamOutput out) throws IOException {
             out.writeOptionalString(advanceTime);
             out.writeOptionalString(skipTime);
             out.writeBoolean(waitForNormalization);
+            if (out.getTransportVersion().onOrAfter(TransportVersion.V_8_500_012)) {
+                out.writeBoolean(refreshRequired);
+            }
         }
 
         public Request(String jobId) {
@@ -138,18 +146,32 @@ public boolean isWaitForNormalization() {
             return waitForNormalization;
         }
 
+        public boolean isRefreshRequired() {
+            return refreshRequired;
+        }
+
         /**
-         * Used internally. Datafeeds do not need to wait renormalization to complete before continuing.
+         * Used internally. Datafeeds do not need to wait for renormalization to complete before continuing.
          *
          * For large jobs, renormalization can take minutes, causing datafeeds to needlessly pause execution.
          */
         public void setWaitForNormalization(boolean waitForNormalization) {
             this.waitForNormalization = waitForNormalization;
         }
 
+        /**
+         * Used internally. For datafeeds, there is no need for the results to be searchable after the flush,
+         * as the datafeed itself does not search them immediately.
+         *
+         * Particularly for short bucket spans these refreshes could be a significant cost.
+         **/
+        public void setRefreshRequired(boolean refreshRequired) {
+            this.refreshRequired = refreshRequired;
+        }
+
         @Override
         public int hashCode() {
-            return Objects.hash(jobId, calcInterim, start, end, advanceTime, skipTime, waitForNormalization);
+            return Objects.hash(jobId, calcInterim, start, end, advanceTime, skipTime, waitForNormalization, refreshRequired);
         }
 
         @Override
@@ -164,6 +186,7 @@ public boolean equals(Object obj) {
             return Objects.equals(jobId, other.jobId)
                 && calcInterim == other.calcInterim
                 && waitForNormalization == other.waitForNormalization
+                && refreshRequired == other.refreshRequired
                 && Objects.equals(start, other.start)
                 && Objects.equals(end, other.end)
                 && Objects.equals(advanceTime, other.advanceTime)

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/job/process/autodetect/output/FlushAcknowledgement.java

@@ -6,6 +6,7 @@
  */
 package org.elasticsearch.xpack.core.ml.job.process.autodetect.output;
 
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
@@ -28,44 +29,57 @@ public class FlushAcknowledgement implements ToXContentObject, Writeable {
     public static final ParseField TYPE = new ParseField("flush");
     public static final ParseField ID = new ParseField("id");
     public static final ParseField LAST_FINALIZED_BUCKET_END = new ParseField("last_finalized_bucket_end");
+    public static final ParseField REFRESH_REQUIRED = new ParseField("refresh_required");
 
     public static final ConstructingObjectParser<FlushAcknowledgement, Void> PARSER = new ConstructingObjectParser<>(
         TYPE.getPreferredName(),
-        a -> new FlushAcknowledgement((String) a[0], (Long) a[1])
+        a -> new FlushAcknowledgement((String) a[0], (Long) a[1], (Boolean) a[2])
     );
 
     static {
         PARSER.declareString(ConstructingObjectParser.constructorArg(), ID);
         PARSER.declareLong(ConstructingObjectParser.optionalConstructorArg(), LAST_FINALIZED_BUCKET_END);
+        PARSER.declareBoolean(ConstructingObjectParser.optionalConstructorArg(), REFRESH_REQUIRED);
     }
 
     private final String id;
     private final Instant lastFinalizedBucketEnd;
+    private final boolean refreshRequired;
 
-    public FlushAcknowledgement(String id, Long lastFinalizedBucketEndMs) {
+    public FlushAcknowledgement(String id, Long lastFinalizedBucketEndMs, Boolean refreshRequired) {
         this.id = id;
         // The C++ passes 0 when last finalized bucket end is not available, so treat 0 as null
         this.lastFinalizedBucketEnd = (lastFinalizedBucketEndMs != null && lastFinalizedBucketEndMs > 0)
             ? Instant.ofEpochMilli(lastFinalizedBucketEndMs)
             : null;
+        this.refreshRequired = refreshRequired == null || refreshRequired;
     }
 
-    public FlushAcknowledgement(String id, Instant lastFinalizedBucketEnd) {
+    public FlushAcknowledgement(String id, Instant lastFinalizedBucketEnd, Boolean refreshRequired) {
         this.id = id;
         // Round to millisecond accuracy to ensure round-tripping via XContent results in an equal object
         long epochMillis = (lastFinalizedBucketEnd != null) ? lastFinalizedBucketEnd.toEpochMilli() : 0;
         this.lastFinalizedBucketEnd = (epochMillis > 0) ? Instant.ofEpochMilli(epochMillis) : null;
+        this.refreshRequired = refreshRequired == null || refreshRequired;
     }
 
     public FlushAcknowledgement(StreamInput in) throws IOException {
         id = in.readString();
         lastFinalizedBucketEnd = in.readOptionalInstant();
+        if (in.getTransportVersion().onOrAfter(TransportVersion.V_8_500_012)) {
+            refreshRequired = in.readBoolean();
+        } else {
+            refreshRequired = true;
+        }
     }
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         out.writeString(id);
         out.writeOptionalInstant(lastFinalizedBucketEnd);
+        if (out.getTransportVersion().onOrAfter(TransportVersion.V_8_500_012)) {
+            out.writeBoolean(refreshRequired);
+        }
     }
 
     public String getId() {
@@ -76,6 +90,10 @@ public Instant getLastFinalizedBucketEnd() {
         return lastFinalizedBucketEnd;
     }
 
+    public boolean getRefreshRequired() {
+        return refreshRequired;
+    }
+
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         builder.startObject();
@@ -87,13 +105,14 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
                 lastFinalizedBucketEnd.toEpochMilli()
             );
         }
+        builder.field(REFRESH_REQUIRED.getPreferredName(), refreshRequired);
         builder.endObject();
         return builder;
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(id, lastFinalizedBucketEnd);
+        return Objects.hash(id, lastFinalizedBucketEnd, refreshRequired);
     }
 
     @Override
@@ -105,6 +124,8 @@ public boolean equals(Object obj) {
             return false;
         }
         FlushAcknowledgement other = (FlushAcknowledgement) obj;
-        return Objects.equals(id, other.id) && Objects.equals(lastFinalizedBucketEnd, other.lastFinalizedBucketEnd);
+        return Objects.equals(id, other.id)
+            && Objects.equals(lastFinalizedBucketEnd, other.lastFinalizedBucketEnd)
+            && refreshRequired == other.refreshRequired;
     }
 }

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ml/action/FlushJobActionRequestTests.java

@@ -19,6 +19,9 @@ protected Request createTestInstance() {
         if (randomBoolean()) {
             request.setWaitForNormalization(randomBoolean());
         }
+        if (randomBoolean()) {
+            request.setRefreshRequired(randomBoolean());
+        }
         if (randomBoolean()) {
             request.setCalcInterim(randomBoolean());
         }
@@ -49,6 +52,9 @@ protected Writeable.Reader<Request> instanceReader() {
 
     @Override
     protected Request mutateInstanceForVersion(Request instance, TransportVersion version) {
+        if (version.before(TransportVersion.V_8_500_012)) {
+            instance.setRefreshRequired(true);
+        }
         return instance;
     }
 }

x-pack/plugin/ml/src/internalClusterTest/java/org/elasticsearch/xpack/ml/integration/AutodetectResultProcessorIT.java

@@ -575,7 +575,7 @@ private static Quantiles createQuantiles() {
     }
 
     private static FlushAcknowledgement createFlushAcknowledgement() {
-        return new FlushAcknowledgement(randomAlphaOfLength(5), randomInstant());
+        return new FlushAcknowledgement(randomAlphaOfLength(5), randomInstant(), true);
     }
 
     private static class ResultsBuilder {

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportFlushJobAction.java

@@ -51,6 +51,7 @@ protected void taskOperation(
         FlushJobParams.Builder paramsBuilder = FlushJobParams.builder();
         paramsBuilder.calcInterim(request.getCalcInterim());
         paramsBuilder.waitForNormalization(request.isWaitForNormalization());
+        paramsBuilder.refreshRequired(request.isRefreshRequired());
         if (request.getAdvanceTime() != null) {
             paramsBuilder.advanceTime(request.getAdvanceTime());
         }

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/datafeed/DatafeedJob.java

@@ -175,6 +175,7 @@ Long runLookBack(long startTime, Long endTime) throws Exception {
 
         FlushJobAction.Request request = new FlushJobAction.Request(jobId);
         request.setCalcInterim(true);
+        request.setRefreshRequired(false);
         run(lookbackStartTimeMs, lookbackEnd, request);
         if (shouldPersistAfterLookback(isLookbackOnly)) {
             sendPersistRequest();
@@ -205,6 +206,7 @@ private long skipToStartTime(long startTime) {
             // start time is after last checkpoint, thus we need to skip time
             FlushJobAction.Request request = new FlushJobAction.Request(jobId);
             request.setSkipTime(String.valueOf(startTime));
+            request.setRefreshRequired(false);
             FlushJobAction.Response flushResponse = flushJob(request);
             LOGGER.info("[{}] Skipped to time [{}]", jobId, flushResponse.getLastFinalizedBucketEnd().toEpochMilli());
             return flushResponse.getLastFinalizedBucketEnd().toEpochMilli();
@@ -218,6 +220,7 @@ long runRealtime() throws Exception {
         long end = toIntervalStartEpochMs(nowMinusQueryDelay);
         FlushJobAction.Request request = new FlushJobAction.Request(jobId);
         request.setWaitForNormalization(false);
+        request.setRefreshRequired(false);
         request.setCalcInterim(true);
         request.setAdvanceTime(String.valueOf(end));
         run(start, end, request);

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/process/autodetect/BlackHoleAutodetectProcess.java

@@ -101,7 +101,7 @@ public void writeUpdateScheduledEventsMessage(List<ScheduledEvent> events, TimeV
      */
     @Override
     public String flushJob(FlushJobParams params) {
-        FlushAcknowledgement flushAcknowledgement = new FlushAcknowledgement(FLUSH_ID, 0L);
+        FlushAcknowledgement flushAcknowledgement = new FlushAcknowledgement(FLUSH_ID, 0L, true);
         AutodetectResult result = new AutodetectResult(
             null,
             null,

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/process/autodetect/output/AutodetectResultProcessor.java

@@ -394,10 +394,12 @@ void processResult(AutodetectResult result) {
             try {
                 bulkResultsPersister.executeRequest();
                 bulkAnnotationsPersister.executeRequest();
-                persister.commitWrites(
-                    jobId,
-                    EnumSet.of(JobResultsPersister.CommitType.RESULTS, JobResultsPersister.CommitType.ANNOTATIONS)
-                );
+                if (flushAcknowledgement.getRefreshRequired()) {
+                    persister.commitWrites(
+                        jobId,
+                        EnumSet.of(JobResultsPersister.CommitType.RESULTS, JobResultsPersister.CommitType.ANNOTATIONS)
+                    );
+                }
             } catch (Exception e) {
                 logger.error(
                     "["

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/process/autodetect/output/FlushListener.java

@@ -70,7 +70,7 @@ private static class FlushAcknowledgementHolder {
         private volatile Exception flushException;
 
         private FlushAcknowledgementHolder(String flushId) {
-            this.flushAcknowledgement = new FlushAcknowledgement(flushId, 0L);
+            this.flushAcknowledgement = new FlushAcknowledgement(flushId, 0L, true);
             this.latch = new CountDownLatch(1);
         }
     }