REST high-level client: add simulate pipeline API

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

@@ -24,6 +24,8 @@
 import org.elasticsearch.action.ingest.GetPipelineRequest;
 import org.elasticsearch.action.ingest.GetPipelineResponse;
 import org.elasticsearch.action.ingest.PutPipelineRequest;
+import org.elasticsearch.action.ingest.SimulatePipelineRequest;
+import org.elasticsearch.action.ingest.SimulatePipelineResponse;
 import org.elasticsearch.action.ingest.WritePipelineResponse;
 
 import java.io.IOException;
@@ -125,4 +127,37 @@ public void deletePipelineAsync(DeletePipelineRequest request, RequestOptions op
         restHighLevelClient.performRequestAsyncAndParseEntity( request, RequestConverters::deletePipeline, options,
             WritePipelineResponse::fromXContent, listener, emptySet());
     }
+
+    /**
+     * Simulate a pipeline on a set of documents provided in the request
+     * <p>
+     * See
+     * <a href="https://www.elastic.co/guide/en/elasticsearch/reference/master/simulate-pipeline-api.html">
+     *     Simulate Pipeline API on elastic.co</a>
+     * @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 SimulatePipelineResponse simulatePipeline(SimulatePipelineRequest request, RequestOptions options) throws IOException {
+        return restHighLevelClient.performRequestAndParseEntity( request, RequestConverters::simulatePipeline, options,
+            SimulatePipelineResponse::fromXContent, emptySet());
+    }
+
+    /**
+     * Asynchronously simulate a pipeline on a set of documents provided in the request
+     * <p>
+     * See
+     * <a href="https://www.elastic.co/guide/en/elasticsearch/reference/master/simulate-pipeline-api.html">
+     *     Simulate Pipeline API on elastic.co</a>
+     * @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 simulatePipelineAsync(SimulatePipelineRequest request,
+                                      RequestOptions options,
+                                      ActionListener<SimulatePipelineResponse> listener) {
+        restHighLevelClient.performRequestAsyncAndParseEntity( request, RequestConverters::simulatePipeline, options,
+            SimulatePipelineResponse::fromXContent, listener, emptySet());
+    }
 }

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

@@ -71,6 +71,7 @@
 import org.elasticsearch.action.ingest.DeletePipelineRequest;
 import org.elasticsearch.action.ingest.PutPipelineRequest;
 import org.elasticsearch.action.ingest.GetPipelineRequest;
+import org.elasticsearch.action.ingest.SimulatePipelineRequest;
 import org.elasticsearch.action.search.ClearScrollRequest;
 import org.elasticsearch.action.search.MultiSearchRequest;
 import org.elasticsearch.action.search.SearchRequest;
@@ -886,6 +887,20 @@ static Request validateQuery(ValidateQueryRequest validateQueryRequest) throws I
         return request;
     }
 
+    static Request simulatePipeline(SimulatePipelineRequest simulatePipelineRequest) throws IOException {
+        EndpointBuilder builder = new EndpointBuilder().addPathPartAsIs("_ingest/pipeline");
+        if (simulatePipelineRequest.getId() != null && !simulatePipelineRequest.getId().isEmpty()) {
+            builder.addPathPart(simulatePipelineRequest.getId());
+        }
+        builder.addPathPartAsIs("_simulate");
+        String endpoint = builder.build();
+        Request request = new Request(HttpPost.METHOD_NAME, endpoint);
+        Params params = new Params(request);
+        params.putParam("verbose", Boolean.toString(simulatePipelineRequest.isVerbose()));
+        request.setEntity(createEntity(simulatePipelineRequest, REQUEST_BODY_CONTENT_TYPE));
+        return request;
+    }
+
     static Request getAlias(GetAliasesRequest getAliasesRequest) {
         String[] indices = getAliasesRequest.indices() == null ? Strings.EMPTY_ARRAY : getAliasesRequest.indices();
         String[] aliases = getAliasesRequest.aliases() == null ? Strings.EMPTY_ARRAY : getAliasesRequest.aliases();

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

@@ -85,9 +85,7 @@ private HighLevelClient(RestClient restClient) {
         }
     }
 
-    protected static XContentBuilder buildRandomXContentPipeline() throws IOException {
-        XContentType xContentType = randomFrom(XContentType.values());
-        XContentBuilder pipelineBuilder = XContentBuilder.builder(xContentType.xContent());
+    protected static XContentBuilder buildRandomXContentPipeline(XContentBuilder pipelineBuilder) throws IOException {
         pipelineBuilder.startObject();
         {
             pipelineBuilder.field(Pipeline.DESCRIPTION_KEY, "some random set of processors");
@@ -114,6 +112,12 @@ protected static XContentBuilder buildRandomXContentPipeline() throws IOExceptio
         return pipelineBuilder;
     }
 
+    protected static XContentBuilder buildRandomXContentPipeline() throws IOException {
+        XContentType xContentType = randomFrom(XContentType.values());
+        XContentBuilder pipelineBuilder = XContentBuilder.builder(xContentType.xContent());
+        return buildRandomXContentPipeline(pipelineBuilder);
+    }
+
     protected static void createPipeline(String pipelineId) throws IOException {
         XContentBuilder builder = buildRandomXContentPipeline();
         createPipeline(new PutPipelineRequest(pipelineId, BytesReference.bytes(builder), builder.contentType()));

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

@@ -23,12 +23,22 @@
 import org.elasticsearch.action.ingest.GetPipelineRequest;
 import org.elasticsearch.action.ingest.GetPipelineResponse;
 import org.elasticsearch.action.ingest.PutPipelineRequest;
+import org.elasticsearch.action.ingest.SimulateDocumentBaseResult;
+import org.elasticsearch.action.ingest.SimulateDocumentResult;
+import org.elasticsearch.action.ingest.SimulateDocumentVerboseResult;
+import org.elasticsearch.action.ingest.SimulatePipelineRequest;
+import org.elasticsearch.action.ingest.SimulatePipelineResponse;
 import org.elasticsearch.action.ingest.WritePipelineResponse;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.ingest.PipelineConfiguration;
 
 import java.io.IOException;
+import java.util.List;
+
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.core.IsInstanceOf.instanceOf;
 
 public class IngestClientIT extends ESRestHighLevelClientTestCase {
 
@@ -80,4 +90,93 @@ public void testDeletePipeline() throws IOException {
             execute(request, highLevelClient().ingest()::deletePipeline, highLevelClient().ingest()::deletePipelineAsync);
         assertTrue(response.isAcknowledged());
     }
+
+    public void testSimulatePipeline() throws IOException {
+        testSimulatePipeline(false, false);
+    }
+
+    public void testSimulatePipelineWithFailure() throws IOException {
+        testSimulatePipeline(false, true);
+    }
+
+    public void testSimulatePipelineVerbose() throws IOException {
+        testSimulatePipeline(true, false);
+    }
+
+    public void testSimulatePipelineVerboseWithFailure() throws IOException {
+        testSimulatePipeline(true, true);
+    }
+
+    private void testSimulatePipeline(boolean isVerbose,
+                                      boolean isFailure) throws IOException {
+        XContentType xContentType = randomFrom(XContentType.values());
+        XContentBuilder builder = XContentBuilder.builder(xContentType.xContent());
+        String rankValue = isFailure ? "non-int" : Integer.toString(1234);
+        builder.startObject();
+        {
+            builder.field("pipeline");
+            buildRandomXContentPipeline(builder);
+            builder.startArray("docs");
+            {
+                builder.startObject()
+                    .field("_index", "index")
+                    .field("_type", "doc")
+                    .field("_id", "doc_" + 1)
+                    .startObject("_source").field("foo", "rab_" + 1).field("rank", rankValue).endObject()
+                    .endObject();
+            }
+            builder.endArray();
+        }
+        builder.endObject();
+
+        SimulatePipelineRequest request = new SimulatePipelineRequest(
+            BytesReference.bytes(builder),
+            builder.contentType()
+        );
+        request.setVerbose(isVerbose);
+        SimulatePipelineResponse response =
+            execute(request, highLevelClient().ingest()::simulatePipeline, highLevelClient().ingest()::simulatePipelineAsync);
+        List<SimulateDocumentResult> results = response.getResults();
+        assertEquals(1, results.size());
+        if (isVerbose) {
+            assertThat(results.get(0), instanceOf(SimulateDocumentVerboseResult.class));
+            SimulateDocumentVerboseResult verboseResult = (SimulateDocumentVerboseResult) results.get(0);
+            assertEquals(2, verboseResult.getProcessorResults().size());
+            if (isFailure) {
+                assertNotNull(verboseResult.getProcessorResults().get(1).getFailure());
+                assertThat(verboseResult.getProcessorResults().get(1).getFailure().getMessage(),
+                    containsString("unable to convert [non-int] to integer"));
+            } else {
+                assertEquals(
+                    verboseResult.getProcessorResults().get(0).getIngestDocument()
+                        .getFieldValue("foo", String.class),
+                    "bar"
+                );
+                assertEquals(
+                    Integer.valueOf(1234),
+                    verboseResult.getProcessorResults().get(1).getIngestDocument()
+                        .getFieldValue("rank", Integer.class)
+                );
+            }
+        } else {
+            assertThat(results.get(0), instanceOf(SimulateDocumentBaseResult.class));
+            SimulateDocumentBaseResult baseResult = (SimulateDocumentBaseResult)results.get(0);
+            if (isFailure) {
+                assertNotNull(baseResult.getFailure());
+                assertThat(baseResult.getFailure().getMessage(),
+                    containsString("unable to convert [non-int] to integer"));
+            } else {
+                assertNotNull(baseResult.getIngestDocument());
+                assertEquals(
+                    baseResult.getIngestDocument().getFieldValue("foo", String.class),
+                    "bar"
+                );
+                assertEquals(
+                    Integer.valueOf(1234),
+                    baseResult.getIngestDocument()
+                        .getFieldValue("rank", Integer.class)
+                );
+            }
+        }
+    }
 }

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

@@ -74,6 +74,7 @@
 import org.elasticsearch.action.ingest.DeletePipelineRequest;
 import org.elasticsearch.action.ingest.GetPipelineRequest;
 import org.elasticsearch.action.ingest.PutPipelineRequest;
+import org.elasticsearch.action.ingest.SimulatePipelineRequest;
 import org.elasticsearch.action.search.ClearScrollRequest;
 import org.elasticsearch.action.search.MultiSearchRequest;
 import org.elasticsearch.action.search.SearchRequest;
@@ -1534,6 +1535,34 @@ public void testDeletePipeline() {
         assertEquals(expectedParams, expectedRequest.getParameters());
     }
 
+    public void testSimulatePipeline() throws IOException {
+        String pipelineId = randomBoolean() ? "some_pipeline_id" : null;
+        boolean verbose = randomBoolean();
+        String json = "{\"pipeline\":{" +
+            "\"description\":\"_description\"," +
+            "\"processors\":[{\"set\":{\"field\":\"field2\",\"value\":\"_value\"}}]}," +
+            "\"docs\":[{\"_index\":\"index\",\"_type\":\"_doc\",\"_id\":\"id\",\"_source\":{\"foo\":\"rab\"}}]}";
+        SimulatePipelineRequest request = new SimulatePipelineRequest(
+            new BytesArray(json.getBytes(StandardCharsets.UTF_8)),
+            XContentType.JSON
+        );
+        request.setId(pipelineId);
+        request.setVerbose(verbose);
+        Map<String, String> expectedParams = new HashMap<>();
+        expectedParams.put("verbose", Boolean.toString(verbose));
+
+        Request expectedRequest = RequestConverters.simulatePipeline(request);
+        StringJoiner endpoint = new StringJoiner("/", "/", "");
+        endpoint.add("_ingest/pipeline");
+        if (pipelineId != null && !pipelineId.isEmpty())
+            endpoint.add(pipelineId);
+        endpoint.add("_simulate");
+        assertEquals(endpoint.toString(), expectedRequest.getEndpoint());
+        assertEquals(HttpPost.METHOD_NAME, expectedRequest.getMethod());
+        assertEquals(expectedParams, expectedRequest.getParameters());
+        assertToXContentBody(request, expectedRequest.getEntity());
+    }
+
     public void testClusterHealth() {
         ClusterHealthRequest healthRequest = new ClusterHealthRequest();
         Map<String, String> expectedParams = new HashMap<>();

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

@@ -25,6 +25,12 @@
 import org.elasticsearch.action.ingest.GetPipelineRequest;
 import org.elasticsearch.action.ingest.GetPipelineResponse;
 import org.elasticsearch.action.ingest.PutPipelineRequest;
+import org.elasticsearch.action.ingest.SimulateDocumentBaseResult;
+import org.elasticsearch.action.ingest.SimulateDocumentResult;
+import org.elasticsearch.action.ingest.SimulateDocumentVerboseResult;
+import org.elasticsearch.action.ingest.SimulatePipelineRequest;
+import org.elasticsearch.action.ingest.SimulatePipelineResponse;
+import org.elasticsearch.action.ingest.SimulateProcessorResult;
 import org.elasticsearch.action.ingest.WritePipelineResponse;
 import org.elasticsearch.client.ESRestHighLevelClientTestCase;
 import org.elasticsearch.client.RequestOptions;
@@ -277,4 +283,109 @@ public void onFailure(Exception e) {
         }
     }
 
+    public void testSimulatePipeline() throws IOException {
+        RestHighLevelClient client = highLevelClient();
+
+        {
+            // tag::simulate-pipeline-request
+            String source =
+                "{\"" +
+                    "pipeline\":{" +
+                        "\"description\":\"_description\"," +
+                        "\"processors\":[{\"set\":{\"field\":\"field2\",\"value\":\"_value\"}}]" +
+                    "}," +
+                    "\"docs\":[" +
+                        "{\"_index\":\"index\",\"_type\":\"_doc\",\"_id\":\"id\",\"_source\":{\"foo\":\"bar\"}}," +
+                        "{\"_index\":\"index\",\"_type\":\"_doc\",\"_id\":\"id\",\"_source\":{\"foo\":\"rab\"}}" +
+                    "]" +
+                "}";
+            SimulatePipelineRequest request = new SimulatePipelineRequest(
+                new BytesArray(source.getBytes(StandardCharsets.UTF_8)), // <1>
+                XContentType.JSON // <2>
+            );
+            // end::simulate-pipeline-request
+
+            // tag::simulate-pipeline-request-pipeline-id
+            request.setId("my-pipeline-id"); // <1>
+            // end::simulate-pipeline-request-pipeline-id
+
+            // For testing we set this back to null
+            request.setId(null);
+
+            // tag::simulate-pipeline-request-verbose
+            request.setVerbose(true); // <1>
+            // end::simulate-pipeline-request-verbose
+
+            // tag::simulate-pipeline-execute
+            SimulatePipelineResponse response = client.ingest().simulatePipeline(request, RequestOptions.DEFAULT); // <1>
+            // end::simulate-pipeline-execute
+
+            // tag::simulate-pipeline-response
+            for (SimulateDocumentResult result: response.getResults()) { // <1>
+                if (request.isVerbose()) {
+                    assert result instanceof SimulateDocumentVerboseResult;
+                    SimulateDocumentVerboseResult verboseResult = (SimulateDocumentVerboseResult)result; // <2>
+                    for (SimulateProcessorResult processorResult: verboseResult.getProcessorResults()) { // <3>
+                        processorResult.getIngestDocument(); // <4>
+                        processorResult.getFailure(); // <5>
+                    }
+                } else {
+                    assert result instanceof SimulateDocumentBaseResult;
+                    SimulateDocumentBaseResult baseResult = (SimulateDocumentBaseResult)result; // <6>
+                    baseResult.getIngestDocument(); // <7>
+                    baseResult.getFailure(); // <8>
+                }
+            }
+            // end::simulate-pipeline-response
+            assert(response.getResults().size() > 0);
+        }
+    }
+
+    public void testSimulatePipelineAsync() throws Exception {
+        RestHighLevelClient client = highLevelClient();
+
+        {
+            String source =
+                "{\"" +
+                    "pipeline\":{" +
+                    "\"description\":\"_description\"," +
+                    "\"processors\":[{\"set\":{\"field\":\"field2\",\"value\":\"_value\"}}]" +
+                    "}," +
+                    "\"docs\":[" +
+                    "{\"_index\":\"index\",\"_type\":\"_doc\",\"_id\":\"id\",\"_source\":{\"foo\":\"bar\"}}," +
+                    "{\"_index\":\"index\",\"_type\":\"_doc\",\"_id\":\"id\",\"_source\":{\"foo\":\"rab\"}}" +
+                    "]" +
+                    "}";
+            SimulatePipelineRequest request = new SimulatePipelineRequest(
+                new BytesArray(source.getBytes(StandardCharsets.UTF_8)),
+                XContentType.JSON
+            );
+
+            // tag::simulate-pipeline-execute-listener
+            ActionListener<SimulatePipelineResponse> listener =
+                new ActionListener<SimulatePipelineResponse>() {
+                    @Override
+                    public void onResponse(SimulatePipelineResponse response) {
+                        // <1>
+                    }
+
+                    @Override
+                    public void onFailure(Exception e) {
+                        // <2>
+                    }
+                };
+            // end::simulate-pipeline-execute-listener
+
+            // Replace the empty listener by a blocking listener in test
+            final CountDownLatch latch = new CountDownLatch(1);
+            listener = new LatchedActionListener<>(listener, latch);
+
+            // tag::simulate-pipeline-execute-async
+            client.ingest().simulatePipelineAsync(request, RequestOptions.DEFAULT, listener); // <1>
+            // end::simulate-pipeline-execute-async
+
+            assertTrue(latch.await(30L, TimeUnit.SECONDS));
+        }
+    }
+
 }