HLRC: ML Stop datafeed API (#33946)

* HLRC: ML stop data feed api

client/rest-high-level/src/main/java/org/elasticsearch/client/MLRequestConverters.java

@@ -49,6 +49,7 @@ import org.elasticsearch.client.ml.PutCalendarRequest;
 import org.elasticsearch.client.ml.PutDatafeedRequest;
 import org.elasticsearch.client.ml.PutJobRequest;
 import org.elasticsearch.client.ml.StartDatafeedRequest;
+import org.elasticsearch.client.ml.StopDatafeedRequest;
 import org.elasticsearch.client.ml.UpdateJobRequest;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesReference;
@@ -245,6 +246,19 @@ final class MLRequestConverters {
         return request;
     }
 
+    static Request stopDatafeed(StopDatafeedRequest stopDatafeedRequest) throws IOException {
+        String endpoint = new EndpointBuilder()
+            .addPathPartAsIs("_xpack")
+            .addPathPartAsIs("ml")
+            .addPathPartAsIs("datafeeds")
+            .addPathPart(Strings.collectionToCommaDelimitedString(stopDatafeedRequest.getDatafeedIds()))
+            .addPathPartAsIs("_stop")
+            .build();
+        Request request = new Request(HttpPost.METHOD_NAME, endpoint);
+        request.setEntity(createEntity(stopDatafeedRequest, REQUEST_BODY_CONTENT_TYPE));
+        return request;
+    }
+
     static Request deleteForecast(DeleteForecastRequest deleteForecastRequest) {
         String endpoint = new EndpointBuilder()
             .addPathPartAsIs("_xpack")

client/rest-high-level/src/main/java/org/elasticsearch/client/MachineLearningClient.java

@@ -60,6 +60,8 @@ import org.elasticsearch.client.ml.PutJobRequest;
 import org.elasticsearch.client.ml.PutJobResponse;
 import org.elasticsearch.client.ml.StartDatafeedRequest;
 import org.elasticsearch.client.ml.StartDatafeedResponse;
+import org.elasticsearch.client.ml.StopDatafeedRequest;
+import org.elasticsearch.client.ml.StopDatafeedResponse;
 import org.elasticsearch.client.ml.UpdateJobRequest;
 import org.elasticsearch.client.ml.job.stats.JobStats;
 
@@ -607,6 +609,46 @@ public final class MachineLearningClient {
             Collections.emptySet());
     }
 
+    /**
+     * Stops the given Machine Learning Datafeed
+     * <p>
+     * For additional info
+     * see <a href="http://www.elastic.co/guide/en/elasticsearch/reference/current/ml-stop-datafeed.html">
+     *     ML Stop Datafeed documentation</a>
+     *
+     * @param request The request to stop the datafeed
+     * @param options  Additional request options (e.g. headers), use {@link RequestOptions#DEFAULT} if nothing needs to be customized
+     * @return action acknowledgement
+     * @throws IOException when there is a serialization issue sending the request or receiving the response
+     */
+    public StopDatafeedResponse stopDatafeed(StopDatafeedRequest request, RequestOptions options) throws IOException {
+        return restHighLevelClient.performRequestAndParseEntity(request,
+            MLRequestConverters::stopDatafeed,
+            options,
+            StopDatafeedResponse::fromXContent,
+            Collections.emptySet());
+    }
+
+    /**
+     * Stops the given Machine Learning Datafeed asynchronously and notifies the listener on completion
+     * <p>
+     * For additional info
+     * see <a href="http://www.elastic.co/guide/en/elasticsearch/reference/current/ml-stop-datafeed.html">
+     *         ML Stop Datafeed documentation</a>
+     *
+     * @param request The request to stop the datafeed
+     * @param options  Additional request options (e.g. headers), use {@link RequestOptions#DEFAULT} if nothing needs to be customized
+     * @param listener Listener to be notified upon request completion
+     */
+    public void stopDatafeedAsync(StopDatafeedRequest request, RequestOptions options, ActionListener<StopDatafeedResponse> listener) {
+        restHighLevelClient.performRequestAsyncAndParseEntity(request,
+            MLRequestConverters::stopDatafeed,
+            options,
+            StopDatafeedResponse::fromXContent,
+            listener,
+            Collections.emptySet());
+    }
+
     /**
      * Updates a Machine Learning {@link org.elasticsearch.client.ml.job.config.Job}
      * <p>

client/rest-high-level/src/main/java/org/elasticsearch/client/ml/StopDatafeedRequest.java

@@ -0,0 +1,195 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.elasticsearch.client.ml;
+
+import org.elasticsearch.action.ActionRequest;
+import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.client.ml.datafeed.DatafeedConfig;
+import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.common.xcontent.ConstructingObjectParser;
+import org.elasticsearch.common.xcontent.ObjectParser;
+import org.elasticsearch.common.xcontent.ToXContentObject;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+
+import java.io.IOException;
+import java.security.InvalidParameterException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * Request to stop Machine Learning Datafeeds
+ */
+public class StopDatafeedRequest extends ActionRequest implements ToXContentObject {
+
+    public static final ParseField TIMEOUT = new ParseField("timeout");
+    public static final ParseField FORCE = new ParseField("force");
+    public static final ParseField ALLOW_NO_DATAFEEDS = new ParseField("allow_no_datafeeds");
+
+    @SuppressWarnings("unchecked")
+    public static final ConstructingObjectParser<StopDatafeedRequest, Void> PARSER = new ConstructingObjectParser<>(
+        "stop_datafeed_request",
+         a -> new StopDatafeedRequest((List<String>) a[0]));
+
+    static {
+        PARSER.declareField(ConstructingObjectParser.constructorArg(),
+            p -> Arrays.asList(Strings.commaDelimitedListToStringArray(p.text())),
+            DatafeedConfig.ID, ObjectParser.ValueType.STRING_ARRAY);
+        PARSER.declareString((obj, val) -> obj.setTimeout(TimeValue.parseTimeValue(val, TIMEOUT.getPreferredName())), TIMEOUT);
+        PARSER.declareBoolean(StopDatafeedRequest::setForce, FORCE);
+        PARSER.declareBoolean(StopDatafeedRequest::setAllowNoDatafeeds, ALLOW_NO_DATAFEEDS);
+    }
+
+    private static final String ALL_DATAFEEDS = "_all";
+
+    private final List<String> datafeedIds;
+    private TimeValue timeout;
+    private Boolean force;
+    private Boolean allowNoDatafeeds;
+
+    /**
+     * Explicitly stop all datafeeds
+     *
+     * @return a {@link StopDatafeedRequest} for all existing datafeeds
+     */
+    public static StopDatafeedRequest stopAllDatafeedsRequest(){
+        return new StopDatafeedRequest(ALL_DATAFEEDS);
+    }
+
+    StopDatafeedRequest(List<String> datafeedIds) {
+        if (datafeedIds.isEmpty()) {
+            throw new InvalidParameterException("datafeedIds must not be empty");
+        }
+        if (datafeedIds.stream().anyMatch(Objects::isNull)) {
+            throw new NullPointerException("datafeedIds must not contain null values");
+        }
+        this.datafeedIds = new ArrayList<>(datafeedIds);
+    }
+
+    /**
+     * Close the specified Datafeeds via their unique datafeedIds
+     *
+     * @param datafeedIds must be non-null and non-empty and each datafeedId must be non-null
+     */
+    public StopDatafeedRequest(String... datafeedIds) {
+        this(Arrays.asList(datafeedIds));
+    }
+
+    /**
+     * All the datafeedIds to be stopped
+     */
+    public List<String> getDatafeedIds() {
+        return datafeedIds;
+    }
+
+    public TimeValue getTimeout() {
+        return timeout;
+    }
+
+    /**
+     * How long to wait for the stop request to complete before timing out.
+     *
+     * @param timeout Default value: 30 minutes
+     */
+    public void setTimeout(TimeValue timeout) {
+        this.timeout = timeout;
+    }
+
+    public Boolean isForce() {
+        return force;
+    }
+
+    /**
+     * Should the stopping be forced.
+     *
+     * Use to forcefully stop a datafeed
+     *
+     * @param force When {@code true} forcefully stop the datafeed. Defaults to {@code false}
+     */
+    public void setForce(boolean force) {
+        this.force = force;
+    }
+
+    public Boolean isAllowNoDatafeeds() {
+        return this.allowNoDatafeeds;
+    }
+
+    /**
+     * Whether to ignore if a wildcard expression matches no datafeeds.
+     *
+     * This includes {@code _all} string.
+     *
+     * @param allowNoDatafeeds When {@code true} ignore if wildcard or {@code _all} matches no datafeeds. Defaults to {@code true}
+     */
+    public void setAllowNoDatafeeds(boolean allowNoDatafeeds) {
+        this.allowNoDatafeeds = allowNoDatafeeds;
+    }
+
+    @Override
+    public ActionRequestValidationException validate() {
+        return null;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(datafeedIds, timeout, force, allowNoDatafeeds);
+    }
+
+    @Override
+    public boolean equals(Object other) {
+        if (this == other) {
+            return true;
+        }
+
+        if (other == null || getClass() != other.getClass()) {
+            return false;
+        }
+
+        StopDatafeedRequest that = (StopDatafeedRequest) other;
+        return Objects.equals(datafeedIds, that.datafeedIds) &&
+            Objects.equals(timeout, that.timeout) &&
+            Objects.equals(force, that.force) &&
+            Objects.equals(allowNoDatafeeds, that.allowNoDatafeeds);
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject();
+        builder.field(DatafeedConfig.ID.getPreferredName(), Strings.collectionToCommaDelimitedString(datafeedIds));
+        if (timeout != null) {
+            builder.field(TIMEOUT.getPreferredName(), timeout.getStringRep());
+        }
+        if (force != null) {
+            builder.field(FORCE.getPreferredName(), force);
+        }
+        if (allowNoDatafeeds != null) {
+            builder.field(ALLOW_NO_DATAFEEDS.getPreferredName(), allowNoDatafeeds);
+        }
+        builder.endObject();
+        return builder;
+    }
+
+    @Override
+    public String toString() {
+        return Strings.toString(this);
+    }
+}

client/rest-high-level/src/main/java/org/elasticsearch/client/ml/StopDatafeedResponse.java

@@ -0,0 +1,93 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.elasticsearch.client.ml;
+
+import org.elasticsearch.action.ActionResponse;
+import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.xcontent.ConstructingObjectParser;
+import org.elasticsearch.common.xcontent.ToXContentObject;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentParser;
+
+import java.io.IOException;
+import java.util.Objects;
+
+/**
+ * Response indicating if the Machine Learning Datafeed is now stopped or not
+ */
+public class StopDatafeedResponse extends ActionResponse implements ToXContentObject {
+
+    private static final ParseField STOPPED = new ParseField("stopped");
+
+    public static final ConstructingObjectParser<StopDatafeedResponse, Void> PARSER =
+        new ConstructingObjectParser<>(
+            "stop_datafeed_response",
+            true,
+            (a) -> new StopDatafeedResponse((Boolean)a[0]));
+
+    static {
+        PARSER.declareBoolean(ConstructingObjectParser.constructorArg(), STOPPED);
+    }
+
+    private final boolean stopped;
+
+    public StopDatafeedResponse(boolean stopped) {
+        this.stopped = stopped;
+    }
+
+    public static StopDatafeedResponse fromXContent(XContentParser parser) throws IOException {
+        return PARSER.parse(parser, null);
+    }
+
+    /**
+     * Has the Datafeed stopped or not
+     *
+     * @return boolean value indicating the Datafeed stopped status
+     */
+    public boolean isStopped() {
+        return stopped;
+    }
+
+    @Override
+    public boolean equals(Object other) {
+        if (this == other) {
+            return true;
+        }
+
+        if (other == null || getClass() != other.getClass()) {
+            return false;
+        }
+
+        StopDatafeedResponse that = (StopDatafeedResponse) other;
+        return isStopped() == that.isStopped();
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(isStopped());
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject();
+        builder.field(STOPPED.getPreferredName(), stopped);
+        builder.endObject();
+        return builder;
+    }
+}

client/rest-high-level/src/test/java/org/elasticsearch/client/MLRequestConvertersTests.java

@@ -46,6 +46,7 @@ import org.elasticsearch.client.ml.PutDatafeedRequest;
 import org.elasticsearch.client.ml.PutJobRequest;
 import org.elasticsearch.client.ml.StartDatafeedRequest;
 import org.elasticsearch.client.ml.StartDatafeedRequestTests;
+import org.elasticsearch.client.ml.StopDatafeedRequest;
 import org.elasticsearch.client.ml.UpdateJobRequest;
 import org.elasticsearch.client.ml.calendars.Calendar;
 import org.elasticsearch.client.ml.calendars.CalendarTests;
@@ -276,6 +277,22 @@ public class MLRequestConvertersTests extends ESTestCase {
         }
     }
 
