[9.1] (backport #345) Support terminate processor.

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 9.1.1
+  - Add `terminate` processor support [#345](https://github.com/elastic/logstash-filter-elastic_integration/pull/345)
+
 ## 9.1.0
   - Introduces `proxy` param to support proxy [#316](https://github.com/elastic/logstash-filter-elastic_integration/pull/316)
   - Embeds Ingest Node components from Elasticsearch 9.1

VERSION

@@ -1 +1 @@
-9.1.0
+9.1.1

spec/integration/elastic_integration_spec.rb

@@ -1134,6 +1134,35 @@ def path; @path; end
       # end
     end
 
+    describe 'with terminate processor' do
+      let(:pipeline_processor) {
+        '{
+          "terminate": {
+            "if": "ctx.error != null",
+            "tag": "terminated_ingest_pipeline"
+          }
+        },
+        {
+          "append": {
+            "field": "append_field",
+            "value": ["integration", "test"]
+          }
+        }'
+      }
+
+      it 'terminates the ingest pipeline' do
+        events = [LogStash::Event.new(
+          "message" => "Send message to pipeline which gets terminated.",
+          "error" => "This is intentionally placed error.",
+          "data_stream" => data_stream)]
+
+        subject.multi_filter(events).each do |event|
+          expect(event.get("[@metadata][target_ingest_pipeline]")).to include("_none")
+          # intentionally placed append processor to check if it is not executed after terminate processor
+          expect(event.get("append_field")).to be_nil
+        end
+      end
+    end
   end
 
   context '#multi-pipeline execution' do

src/main/java/co/elastic/logstash/filters/elasticintegration/EventProcessorBuilder.java

@@ -122,6 +122,7 @@ public EventProcessorBuilder() {
                 org.elasticsearch.ingest.common.SetProcessor.TYPE,
                 org.elasticsearch.ingest.common.SortProcessor.TYPE,
                 org.elasticsearch.ingest.common.SplitProcessor.TYPE,
+                "terminate", // note: upstream constant is package-private
                 org.elasticsearch.ingest.common.TrimProcessor.TYPE,
                 org.elasticsearch.ingest.common.URLDecodeProcessor.TYPE,
                 org.elasticsearch.ingest.common.UppercaseProcessor.TYPE,