Merge branch 'master' into enhancement/remove_node_client_setting

1 files changed, 44 insertions, 77 deletions

diff --git a/core/src/test/java/org/elasticsearch/action/ingest/SimulateExecutionServiceTests.java b/core/src/test/java/org/elasticsearch/action/ingest/SimulateExecutionServiceTests.java
index cf1cab2416..f66dfa81ea 100644
--- a/core/src/test/java/org/elasticsearch/action/ingest/SimulateExecutionServiceTests.java
+++ b/core/src/test/java/org/elasticsearch/action/ingest/SimulateExecutionServiceTests.java
@@ -31,10 +31,8 @@ import org.elasticsearch.threadpool.ThreadPool;
 import org.junit.After;
 import org.junit.Before;
 
-import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Collections;
-import java.util.List;
+import java.util.Map;
 
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.instanceOf;
@@ -46,7 +44,6 @@ public class SimulateExecutionServiceTests extends ESTestCase {
 
     private ThreadPool threadPool;
     private SimulateExecutionService executionService;
-    private Pipeline pipeline;
     private Processor processor;
     private IngestDocument ingestDocument;
 
@@ -59,7 +56,6 @@ public class SimulateExecutionServiceTests extends ESTestCase {
         );
         executionService = new SimulateExecutionService(threadPool);
         processor = new TestProcessor("id", "mock", ingestDocument -> {});
-        pipeline = new Pipeline("_id", "_description", new CompoundProcessor(processor, processor));
         ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
     }
 
@@ -68,74 +64,6 @@ public class SimulateExecutionServiceTests extends ESTestCase {
         threadPool.shutdown();
     }
 
