Add HLRC support for enrich execute policy API (#47991)

This PR also includes HLRC docs for the enrich stats api.

Relates to #32789

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

@@ -21,6 +21,8 @@ package org.elasticsearch.client;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.client.core.AcknowledgedResponse;
 import org.elasticsearch.client.enrich.DeletePolicyRequest;
+import org.elasticsearch.client.enrich.ExecutePolicyRequest;
+import org.elasticsearch.client.enrich.ExecutePolicyResponse;
 import org.elasticsearch.client.enrich.GetPolicyRequest;
 import org.elasticsearch.client.enrich.GetPolicyResponse;
 import org.elasticsearch.client.enrich.PutPolicyRequest;
@@ -224,4 +226,49 @@ public final class EnrichClient {
             Collections.emptySet()
         );
     }
+
+    /**
+     * Executes the execute policy api, which executes an enrich policy.
+     *
+     * See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/enrich-policy-apis.html#execute-policy">
+     * the docs</a> for more.
+     *
+     * @param request the {@link ExecutePolicyRequest}
+     * @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 ExecutePolicyResponse executePolicy(ExecutePolicyRequest request, RequestOptions options) throws IOException {
+        return restHighLevelClient.performRequestAndParseEntity(
+            request,
+            EnrichRequestConverters::executePolicy,
+            options,
+            ExecutePolicyResponse::fromXContent,
+            Collections.emptySet()
+        );
+    }
+
+    /**
+     * Asynchronously executes the execute policy api, which executes an enrich policy.
+     *
+     * See <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/enrich-policy-apis.html#execute-policy">
+     * the docs</a> for more.
+     *
+     * @param request the {@link ExecutePolicyRequest}
+     * @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
+     * @return cancellable that may be used to cancel the request
+     */
+    public Cancellable executePolicyAsync(ExecutePolicyRequest request,
+                                          RequestOptions options,
+                                          ActionListener<ExecutePolicyResponse> listener) {
+        return restHighLevelClient.performRequestAsyncAndParseEntity(
+            request,
+            EnrichRequestConverters::executePolicy,
+            options,
+            ExecutePolicyResponse::fromXContent,
+            listener,
+            Collections.emptySet()
+        );
+    }
 }

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

@@ -20,8 +20,10 @@ package org.elasticsearch.client;
 
 import org.apache.http.client.methods.HttpDelete;
 import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
 import org.apache.http.client.methods.HttpPut;
 import org.elasticsearch.client.enrich.DeletePolicyRequest;
+import org.elasticsearch.client.enrich.ExecutePolicyRequest;
 import org.elasticsearch.client.enrich.GetPolicyRequest;
 import org.elasticsearch.client.enrich.PutPolicyRequest;
 import org.elasticsearch.client.enrich.StatsRequest;
@@ -66,4 +68,17 @@ final class EnrichRequestConverters {
         return new Request(HttpGet.METHOD_NAME, endpoint);
     }
 
+    static Request executePolicy(ExecutePolicyRequest executePolicyRequest) {
+        String endpoint = new RequestConverters.EndpointBuilder()
+            .addPathPartAsIs("_enrich", "policy")
+            .addPathPart(executePolicyRequest.getName())
+            .addPathPartAsIs("_execute")
+            .build();
+        Request request = new Request(HttpPost.METHOD_NAME, endpoint);
+        if (executePolicyRequest.getWaitForCompletion() != null) {
+            request.addParameter("wait_for_completion", executePolicyRequest.getWaitForCompletion().toString());
+        }
+        return request;
+    }
+
 }

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