+    public void testStopDatafeed() throws Exception {
+        StopDatafeedRequest datafeedRequest = new StopDatafeedRequest("datafeed_1", "datafeed_2");
+        datafeedRequest.setForce(true);
+        datafeedRequest.setTimeout(TimeValue.timeValueMinutes(10));
+        datafeedRequest.setAllowNoDatafeeds(true);
+        Request request = MLRequestConverters.stopDatafeed(datafeedRequest);
+        assertEquals(HttpPost.METHOD_NAME, request.getMethod());
+        assertEquals("/_xpack/ml/datafeeds/" +
+            Strings.collectionToCommaDelimitedString(datafeedRequest.getDatafeedIds()) +
+            "/_stop", request.getEndpoint());
+        try (XContentParser parser = createParser(JsonXContent.jsonXContent, request.getEntity().getContent())) {
+            StopDatafeedRequest parsedDatafeedRequest = StopDatafeedRequest.PARSER.apply(parser, null);
+            assertThat(parsedDatafeedRequest, equalTo(datafeedRequest));
+        }
+    }
+
     public void testDeleteForecast() {
         String jobId = randomAlphaOfLength(10);
         DeleteForecastRequest deleteForecastRequest = new DeleteForecastRequest(jobId);

client/rest-high-level/src/test/java/org/elasticsearch/client/MachineLearningIT.java

@@ -57,6 +57,8 @@ import org.elasticsearch.client.ml.PutJobRequest;
 import org.elasticsearch.client.ml.PutJobResponse;
 import org.elasticsearch.client.ml.StartDatafeedRequest;
 import org.elasticsearch.client.ml.StartDatafeedResponse;
+import org.elasticsearch.client.ml.StopDatafeedRequest;
+import org.elasticsearch.client.ml.StopDatafeedResponse;
 import org.elasticsearch.client.ml.UpdateJobRequest;
 import org.elasticsearch.client.ml.calendars.Calendar;
 import org.elasticsearch.client.ml.calendars.CalendarTests;
@@ -497,6 +499,71 @@ public class MachineLearningIT extends ESRestHighLevelClientTestCase {
         }, 30, TimeUnit.SECONDS);
     }
 
