HLRC: Add activate watch action (#33988)

* HLRC: Add activate watcher action

Adds activate watch action to the high level rest client.

Relates #29827

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

@@ -19,6 +19,8 @@
 package org.elasticsearch.client;
 
 import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.client.watcher.ActivateWatchRequest;
+import org.elasticsearch.client.watcher.ActivateWatchResponse;
 import org.elasticsearch.client.watcher.AckWatchRequest;
 import org.elasticsearch.client.watcher.AckWatchResponse;
 import org.elasticsearch.protocol.xpack.watcher.DeleteWatchRequest;
@@ -121,4 +123,31 @@ public final class WatcherClient {
             AckWatchResponse::fromXContent, listener, emptySet());
     }
 
+    /**
+     * Activate a watch from the cluster
+     * See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/watcher-api-activate-watch.html">
+     * the docs</a> for more.
+     * @param request the request
+     * @param options the request options (e.g. headers), use {@link RequestOptions#DEFAULT} if nothing needs to be customized
+     * @return the response
+     * @throws IOException in case there is a problem sending the request or parsing back the response
+     */
+    public ActivateWatchResponse activateWatch(ActivateWatchRequest request, RequestOptions options) throws IOException {
+        return restHighLevelClient.performRequestAndParseEntity(request, WatcherRequestConverters::activateWatch, options,
+            ActivateWatchResponse::fromXContent, singleton(404));
+    }
+
+    /**
+     * Asynchronously activates a watch from the cluster
+     * See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/watcher-api-activate-watch.html">
+     * the docs</a> for more.
+     * @param request the request
+     * @param options the request options (e.g. headers), use {@link RequestOptions#DEFAULT} if nothing needs to be customized
+     * @param listener the listener to be notified upon request completion
+     */
+    public void activateWatchAsync(ActivateWatchRequest request, RequestOptions options, ActionListener<ActivateWatchResponse> listener) {
+        restHighLevelClient.performRequestAsyncAndParseEntity(request, WatcherRequestConverters::activateWatch, options,
+            ActivateWatchResponse::fromXContent, listener, singleton(404));
+    }
+
 }

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

@@ -23,6 +23,7 @@ import org.apache.http.client.methods.HttpDelete;
 import org.apache.http.client.methods.HttpPut;
 import org.apache.http.entity.ByteArrayEntity;
 import org.apache.http.entity.ContentType;
+import org.elasticsearch.client.watcher.ActivateWatchRequest;
 import org.elasticsearch.client.watcher.AckWatchRequest;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.protocol.xpack.watcher.DeleteWatchRequest;
@@ -73,4 +74,16 @@ public class WatcherRequestConverters {
         Request request = new Request(HttpPut.METHOD_NAME, endpoint);
         return request;
     }
+
+    static Request activateWatch(ActivateWatchRequest activateWatchRequest) {
+        String endpoint = new RequestConverters.EndpointBuilder()
+            .addPathPartAsIs("_xpack")
+            .addPathPartAsIs("watcher")
+            .addPathPartAsIs("watch")
+            .addPathPart(activateWatchRequest.getWatchId())
+            .addPathPartAsIs("_activate")
+            .build();
+        Request request = new Request(HttpPut.METHOD_NAME, endpoint);
+        return request;
+    }
 }

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

@@ -0,0 +1,61 @@
+/*
+ * 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.watcher;
+
+import org.elasticsearch.client.Validatable;
+import org.elasticsearch.protocol.xpack.watcher.PutWatchRequest;
+
+import java.util.Objects;
+
+/**
+ * A request to explicitly activate a watch.
+ */
+public final class ActivateWatchRequest implements Validatable {
+
+    private final String watchId;
+
+    public ActivateWatchRequest(String watchId) {
+        this.watchId  = Objects.requireNonNull(watchId, "Watch identifier is required");
+        if (PutWatchRequest.isValidId(this.watchId) == false) {
+            throw new IllegalArgumentException("Watch identifier contains whitespace");
+        }
+    }
+
+    /**
+     * @return The ID of the watch to be activated.
+     */
+    public String getWatchId() {
+        return watchId;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        ActivateWatchRequest that = (ActivateWatchRequest) o;
+        return Objects.equals(watchId, that.watchId);
+    }
+
+    @Override
+    public int hashCode() {
+        int result = Objects.hash(watchId);
+        return result;
+    }
+}

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

