[ML] Add deprecation warning for flush API (#121667)

valeriy42 · web-flow · commit f1b198347616 · 2025-02-07T13:53:43.000+01:00
The anomaly detection job flush API is deprecated since it is only required for the post data API, which was deprecated since 7.11.0. Closes #121506
diff --git a/docs/changelog/121667.yaml b/docs/changelog/121667.yaml
@@ -0,0 +1,11 @@
+pr: 121667
+summary: Add deprecation warning for flush API
+area: Machine Learning
+type: deprecation
+issues:
+ - 121506
+deprecation:
+  title: Add deprecation warning for flush API
+  area: REST API
+  details: The anomaly detection job flush API is deprecated since it is only required for the post data API, which was deprecated since 7.11.0.
+  impact: This should have a minimal impact on users as the flush API is only required for the post data API, which was deprecated since 7.11.0.
diff --git a/x-pack/plugin/ml/qa/basic-multi-node/src/javaRestTest/java/org/elasticsearch/xpack/ml/integration/MlBasicMultiNodeIT.java b/x-pack/plugin/ml/qa/basic-multi-node/src/javaRestTest/java/org/elasticsearch/xpack/ml/integration/MlBasicMultiNodeIT.java
@@ -41,6 +41,15 @@ public class MlBasicMultiNodeIT extends ESRestTestCase {
         )
         .build();
 
+    private static final RequestOptions FLUSH_OPTIONS = RequestOptions.DEFAULT.toBuilder()
+        .setWarningsHandler(
+            warnings -> Collections.singletonList(
+                "Forcing any buffered data to be processed is deprecated, "
+                    + "in a future major version it will be compulsory to use a datafeed"
+            ).equals(warnings) == false
+        )
+        .build();
+
     public void testMachineLearningInstalled() throws Exception {
         Response response = client().performRequest(new Request("GET", "/_xpack"));
         Map<?, ?> features = (Map<?, ?>) entityAsMap(response).get("features");
@@ -93,7 +102,9 @@ public void testMiniFarequote() throws Exception {
         assertEquals(1403481600000L, responseBody.get("earliest_record_timestamp"));
         assertEquals(1403481700000L, responseBody.get("latest_record_timestamp"));
 
-        Response flushResponse = client().performRequest(new Request("POST", BASE_PATH + "anomaly_detectors/" + jobId + "/_flush"));
+        Request flustRequest = new Request("POST", BASE_PATH + "anomaly_detectors/" + jobId + "/_flush");
+        flustRequest.setOptions(FLUSH_OPTIONS);
+        Response flushResponse = client().performRequest(flustRequest);
         assertFlushResponse(flushResponse, true, 1403481600000L);
 
         Request closeRequest = new Request("POST", BASE_PATH + "anomaly_detectors/" + jobId + "/_close");
@@ -191,7 +202,9 @@ public void testMiniFarequoteReopen() throws Exception {
         assertEquals(1403481600000L, responseBody.get("earliest_record_timestamp"));
         assertEquals(1403482000000L, responseBody.get("latest_record_timestamp"));
 
-        Response flushResponse = client().performRequest(new Request("POST", BASE_PATH + "anomaly_detectors/" + jobId + "/_flush"));
+        Request flushRequest = new Request("POST", BASE_PATH + "anomaly_detectors/" + jobId + "/_flush");
+        flushRequest.setOptions(FLUSH_OPTIONS);
+        Response flushResponse = client().performRequest(flushRequest);
         assertFlushResponse(flushResponse, true, 1403481600000L);
 
         Request closeRequest = new Request("POST", BASE_PATH + "anomaly_detectors/" + jobId + "/_close");
diff --git a/x-pack/plugin/ml/qa/native-multi-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/ml/integration/MlJobIT.java b/x-pack/plugin/ml/qa/native-multi-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/ml/integration/MlJobIT.java
@@ -58,6 +58,14 @@ public class MlJobIT extends ESRestTestCase {
             ).equals(warnings) == false
         )
         .build();
+    private static final RequestOptions FLUSH_OPTIONS = RequestOptions.DEFAULT.toBuilder()
+        .setWarningsHandler(
+            warnings -> Collections.singletonList(
+                "Forcing any buffered data to be processed is deprecated, "
+                    + "in a future major version it will be compulsory to use a datafeed"
+            ).equals(warnings) == false
+        )
+        .build();
 
     @Override
     protected Settings restClientSettings() {
@@ -534,9 +542,9 @@ public void testOutOfOrderData() throws Exception {
         postDataRequest.setJsonEntity("{ \"airline\":\"LOT\", \"responsetime\":100, \"time\":\"2019-07-01 00:10:00Z\" }");
         client().performRequest(postDataRequest);
 
-        Response flushResponse = client().performRequest(
-            new Request("POST", MachineLearning.BASE_PATH + "anomaly_detectors/" + jobId + "/_flush")
-        );
+        Request flushRequest = new Request("POST", MachineLearning.BASE_PATH + "anomaly_detectors/" + jobId + "/_flush");
+        flushRequest.setOptions(FLUSH_OPTIONS);
+        Response flushResponse = client().performRequest(flushRequest);
         assertThat(entityAsMap(flushResponse), hasEntry("flushed", true));
 
         closeJob(jobId);
@@ -574,9 +582,9 @@ public void testDeleteJob_TimingStatsDocumentIsDeleted() throws Exception {
             { "airline":"LOT", "response_time":100, "time":"2019-07-01 02:00:00Z" }""");
         client().performRequest(postDataRequest);
 
-        Response flushResponse = client().performRequest(
-            new Request("POST", MachineLearning.BASE_PATH + "anomaly_detectors/" + jobId + "/_flush")
-        );
+        Request flushRequest = new Request("POST", MachineLearning.BASE_PATH + "anomaly_detectors/" + jobId + "/_flush");
+        flushRequest.setOptions(FLUSH_OPTIONS);
+        Response flushResponse = client().performRequest(flushRequest);
         assertThat(entityAsMap(flushResponse), hasEntry("flushed", true));
 
         closeJob(jobId);
diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/rest/job/RestFlushJobAction.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/rest/job/RestFlushJobAction.java
@@ -34,7 +34,9 @@ public class RestFlushJobAction extends BaseRestHandler {
 
     @Override
     public List<Route> routes() {
-        return List.of(new Route(POST, BASE_PATH + "anomaly_detectors/{" + ID + "}/_flush"));
+        final String msg = "Forcing any buffered data to be processed is deprecated, "
+            + "in a future major version it will be compulsory to use a datafeed";
+        return List.of(Route.builder(POST, BASE_PATH + "anomaly_detectors/{" + ID + "}/_flush").deprecateAndKeep(msg).build());
     }
 
     @Override
diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/rest/job/RestPostDataAction.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/rest/job/RestPostDataAction.java
@@ -8,7 +8,6 @@
 
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.client.internal.node.NodeClient;
-import org.elasticsearch.core.UpdateForV9;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestStatus;
@@ -27,19 +26,11 @@ public class RestPostDataAction extends BaseRestHandler {
     private static final String DEFAULT_RESET_START = "";
     private static final String DEFAULT_RESET_END = "";
 
-    @UpdateForV9(owner = UpdateForV9.Owner.MACHINE_LEARNING)
-    // these routes were ".deprecated" in RestApiVersion.V_8 which will require use of REST API compatibility headers to access
-    // this API in v9. It is unclear if this was intentional for v9, and the code has been updated to ".deprecateAndKeep" which will
-    // continue to emit deprecations warnings but will not require any special headers to access the API in v9.
-    // Please review and update the code and tests as needed. The original code remains commented out below for reference.
     @Override
     public List<Route> routes() {
         final String msg = "Posting data directly to anomaly detection jobs is deprecated, "
             + "in a future major version it will be compulsory to use a datafeed";
-        return List.of(
-            // Route.builder(POST, BASE_PATH + "anomaly_detectors/{" + Job.ID + "}/_data").deprecated(msg, RestApiVersion.V_8).build(),
-            Route.builder(POST, BASE_PATH + "anomaly_detectors/{" + Job.ID + "}/_data").deprecateAndKeep(msg).build()
-        );
+        return List.of(Route.builder(POST, BASE_PATH + "anomaly_detectors/{" + Job.ID + "}/_data").deprecateAndKeep(msg).build());
     }
 
     @Override
diff --git a/x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/ml/job_cat_apis.yml b/x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/ml/job_cat_apis.yml
@@ -87,6 +87,8 @@ setup:
           {"airline":"JZA","responsetime":"244.1276","time":"1403485200"}
 
   - do:
+      warnings:
+        - 'Forcing any buffered data to be processed is deprecated, in a future major version it will be compulsory to use a datafeed'
       ml.flush_job:
         job_id: job-stats-test
   - match: { flushed: true }
diff --git a/x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/ml/jobs_get_stats.yml b/x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/ml/jobs_get_stats.yml
@@ -87,6 +87,8 @@ setup:
           {"airline":"JZA","responsetime":"244.1276","time":"1403485200"}
 
   - do:
+      warnings:
+        - 'Forcing any buffered data to be processed is deprecated, in a future major version it will be compulsory to use a datafeed'
       ml.flush_job:
         job_id: job-stats-test
   - match: { flushed: true }
@@ -131,6 +133,8 @@ setup:
           {"airline":"JZA","responsetime":"244.1276","time":"1403485200"}
 
   - do:
+      warnings:
+        - 'Forcing any buffered data to be processed is deprecated, in a future major version it will be compulsory to use a datafeed'
       ml.flush_job:
         job_id: job-stats-test
   - match: { flushed: true }
diff --git a/x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/ml/post_data.yml b/x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/ml/post_data.yml
@@ -77,6 +77,8 @@ setup:
   - match: { latest_record_timestamp: 1403481700000}
 
   - do:
+      warnings:
+        - 'Forcing any buffered data to be processed is deprecated, in a future major version it will be compulsory to use a datafeed'
       ml.flush_job:
         job_id: post-data-job
   - match: { flushed: true }
@@ -110,7 +112,12 @@ setup:
 
 ---
 "Test flush and close job WITHOUT sending any data":
+  - skip:
+      features:
+        - "warnings"
   - do:
+      warnings:
+        - 'Forcing any buffered data to be processed is deprecated, in a future major version it will be compulsory to use a datafeed'
       ml.flush_job:
         job_id: post-data-job
   - match: { flushed: true }
@@ -150,6 +157,8 @@ setup:
 
   # Skip a bucket
   - do:
+      warnings:
+        - 'Forcing any buffered data to be processed is deprecated, in a future major version it will be compulsory to use a datafeed'
       ml.flush_job:
         job_id: post-data-job
         skip_time: 1403488700
@@ -266,26 +275,36 @@ setup:
   - skip:
       reason: "https://github.com/elastic/elasticsearch/issues/34747"
       cluster_features: ["gte_v6.5.0"]
+      features:
+        - "warnings"
 
   - do:
       catch: missing
+      warnings:
+        - 'Forcing any buffered data to be processed is deprecated, in a future major version it will be compulsory to use a datafeed'
       ml.flush_job:
         job_id: not_a_job
 
   - do:
       catch: /parse_exception/
+      warnings:
+        - 'Forcing any buffered data to be processed is deprecated, in a future major version it will be compulsory to use a datafeed'
       ml.flush_job:
         job_id: post-data-job
         start: not_a_date
 
   - do:
       catch: /parse_exception/
+      warnings:
+        - 'Forcing any buffered data to be processed is deprecated, in a future major version it will be compulsory to use a datafeed'
       ml.flush_job:
         job_id: post-data-job
         end: end_not_a_date
 
   - do:
       catch: /parse_exception/
+      warnings:
+        - 'Forcing any buffered data to be processed is deprecated, in a future major version it will be compulsory to use a datafeed'
       ml.flush_job:
         job_id: post-data-job
         advance_time: advance_time_not_a_date
@@ -311,6 +330,8 @@ setup:
 
   - do:
       catch: /status_exception/
+      warnings:
+        - 'Forcing any buffered data to be processed is deprecated, in a future major version it will be compulsory to use a datafeed'
       ml.flush_job:
         job_id: post-data-closed-job
 
diff --git a/x-pack/qa/rolling-upgrade/src/test/java/org/elasticsearch/upgrades/MlJobSnapshotUpgradeIT.java b/x-pack/qa/rolling-upgrade/src/test/java/org/elasticsearch/upgrades/MlJobSnapshotUpgradeIT.java
@@ -338,7 +338,23 @@ protected Response postData(String jobId, String data) throws IOException {
     }
 
     protected void flushJob(String jobId) throws IOException {
-        client().performRequest(new Request("POST", "/_ml/anomaly_detectors/" + jobId + "/_flush"));
+        // Flush job is deprecated, so a deprecation warning is possible (depending on the old version)
+        RequestOptions flushOptions = RequestOptions.DEFAULT.toBuilder().setWarningsHandler(warnings -> {
+            if (warnings.isEmpty()) {
+                // No warning is OK - it means we hit an old node where flush is not deprecated
+                return false;
+            } else if (warnings.size() > 1) {
+                return true;
+            }
+            return warnings.get(0)
+                .equals(
+                    "Forcing any buffered data to be processed is deprecated, "
+                        + "in a future major version it will be compulsory to use a datafeed"
+                ) == false;
+        }).build();
+        Request flushRequest = new Request("POST", "/_ml/anomaly_detectors/" + jobId + "/_flush");
+        flushRequest.setOptions(flushOptions);
+        client().performRequest(flushRequest);
     }
 
     private void closeJob(String jobId) throws IOException {

Original file line number	Diff line number	Diff line change
`@@ -34,7 +34,9 @@ public class RestFlushJobAction extends BaseRestHandler {`
`34`	`34`
`35`	`35`	`@Override`
`36`	`36`	`public List<Route> routes() {`
`37`		`- return List.of(new Route(POST, BASE_PATH + "anomaly_detectors/{" + ID + "}/_flush"));`
	`37`	`+ final String msg = "Forcing any buffered data to be processed is deprecated, "`
	`38`	`+ + "in a future major version it will be compulsory to use a datafeed";`
	`39`	`+ return List.of(Route.builder(POST, BASE_PATH + "anomaly_detectors/{" + ID + "}/_flush").deprecateAndKeep(msg).build());`
`38`	`40`	`}`
`39`	`41`
`40`	`42`	`@Override`