Do not wrap ingest processor exception with IAE (#48816)

The problem with wrapping here is that it converts any exception into an
IAE, which we treat as a client error (400 status) whereas the exception
being wrapped here could be a server error (e.g., NPE). This commit
stops wrapping all ingest processor exceptions as IAEs.

Author: jasontedor

modules/ingest-common/src/test/resources/rest-api-spec/test/ingest/210_pipeline_processor.yml

@@ -107,4 +107,4 @@ teardown:
       pipeline: "outer"
       body: {}
 - match: { error.root_cause.0.type: "exception" }
-- match: { error.root_cause.0.reason: "java.lang.IllegalArgumentException: java.lang.IllegalStateException: Cycle detected for pipeline: inner" }
+- match: { error.root_cause.0.reason: "java.lang.IllegalStateException: Cycle detected for pipeline: inner" }

modules/ingest-common/src/test/resources/rest-api-spec/test/ingest/90_simulate.yml

@@ -668,8 +668,8 @@ teardown:
         }
 - length: { docs: 1 }
 - length: { docs.0.processor_results: 1 }
-- match: { docs.0.processor_results.0.error.reason: "java.lang.IllegalArgumentException: java.lang.IllegalStateException: Cycle detected for pipeline: outer" }
-- match: { docs.0.processor_results.0.error.caused_by.caused_by.reason: "Cycle detected for pipeline: outer" }
+- match: { docs.0.processor_results.0.error.reason: "java.lang.IllegalStateException: Cycle detected for pipeline: outer" }
+- match: { docs.0.processor_results.0.error.caused_by.reason: "Cycle detected for pipeline: outer" }
 
 ---
 "Test verbose simulate with Pipeline Processor with Multiple Pipelines":

server/src/main/java/org/elasticsearch/ingest/CompoundProcessor.java

@@ -212,7 +212,7 @@ private ElasticsearchException newCompoundProcessorException(Exception e, String
             return (ElasticsearchException) e;
         }
 
-        ElasticsearchException exception = new ElasticsearchException(new IllegalArgumentException(e));
+        ElasticsearchException exception = new ElasticsearchException(e);
 
         if (processorType != null) {
             exception.addHeader("processor_type", processorType);

server/src/main/java/org/elasticsearch/ingest/TrackingResultProcessor.java

@@ -53,7 +53,7 @@ public void execute(IngestDocument ingestDocument, BiConsumer<IngestDocument, Ex
                 if (e instanceof ElasticsearchException) {
                     ElasticsearchException elasticsearchException = (ElasticsearchException) e;
                     //else do nothing, let the tracking processors throw the exception while recording the path up to the failure
-                    if (elasticsearchException.getCause().getCause() instanceof IllegalStateException) {
+                    if (elasticsearchException.getCause() instanceof IllegalStateException) {
                         if (ignoreFailure) {
                             processorResultList.add(new SimulateProcessorResult(pipelineProcessor.getTag(),
                                 new IngestDocument(ingestDocument), e));

server/src/test/java/org/elasticsearch/action/ingest/SimulateExecutionServiceTests.java

@@ -260,7 +260,7 @@ public void testExecuteItemWithFailure() throws Exception {
         assertThat(simulateDocumentBaseResult.getFailure(), instanceOf(RuntimeException.class));
         Exception exception = simulateDocumentBaseResult.getFailure();
         assertThat(exception, instanceOf(ElasticsearchException.class));
-        assertThat(exception.getMessage(), equalTo("java.lang.IllegalArgumentException: java.lang.RuntimeException: processor failed"));
+        assertThat(exception.getMessage(), equalTo("java.lang.RuntimeException: processor failed"));
     }
 
     public void testDropDocument() throws Exception {

server/src/test/java/org/elasticsearch/ingest/IngestServiceTests.java

@@ -644,9 +644,8 @@ public String getType() {
         final SetOnce<Boolean> failure = new SetOnce<>();
         final IndexRequest indexRequest = new IndexRequest("_index", "_type", "_id").source(emptyMap()).setPipeline(id);
         final BiConsumer<Integer, Exception> failureHandler = (slot, e) -> {
-            assertThat(e.getCause(), instanceOf(IllegalArgumentException.class));
-            assertThat(e.getCause().getCause(), instanceOf(IllegalStateException.class));
-            assertThat(e.getCause().getCause().getMessage(), equalTo("error"));
+            assertThat(e.getCause(), instanceOf(IllegalStateException.class));
+            assertThat(e.getCause().getMessage(), equalTo("error"));
             failure.set(true);
         };
 
@@ -936,7 +935,7 @@ public void testBulkRequestExecutionWithFailures() throws Exception {
         verify(requestItemErrorHandler, times(numIndexRequests)).accept(anyInt(), argThat(new ArgumentMatcher<Exception>() {
             @Override
             public boolean matches(final Object o) {
-                return ((Exception)o).getCause().getCause().equals(error);
+                return ((Exception)o).getCause().equals(error);
             }
         }));
         verify(completionHandler, times(1)).accept(Thread.currentThread(), null);

server/src/test/java/org/elasticsearch/ingest/TrackingResultProcessorTests.java

@@ -457,8 +457,7 @@ public void testActualPipelineProcessorWithCycle() throws Exception {
         Exception[] holder = new Exception[1];
         trackingProcessor.execute(ingestDocument, (result, e) -> holder[0] = e);
         ElasticsearchException exception = (ElasticsearchException) holder[0];
-        assertThat(exception.getCause(), instanceOf(IllegalArgumentException.class));
-        assertThat(exception.getCause().getCause(), instanceOf(IllegalStateException.class));
+        assertThat(exception.getCause(), instanceOf(IllegalStateException.class));
         assertThat(exception.getMessage(), containsString("Cycle detected for pipeline: pipeline1"));
     }