diff --git a/client/rest-high-level/src/main/java/org/elasticsearch/client/WatcherClient.java b/client/rest-high-level/src/main/java/org/elasticsearch/client/WatcherClient.java
index b1a3eb3f87bf..c182cc27e84e 100644
--- a/client/rest-high-level/src/main/java/org/elasticsearch/client/WatcherClient.java
+++ b/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 final class WatcherClient {
restHighLevelClient.performRequestAsyncAndParseEntity(request, WatcherRequestConverters::deleteWatch, options,
DeleteWatchResponse::fromXContent, listener, singleton(404));
}
+
+ /**
+ * Acknowledges a watch.
+ * See
+ * the docs 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
+ * the docs 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 listener) {
+ restHighLevelClient.performRequestAsyncAndParseEntity(request, WatcherRequestConverters::ackWatch, options,
+ AckWatchResponse::fromXContent, listener, emptySet());
+ }
+
}
diff --git a/client/rest-high-level/src/main/java/org/elasticsearch/client/WatcherRequestConverters.java b/client/rest-high-level/src/main/java/org/elasticsearch/client/WatcherRequestConverters.java
index 3b52d1c7b994..7a8fa19633ef 100644
--- a/client/rest-high-level/src/main/java/org/elasticsearch/client/WatcherRequestConverters.java
+++ b/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.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 @@ public class WatcherRequestConverters {
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;
+ }
}
diff --git a/client/rest-high-level/src/main/java/org/elasticsearch/client/watcher/AckWatchRequest.java b/client/rest-high-level/src/main/java/org/elasticsearch/client/watcher/AckWatchRequest.java
new file mode 100644
index 000000000000..1381544744da
--- /dev/null
+++ b/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();
+ }
+}
diff --git a/client/rest-high-level/src/main/java/org/elasticsearch/client/watcher/AckWatchResponse.java b/client/rest-high-level/src/main/java/org/elasticsearch/client/watcher/AckWatchResponse.java
new file mode 100644
index 000000000000..5c6750193a7c
--- /dev/null
+++ b/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 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);
+ }
+}
diff --git a/client/rest-high-level/src/test/java/org/elasticsearch/client/WatcherIT.java b/client/rest-high-level/src/test/java/org/elasticsearch/client/WatcherIT.java
index 491992735afb..4964fc4be50c 100644
--- a/client/rest-high-level/src/test/java/org/elasticsearch/client/WatcherIT.java
+++ b/client/rest-high-level/src/test/java/org/elasticsearch/client/WatcherIT.java
@@ -18,6 +18,11 @@
*/
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;
@@ -25,6 +30,7 @@ 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 class WatcherIT extends ESRestHighLevelClientTestCase {
}
}
+ 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());
+ }
}
diff --git a/client/rest-high-level/src/test/java/org/elasticsearch/client/WatcherRequestConvertersTests.java b/client/rest-high-level/src/test/java/org/elasticsearch/client/WatcherRequestConvertersTests.java
index cf5af1dd5949..d6227e93941e 100644
--- a/client/rest-high-level/src/test/java/org/elasticsearch/client/WatcherRequestConvertersTests.java
+++ b/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.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 org.elasticsearch.test.ESTestCase;
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 class WatcherRequestConvertersTests extends ESTestCase {
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());
+ }
}
diff --git a/client/rest-high-level/src/test/java/org/elasticsearch/client/documentation/WatcherDocumentationIT.java b/client/rest-high-level/src/test/java/org/elasticsearch/client/documentation/WatcherDocumentationIT.java
index 707997d1f310..48052f86a006 100644
--- a/client/rest-high-level/src/test/java/org/elasticsearch/client/documentation/WatcherDocumentationIT.java
+++ b/client/rest-high-level/src/test/java/org/elasticsearch/client/documentation/WatcherDocumentationIT.java
@@ -21,8 +21,15 @@ package org.elasticsearch.client.documentation;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.LatchedActionListener;
import org.elasticsearch.client.ESRestHighLevelClientTestCase;
+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.AckWatchRequest;
+import org.elasticsearch.client.watcher.AckWatchResponse;
+import org.elasticsearch.client.watcher.ActionStatus;
+import org.elasticsearch.client.watcher.ActionStatus.AckStatus;
+import org.elasticsearch.client.watcher.WatchStatus;
import org.elasticsearch.common.bytes.BytesArray;
import org.elasticsearch.common.bytes.BytesReference;
import org.elasticsearch.common.xcontent.XContentType;
@@ -30,6 +37,7 @@ 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 java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -132,4 +140,67 @@ public class WatcherDocumentationIT extends ESRestHighLevelClientTestCase {
}
}
+ public void testAckWatch() 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 putWatchRequest = new PutWatchRequest("my_watch_id", watch, XContentType.JSON);
+ client.watcher().putWatch(putWatchRequest, RequestOptions.DEFAULT);
+
+ // TODO: use the high-level REST client here once it supports 'execute watch'.
+ Request executeWatchRequest = new Request("POST", "_xpack/watcher/watch/my_watch_id/_execute");
+ executeWatchRequest.setJsonEntity("{ \"record_execution\": true }");
+ Response executeResponse = client().performRequest(executeWatchRequest);
+ assertEquals(RestStatus.OK.getStatus(), executeResponse.getStatusLine().getStatusCode());
+ }
+
+ {
+ //tag::ack-watch-execute
+ AckWatchRequest request = new AckWatchRequest("my_watch_id", // <1>
+ "logme", "emailme"); // <2>
+ AckWatchResponse response = client.watcher().ackWatch(request, RequestOptions.DEFAULT);
+ //end::ack-watch-execute
+
+ //tag::ack-watch-response
+ WatchStatus watchStatus = response.getStatus();
+ ActionStatus actionStatus = watchStatus.actionStatus("logme"); // <1>
+ AckStatus.State ackState = actionStatus.ackStatus().state(); // <2>
+ //end::ack-watch-response
+
+ assertEquals(AckStatus.State.ACKED, ackState);
+ }
+
+ {
+ AckWatchRequest request = new AckWatchRequest("my_watch_id");
+ // tag::ack-watch-execute-listener
+ ActionListener listener = new ActionListener() {
+ @Override
+ public void onResponse(AckWatchResponse response) {
+ // <1>
+ }
+
+ @Override
+ public void onFailure(Exception e) {
+ // <2>
+ }
+ };
+ // end::ack-watch-execute-listener
+
+ // For testing, replace the empty listener by a blocking listener.
+ final CountDownLatch latch = new CountDownLatch(1);
+ listener = new LatchedActionListener<>(listener, latch);
+
+ // tag::ack-watch-execute-async
+ client.watcher().ackWatchAsync(request, RequestOptions.DEFAULT, listener); // <1>
+ // end::ack-watch-execute-async
+
+ assertTrue(latch.await(30L, TimeUnit.SECONDS));
+ }
+ }
+
}
diff --git a/client/rest-high-level/src/test/java/org/elasticsearch/client/watcher/AckWatchResponseTests.java b/client/rest-high-level/src/test/java/org/elasticsearch/client/watcher/AckWatchResponseTests.java
new file mode 100644
index 000000000000..6d6d14e13ede
--- /dev/null
+++ b/client/rest-high-level/src/test/java/org/elasticsearch/client/watcher/AckWatchResponseTests.java
@@ -0,0 +1,105 @@
+/*
+ * 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 AckWatchResponse}.
+ *
+ * Note that we only sanity check watch status parsing here, as there
+ * are dedicated tests for it in {@link WatchStatusTests}.
+ */
+public class AckWatchResponseTests 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)
+ .endObject()
+ .endObject();
+ BytesReference bytes = BytesReference.bytes(builder);
+
+ AckWatchResponse response = parse(builder.contentType(), bytes);
+ WatchStatus status = response.getStatus();
+ assertNotNull(status);
+ assertEquals(42, status.version());
+ assertEquals(ExecutionState.ACKNOWLEDGED, status.getExecutionState());
+ }
+
+ 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)
+ .endObject()
+ .endObject();
+ BytesReference bytes = BytesReference.bytes(builder);
+
+ Predicate excludeFilter = field -> field.equals("status.actions");
+ BytesReference bytesWithRandomFields = XContentTestUtils.insertRandomFields(
+ builder.contentType(), bytes, excludeFilter, random());
+
+ AckWatchResponse response = parse(builder.contentType(), bytesWithRandomFields);
+ WatchStatus status = response.getStatus();
+ assertNotNull(status);
+ assertEquals(42, status.version());
+ assertEquals(ExecutionState.ACKNOWLEDGED, status.getExecutionState());
+ }
+
+ private AckWatchResponse parse(XContentType contentType, BytesReference bytes) throws IOException {
+ XContentParser parser = XContentFactory.xContent(contentType)
+ .createParser(NamedXContentRegistry.EMPTY, null, bytes.streamInput());
+ parser.nextToken();
+ return AckWatchResponse.fromXContent(parser);
+ }
+}
diff --git a/client/rest-high-level/src/test/java/org/elasticsearch/client/watcher/WatchRequestValidationTests.java b/client/rest-high-level/src/test/java/org/elasticsearch/client/watcher/WatchRequestValidationTests.java
new file mode 100644
index 000000000000..d75e36f7a366
--- /dev/null
+++ b/client/rest-high-level/src/test/java/org/elasticsearch/client/watcher/WatchRequestValidationTests.java
@@ -0,0 +1,98 @@
+/*
+ * 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.action.ActionRequestValidationException;
+import org.elasticsearch.client.ValidationException;
+import org.elasticsearch.common.bytes.BytesArray;
+import org.elasticsearch.common.xcontent.XContentType;
+import org.elasticsearch.protocol.xpack.watcher.DeleteWatchRequest;
+import org.elasticsearch.protocol.xpack.watcher.PutWatchRequest;
+import org.elasticsearch.test.ESTestCase;
+
+import java.util.Optional;
+
+import static org.hamcrest.Matchers.hasItem;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.notNullValue;
+
+public class WatchRequestValidationTests extends ESTestCase {
+
+ public void testAcknowledgeWatchInvalidWatchId() {
+ ValidationException e = expectThrows(ValidationException.class,
+ () -> new AckWatchRequest("id with whitespaces"));
+ assertThat(e.validationErrors(), hasItem("watch id contains whitespace"));
+ }
+
+ public void testAcknowledgeWatchInvalidActionId() {
+ ValidationException e = expectThrows(ValidationException.class,
+ () -> new AckWatchRequest("_id", "action id with whitespaces"));
+ assertThat(e.validationErrors(), hasItem("action id [action id with whitespaces] contains whitespace"));
+ }
+
+ public void testAcknowledgeWatchNullActionArray() {
+ // need this to prevent some compilation errors, i.e. in 1.8.0_91
+ String[] nullArray = null;
+ Optional e = new AckWatchRequest("_id", nullArray).validate();
+ assertFalse(e.isPresent());
+ }
+
+ public void testAcknowledgeWatchNullActionId() {
+ ValidationException e = expectThrows(ValidationException.class,
+ () -> new AckWatchRequest("_id", new String[] {null}));
+ assertThat(e.validationErrors(), hasItem("action id may not be null"));
+ }
+
+ public void testDeleteWatchInvalidWatchId() {
+ ActionRequestValidationException e = new DeleteWatchRequest("id with whitespaces").validate();
+ assertThat(e, is(notNullValue()));
+ assertThat(e.validationErrors(), hasItem("watch id contains whitespace"));
+ }
+
+ public void testDeleteWatchNullId() {
+ ActionRequestValidationException e = new DeleteWatchRequest(null).validate();
+ assertThat(e, is(notNullValue()));
+ assertThat(e.validationErrors(), hasItem("watch id is missing"));
+ }
+
+ public void testPutWatchInvalidWatchId() {
+ ActionRequestValidationException e = new PutWatchRequest("id with whitespaces", BytesArray.EMPTY, XContentType.JSON).validate();
+ assertThat(e, is(notNullValue()));
+ assertThat(e.validationErrors(), hasItem("watch id contains whitespace"));
+ }
+
+ public void testPutWatchNullId() {
+ ActionRequestValidationException e = new PutWatchRequest(null, BytesArray.EMPTY, XContentType.JSON).validate();
+ assertThat(e, is(notNullValue()));
+ assertThat(e.validationErrors(), hasItem("watch id is missing"));
+ }
+
+ public void testPutWatchSourceNull() {
+ ActionRequestValidationException e = new PutWatchRequest("foo", null, XContentType.JSON).validate();
+ assertThat(e, is(notNullValue()));
+ assertThat(e.validationErrors(), hasItem("watch source is missing"));
+ }
+
+ public void testPutWatchContentNull() {
+ ActionRequestValidationException e = new PutWatchRequest("foo", BytesArray.EMPTY, null).validate();
+ assertThat(e, is(notNullValue()));
+ assertThat(e.validationErrors(), hasItem("request body is missing"));
+ }
+}
diff --git a/docs/java-rest/high-level/supported-apis.asciidoc b/docs/java-rest/high-level/supported-apis.asciidoc
index cfac4a3c2938..bb326cbb9c66 100644
--- a/docs/java-rest/high-level/supported-apis.asciidoc
+++ b/docs/java-rest/high-level/supported-apis.asciidoc
@@ -310,9 +310,11 @@ The Java High Level REST Client supports the following Watcher APIs:
* <>
* <>
+* <>
include::watcher/put-watch.asciidoc[]
include::watcher/delete-watch.asciidoc[]
+include::watcher/ack-watch.asciidoc[]
== Graph APIs
diff --git a/docs/java-rest/high-level/watcher/ack-watch.asciidoc b/docs/java-rest/high-level/watcher/ack-watch.asciidoc
new file mode 100644
index 000000000000..13b62ba3be81
--- /dev/null
+++ b/docs/java-rest/high-level/watcher/ack-watch.asciidoc
@@ -0,0 +1,57 @@
+[[java-rest-high-watcher-ack-watch]]
+=== Ack Watch API
+
+[[java-rest-high-watcher-ack-watch-execution]]
+==== Execution
+
+{xpack-ref}/actions.html#actions-ack-throttle[Acknowledging a watch] enables you
+to manually throttle execution of a watch's actions. A watch can be acknowledged
+through the following request:
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests}/WatcherDocumentationIT.java[ack-watch-execute]
+--------------------------------------------------
+<1> The ID of the watch to ack.
+<2> An optional list of IDs representing the watch actions that should be acked.
+If no action IDs are provided, then all of the watch's actions will be acked.
+
+[[java-rest-high-watcher-ack-watch-response]]
+==== Response
+
+The returned `AckWatchResponse` contains the new status of the requested watch:
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests}/WatcherDocumentationIT.java[ack-watch-response]
+--------------------------------------------------
+<1> The status of a specific action that was acked.
+<2> The acknowledgement state of the action. If the action was successfully
+acked, this state will be equal to `AckStatus.State.ACKED`.
+
+[[java-rest-high-watcher-ack-watch-async]]
+==== Asynchronous Execution
+
+This request can be executed asynchronously:
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests}/WatcherDocumentationIT.java[ack-watch-execute-async]
+--------------------------------------------------
+<1> The `AckWatchRequest` to execute and the `ActionListener` to use when
+the execution completes.
+
+The asynchronous method does not block and returns immediately. Once the request
+completes, the `ActionListener` is called back using the `onResponse` method
+if the execution successfully completed or using the `onFailure` method if
+it failed.
+
+A listener for `AckWatchResponse` can be constructed as follows:
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests}/WatcherDocumentationIT.java[ack-watch-execute-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.