-    public void testExecuteVerboseDocumentSimple() throws Exception {
-        List<SimulateProcessorResult> processorResultList = new ArrayList<>();
-        executionService.executeVerboseDocument(processor, ingestDocument, processorResultList);
-        SimulateProcessorResult result = new SimulateProcessorResult("id", ingestDocument);
-        assertThat(processorResultList.size(), equalTo(1));
-        assertThat(processorResultList.get(0).getProcessorTag(), equalTo(result.getProcessorTag()));
-        assertThat(processorResultList.get(0).getIngestDocument(), equalTo(result.getIngestDocument()));
-        assertThat(processorResultList.get(0).getFailure(), nullValue());
-    }
-
-    public void testExecuteVerboseDocumentSimpleException() throws Exception {
-        RuntimeException exception = new RuntimeException("mock_exception");
-        TestProcessor processor = new TestProcessor("id", "mock", ingestDocument -> { throw exception; });
-        List<SimulateProcessorResult> processorResultList = new ArrayList<>();
-        try {
-            executionService.executeVerboseDocument(processor, ingestDocument, processorResultList);
-            fail("should throw exception");
-        } catch (RuntimeException e) {
-            assertThat(e.getMessage(), equalTo("mock_exception"));
-        }
-        SimulateProcessorResult result = new SimulateProcessorResult("id", exception);
-        assertThat(processorResultList.size(), equalTo(1));
-        assertThat(processorResultList.get(0).getProcessorTag(), equalTo(result.getProcessorTag()));
-        assertThat(processorResultList.get(0).getFailure(), equalTo(result.getFailure()));
-    }
-
-    public void testExecuteVerboseDocumentCompoundSuccess() throws Exception {
-        TestProcessor processor1 = new TestProcessor("p1", "mock", ingestDocument -> { });
-        TestProcessor processor2 = new TestProcessor("p2", "mock", ingestDocument -> { });
-
-        Processor compoundProcessor = new CompoundProcessor(processor1, processor2);
-        List<SimulateProcessorResult> processorResultList = new ArrayList<>();
-        executionService.executeVerboseDocument(compoundProcessor, ingestDocument, processorResultList);
-        assertThat(processor1.getInvokedCounter(), equalTo(1));
-        assertThat(processor2.getInvokedCounter(), equalTo(1));
-        assertThat(processorResultList.size(), equalTo(2));
-        assertThat(processorResultList.get(0).getProcessorTag(), equalTo("p1"));
-        assertThat(processorResultList.get(0).getIngestDocument(), equalTo(ingestDocument));
-        assertThat(processorResultList.get(0).getFailure(), nullValue());
-        assertThat(processorResultList.get(1).getProcessorTag(), equalTo("p2"));
-        assertThat(processorResultList.get(1).getIngestDocument(), equalTo(ingestDocument));
-        assertThat(processorResultList.get(1).getFailure(), nullValue());
-    }
-
-    public void testExecuteVerboseDocumentCompoundOnFailure() throws Exception {
-        TestProcessor processor1 = new TestProcessor("p1", "mock", ingestDocument -> { });
-        TestProcessor processor2 = new TestProcessor("p2", "mock", ingestDocument -> { throw new RuntimeException("p2_exception"); });
-        TestProcessor onFailureProcessor1 = new TestProcessor("fail_p1", "mock", ingestDocument -> { });
-        TestProcessor onFailureProcessor2 = new TestProcessor("fail_p2", "mock", ingestDocument -> { throw new RuntimeException("fail_p2_exception"); });
-        TestProcessor onFailureProcessor3 = new TestProcessor("fail_p3", "mock", ingestDocument -> { });
-        CompoundProcessor onFailureCompoundProcessor = new CompoundProcessor(Collections.singletonList(onFailureProcessor2), Collections.singletonList(onFailureProcessor3));
-
-        Processor compoundProcessor = new CompoundProcessor(Arrays.asList(processor1, processor2), Arrays.asList(onFailureProcessor1, onFailureCompoundProcessor));
-        List<SimulateProcessorResult> processorResultList = new ArrayList<>();
-        executionService.executeVerboseDocument(compoundProcessor, ingestDocument, processorResultList);
-        assertThat(processor1.getInvokedCounter(), equalTo(1));
-        assertThat(processor2.getInvokedCounter(), equalTo(1));
-        assertThat(onFailureProcessor1.getInvokedCounter(), equalTo(1));
-        assertThat(onFailureProcessor2.getInvokedCounter(), equalTo(1));
-        assertThat(onFailureProcessor3.getInvokedCounter(), equalTo(1));
-        assertThat(processorResultList.size(), equalTo(5));
-        assertThat(processorResultList.get(0).getProcessorTag(), equalTo("p1"));
-        assertThat(processorResultList.get(1).getProcessorTag(), equalTo("p2"));
-        assertThat(processorResultList.get(2).getProcessorTag(), equalTo("fail_p1"));
-        assertThat(processorResultList.get(3).getProcessorTag(), equalTo("fail_p2"));
-        assertThat(processorResultList.get(4).getProcessorTag(), equalTo("fail_p3"));
-    }
-
     public void testExecuteVerboseItem() throws Exception {
         TestProcessor processor = new TestProcessor("test-id", "mock", ingestDocument -> {});
         Pipeline pipeline = new Pipeline("_id", "_description", new CompoundProcessor(processor, processor));
@@ -170,16 +98,43 @@ public class SimulateExecutionServiceTests extends ESTestCase {
         assertThat(simulateDocumentBaseResult.getFailure(), nullValue());
     }
 
-    public void testExecuteVerboseItemWithFailure() throws Exception {
+    public void testExecuteVerboseItemExceptionWithoutOnFailure() throws Exception {
+        TestProcessor processor1 = new TestProcessor("processor_0", "mock", ingestDocument -> {});
+        TestProcessor processor2 = new TestProcessor("processor_1", "mock", ingestDocument -> { throw new RuntimeException("processor failed"); });
+        TestProcessor processor3 = new TestProcessor("processor_2", "mock", ingestDocument -> {});
+        Pipeline pipeline = new Pipeline("_id", "_description", new CompoundProcessor(processor1, processor2, processor3));
+        SimulateDocumentResult actualItemResponse = executionService.executeDocument(pipeline, ingestDocument, true);
+        assertThat(processor1.getInvokedCounter(), equalTo(1));
+        assertThat(processor2.getInvokedCounter(), equalTo(1));
+        assertThat(processor3.getInvokedCounter(), equalTo(0));
+        assertThat(actualItemResponse, instanceOf(SimulateDocumentVerboseResult.class));
+        SimulateDocumentVerboseResult simulateDocumentVerboseResult = (SimulateDocumentVerboseResult) actualItemResponse;
+        assertThat(simulateDocumentVerboseResult.getProcessorResults().size(), equalTo(2));
+        assertThat(simulateDocumentVerboseResult.getProcessorResults().get(0).getProcessorTag(), equalTo("processor_0"));
+        assertThat(simulateDocumentVerboseResult.getProcessorResults().get(0).getFailure(), nullValue());
+        assertThat(simulateDocumentVerboseResult.getProcessorResults().get(0).getIngestDocument(), not(sameInstance(ingestDocument)));
+        assertThat(simulateDocumentVerboseResult.getProcessorResults().get(0).getIngestDocument(), equalTo(ingestDocument));
+        assertThat(simulateDocumentVerboseResult.getProcessorResults().get(0).getIngestDocument().getSourceAndMetadata(), not(sameInstance(ingestDocument.getSourceAndMetadata())));
+        assertThat(simulateDocumentVerboseResult.getProcessorResults().get(1).getProcessorTag(), equalTo("processor_1"));
+        assertThat(simulateDocumentVerboseResult.getProcessorResults().get(1).getIngestDocument(), nullValue());
+        assertThat(simulateDocumentVerboseResult.getProcessorResults().get(1).getFailure(), instanceOf(RuntimeException.class));
+        RuntimeException runtimeException = (RuntimeException) simulateDocumentVerboseResult.getProcessorResults().get(1).getFailure();
+        assertThat(runtimeException.getMessage(), equalTo("processor failed"));
+    }
+
+    public void testExecuteVerboseItemWithOnFailure() throws Exception {
         TestProcessor processor1 = new TestProcessor("processor_0", "mock", ingestDocument -> { throw new RuntimeException("processor failed"); });
         TestProcessor processor2 = new TestProcessor("processor_1", "mock", ingestDocument -> {});
-        Pipeline pipeline = new Pipeline("_id", "_description", new CompoundProcessor(Collections.singletonList(processor1), Collections.singletonList(processor2)));
+        TestProcessor processor3 = new TestProcessor("processor_2", "mock", ingestDocument -> {});
+        Pipeline pipeline = new Pipeline("_id", "_description",
+                new CompoundProcessor(new CompoundProcessor(Collections.singletonList(processor1),
+                                Collections.singletonList(processor2)), processor3));
         SimulateDocumentResult actualItemResponse = executionService.executeDocument(pipeline, ingestDocument, true);
         assertThat(processor1.getInvokedCounter(), equalTo(1));
         assertThat(processor2.getInvokedCounter(), equalTo(1));
         assertThat(actualItemResponse, instanceOf(SimulateDocumentVerboseResult.class));
         SimulateDocumentVerboseResult simulateDocumentVerboseResult = (SimulateDocumentVerboseResult) actualItemResponse;
-        assertThat(simulateDocumentVerboseResult.getProcessorResults().size(), equalTo(2));
+        assertThat(simulateDocumentVerboseResult.getProcessorResults().size(), equalTo(3));
         assertThat(simulateDocumentVerboseResult.getProcessorResults().get(0).getProcessorTag(), equalTo("processor_0"));
         assertThat(simulateDocumentVerboseResult.getProcessorResults().get(0).getIngestDocument(), nullValue());
         assertThat(simulateDocumentVerboseResult.getProcessorResults().get(0).getFailure(), instanceOf(RuntimeException.class));
@@ -187,8 +142,20 @@ public class SimulateExecutionServiceTests extends ESTestCase {
         assertThat(runtimeException.getMessage(), equalTo("processor failed"));
         assertThat(simulateDocumentVerboseResult.getProcessorResults().get(1).getProcessorTag(), equalTo("processor_1"));
         assertThat(simulateDocumentVerboseResult.getProcessorResults().get(1).getIngestDocument(), not(sameInstance(ingestDocument)));
-        assertThat(simulateDocumentVerboseResult.getProcessorResults().get(1).getIngestDocument(), equalTo(ingestDocument));
+
+        IngestDocument ingestDocumentWithOnFailureMetadata = new IngestDocument(ingestDocument);
+        Map<String, String> metadata = ingestDocumentWithOnFailureMetadata.getIngestMetadata();
+        metadata.put(CompoundProcessor.ON_FAILURE_PROCESSOR_TYPE_FIELD, "mock");
+        metadata.put(CompoundProcessor.ON_FAILURE_PROCESSOR_TAG_FIELD, "processor_0");
+        metadata.put(CompoundProcessor.ON_FAILURE_MESSAGE_FIELD, "processor failed");
+        assertThat(simulateDocumentVerboseResult.getProcessorResults().get(1).getIngestDocument(), equalTo(ingestDocumentWithOnFailureMetadata));
+
         assertThat(simulateDocumentVerboseResult.getProcessorResults().get(1).getFailure(), nullValue());
+
+        assertThat(simulateDocumentVerboseResult.getProcessorResults().get(2).getProcessorTag(), equalTo("processor_2"));
+        assertThat(simulateDocumentVerboseResult.getProcessorResults().get(2).getIngestDocument(), not(sameInstance(ingestDocument)));
+        assertThat(simulateDocumentVerboseResult.getProcessorResults().get(2).getIngestDocument(), equalTo(ingestDocument));
+        assertThat(simulateDocumentVerboseResult.getProcessorResults().get(2).getFailure(), nullValue());
     }
 
     public void testExecuteItemWithFailure() throws Exception {