Fix thread safety in dissect processor by eliminating mutable shared state

data-prepper-plugins/dissect-processor/build.gradle

@@ -9,6 +9,7 @@
 
 plugins {
     id 'java'
+    id 'data-prepper.jmh'
 }
 
 
@@ -21,8 +22,17 @@ dependencies {
     testImplementation project(':data-prepper-plugins:log-generator-source')
     testImplementation project(':data-prepper-test:test-common')
     implementation libs.commons.lang3
+    jmh project(':data-prepper-api')
 }
 
 test {
     useJUnitPlatform()
+}
+
+tasks.withType(Zip).configureEach {
+    zip64 = true
+}
+
+jmh {
+    profilers = ['gc']
 }

data-prepper-plugins/dissect-processor/src/jmh/java/org/opensearch/dataprepper/plugins/processor/dissect/DissectProcessorBenchmark.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.dataprepper.plugins.processor.dissect;
+
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+
+@Fork(2)
+@Threads(4)
+@Warmup(iterations = 1, time = 2)
+@Measurement(iterations = 5, time = 10)
+public class DissectProcessorBenchmark {
+
+    @State(Scope.Benchmark)
+    public static class BenchmarkState {
+        Dissector dissector;
+        String input;
+
+        @Setup
+        public void setUp() {
+            dissector = new Dissector("%{timestamp} %{level} %{message}");
+            input = "2024-01-15 ERROR service crashed";
+        }
+    }
+
+    @Benchmark
+    @BenchmarkMode(Mode.Throughput)
+    public Object benchmark_dissect(BenchmarkState state) {
+        return state.dissector.dissectText(state.input);
+    }
+}

data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/Delimiter.java

@@ -11,48 +11,11 @@
 
 public class Delimiter {
     private final String delimiterString;
-    private int start = -1;
-    private int end = -1;
-    private Delimiter next = null;
-
-    private Delimiter prev = null;
 
     public Delimiter(String delimiterString) {
         this.delimiterString = delimiterString;
     }
 
-    public int getStart() {
-        return start;
-    }
-
-    public void setStart(int ind) {
-        start = ind;
-    }
-
-    public int getEnd() {
-        return end;
-    }
-
-    public void setEnd(int ind) {
-        end = ind;
-    }
-
-    public Delimiter getNext() {
-        return next;
-    }
-
-    public void setNext(Delimiter nextDelimiter) {
-        next = nextDelimiter;
-    }
-
-    public Delimiter getPrev() {
-        return prev;
-    }
-
-    public void setPrev(Delimiter prevDelimiter) {
-        prev = prevDelimiter;
-    }
-
     @Override
     public String toString() {
         return delimiterString;

data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/DissectProcessor.java

@@ -15,26 +15,21 @@
 import org.opensearch.dataprepper.metrics.PluginMetrics;
 import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin;
 import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor;
-import org.opensearch.dataprepper.model.annotations.SingleThread;
 import org.opensearch.dataprepper.model.event.Event;
 import org.opensearch.dataprepper.model.plugin.InvalidPluginConfigurationException;
 import org.opensearch.dataprepper.model.processor.AbstractProcessor;
 import org.opensearch.dataprepper.model.processor.Processor;
 import org.opensearch.dataprepper.model.record.Record;
-import org.opensearch.dataprepper.plugins.processor.dissect.Fields.Field;
 import org.opensearch.dataprepper.plugins.processor.mutateevent.TargetType;
 import org.opensearch.dataprepper.typeconverter.TypeConverter;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.Collection;
 import java.util.HashMap;
-import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 
-
-@SingleThread
 @DataPrepperPlugin(name = "dissect", pluginType = Processor.class, pluginConfigurationType = DissectProcessorConfig.class)
 public class DissectProcessor extends AbstractProcessor<Record<Event>, Record<Event>> {
     private static final Logger LOG = LoggerFactory.getLogger(DissectProcessor.class);
@@ -96,16 +91,15 @@ private void dissectField(Event event, String field){
         Dissector dissector = dissectorMap.get(field);
         boolean isDeleteSourceOnSuccessfulDissect = dissectConfig.isDeleteSourceRequested();
         String text = event.get(field, String.class);
-        if (dissector.dissectText(text)) {
-            List<Field> dissectedFields = dissector.getDissectedFields();
-            for(Field disectedField: dissectedFields) {
-                String dissectFieldName = disectedField.getKey();
-                Object dissectFieldValue = convertTargetType(dissectFieldName,disectedField.getValue());
-                event.put(disectedField.getKey(), dissectFieldValue);
-            }
-            if (isDeleteSourceOnSuccessfulDissect) {
-                event.delete(field);
-            }
+        Map<String, String> dissectedFields = dissector.dissectText(text);
+        if (dissectedFields == null) {
+            return;
+        }
+        for(Map.Entry<String, String> entry: dissectedFields.entrySet()) {
+            event.put(entry.getKey(), convertTargetType(entry.getKey(), entry.getValue()));
+        }
+        if (isDeleteSourceOnSuccessfulDissect) {
+            event.delete(field);
         }
     }
 
@@ -126,8 +120,6 @@ private Object convertTargetType(String fieldKey, String fieldValue){
         }
     }
 
-
-
     @Override
     public void prepareForShutdown() {