+    public void testStopDatafeed() throws Exception {
+        String jobId1 = "test-stop-datafeed1";
+        String jobId2 = "test-stop-datafeed2";
+        String jobId3 = "test-stop-datafeed3";
+        String indexName = "stop_data_1";
+
+        // Set up the index
+        CreateIndexRequest createIndexRequest = new CreateIndexRequest(indexName);
+        createIndexRequest.mapping("doc", "timestamp", "type=date", "total", "type=long");
+        highLevelClient().indices().create(createIndexRequest, RequestOptions.DEFAULT);
+
+        // create the job and the datafeed
+        Job job1 = buildJob(jobId1);
+        putJob(job1);
+        openJob(job1);
+
+        Job job2 = buildJob(jobId2);
+        putJob(job2);
+        openJob(job2);
+
+        Job job3 = buildJob(jobId3);
+        putJob(job3);
+        openJob(job3);
+
+        String datafeedId1 = createAndPutDatafeed(jobId1, indexName);
+        String datafeedId2 = createAndPutDatafeed(jobId2, indexName);
+        String datafeedId3 = createAndPutDatafeed(jobId3, indexName);
+
+        MachineLearningClient machineLearningClient = highLevelClient().machineLearning();
+
+        machineLearningClient.startDatafeed(new StartDatafeedRequest(datafeedId1), RequestOptions.DEFAULT);
+        machineLearningClient.startDatafeed(new StartDatafeedRequest(datafeedId2), RequestOptions.DEFAULT);
+        machineLearningClient.startDatafeed(new StartDatafeedRequest(datafeedId3), RequestOptions.DEFAULT);
+
+        {
+            StopDatafeedRequest request = new StopDatafeedRequest(datafeedId1);
+            request.setAllowNoDatafeeds(false);
+            StopDatafeedResponse stopDatafeedResponse = execute(request,
+                machineLearningClient::stopDatafeed,
+                machineLearningClient::stopDatafeedAsync);
+            assertTrue(stopDatafeedResponse.isStopped());
+        }
+        {
+            StopDatafeedRequest request = new StopDatafeedRequest(datafeedId2, datafeedId3);
+            request.setAllowNoDatafeeds(false);
+            StopDatafeedResponse stopDatafeedResponse = execute(request,
+                machineLearningClient::stopDatafeed,
+                machineLearningClient::stopDatafeedAsync);
+            assertTrue(stopDatafeedResponse.isStopped());
+        }
+        {
+            StopDatafeedResponse stopDatafeedResponse = execute(new StopDatafeedRequest("datafeed_that_doesnot_exist*"),
+                machineLearningClient::stopDatafeed,
+                machineLearningClient::stopDatafeedAsync);
+            assertTrue(stopDatafeedResponse.isStopped());
+        }
+        {
+            StopDatafeedRequest request = new StopDatafeedRequest("datafeed_that_doesnot_exist*");
+            request.setAllowNoDatafeeds(false);
+            ElasticsearchStatusException exception = expectThrows(ElasticsearchStatusException.class,
+                () -> execute(request, machineLearningClient::stopDatafeed, machineLearningClient::stopDatafeedAsync));
+            assertThat(exception.status().getStatus(), equalTo(404));
+        }
+    }
+
     public void testDeleteForecast() throws Exception {
         String jobId = "test-delete-forecast";
 
@@ -642,4 +709,23 @@ public class MachineLearningIT extends ESRestHighLevelClientTestCase {
 
         return builder.build();
     }
+
+    private void putJob(Job job) throws IOException {
+        highLevelClient().machineLearning().putJob(new PutJobRequest(job), RequestOptions.DEFAULT);
+    }
+
+    private void openJob(Job job) throws IOException {
+        highLevelClient().machineLearning().openJob(new OpenJobRequest(job.getId()), RequestOptions.DEFAULT);
+    }
+
+    private String createAndPutDatafeed(String jobId, String indexName) throws IOException {
+        String datafeedId = jobId + "-feed";
+        DatafeedConfig datafeed = DatafeedConfig.builder(datafeedId, jobId)
+            .setIndices(indexName)
+            .setQueryDelay(TimeValue.timeValueSeconds(1))
+            .setTypes(Arrays.asList("doc"))
+            .setFrequency(TimeValue.timeValueSeconds(1)).build();
+        highLevelClient().machineLearning().putDatafeed(new PutDatafeedRequest(datafeed), RequestOptions.DEFAULT);
+        return datafeedId;
+    }
 }

