fix: no other processors should be executed after on_failure is called in a compound processor (#19545)

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

@@ -109,6 +109,7 @@ public class CompoundProcessor implements Processor {
                     throw compoundProcessorException;
                 } else {
                     executeOnFailure(ingestDocument, compoundProcessorException);
+                    break;
                 }
             }
         }

core/src/test/java/org/elasticsearch/ingest/CompoundProcessorTests.java

@@ -193,4 +193,17 @@ public class CompoundProcessorTests extends ESTestCase {
         assertThat(firstProcessor.getInvokedCounter(), equalTo(1));
         assertThat(secondProcessor.getInvokedCounter(), equalTo(1));
     }
+
+    public void testBreakOnFailure() throws Exception {
+        TestProcessor firstProcessor = new TestProcessor("id1", "first", ingestDocument -> {throw new RuntimeException("error1");});
+        TestProcessor secondProcessor = new TestProcessor("id2", "second", ingestDocument -> {throw new RuntimeException("error2");});
+        TestProcessor onFailureProcessor = new TestProcessor("id2", "on_failure", ingestDocument -> {});
+        CompoundProcessor pipeline = new CompoundProcessor(false, Arrays.asList(firstProcessor, secondProcessor),
+            Collections.singletonList(onFailureProcessor));
+        pipeline.execute(ingestDocument);
+        assertThat(firstProcessor.getInvokedCounter(), equalTo(1));
+        assertThat(secondProcessor.getInvokedCounter(), equalTo(0));
+        assertThat(onFailureProcessor.getInvokedCounter(), equalTo(1));
+
+    }
 }

modules/ingest-common/src/test/resources/rest-api-spec/test/ingest/50_on_failure.yaml

@@ -19,6 +19,11 @@
                   "target_field" : "date",
                   "formats" : ["yyyy"]
                 }
+              },
+              {
+                "uppercase" : {
+                  "field": "field1"
+                }
               }
             ],
             "on_failure" : [