@@ -0,0 +1,43 @@
+/*
+ * 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.enrich;
+
+import org.elasticsearch.client.Validatable;
+
+public final class ExecutePolicyRequest implements Validatable {
+
+    private final String name;
+    private Boolean waitForCompletion;
+
+    public ExecutePolicyRequest(String name) {
+        this.name = name;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public Boolean getWaitForCompletion() {
+        return waitForCompletion;
+    }
+
+    public void setWaitForCompletion(boolean waitForCompletion) {
+        this.waitForCompletion = waitForCompletion;
+    }
+}

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

@@ -0,0 +1,85 @@
+/*
+ * 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.enrich;
+
+import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.xcontent.ConstructingObjectParser;
+import org.elasticsearch.common.xcontent.XContentParser;
+
+public final class ExecutePolicyResponse {
+
+    private static final ParseField TASK_FIELD = new ParseField("task");
+    private static final ParseField STATUS_FIELD = new ParseField("status");
+
+    private static final ConstructingObjectParser<ExecutePolicyResponse, Void> PARSER = new ConstructingObjectParser<>(
+        "execute_policy_response",
+        true,
+        args -> new ExecutePolicyResponse((String) args[0], (ExecutionStatus) args[1])
+    );
+
+    static {
+        PARSER.declareString(ConstructingObjectParser.optionalConstructorArg(), TASK_FIELD);
+        PARSER.declareObject(ConstructingObjectParser.optionalConstructorArg(), ExecutionStatus.PARSER, STATUS_FIELD);
+    }
+
+    public static ExecutePolicyResponse fromXContent(XContentParser parser) {
+        return PARSER.apply(parser, null);
+    }
+
+    private final String taskId;
+    private final ExecutionStatus executionStatus;
+
+    ExecutePolicyResponse(String taskId, ExecutionStatus executionStatus) {
+        this.taskId = taskId;
+        this.executionStatus = executionStatus;
+    }
+
+    public String getTaskId() {
+        return taskId;
+    }
+
+    public ExecutionStatus getExecutionStatus() {
+        return executionStatus;
+    }
+
+    public static final class ExecutionStatus {
+
+        private static final ParseField PHASE_FIELD = new ParseField("phase");
+
+        private static final ConstructingObjectParser<ExecutionStatus, Void> PARSER = new ConstructingObjectParser<>(
+            "execution_status",
+            true,
+            args -> new ExecutionStatus((String) args[0])
+        );
+
+        static {
+            PARSER.declareString(ConstructingObjectParser.constructorArg(), PHASE_FIELD);
+        }
+
+        private final String phase;
+
+        ExecutionStatus(String phase) {
+            this.phase = phase;
+        }
+
+        public String getPhase() {
+            return phase;
+        }
+    }
+}

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

@@ -20,13 +20,17 @@ package org.elasticsearch.client;
 
 import org.elasticsearch.client.core.AcknowledgedResponse;
 import org.elasticsearch.client.enrich.DeletePolicyRequest;
+import org.elasticsearch.client.enrich.ExecutePolicyRequest;
+import org.elasticsearch.client.enrich.ExecutePolicyResponse;
 import org.elasticsearch.client.enrich.GetPolicyRequest;
 import org.elasticsearch.client.enrich.GetPolicyResponse;
 import org.elasticsearch.client.enrich.PutPolicyRequest;
 import org.elasticsearch.client.enrich.StatsRequest;
 import org.elasticsearch.client.enrich.StatsResponse;
+import org.elasticsearch.client.indices.CreateIndexRequest;
 
 import java.util.List;
+import java.util.Map;
 
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
@@ -36,6 +40,10 @@ import static org.hamcrest.Matchers.notNullValue;
 public class EnrichIT extends ESRestHighLevelClientTestCase {
 
     public void testCRUD() throws Exception {
+        CreateIndexRequest createIndexRequest = new CreateIndexRequest("my-index")
+            .mapping(Map.of("properties", Map.of("enrich_key", Map.of("type", "keyword"))));
+        highLevelClient().indices().create(createIndexRequest, RequestOptions.DEFAULT);
+
         final EnrichClient enrichClient = highLevelClient().enrich();
         PutPolicyRequest putPolicyRequest =
             new PutPolicyRequest("my-policy", "match", List.of("my-index"), "enrich_key", List.of("enrich_value"));
@@ -60,6 +68,11 @@ public class EnrichIT extends ESRestHighLevelClientTestCase {
         assertThat(statsResponse.getCoordinatorStats().get(0).getRemoteRequestsTotal(), greaterThanOrEqualTo(0L));
         assertThat(statsResponse.getCoordinatorStats().get(0).getExecutedSearchesTotal(), greaterThanOrEqualTo(0L));
 
+        ExecutePolicyRequest executePolicyRequest = new ExecutePolicyRequest("my-policy");
+        ExecutePolicyResponse executePolicyResponse =
+            execute(executePolicyRequest, enrichClient::executePolicy, enrichClient::executePolicyAsync);
+        assertThat(executePolicyResponse.getExecutionStatus().getPhase(), equalTo("COMPLETE"));
+
         DeletePolicyRequest deletePolicyRequest = new DeletePolicyRequest("my-policy");
         AcknowledgedResponse deletePolicyResponse =
             execute(deletePolicyRequest, enrichClient::deletePolicy, enrichClient::deletePolicyAsync);

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

@@ -20,8 +20,10 @@ package org.elasticsearch.client;
 
 import org.apache.http.client.methods.HttpDelete;
 import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
 import org.apache.http.client.methods.HttpPut;
 import org.elasticsearch.client.enrich.DeletePolicyRequest;
+import org.elasticsearch.client.enrich.ExecutePolicyRequest;
 import org.elasticsearch.client.enrich.GetPolicyRequest;
 import org.elasticsearch.client.enrich.PutPolicyRequest;
 import org.elasticsearch.client.enrich.PutPolicyRequestTests;
@@ -89,4 +91,24 @@ public class EnrichRequestConvertersTests extends ESTestCase {
         assertThat(result.getEntity(), nullValue());
     }
 
+    public void testExecutePolicy() {
+        ExecutePolicyRequest request = new ExecutePolicyRequest(randomAlphaOfLength(4));
+        Request result = EnrichRequestConverters.executePolicy(request);
+
+        assertThat(result.getMethod(), equalTo(HttpPost.METHOD_NAME));
+        assertThat(result.getEndpoint(), equalTo("/_enrich/policy/" + request.getName() + "/_execute"));
+        assertThat(result.getParameters().size(), equalTo(0));
+        assertThat(result.getEntity(), nullValue());
+
+        request = new ExecutePolicyRequest(randomAlphaOfLength(4));
+        request.setWaitForCompletion(randomBoolean());
+        result = EnrichRequestConverters.executePolicy(request);
+
+        assertThat(result.getMethod(), equalTo(HttpPost.METHOD_NAME));
+        assertThat(result.getEndpoint(), equalTo("/_enrich/policy/" + request.getName() + "/_execute"));
+        assertThat(result.getParameters().size(), equalTo(1));
+        assertThat(result.getParameters().get("wait_for_completion"), equalTo(request.getWaitForCompletion().toString()));
+        assertThat(result.getEntity(), nullValue());
+    }
+
 }

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

@@ -25,6 +25,8 @@ import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.RestHighLevelClient;
 import org.elasticsearch.client.core.AcknowledgedResponse;
 import org.elasticsearch.client.enrich.DeletePolicyRequest;
+import org.elasticsearch.client.enrich.ExecutePolicyRequest;
+import org.elasticsearch.client.enrich.ExecutePolicyResponse;
 import org.elasticsearch.client.enrich.NamedPolicy;
 import org.elasticsearch.client.enrich.GetPolicyRequest;
 import org.elasticsearch.client.enrich.GetPolicyResponse;
@@ -33,9 +35,11 @@ import org.elasticsearch.client.enrich.StatsRequest;
 import org.elasticsearch.client.enrich.StatsResponse;
 import org.elasticsearch.client.enrich.StatsResponse.CoordinatorStats;
 import org.elasticsearch.client.enrich.StatsResponse.ExecutingPolicy;
+import org.elasticsearch.client.indices.CreateIndexRequest;
 import org.junit.After;
 
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
@@ -152,7 +156,7 @@ public class EnrichDocumentationIT extends ESRestHighLevelClientTestCase {
         RestHighLevelClient client = highLevelClient();
 
         PutPolicyRequest putPolicyRequest = new PutPolicyRequest(
-            "users-policy", "exact_match", List.of("users"),
+            "users-policy", "match", List.of("users"),
             "email", List.of("address", "zip", "city", "state"));
         client.enrich().putPolicy(putPolicyRequest, RequestOptions.DEFAULT);
 
@@ -245,4 +249,59 @@ public class EnrichDocumentationIT extends ESRestHighLevelClientTestCase {
         assertTrue(latch.await(30L, TimeUnit.SECONDS));
     }
 
+    public void testExecutePolicy() throws Exception {
+        RestHighLevelClient client = highLevelClient();
+
+        {
+            CreateIndexRequest createIndexRequest = new CreateIndexRequest("users")
+                .mapping(Map.of("properties", Map.of("email", Map.of("type", "keyword"))));
+            client.indices().create(createIndexRequest, RequestOptions.DEFAULT);
+            PutPolicyRequest putPolicyRequest = new PutPolicyRequest(
+                "users-policy", "match", List.of("users"),
+                "email", List.of("address", "zip", "city", "state"));
+            client.enrich().putPolicy(putPolicyRequest, RequestOptions.DEFAULT);
+        }
+
+        // tag::enrich-execute-policy-request
+        ExecutePolicyRequest request =
+            new ExecutePolicyRequest("users-policy");
+        // end::enrich-execute-policy-request
+
+        // tag::enrich-execute-policy-execute
+        ExecutePolicyResponse response =
+            client.enrich().executePolicy(request, RequestOptions.DEFAULT);
+        // end::enrich-execute-policy-execute
+
+        // tag::enrich-execute-policy-response
+        ExecutePolicyResponse.ExecutionStatus status =
+            response.getExecutionStatus();
+        // end::enrich-execute-policy-response
+
+        // tag::enrich-execute-policy-execute-listener
+        ActionListener<ExecutePolicyResponse> listener = new ActionListener<>() {
+            @Override
+            public void onResponse(ExecutePolicyResponse response) { // <1>
+                ExecutePolicyResponse.ExecutionStatus status =
+                    response.getExecutionStatus();
+            }
+
+            @Override
+            public void onFailure(Exception e) {
+                // <2>
+            }
+        };
+        // end::enrich-execute-policy-execute-listener
+
+        // Replace the empty listener by a blocking listener in test
+        final CountDownLatch latch = new CountDownLatch(1);
+        listener = new LatchedActionListener<>(listener, latch);
+
+        // tag::enrich-execute-policy-execute-async
+        client.enrich().executePolicyAsync(request, RequestOptions.DEFAULT,
+            listener); // <1>
+        // end::enrich-execute-policy-execute-async
+
+        assertTrue(latch.await(30L, TimeUnit.SECONDS));
+    }
+
 }

client/rest-high-level/src/test/java/org/elasticsearch/client/enrich/ExecutePolicyResponseTests.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.enrich;
+
+import org.elasticsearch.client.AbstractResponseTestCase;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.common.xcontent.XContentType;
+import org.elasticsearch.tasks.TaskId;
+import org.elasticsearch.xpack.core.enrich.action.ExecuteEnrichPolicyAction;
+import org.elasticsearch.xpack.core.enrich.action.ExecuteEnrichPolicyStatus;
+
+import java.io.IOException;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.nullValue;
+
+public class ExecutePolicyResponseTests extends AbstractResponseTestCase<ExecuteEnrichPolicyAction.Response, ExecutePolicyResponse> {
+
+    @Override
+    protected ExecuteEnrichPolicyAction.Response createServerTestInstance(XContentType xContentType) {
+        if (randomBoolean()) {
+            return new ExecuteEnrichPolicyAction.Response(new ExecuteEnrichPolicyStatus(randomAlphaOfLength(4)));
+        } else {
+            return new ExecuteEnrichPolicyAction.Response(new TaskId(randomAlphaOfLength(4), randomNonNegativeLong()));
+        }
+    }
+
+    @Override
+    protected ExecutePolicyResponse doParseToClientInstance(XContentParser parser) throws IOException {
+        return ExecutePolicyResponse.fromXContent(parser);
+    }
+
+    @Override
+    protected void assertInstances(ExecuteEnrichPolicyAction.Response serverTestInstance, ExecutePolicyResponse clientInstance) {
+        if (serverTestInstance.getStatus() != null) {
+            assertThat(clientInstance.getExecutionStatus().getPhase(), equalTo(serverTestInstance.getStatus().getPhase()));
+            assertThat(clientInstance.getTaskId(), nullValue());
+        } else if (serverTestInstance.getTaskId() != null) {
+            assertThat(clientInstance.getTaskId(), equalTo(clientInstance.getTaskId()));
+            assertThat(clientInstance.getExecutionStatus(), nullValue());
+        } else {
+            assert false;
+        }
+    }
+}

docs/java-rest/high-level/enrich/execute_policy.asciidoc

@@ -0,0 +1,30 @@
+--
+:api: enrich-execute-policy
+:request: ExecutePolicyRequest
+:response: ExecutePolicyResponse
+--
+
+[id="{upid}-{api}"]
+=== Execute Policy API
+
+[id="{upid}-{api}-request"]
+==== Request
+
+The Execute Policy API allows to execute an enrich policy by name.
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests-file}[{api}-request]
+--------------------------------------------------
+
+[id="{upid}-{api}-response"]
+==== Response
+
+The returned +{response}+ includes either the status or task id.
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests-file}[{api}-response]
+--------------------------------------------------
+
+include::../execution.asciidoc[]

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

@@ -614,8 +614,10 @@ The Java High Level REST Client supports the following Enrich APIs:
 * <<{upid}-enrich-delete-policy>>
 * <<{upid}-enrich-get-policy>>
 * <<{upid}-enrich-stats-policy>>
+* <<{upid}-enrich-execute-policy>>
 
 include::enrich/put_policy.asciidoc[]
 include::enrich/delete_policy.asciidoc[]
 include::enrich/get_policy.asciidoc[]
 include::enrich/stats.asciidoc[]
+include::enrich/execute_policy.asciidoc[]