Add support for 'ack watch' to the HLRC.

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.AckWatchRequest;
+import org.elasticsearch.client.watcher.AckWatchResponse;
 import org.elasticsearch.protocol.xpack.watcher.DeleteWatchRequest;
 import org.elasticsearch.protocol.xpack.watcher.DeleteWatchResponse;
 import org.elasticsearch.protocol.xpack.watcher.PutWatchRequest;
@@ -91,4 +93,32 @@ public void deleteWatchAsync(DeleteWatchRequest request, RequestOptions options,
         restHighLevelClient.performRequestAsyncAndParseEntity(request, WatcherRequestConverters::deleteWatch, options,
             DeleteWatchResponse::fromXContent, listener, singleton(404));
     }
+
+    /**
+     * Acknowledges a watch.
+     * See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/watcher-api-ack-watch.html">
+     * the docs</a> for more information.
+     * @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 if there is a problem sending the request or parsing back the response
+     */
+    public AckWatchResponse ackWatch(AckWatchRequest request, RequestOptions options) throws IOException {
+        return restHighLevelClient.performRequestAndParseEntity(request, WatcherRequestConverters::ackWatch, options,
+            AckWatchResponse::fromXContent, emptySet());
+    }
+
+    /**
+     * Asynchronously acknowledges a watch.
+     * See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/watcher-api-ack-watch.html">
+     * the docs</a> for more information.
+     * @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 completion of the request
+     */
+    public void ackWatchAsync(AckWatchRequest request, RequestOptions options, ActionListener<AckWatchResponse> listener) {
+        restHighLevelClient.performRequestAsyncAndParseEntity(request, WatcherRequestConverters::ackWatch, options,
+            AckWatchResponse::fromXContent, listener, emptySet());
+    }
+
 }

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

@@ -23,6 +23,7 @@
 import org.apache.http.client.methods.HttpPut;
 import org.apache.http.entity.ByteArrayEntity;
 import org.apache.http.entity.ContentType;
+import org.elasticsearch.client.watcher.AckWatchRequest;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.protocol.xpack.watcher.DeleteWatchRequest;
 import org.elasticsearch.protocol.xpack.watcher.PutWatchRequest;
@@ -59,4 +60,17 @@ static Request deleteWatch(DeleteWatchRequest deleteWatchRequest) {
         Request request = new Request(HttpDelete.METHOD_NAME, endpoint);
         return request;
     }
+
+    public static Request ackWatch(AckWatchRequest ackWatchRequest) {
+        String endpoint = new RequestConverters.EndpointBuilder()
+            .addPathPartAsIs("_xpack")
+            .addPathPartAsIs("watcher")
+            .addPathPartAsIs("watch")
+            .addPathPart(ackWatchRequest.getWatchId())
+            .addPathPartAsIs("_ack")
+            .addCommaSeparatedPathParts(ackWatchRequest.getActionIds())
+            .build();
+        Request request = new Request(HttpPut.METHOD_NAME, endpoint);
+        return request;
+    }
 }

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

@@ -0,0 +1,96 @@
+/*
+ * 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.client.ValidationException;
+import org.elasticsearch.protocol.xpack.watcher.PutWatchRequest;
+
+import java.util.Locale;
+
+/**
+ * A request to explicitly acknowledge a watch.
+ */
+public class AckWatchRequest implements Validatable {
+
+    private final String watchId;
+    private final String[] actionIds;
+
+    public AckWatchRequest(String watchId, String... actionIds) {
+        validateIds(watchId, actionIds);
+        this.watchId = watchId;
+        this.actionIds = actionIds;
+    }
+
+    private void validateIds(String watchId, String... actionIds) {
+        ValidationException exception = new ValidationException();
+        if (watchId == null) {
+            exception.addValidationError("watch id is missing");
+        } else if (PutWatchRequest.isValidId(watchId) == false) {
+            exception.addValidationError("watch id contains whitespace");
+        }
+
+        if (actionIds != null) {
+            for (String actionId : actionIds) {
+                if (actionId == null) {
+                    exception.addValidationError(String.format(Locale.ROOT, "action id may not be null"));
+                } else if (PutWatchRequest.isValidId(actionId) == false) {
+                    exception.addValidationError(
+                        String.format(Locale.ROOT, "action id [%s] contains whitespace", actionId));
+                }
+            }
+        }
+
+        if (!exception.validationErrors().isEmpty()) {
+            throw exception;
+        }
+    }
+
+    /**
+     * @return The ID of the watch to be acked.
+     */
+    public String getWatchId() {
+        return watchId;
+    }
+
+    /**
+     * @return The IDs of the actions to be acked. If omitted,
+     * all actions for the given watch will be acknowledged.
+     */
+    public String[] getActionIds() {
+        return actionIds;
+    }
+
+    @Override
+    public String toString() {
+        StringBuilder sb = new StringBuilder("ack [").append(watchId).append("]");
+        if (actionIds.length > 0) {
+            sb.append("[");
+            for (int i = 0; i < actionIds.length; i++) {
+                if (i > 0) {
+                    sb.append(", ");
+                }
+                sb.append(actionIds[i]);
+            }
+            sb.append("]");
+        }
+        return sb.toString();
+    }
+}