@@ -0,0 +1,71 @@
+/*
+ * 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.watcher;
+
+import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.xcontent.ConstructingObjectParser;
+import org.elasticsearch.common.xcontent.XContentParser;
+
+import java.io.IOException;
+import java.util.Objects;
+
+/**
+ * Response from an 'activate watch' request.
+ */
+public final class ActivateWatchResponse {
+
+    private static final ParseField STATUS_FIELD = new ParseField("status");
+    private static ConstructingObjectParser<ActivateWatchResponse, Void> PARSER =
+        new ConstructingObjectParser<>("activate_watch_response", true,
+            a -> new ActivateWatchResponse((WatchStatus) a[0]));
+
+    static {
+        PARSER.declareObject(ConstructingObjectParser.constructorArg(),
+            (parser, context) -> WatchStatus.parse(parser),
+            STATUS_FIELD);
+    }
+
+    private final WatchStatus status;
+
+    public ActivateWatchResponse(WatchStatus status) {
+        this.status = status;
+    }
+
+    public WatchStatus getStatus() {
+        return status;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        ActivateWatchResponse that = (ActivateWatchResponse) o;
+        return Objects.equals(status, that.status);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(status);
+    }
+
+    public static ActivateWatchResponse fromXContent(XContentParser parser) throws IOException {
+        return PARSER.parse(parser, null);
+    }
+}

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

@@ -19,6 +19,8 @@
 package org.elasticsearch.client;
 
 import org.elasticsearch.ElasticsearchStatusException;
+import org.elasticsearch.client.watcher.ActivateWatchRequest;
+import org.elasticsearch.client.watcher.ActivateWatchResponse;
 import org.elasticsearch.client.watcher.AckWatchRequest;
 import org.elasticsearch.client.watcher.AckWatchResponse;
 import org.elasticsearch.client.watcher.ActionStatus;
@@ -33,6 +35,7 @@ import org.elasticsearch.protocol.xpack.watcher.PutWatchResponse;
 import org.elasticsearch.rest.RestStatus;
 
 import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.lessThan;
 
 public class WatcherIT extends ESRestHighLevelClientTestCase {
 
@@ -108,4 +111,26 @@ public class WatcherIT extends ESRestHighLevelClientTestCase {
                 new AckWatchRequest("nonexistent"), RequestOptions.DEFAULT));
         assertEquals(RestStatus.NOT_FOUND, exception.status());
     }
+
+    public void testActivateWatchThatExists() throws Exception {
+        String watchId = randomAlphaOfLength(10);
+        createWatch(watchId);
+        ActivateWatchResponse activateWatchResponse1 = highLevelClient().watcher().activateWatch(new ActivateWatchRequest(watchId),
+            RequestOptions.DEFAULT);
+        assertThat(activateWatchResponse1.getStatus().state().isActive(), is(true));
+
+        ActivateWatchResponse activateWatchResponse2 = highLevelClient().watcher().activateWatch(new ActivateWatchRequest(watchId),
+            RequestOptions.DEFAULT);
+        assertThat(activateWatchResponse2.getStatus().state().isActive(), is(true));
+        assertThat(activateWatchResponse1.getStatus().state().getTimestamp(),
+            lessThan(activateWatchResponse2.getStatus().state().getTimestamp()));
+    }
+
+    public void testActivateWatchThatDoesNotExist() throws Exception {
+        String watchId = randomAlphaOfLength(10);
+        // exception when activating a not existing watcher
+        ElasticsearchStatusException exception =  expectThrows(ElasticsearchStatusException.class, () ->
+            highLevelClient().watcher().activateWatch(new ActivateWatchRequest(watchId), RequestOptions.DEFAULT));
+        assertEquals(RestStatus.NOT_FOUND, exception.status());
+    }
 }

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

@@ -21,6 +21,7 @@ package org.elasticsearch.client;
 
 import org.apache.http.client.methods.HttpDelete;
 import org.apache.http.client.methods.HttpPut;
+import org.elasticsearch.client.watcher.ActivateWatchRequest;
 import org.elasticsearch.client.watcher.AckWatchRequest;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.xcontent.XContentType;
@@ -97,4 +98,14 @@ public class WatcherRequestConvertersTests extends ESTestCase {
         assertEquals(expectedEndpoint.toString(), request.getEndpoint());
         assertThat(request.getEntity(), nullValue());
     }
+
+    public void testActivateWatchRequestConversion() {
+        String watchId = randomAlphaOfLength(10);
+        ActivateWatchRequest activateWatchRequest = new ActivateWatchRequest(watchId);
+
+        Request request = WatcherRequestConverters.activateWatch(activateWatchRequest);
+        assertEquals(HttpPut.METHOD_NAME, request.getMethod());
+        assertEquals("/_xpack/watcher/watch/" + watchId + "/_activate", request.getEndpoint());
+        assertThat(request.getEntity(), nullValue());
+    }
 }

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