client/rest-high-level/src/test/java/org/elasticsearch/client/documentation/MlClientDocumentationIT.java

@@ -73,6 +73,8 @@ import org.elasticsearch.client.ml.PutJobRequest;
 import org.elasticsearch.client.ml.PutJobResponse;
 import org.elasticsearch.client.ml.StartDatafeedRequest;
 import org.elasticsearch.client.ml.StartDatafeedResponse;
+import org.elasticsearch.client.ml.StopDatafeedRequest;
+import org.elasticsearch.client.ml.StopDatafeedResponse;
 import org.elasticsearch.client.ml.UpdateJobRequest;
 import org.elasticsearch.client.ml.calendars.Calendar;
 import org.elasticsearch.client.ml.datafeed.ChunkingConfig;
@@ -769,6 +771,57 @@ public class MlClientDocumentationIT extends ESRestHighLevelClientTestCase {
         }
     }
 
+    public void testStopDatafeed() throws Exception {
+        RestHighLevelClient client = highLevelClient();
+
+        {
+            //tag::x-pack-ml-stop-datafeed-request
+            StopDatafeedRequest request = new StopDatafeedRequest("datafeed_id1", "datafeed_id*"); // <1>
+            //end::x-pack-ml-stop-datafeed-request
+            request = StopDatafeedRequest.stopAllDatafeedsRequest();
+
+            //tag::x-pack-ml-stop-datafeed-request-options
+            request.setAllowNoDatafeeds(true); // <1>
+            request.setForce(true); // <2>
+            request.setTimeout(TimeValue.timeValueMinutes(10)); // <3>
+            //end::x-pack-ml-stop-datafeed-request-options
+
+            //tag::x-pack-ml-stop-datafeed-execute
+            StopDatafeedResponse response = client.machineLearning().stopDatafeed(request, RequestOptions.DEFAULT);
+            boolean stopped = response.isStopped(); // <1>
+            //end::x-pack-ml-stop-datafeed-execute
+
+            assertTrue(stopped);
+        }
+        {
+            StopDatafeedRequest request = StopDatafeedRequest.stopAllDatafeedsRequest();
+
+            // tag::x-pack-ml-stop-datafeed-listener
+            ActionListener<StopDatafeedResponse> listener = new ActionListener<StopDatafeedResponse>() {
+                @Override
+                public void onResponse(StopDatafeedResponse response) {
+                    // <1>
+                }
+
+                @Override
+                public void onFailure(Exception e) {
+                    // <2>
+                }
+            };
+            // end::x-pack-ml-stop-datafeed-listener
+
+            // Replace the empty listener by a blocking listener in test
+            final CountDownLatch latch = new CountDownLatch(1);
+            listener = new LatchedActionListener<>(listener, latch);
+
+            // tag::x-pack-ml-stop-datafeed-execute-async
+            client.machineLearning().stopDatafeedAsync(request, RequestOptions.DEFAULT, listener); // <1>
+            // end::x-pack-ml-stop-datafeed-execute-async
+
+            assertTrue(latch.await(30L, TimeUnit.SECONDS));
+        }
+    }
+
     public void testGetBuckets() throws IOException, InterruptedException {
         RestHighLevelClient client = highLevelClient();
 

client/rest-high-level/src/test/java/org/elasticsearch/client/ml/StopDatafeedRequestTests.java

@@ -0,0 +1,81 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.elasticsearch.client.ml;
+
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.test.AbstractXContentTestCase;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+public class StopDatafeedRequestTests extends AbstractXContentTestCase<StopDatafeedRequest> {
+
+    public void testCloseAllDatafeedsRequest() {
+        StopDatafeedRequest request = StopDatafeedRequest.stopAllDatafeedsRequest();
+        assertEquals(request.getDatafeedIds().size(), 1);
+        assertEquals(request.getDatafeedIds().get(0), "_all");
+    }
+
+    public void testWithNullDatafeedIds() {
+        Exception exception = expectThrows(IllegalArgumentException.class, StopDatafeedRequest::new);
+        assertEquals(exception.getMessage(), "datafeedIds must not be empty");
+
+        exception = expectThrows(NullPointerException.class, () -> new StopDatafeedRequest("datafeed1", null));
+        assertEquals(exception.getMessage(), "datafeedIds must not contain null values");
+    }
+
+
+    @Override
+    protected StopDatafeedRequest createTestInstance() {
+        int datafeedCount = randomIntBetween(1, 10);
+        List<String> datafeedIds = new ArrayList<>(datafeedCount);
+
+        for (int i = 0; i < datafeedCount; i++) {
+            datafeedIds.add(randomAlphaOfLength(10));
+        }
+
+        StopDatafeedRequest request = new StopDatafeedRequest(datafeedIds.toArray(new String[0]));
+
+        if (randomBoolean()) {
+            request.setAllowNoDatafeeds(randomBoolean());
+        }
+
+        if (randomBoolean()) {
+            request.setTimeout(TimeValue.timeValueMinutes(randomIntBetween(1, 10)));
+        }
+
+        if (randomBoolean()) {
+            request.setForce(randomBoolean());
+        }
+
+        return request;
+    }
+
+    @Override
+    protected StopDatafeedRequest doParseInstance(XContentParser parser) throws IOException {
+        return StopDatafeedRequest.PARSER.parse(parser, null);
+    }
+
+    @Override
+    protected boolean supportsUnknownFields() {
+        return false;
+    }
+}

client/rest-high-level/src/test/java/org/elasticsearch/client/ml/StopDatafeedResponseTests.java

@@ -0,0 +1,42 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.elasticsearch.client.ml;
+
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.test.AbstractXContentTestCase;
+
+import java.io.IOException;
+
+public class StopDatafeedResponseTests extends AbstractXContentTestCase<StopDatafeedResponse> {
+
+    @Override
+    protected StopDatafeedResponse createTestInstance() {
+        return new StopDatafeedResponse(randomBoolean());
+    }
+
+    @Override
+    protected StopDatafeedResponse doParseInstance(XContentParser parser) throws IOException {
+        return StopDatafeedResponse.fromXContent(parser);
+    }
+
+    @Override
+    protected boolean supportsUnknownFields() {
+        return true;
+    }
+}

docs/java-rest/high-level/ml/stop-datafeed.asciidoc

@@ -0,0 +1,66 @@
+[[java-rest-high-x-pack-ml-stop-datafeed]]
+=== Stop Datafeed API
+
+The Stop Datafeed API provides the ability to stop a {ml} datafeed in the cluster.
+It accepts a `StopDatafeedRequest` object and responds
+with a `StopDatafeedResponse` object.
+
+[[java-rest-high-x-pack-ml-stop-datafeed-request]]
+==== Stop Datafeed Request
+
+A `StopDatafeedRequest` object is created referencing any number of non-null `datafeedId` entries.
+Wildcards and `_all` are also accepted.
+All other fields are optional for the request.
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests}/MlClientDocumentationIT.java[x-pack-ml-stop-datafeed-request]
+--------------------------------------------------
+<1> Constructing a new request referencing existing `datafeedId` entries.
+
+==== Optional Arguments
+
+The following arguments are optional.
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests}/MlClientDocumentationIT.java[x-pack-ml-stop-datafeed-request-options]
+--------------------------------------------------
+<1> Whether to ignore if a wildcard expression matches no datafeeds. (This includes `_all` string)
+<2> If true, the datafeed is stopped forcefully.
+<3> Controls the amount of time to wait until a datafeed stops. The default value is 20 seconds.
+
+[[java-rest-high-x-pack-ml-stop-datafeed-execution]]
+==== Execution
+
+The request can be executed through the `MachineLearningClient` contained
+in the `RestHighLevelClient` object, accessed via the `machineLearningClient()` method.
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests}/MlClientDocumentationIT.java[x-pack-ml-stop-datafeed-execute]
+--------------------------------------------------
+<1> Did the datafeed successfully stop?
+
+[[java-rest-high-x-pack-ml-stop-datafeed-execution-async]]
+==== Asynchronous Execution
+
+The request can also be executed asynchronously:
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests}/MlClientDocumentationIT.java[x-pack-ml-stop-datafeed-execute-async]
+--------------------------------------------------
+<1> The `StopDatafeedRequest` to execute and the `ActionListener` to use when
+the execution completes
+
+The method does not block and returns immediately. The passed `ActionListener` is used
+to notify the caller of completion. A typical `ActionListener` for `StopDatafeedResponse` may
+look like
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests}/MlClientDocumentationIT.java[x-pack-ml-stop-datafeed-listener]
+--------------------------------------------------
+<1> `onResponse` is called back when the action is completed successfully
+<2> `onFailure` is called back when some unexpected error occurs

docs/java-rest/high-level/supported-apis.asciidoc

@@ -226,6 +226,7 @@ The Java High Level REST Client supports the following Machine Learning APIs:
 * <<java-rest-high-x-pack-ml-get-datafeed>>
 * <<java-rest-high-x-pack-ml-delete-datafeed>>
 * <<java-rest-high-x-pack-ml-start-datafeed>>
+* <<java-rest-high-x-pack-ml-stop-datafeed>>
 * <<java-rest-high-x-pack-ml-forecast-job>>
 * <<java-rest-high-x-pack-ml-delete-forecast>>
 * <<java-rest-high-x-pack-ml-get-buckets>>
@@ -249,6 +250,7 @@ include::ml/put-datafeed.asciidoc[]
 include::ml/get-datafeed.asciidoc[]
 include::ml/delete-datafeed.asciidoc[]
 include::ml/start-datafeed.asciidoc[]
+include::ml/stop-datafeed.asciidoc[]
 include::ml/get-job-stats.asciidoc[]
 include::ml/forecast-job.asciidoc[]
 include::ml/delete-forecast.asciidoc[]