client/rest-high-level/src/main/java/org/elasticsearch/client/watcher/AckWatchResponse.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.common.ParseField;
+import org.elasticsearch.common.xcontent.ConstructingObjectParser;
+import org.elasticsearch.common.xcontent.XContentParser;
+
+import java.io.IOException;
+
+/**
+ * The response from an 'ack watch' request.
+ */
+public class AckWatchResponse {
+
+    private final WatchStatus status;
+
+    public AckWatchResponse(WatchStatus status) {
+        this.status = status;
+    }
+
+    /**
+     * @return the status of the requested watch. If an action was
+     * successfully acknowledged, this will be reflected in its status.
+     */
+    public WatchStatus getStatus() {
+        return status;
+    }
+
+    private static final ParseField STATUS_FIELD = new ParseField("status");
+    private static ConstructingObjectParser<AckWatchResponse, Void> PARSER =
+        new ConstructingObjectParser<>("ack_watch_response", true,
+            a -> new AckWatchResponse((WatchStatus) a[0]));
+
+    static {
+        PARSER.declareObject(ConstructingObjectParser.constructorArg(),
+            (parser, context) -> WatchStatus.parse(parser),
+            STATUS_FIELD);
+    }
+
+    public static AckWatchResponse fromXContent(XContentParser parser) throws IOException {
+        return PARSER.parse(parser, null);
+    }
+}

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

@@ -18,13 +18,19 @@
  */
 package org.elasticsearch.client;
 
+import org.elasticsearch.ElasticsearchStatusException;
+import org.elasticsearch.client.watcher.AckWatchRequest;
+import org.elasticsearch.client.watcher.AckWatchResponse;
+import org.elasticsearch.client.watcher.ActionStatus;
+import org.elasticsearch.client.watcher.ActionStatus.AckStatus;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.protocol.xpack.watcher.DeleteWatchRequest;
 import org.elasticsearch.protocol.xpack.watcher.DeleteWatchResponse;
 import org.elasticsearch.protocol.xpack.watcher.PutWatchRequest;
 import org.elasticsearch.protocol.xpack.watcher.PutWatchResponse;
+import org.elasticsearch.rest.RestStatus;
 
 import static org.hamcrest.Matchers.is;
 
@@ -72,4 +78,34 @@ public void testDeleteWatch() throws Exception {
         }
     }
 
+    public void testAckWatch() throws Exception {
+        String watchId = randomAlphaOfLength(10);
+        String actionId = "logme";
+
+        PutWatchResponse putWatchResponse = createWatch(watchId);
+        assertThat(putWatchResponse.isCreated(), is(true));
+
+        AckWatchResponse response = highLevelClient().watcher().ackWatch(
+            new AckWatchRequest(watchId, actionId), RequestOptions.DEFAULT);
+
+        ActionStatus actionStatus = response.getStatus().actionStatus(actionId);
+        assertEquals(AckStatus.State.AWAITS_SUCCESSFUL_EXECUTION, actionStatus.ackStatus().state());
+
+        // TODO: use the high-level REST client here once it supports 'execute watch'.
+        Request executeWatchRequest = new Request("POST", "_xpack/watcher/watch/" + watchId + "/_execute");
+        executeWatchRequest.setJsonEntity("{ \"record_execution\": true }");
+        Response executeResponse = client().performRequest(executeWatchRequest);
+        assertEquals(RestStatus.OK.getStatus(), executeResponse.getStatusLine().getStatusCode());
+
+        response = highLevelClient().watcher().ackWatch(
+            new AckWatchRequest(watchId, actionId), RequestOptions.DEFAULT);
+
+        actionStatus = response.getStatus().actionStatus(actionId);
+        assertEquals(AckStatus.State.ACKED, actionStatus.ackStatus().state());
+
+        ElasticsearchStatusException exception = expectThrows(ElasticsearchStatusException.class,
+            () -> highLevelClient().watcher().ackWatch(
+                new AckWatchRequest("nonexistent"), RequestOptions.DEFAULT));
+        assertEquals(RestStatus.NOT_FOUND, exception.status());
+    }
 }

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

@@ -21,6 +21,7 @@
 
 import org.apache.http.client.methods.HttpDelete;
 import org.apache.http.client.methods.HttpPut;
+import org.elasticsearch.client.watcher.AckWatchRequest;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.protocol.xpack.watcher.DeleteWatchRequest;
@@ -30,6 +31,7 @@
 import java.io.ByteArrayOutputStream;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.StringJoiner;
 
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.nullValue;
@@ -75,4 +77,24 @@ public void testDeleteWatch() {
         assertEquals("/_xpack/watcher/watch/" + watchId, request.getEndpoint());
         assertThat(request.getEntity(), nullValue());
     }
+
+    public void testAckWatch() {
+        String watchId = randomAlphaOfLength(10);
+        String[] actionIds = generateRandomStringArray(5, 10, false, true);
+
+        AckWatchRequest ackWatchRequest = new AckWatchRequest(watchId, actionIds);
+        Request request = WatcherRequestConverters.ackWatch(ackWatchRequest);
+
+        assertEquals(HttpPut.METHOD_NAME, request.getMethod());
+
+        StringJoiner expectedEndpoint = new StringJoiner("/", "/", "")
+            .add("_xpack").add("watcher").add("watch").add(watchId).add("_ack");
+        if (ackWatchRequest.getActionIds().length > 0) {
+            String actionsParam = String.join(",", ackWatchRequest.getActionIds());
+            expectedEndpoint.add(actionsParam);
+        }
+
+        assertEquals(expectedEndpoint.toString(), request.getEndpoint());
+        assertThat(request.getEntity(), nullValue());
+    }
 }