@@ -25,6 +25,8 @@ import org.elasticsearch.client.Request;
 import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.Response;
 import org.elasticsearch.client.RestHighLevelClient;
+import org.elasticsearch.client.watcher.ActivateWatchRequest;
+import org.elasticsearch.client.watcher.ActivateWatchResponse;
 import org.elasticsearch.client.watcher.AckWatchRequest;
 import org.elasticsearch.client.watcher.AckWatchResponse;
 import org.elasticsearch.client.watcher.ActionStatus;
@@ -203,4 +205,60 @@ public class WatcherDocumentationIT extends ESRestHighLevelClientTestCase {
         }
     }
 
+    public void testActivateWatch() throws Exception {
+        RestHighLevelClient client = highLevelClient();
+
+        {
+            BytesReference watch = new BytesArray("{ \n" +
+                "  \"trigger\": { \"schedule\": { \"interval\": \"10h\" } },\n" +
+                "  \"input\": { \"simple\": { \"foo\" : \"bar\" } },\n" +
+                "  \"actions\": { \"logme\": { \"logging\": { \"text\": \"{{ctx.payload}}\" } } }\n" +
+                "}");
+            PutWatchRequest request = new PutWatchRequest("my_watch_id", watch, XContentType.JSON);
+            request.setActive(false); // <1>
+            PutWatchResponse response = client.watcher().putWatch(request, RequestOptions.DEFAULT);
+        }
+
+        {
+            //tag::activate-watch-request
+            ActivateWatchRequest request = new ActivateWatchRequest("my_watch_id");
+            ActivateWatchResponse response = client.watcher().activateWatch(request, RequestOptions.DEFAULT);
+            //end::activate-watch-request
+
+            //tag::activate-watch-request
+            WatchStatus watchStatus = response.getStatus(); // <1>
+            //end::activate-watch-request
+
+            assertTrue(watchStatus.state().isActive());
+        }
+
+        {
+            ActivateWatchRequest request = new ActivateWatchRequest("my_watch_id");
+            //tag::activate-watch-request-listener
+            ActionListener<ActivateWatchResponse> listener = new ActionListener<ActivateWatchResponse>() {
+                @Override
+                public void onResponse(ActivateWatchResponse response) {
+                    // <1>
+                }
+
+                @Override
+                public void onFailure(Exception e) {
+                    // <2>
+                }
+            };
+            //end::activate-watch-request-listener
+
+            //Replace the empty listener by a blocking listener in test
+            final CountDownLatch latch = new CountDownLatch(1);
+            listener = new LatchedActionListener<>(listener, latch);
+
+            //tag::activate-watch-request-async
+            client.watcher().activateWatchAsync(request, RequestOptions.DEFAULT, listener); // <1>
+            //end::activate-watch-request-async
+
+            assertTrue(latch.await(30L, TimeUnit.SECONDS));
+
+        }
+    }
+
 }

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

@@ -0,0 +1,113 @@
+/*
+ * 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.watcher;
+
+import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.xcontent.NamedXContentRegistry;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentFactory;
+import org.elasticsearch.common.xcontent.XContentParseException;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.common.xcontent.XContentType;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.test.XContentTestUtils;
+
+import java.io.IOException;
+import java.util.function.Predicate;
+
+/**
+ * Basic unit tests for {@link ActivateWatchResponse}.
+ *
+ * Note that we only sanity check watch status parsing here, as there
+ * are dedicated tests for it in {@link WatchStatusTests}.
+ */
+public class ActivateWatchResponseTests extends ESTestCase {
+
+    public void testBasicParsing() throws IOException {
+        XContentType contentType = randomFrom(XContentType.values());
+        XContentBuilder builder = XContentFactory.contentBuilder(contentType).startObject()
+            .startObject("status")
+                .field("version", 42)
+                .field("execution_state", ExecutionState.ACKNOWLEDGED)
+                .startObject("state")
+                   .field("active", false)
+                .endObject()
+            .endObject()
+        .endObject();
+        BytesReference bytes = BytesReference.bytes(builder);
+
+        ActivateWatchResponse response = parse(builder.contentType(), bytes);
+        WatchStatus status = response.getStatus();
+        assertNotNull(status);
+        assertEquals(42, status.version());
+        assertEquals(ExecutionState.ACKNOWLEDGED, status.getExecutionState());
+        assertFalse(status.state().isActive());
+    }
+
+    public void testParsingWithMissingStatus() throws IOException {
+        XContentType contentType = randomFrom(XContentType.values());
+        XContentBuilder builder = XContentFactory.contentBuilder(contentType).startObject().endObject();
+        BytesReference bytes = BytesReference.bytes(builder);
+
+        expectThrows(IllegalArgumentException.class, () -> parse(builder.contentType(), bytes));
+    }
+
+    public void testParsingWithNullStatus() throws IOException {
+        XContentType contentType = randomFrom(XContentType.values());
+        XContentBuilder builder = XContentFactory.contentBuilder(contentType).startObject()
+            .nullField("status")
+        .endObject();
+        BytesReference bytes = BytesReference.bytes(builder);
+
+        expectThrows(XContentParseException.class, () -> parse(builder.contentType(), bytes));
+    }
+
+    public void testParsingWithUnknownKeys() throws IOException {
+        XContentType contentType = randomFrom(XContentType.values());
+        XContentBuilder builder = XContentFactory.contentBuilder(contentType).startObject()
+            .startObject("status")
+                .field("version", 42)
+                .field("execution_state", ExecutionState.ACKNOWLEDGED)
+                .startObject("state")
+                  .field("active", true)
+                .endObject()
+            .endObject()
+        .endObject();
+        BytesReference bytes = BytesReference.bytes(builder);
+
+        Predicate<String> excludeFilter = field -> field.equals("status.actions");
+        BytesReference bytesWithRandomFields = XContentTestUtils.insertRandomFields(
+            builder.contentType(), bytes, excludeFilter, random());
+
+        ActivateWatchResponse response = parse(builder.contentType(), bytesWithRandomFields);
+        WatchStatus status = response.getStatus();
+        assertNotNull(status);
+        assertEquals(42, status.version());
+        assertEquals(ExecutionState.ACKNOWLEDGED, status.getExecutionState());
+        assertTrue(status.state().isActive());
+    }
+
+    private ActivateWatchResponse parse(XContentType contentType, BytesReference bytes) throws IOException {
+        XContentParser parser = XContentFactory.xContent(contentType)
+            .createParser(NamedXContentRegistry.EMPTY, null, bytes.streamInput());
+        parser.nextToken();
+        return ActivateWatchResponse.fromXContent(parser);
+    }
+}

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

@@ -306,15 +306,20 @@ include::security/change-password.asciidoc[]
 
 == Watcher APIs
 
+:upid: {mainid}-document
+:doc-tests-file: {doc-tests}/WatcherDocumentationIT.java
+
 The Java High Level REST Client supports the following Watcher APIs:
 
 * <<java-rest-high-x-pack-watcher-put-watch>>
 * <<java-rest-high-x-pack-watcher-delete-watch>>
 * <<java-rest-high-watcher-ack-watch>>
+* <<{upid}-activate-watch>>
 
 include::watcher/put-watch.asciidoc[]
 include::watcher/delete-watch.asciidoc[]
 include::watcher/ack-watch.asciidoc[]
+include::watcher/activate-watch.asciidoc[]
 
 == Graph APIs
 

docs/java-rest/high-level/watcher/activate-watch.asciidoc

@@ -0,0 +1,56 @@
+--
+:api: activate-watch
+:request: ActivateWatchRequest
+:response: ActivateWatchResponse
+--
+
+[id="{upid}-{api}"]
+=== Activate Watch API
+
+[id="{upid}-{api}-request"]
+==== Execution
+
+A watch can be activated as follows:
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests-file}[{api}-request]
+--------------------------------------------------
+
+[id="{upid}-{api}-response"]
+==== Response
+
+The returned +{response}+ contains the new status of the activated watch.
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests-file}[{api}-response]
+--------------------------------------------------
+<1> `watchStatus` contains status of the watch
+
+[id="{upid}-{api}-request-async"]
+==== Asynchronous Execution
+
+This request can be executed asynchronously:
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests-file}[{api}-request-async]
+--------------------------------------------------
+<1> The +{request}+ to execute and the `ActionListener` to use when
+the execution completes
+
+The asynchronous method does not block and returns immediately. Once it is
+completed the `ActionListener` is called back using the `onResponse` method
+if the execution successfully completed or using the `onFailure` method if
+it failed.
+
+A typical listener for +{response}+ looks like:
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests-file}[{api}-request-listener]
+--------------------------------------------------
+<1> Called when the execution is successfully completed. The response is
+provided as an argument
+<2> Called in case of failure. The raised exception is provided as an argument