diff --git a/data-prepper-plugins/dissect-processor/README.md b/data-prepper-plugins/dissect-processor/README.md
new file mode 100644
index 0000000000..84bd7c286e
--- /dev/null
+++ b/data-prepper-plugins/dissect-processor/README.md
@@ -0,0 +1,129 @@
+# Dissect Processor
+
+The Dissect processor is useful when dealing with log files or messages that have a known pattern or structure. It extracts specific pieces of information from the text and map them to individual fields based on the defined Dissect patterns.
+
+
+## Basic Usage
+
+To get started with dissect processor using Data Prepper, create the following `pipeline.yaml`.
+```yaml
+dissect-pipeline:
+  source:
+    file:
+      path: "/full/path/to/dissect_logs_json.log"
+      record_type: "event"
+      format: "json"
+  processor:
+    - dissect:
+        map:
+          log: "%{Date} %{Time} %{Log_Type}: %{Message}"
+  sink:
+    - stdout:
+```
+
+Create the following file named `dissect_logs_json.log` and replace the `path` in the file source of your `pipeline.yaml` with the path of this file.
+
+```
+{"log": "07-25-2023 10:00:00 ERROR: Some error"}
+```
+
+The Dissect processor will retrieve the necessary fields from the `log` message, such as `Date`, `Time`, `Log_Type`, and `Message`, with the help of the pattern `%{Date} %{Time} %{Type}: %{Message}`, configured in the pipeline.
+
+When you run Data Prepper with this `pipeline.yaml` passed in, you should see the following standard output.
+
+```
+{
+    "log" : "07-25-2023 10:00:00 ERROR: Some error",
+    "Date" : "07-25-2023"
+    "Time" : "10:00:00"
+    "Log_Type" : "ERROR"
+    "Message" : "Some error"
+}
+```
+
+The fields `Date`, `Time`, `Log_Type`, and `Message` have been extracted from `log` value.
+
+## Configuration
+* `map` (Required): `map` is required to specify the dissect patterns. It takes a `Map<String, String>` with fields as keys and respective dissect patterns as values.
+
+
+* `target_types` (Optional): A `Map<String, String>` that specifies what the target type of specific field should be. Valid options are `integer`, `double`, `string`, and `boolean`. By default, all the values are `string`. Target types will be changed after the dissection process. 
+
+
+* `dissect_when` (Optional): A Data Prepper Expression string following the [Data Prepper Expression syntax](../../docs/expression_syntax.md). When configured, the processor will evaluate the expression before proceeding with the dissection process and perform the dissection if the expression evaluates to `true`.
+
+## Field Notations
+
+Symbols like `?, +, ->, /, &`  can be  used to perform logical extraction of data.
+
+* **Normal Field** : The field without a suffix or prefix. The field will be directly added to the output Event.
+
+    Ex: `%{field_name}`
+
+
+* **Skip Field** : ? can be used as a prefix to key to skip that field in the output JSON.  
+    * Skip Field : `%{}`  
+    * Named skip field  : `%{?field_name}` 
+
+    
+
+
+* **Append Field** : To append multiple values and put the final value in the field, we can use + before the field name in the dissect pattern
+    * **Usage**:
+
+            Pattern : "%{+field_name}, %{+field_name}"
+            Text : "foo, bar"  
+  
+            Output : {"field_name" : "foobar"}
+
+    We can also define the order the concatenation with the help of suffix `/<digits>` .
+
+    * **Usage**:
+
+            Pattern : "%{+field_name/2}, %{+field_name/1}"
+            Text : "foo, bar"
+        
+            Output : {"field_name" : "barfoo"}
+
+    If the order is not mentioned, the append operation will take place in the order of fields specified in the dissect pattern.<br><br>
+
+* **Indirect Field** : While defining a pattern, prefix the field with a `&` to assign the value found with this field to the value of another field found as the key.
+    * **Usage**:
+
+            Pattern : "%{?field_name}, %{&field_name}"
+            Text: "foo, bar"  
+    
+            Output : {“foo” : “bar”}
+
+  Here we can see that `foo` which was captured from the skip field `%{?field_name}` is made the key to value captured form the field `%{&field_name}`
+    * **Usage**:
+  
+            Pattern : %{field_name}, %{&field_name}
+            Text: "foo, bar"
+    
+            Output : {“field_name”:“foo”, “foo”:“bar”}
+
+    We can also indirectly assign the value to an appended field, along with `normal` field and `skip` field.
+
+### Padding
+
+* `->` operator can be used as a suffix to a field to indicate that white spaces after this field can be ignored.
+    * **Usage**:
+
+            Pattern : %{field1→} %{field2}
+            Text : “firstname               lastname”
+  
+            Output : {“field1” : “firstname”, “field2” : “lastname”}
+
+* This operator should be used as the right most suffix.
+    * **Usage**:
+
+          Pattern : %{fieldname/1->} %{fieldname/2}
+
+    If we use `->` before `/<digit>`, the `->` operator will also be considered part of the field name.
+
+
+## Developer Guide
+This plugin is compatible with Java 14. See
+- [CONTRIBUTING](https://github.com/opensearch-project/data-prepper/blob/main/CONTRIBUTING.md)
+- [monitoring](https://github.com/opensearch-project/data-prepper/blob/main/docs/monitoring.md)
diff --git a/data-prepper-plugins/dissect-processor/build.gradle b/data-prepper-plugins/dissect-processor/build.gradle
new file mode 100644
index 0000000000..5139ebbb07
--- /dev/null
+++ b/data-prepper-plugins/dissect-processor/build.gradle
@@ -0,0 +1,24 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+plugins {
+    id 'java'
+}
+
+
+dependencies {
+    implementation project(':data-prepper-api')
+    implementation 'com.fasterxml.jackson.dataformat:jackson-dataformat-yaml:2.15.0'
+    implementation 'io.micrometer:micrometer-core'
+    implementation project(path: ':data-prepper-api')
+    implementation project(path: ':data-prepper-plugins:mutate-event-processors')
+    testImplementation project(':data-prepper-plugins:log-generator-source')
+    testImplementation project(':data-prepper-test-common')
+    implementation 'org.apache.commons:commons-lang3:3.12.0'
+}
+
+test {
+    useJUnitPlatform()
+}
\ No newline at end of file
diff --git a/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/Delimiter.java b/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/Delimiter.java
new file mode 100644
index 0000000000..0654d6fd94
--- /dev/null
+++ b/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/Delimiter.java
@@ -0,0 +1,56 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.processor.dissect;
+
+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;
+    }
+}
diff --git a/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/DissectProcessor.java b/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/DissectProcessor.java
new file mode 100644
index 0000000000..5ff7f4ad56
--- /dev/null
+++ b/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/DissectProcessor.java
@@ -0,0 +1,122 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.processor.dissect;
+
+import org.opensearch.dataprepper.expression.ExpressionEvaluator;
+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.event.Event;
+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;
+
+import static org.opensearch.dataprepper.logging.DataPrepperMarkers.EVENT;
+
+
+@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);
+    private final DissectProcessorConfig dissectConfig;
+    private final Map<String, Dissector> dissectorMap = new HashMap<>();
+    private final Map<String, TargetType> targetTypeMap;
+    private final ExpressionEvaluator expressionEvaluator;
+
+    @DataPrepperPluginConstructor
+    public DissectProcessor(PluginMetrics pluginMetrics, final DissectProcessorConfig dissectConfig, final ExpressionEvaluator expressionEvaluator) {
+        super(pluginMetrics);
+        this.dissectConfig = dissectConfig;
+        this.expressionEvaluator = expressionEvaluator;
+        this.targetTypeMap = dissectConfig.getTargetTypes();
+
+        Map<String, String> patternsMap = dissectConfig.getMap();
+        for (String key : patternsMap.keySet()) {
+            Dissector dissector = new Dissector(patternsMap.get(key));
+            dissectorMap.put(key, dissector);
+        }
+
+    }
+
+    @Override
+    public Collection<Record<Event>> doExecute(Collection<Record<Event>> records) {
+        for (final Record<Event> record : records) {
+            Event event = record.getData();
+            String dissectWhen = dissectConfig.getDissectWhen();
+            if (Objects.nonNull(dissectWhen) && !expressionEvaluator.evaluateConditional(dissectWhen, event)) {
+                continue;
+            }
+            try{
+                for(String field: dissectorMap.keySet()){
+                    if(event.containsKey(field)){
+                        dissectField(event, field);
+                    }
+                }
+            }
+            catch (Exception ex){
+                LOG.error(EVENT, "Error dissecting the event [{}] ", record.getData(), ex);
+            }
+        }
+        return records;
+    }
+
+    private void dissectField(Event event, String field){
+        Dissector dissector = dissectorMap.get(field);
+        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);
+            }
+        }
+    }
+
+    private Object convertTargetType(String fieldKey, String fieldValue){
+        if(targetTypeMap == null){
+            return fieldValue;
+        }
+        try{
+            if(targetTypeMap.containsKey(fieldKey)){
+                TypeConverter converter = targetTypeMap.get(fieldKey).getTargetConverter();
+                return converter.convert(fieldValue);
+            } else {
+                return fieldValue;
+            }
+        } catch (NumberFormatException ex){
+            LOG.error("Unable to convert [{}] to the target type mentioned", fieldKey);
+            return fieldValue;
+        }
+    }
+
+
+
+    @Override
+    public void prepareForShutdown() {
+
+    }
+
+    @Override
+    public boolean isReadyForShutdown() {
+        return true;
+    }
+
+    @Override
+    public void shutdown() {
+
+    }
+}
diff --git a/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/DissectProcessorConfig.java b/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/DissectProcessorConfig.java
new file mode 100644
index 0000000000..bd1ca0d910
--- /dev/null
+++ b/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/DissectProcessorConfig.java
@@ -0,0 +1,28 @@
+package org.opensearch.dataprepper.plugins.processor.dissect;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import jakarta.validation.constraints.NotNull;
+import org.opensearch.dataprepper.plugins.processor.mutateevent.TargetType;
+
+import java.util.Map;
+
+public class DissectProcessorConfig {
+    @NotNull
+    @JsonProperty("map")
+    private Map<String, String> map;
+    @JsonProperty("target_types")
+    private Map<String, TargetType> targetTypes;
+    @JsonProperty("dissect_when")
+    private String dissectWhen;
+
+    public String getDissectWhen(){
+        return dissectWhen;
+    }
+
+    public Map<String, String> getMap() {
+        return map;
+    }
+
+    public Map<String, TargetType> getTargetTypes() { return targetTypes; }
+
+}
diff --git a/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/Dissector.java b/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/Dissector.java
new file mode 100644
index 0000000000..befb168f7e
--- /dev/null
+++ b/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/Dissector.java
@@ -0,0 +1,204 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.processor.dissect;
+
+import org.opensearch.dataprepper.plugins.processor.dissect.Fields.AppendField;
+import org.opensearch.dataprepper.plugins.processor.dissect.Fields.Field;
+import org.opensearch.dataprepper.plugins.processor.dissect.Fields.FieldHelper;
+import org.opensearch.dataprepper.plugins.processor.dissect.Fields.IndirectField;
+import org.opensearch.dataprepper.plugins.processor.dissect.Fields.NormalField;
+import org.opensearch.dataprepper.plugins.processor.dissect.Fields.SkipField;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+public class Dissector {
+    private Map<String, SkipField> skipFieldMap;
+    private Map<String, NormalField> normalFieldMap;
+    private Map<String, IndirectField> indirectFieldMap;
+    private Map<String, List<AppendField>> unAppendedFieldsMap;
+    private final FieldHelper fieldHelper = new FieldHelper();
+    private final LinkedList<Field> fieldsList = new LinkedList<>();
+    private final LinkedList<Delimiter> delimiterList = new LinkedList<>();
+    private static final Pattern DISSECT_PATTERN = Pattern.compile("%\\{([^}]*)}");
+
+    public Dissector(String dissectPatternString){
+        int maxLength = dissectPatternString.length()/3;
+        String[] delimiterArray = new String[maxLength];
+        String[] fieldsArray = new String[maxLength];
+
+        int lastIndex = 0;
+        int fieldIndex = 0;
+        int delimIndex = 0;
+
+        Matcher matcher = DISSECT_PATTERN.matcher(dissectPatternString);
+        while (matcher.find()) {
+            int matchStart = matcher.start();
+            int matchEnd = matcher.end();
+
+            delimiterArray[delimIndex] = dissectPatternString.substring(lastIndex, matchStart);
+            fieldsArray[fieldIndex] = matcher.group(1);
+
+            delimIndex++;
+            fieldIndex++;
+            lastIndex = matchEnd;
+        }
+
+        if (lastIndex < dissectPatternString.length()) {
+            delimiterArray[delimIndex] = dissectPatternString.substring(lastIndex);
+        }
+        parseFields(fieldsArray);
+        parseDelimiters(delimiterArray);
+        setFieldsMaps();
+    }
+
+    public boolean dissectText(String text){
+        try {
+            if (!setDelimiterIndexes(text)) {
+                return false;
+            }
+            Field head = fieldsList.getFirst();
+            for (final Delimiter delimiter : delimiterList) {
+                int fieldStart = 0;
+                int fieldEnd = delimiter.getStart();
+                if (delimiter.getPrev() == null && delimiter.getStart() == 0) {
+                    continue;
+                }
+                if (delimiter.getPrev() != null || delimiter.getStart() == 0) {
+                    fieldStart = delimiter.getPrev().getEnd() + 1;
+                }
+                head.setValue(text.substring(fieldStart, fieldEnd));
+                head = head.getNext();
+            }
+            if (delimiterList.getLast().getEnd() != text.length() - 1) {
+                int fieldStart = delimiterList.getLast().getEnd() + 1;
+                int fieldEnd = text.length();
+                head.setValue(text.substring(fieldStart, fieldEnd));
+            }
+            return true;
+        } catch (Exception e) {
+            return false;
+        }
+    }
+
+    public List<Field> getDissectedFields(){
+        final List<Field> dissectedFields = new ArrayList<>();
+        Map<String, AppendField> appendFieldMap = getAppendedFields(unAppendedFieldsMap);
+
+        dissectedFields.addAll(normalFieldMap.values());
+        dissectedFields.addAll(appendFieldMap.values());
+
+        for(final Field indirectField : indirectFieldMap.values()){
+            if(normalFieldMap.containsKey(indirectField.getKey())){
+                indirectField.setKey(normalFieldMap.get(indirectField.getKey()).getValue());
+            }
+            if(skipFieldMap.containsKey(indirectField.getKey())){
+                indirectField.setKey(skipFieldMap.get(indirectField.getKey()).getValue());
+            }
+            if(appendFieldMap.containsKey(indirectField.getKey())){
+                indirectField.setKey(appendFieldMap
+                                             .get(indirectField.getKey()).getValue());
+            }
+            dissectedFields.add(indirectField);
+        }
+
+        return dissectedFields;
+    }
+
+    private void setFieldsMaps(){
+        this.normalFieldMap = fieldHelper.getNormalFieldMap();
+        this.skipFieldMap = fieldHelper.getSkipFieldMap();
+        this.indirectFieldMap = fieldHelper.getIndirectFieldMap();
+        this.unAppendedFieldsMap = fieldHelper.getAppendFieldMap();
+    }
+
+    private void parseFields(String[] fieldsArray){
+        for(final String fieldString : fieldsArray){
+            if(fieldString==null) {
+                return;
+            }
+            Field field = fieldHelper.getField(fieldString);
+            if(fieldsList.size()==0) {
+                fieldsList.addLast(field);
+            }
+            else{
+                fieldsList.getLast().setNext(field);
+                fieldsList.addLast(field);
+            }
+        }
+    }
+
+    private void parseDelimiters(String[] delimiterArray) {
+        for (final String delimiterString : delimiterArray) {
+            if (delimiterString == null) {
+                return;
+            }
+            if (delimiterString.length() == 0) {
+                continue;
+            }
+            Delimiter delimiter = new Delimiter(delimiterString);
+            if (delimiterList.size() == 0) {
+                delimiterList.addLast(delimiter);
+            } else {
+                delimiterList.getLast().setNext(delimiter);
+                delimiter.setPrev(delimiterList.getLast());
+                delimiterList.addLast(delimiter);
+            }
+        }
+    }
+
+    private boolean setDelimiterIndexes(String text){
+        for (Delimiter delimiter : delimiterList) {
+            int prevEnd = 0;
+            if (delimiter.getPrev() != null) {
+                prevEnd = delimiter.getPrev().getEnd() + 1;
+            }
+            String delimiterString = delimiter.toString();
+            int start = text.indexOf(delimiterString, prevEnd);
+            if (delimiterString.trim().isEmpty()) {
+                start = start + findLastWhitespaceIndex(text.substring(start), delimiterString.length());
+            }
+            int end = start + delimiterString.length() -1;
+            if (start < 0 || end > text.length()) {
+                return false;
+            }
+            delimiter.setStart(start);
+            delimiter.setEnd(end);
+        }
+        return true;
+    }
+
+    private Map<String, AppendField> getAppendedFields(Map<String, List<AppendField>> unAppendedFieldsMap){
+        final Map<String, AppendField> appendFieldMap = new HashMap<>();
+        for(final String key : unAppendedFieldsMap.keySet()){
+            List<AppendField> appendFields = unAppendedFieldsMap.get(key);
+            Collections.sort(appendFields);
+            String value = appendFields.stream().map(AppendField::getValue).collect(Collectors.joining());
+            AppendField sortedField = new AppendField(key);
+            sortedField.setValue(value);
+            appendFieldMap.put(sortedField.getKey(), sortedField);
+        }
+        return appendFieldMap;
+    }
+
+    private int findLastWhitespaceIndex(String s, int w) {
+
+        final String[] leadingSpaces = s.split("\\S", 2);
+
+        if (leadingSpaces.length > 0 && leadingSpaces[0].length() >= w) {
+            return leadingSpaces[0].length() - w;
+        }
+
+        return 0;
+    }
+}
diff --git a/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/Fields/AppendField.java b/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/Fields/AppendField.java
new file mode 100644
index 0000000000..b032e251f3
--- /dev/null
+++ b/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/Fields/AppendField.java
@@ -0,0 +1,22 @@
+package org.opensearch.dataprepper.plugins.processor.dissect.Fields;
+
+public class AppendField extends Field implements Comparable<AppendField> {
+    private int index;
+    public AppendField(String key) {
+        this.setKey(key);
+    }
+
+    public void setIndex(int index){
+        this.index = index;
+    }
+
+    public int getIndex(){
+        return index;
+    }
+
+
+    @Override
+    public int compareTo(AppendField appendField) {
+        return this.index - appendField.getIndex();
+    }
+}
diff --git a/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/Fields/Field.java b/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/Fields/Field.java
new file mode 100644
index 0000000000..49e0c4852f
--- /dev/null
+++ b/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/Fields/Field.java
@@ -0,0 +1,38 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.processor.dissect.Fields;
+
+public abstract class Field {
+    boolean stripTrailing = false;
+    private String key;
+    private String value;
+    private Field next;
+
+    public String getValue() {
+        return value;
+    }
+
+    public String getKey() {
+        return key;
+    }
+
+    public Field getNext() {
+        return next;
+    }
+
+    public void setKey(String key) {
+        this.key = key;
+    }
+
+    public void setNext(Field next) {
+        this.next = next;
+    }
+
+    public void setValue(String value) {
+        this.value = stripTrailing ? value.stripTrailing() : value;
+    }
+
+}
diff --git a/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/Fields/FieldHelper.java b/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/Fields/FieldHelper.java
new file mode 100644
index 0000000000..eefbe4e50f
--- /dev/null
+++ b/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/Fields/FieldHelper.java
@@ -0,0 +1,112 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.processor.dissect.Fields;
+
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+public class FieldHelper {
+    private final Pattern appendPattern = Pattern.compile("^(.*?)->");
+    private final Pattern prefixPattern = Pattern.compile("([+&?])(.+)");
+    private final Pattern indexPattern = Pattern.compile("/(\\d+)$");
+    private final Map<String, SkipField> skipFieldMap = new HashMap<>();
+    private final Map<String, NormalField> normalFieldMap = new HashMap<>();
+    private final Map<String, IndirectField> indirectFieldMap = new HashMap<>();
+    private final Map<String, List<AppendField>> appendFieldMap = new HashMap<>();
+
+    public Map<String, SkipField> getSkipFieldMap(){
+        return skipFieldMap;
+    }
+    public Map<String, NormalField> getNormalFieldMap(){
+        return normalFieldMap;
+    }
+    public Map<String, IndirectField> getIndirectFieldMap(){
+        return indirectFieldMap;
+    }
+    public Map<String, List<AppendField>> getAppendFieldMap(){
+        return appendFieldMap;
+    }
+
+    public Field getField(String fieldString) {
+        if (fieldString == null) {
+            return null;
+        }
+        if(fieldString.trim().isEmpty()){
+            return new AppendField("");
+        }
+
+        Field field = null;
+
+        Matcher matcher = prefixPattern.matcher(fieldString);
+        if (matcher.matches()) {
+            final String notation = matcher.group(1);
+            final String key = matcher.group(2);
+            if (Objects.equals(notation, "+")) {
+                field = new AppendField(key);
+                setAppendIndex((AppendField) field);
+                setStripTrailing(field);
+                putInAppendMap((AppendField) field);
+            } else if (Objects.equals(notation, "?")) {
+                field = new SkipField(key);
+                setStripTrailing(field);
+                skipFieldMap.put(field.getKey(), (SkipField) field);
+            } else if (Objects.equals(notation, "&")) {
+                field = new IndirectField(key);
+                setStripTrailing(field);
+                indirectFieldMap.put(field.getKey(), (IndirectField) field);
+            }
+        } else {
+            field = new NormalField(fieldString);
+            setStripTrailing(field);
+            normalFieldMap.put(field.getKey(), (NormalField) field);
+        }
+        return field;
+    }
+
+    private void setAppendIndex(AppendField field) {
+        String fieldString = field.getKey();
+        Matcher matcher = indexPattern.matcher(fieldString);
+
+        if (matcher.find()) {
+            String key = fieldString.substring(0, matcher.start());
+            int index = Integer.parseInt(matcher.group(1));
+            field.setKey(key);
+            field.setIndex(index);
+        }
+    }
+
+    private void setStripTrailing(Field field) {
+        if (field == null) {
+            return;
+        }
+
+        String fieldString = field.getKey();
+
+        Matcher matcher = appendPattern.matcher(fieldString);
+        if (matcher.find()) {
+            field.setKey(matcher.group(1));
+            field.stripTrailing = true;
+        }
+    }
+
+    private void putInAppendMap(AppendField field) {
+        String key = field.getKey();
+        if (appendFieldMap.containsKey(key)) {
+            appendFieldMap.get(key).add(field);
+        } else {
+            List<AppendField> appendFields = new ArrayList<>();
+            appendFields.add(field);
+            appendFieldMap.put(key, appendFields);
+        }
+    }
+
+}
diff --git a/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/Fields/IndirectField.java b/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/Fields/IndirectField.java
new file mode 100644
index 0000000000..6d15492a5e
--- /dev/null
+++ b/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/Fields/IndirectField.java
@@ -0,0 +1,10 @@
+package org.opensearch.dataprepper.plugins.processor.dissect.Fields;
+
+public class IndirectField extends Field {
+
+
+    public IndirectField(String key) {
+        this.setKey(key);
+    }
+
+}
diff --git a/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/Fields/NormalField.java b/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/Fields/NormalField.java
new file mode 100644
index 0000000000..9552e16b32
--- /dev/null
+++ b/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/Fields/NormalField.java
@@ -0,0 +1,9 @@
+package org.opensearch.dataprepper.plugins.processor.dissect.Fields;
+
+public class NormalField extends Field {
+
+    public NormalField(String key) {
+        this.setKey(key);
+    }
+
+}
diff --git a/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/Fields/SkipField.java b/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/Fields/SkipField.java
new file mode 100644
index 0000000000..68e569f7c7
--- /dev/null
+++ b/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/Fields/SkipField.java
@@ -0,0 +1,9 @@
+package org.opensearch.dataprepper.plugins.processor.dissect.Fields;
+
+public class SkipField extends Field {
+
+    public SkipField(String key) {
+        this.setKey(key);
+    }
+
+}
diff --git a/data-prepper-plugins/dissect-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/dissect/DelimiterTest.java b/data-prepper-plugins/dissect-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/dissect/DelimiterTest.java
new file mode 100644
index 0000000000..e52da64fe6
--- /dev/null
+++ b/data-prepper-plugins/dissect-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/dissect/DelimiterTest.java
@@ -0,0 +1,49 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.processor.dissect;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+
+class DelimiterTest {
+
+    private static final String DELIMITER_STRING = "test_string";
+    private static final int START_END_INDEX = 3;
+    private static final Delimiter DELIMITER = new Delimiter("test_delimiter");
+    private Delimiter delimiter;
+
+    @BeforeEach
+    public void setUp() {
+        delimiter = new Delimiter(DELIMITER_STRING);
+    }
+
+    @Test
+    public void testSetAndGetStart() {
+        delimiter.setStart(START_END_INDEX);
+        assertThat(delimiter.getStart(), is(START_END_INDEX));
+    }
+
+    @Test
+    public void testSetAndGetEnd() {
+        delimiter.setEnd(START_END_INDEX);
+        assertThat(delimiter.getEnd(), is(START_END_INDEX));
+    }
+
+    @Test
+    public void testSetAndGetNext() {
+        delimiter.setNext(DELIMITER);
+        assertThat(delimiter.getNext(), is(DELIMITER));
+    }
+
+    @Test
+    public void testSetAndGetPrev() {
+        delimiter.setPrev(DELIMITER);
+        assertThat(delimiter.getPrev(), is(DELIMITER));
+    }
+}
\ No newline at end of file
diff --git a/data-prepper-plugins/dissect-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/dissect/DissectProcessorConfigTest.java b/data-prepper-plugins/dissect-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/dissect/DissectProcessorConfigTest.java
new file mode 100644
index 0000000000..01061ae5fc
--- /dev/null
+++ b/data-prepper-plugins/dissect-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/dissect/DissectProcessorConfigTest.java
@@ -0,0 +1,45 @@
+package org.opensearch.dataprepper.plugins.processor.dissect;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.opensearch.dataprepper.plugins.processor.mutateevent.TargetType;
+
+import java.util.Map;
+
+import static org.hamcrest.core.Is.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.opensearch.dataprepper.test.helper.ReflectivelySetField.setField;
+
+class DissectProcessorConfigTest {
+    private DissectProcessorConfig dissectProcessorConfig;
+    private DissectProcessorConfig createObjectUnderTest() {
+        return new DissectProcessorConfig();
+    }
+
+    @BeforeEach
+    void setup(){
+        dissectProcessorConfig = createObjectUnderTest();
+    }
+
+    @Test
+    void test_get_map() throws NoSuchFieldException, IllegalAccessException {
+        Map<String, String> dissectMap = Map.of("key1", "%{field1}");
+        setField(DissectProcessorConfig.class, dissectProcessorConfig, "map", dissectMap);
+        assertThat(dissectProcessorConfig.getMap(), is(dissectMap));
+    }
+
+    @Test
+    void test_get_dissect_when() throws NoSuchFieldException, IllegalAccessException {
+        String dissectWhen = "/test!=null";
+        setField(DissectProcessorConfig.class, dissectProcessorConfig, "dissectWhen", dissectWhen);
+        assertThat(dissectProcessorConfig.getDissectWhen(), is(dissectWhen));
+    }
+
+    @Test
+    void test_get_targets_types() throws NoSuchFieldException, IllegalAccessException {
+        Map<String, TargetType> targetTypeMap = Map.of("field1", TargetType.INTEGER);
+        setField(DissectProcessorConfig.class, dissectProcessorConfig, "targetTypes", targetTypeMap);
+        assertThat(dissectProcessorConfig.getTargetTypes(), is(targetTypeMap));
+    }
+
+}
\ No newline at end of file
diff --git a/data-prepper-plugins/dissect-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/dissect/DissectProcessorTest.java b/data-prepper-plugins/dissect-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/dissect/DissectProcessorTest.java
new file mode 100644
index 0000000000..b344bda68e
--- /dev/null
+++ b/data-prepper-plugins/dissect-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/dissect/DissectProcessorTest.java
@@ -0,0 +1,256 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.processor.dissect;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.opensearch.dataprepper.expression.ExpressionEvaluator;
+import org.opensearch.dataprepper.metrics.PluginMetrics;
+import org.opensearch.dataprepper.model.event.Event;
+import org.opensearch.dataprepper.model.event.JacksonEvent;
+import org.opensearch.dataprepper.model.record.Record;
+import org.opensearch.dataprepper.plugins.processor.dissect.Fields.AppendField;
+import org.opensearch.dataprepper.plugins.processor.dissect.Fields.Field;
+import org.opensearch.dataprepper.plugins.processor.dissect.Fields.IndirectField;
+import org.opensearch.dataprepper.plugins.processor.dissect.Fields.NormalField;
+import org.opensearch.dataprepper.plugins.processor.mutateevent.TargetType;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class DissectProcessorTest {
+    @Mock
+    private PluginMetrics pluginMetrics;
+    @Mock
+    private ExpressionEvaluator expressionEvaluator;
+    @Mock
+    private DissectProcessorConfig dissectConfig;
+
+    @Mock
+    private Dissector dissector;
+
+    @BeforeEach
+    void setUp() {
+        when(dissectConfig.getMap()).thenReturn(Map.of());
+    }
+
+    @Test
+    void test_normal_fields_dissect_succeeded() throws NoSuchFieldException, IllegalAccessException {
+
+        Field field1 = new NormalField("field1");
+        Field field2 = new NormalField("field2");
+        field1.setValue("foo");
+        field2.setValue("bar");
+
+        when(dissector.dissectText(any(String.class))).thenReturn(true);
+        when(dissector.getDissectedFields()).thenReturn(List.of(field1, field2));
+
+        final DissectProcessor processor = createObjectUnderTest();
+        reflectivelySetDissectorMap(processor);
+        final Record<Event> record = getEvent("");
+        final List<Record<Event>> dissectedRecords = (List<Record<Event>>) processor.doExecute(Collections.singletonList(record));
+
+        assertTrue(dissectedRecords.get(0).getData().containsKey("field1"));
+        assertTrue(dissectedRecords.get(0).getData().containsKey("field2"));
+
+        assertThat(dissectedRecords.get(0).getData().get("field1", String.class), is("foo"));
+        assertThat(dissectedRecords.get(0).getData().get("field2", String.class), is("bar"));
+    }
+
+    @Test
+    void test_append_fields_dissect_succeeded() throws NoSuchFieldException, IllegalAccessException {
+
+        Field field1 = new AppendField("field1");
+        Field field2 = new AppendField("field2");
+        field1.setValue("foo");
+        field2.setValue("bar");
+
+        when(dissector.dissectText(any(String.class))).thenReturn(true);
+        when(dissector.getDissectedFields()).thenReturn(List.of(field1, field2));
+
+        final DissectProcessor processor = createObjectUnderTest();
+        reflectivelySetDissectorMap(processor);
+        final Record<Event> record = getEvent("");
+        final List<Record<Event>> dissectedRecords = (List<Record<Event>>) processor.doExecute(Collections.singletonList(record));
+
+        assertTrue(dissectedRecords.get(0).getData().containsKey("field1"));
+        assertTrue(dissectedRecords.get(0).getData().containsKey("field2"));
+
+        assertThat(dissectedRecords.get(0).getData().get("field1", String.class), is("foo"));
+        assertThat(dissectedRecords.get(0).getData().get("field2", String.class), is("bar"));
+    }
+
+    @Test
+    void test_indirect_fields_dissect_succeeded() throws NoSuchFieldException, IllegalAccessException {
+
+        Field field1 = new IndirectField("field1");
+        Field field2 = new IndirectField("field2");
+        field1.setValue("foo");
+        field2.setValue("bar");
+
+        when(dissector.dissectText(any(String.class))).thenReturn(true);
+        when(dissector.getDissectedFields()).thenReturn(List.of(field1, field2));
+
+        final DissectProcessor processor = createObjectUnderTest();
+        reflectivelySetDissectorMap(processor);
+        final Record<Event> record = getEvent("");
+        final List<Record<Event>> dissectedRecords = (List<Record<Event>>) processor.doExecute(Collections.singletonList(record));
+
+        assertTrue(dissectedRecords.get(0).getData().containsKey("field1"));
+        assertTrue(dissectedRecords.get(0).getData().containsKey("field2"));
+
+        assertThat(dissectedRecords.get(0).getData().get("field1", String.class), is("foo"));
+        assertThat(dissectedRecords.get(0).getData().get("field2", String.class), is("bar"));
+    }
+
+    @Test
+    void test_dissectText_returns_false() throws NoSuchFieldException, IllegalAccessException {
+
+        when(dissector.dissectText(any(String.class))).thenReturn(false);
+
+        final DissectProcessor processor = createObjectUnderTest();
+        reflectivelySetDissectorMap(processor);
+        final Record<Event> record = getEvent("");
+        final List<Record<Event>> dissectedRecords = (List<Record<Event>>) processor.doExecute(Collections.singletonList(record));
+
+        // assert event is not modified
+        assertThat(dissectedRecords.get(0).getData(), is(record.getData()));
+    }
+
+    @Test
+    void test_dissectText_throws_exception() throws NoSuchFieldException, IllegalAccessException {
+        when(dissector.dissectText(any(String.class))).thenThrow(RuntimeException.class);
+
+        final DissectProcessor processor = createObjectUnderTest();
+        reflectivelySetDissectorMap(processor);
+        final Record<Event> record = getEvent("");
+        final List<Record<Event>> dissectedRecords = (List<Record<Event>>) processor.doExecute(Collections.singletonList(record));
+
+        // assert event is not modified
+        assertThat(dissectedRecords.get(0).getData(), is(record.getData()));
+    }
+
+    @Test
+    void test_target_type_int() throws NoSuchFieldException, IllegalAccessException {
+
+        Field field1 = new IndirectField("field1");
+        Field field2 = new IndirectField("field2");
+        field1.setValue("20");
+        field2.setValue("30");
+        when(dissector.dissectText(any(String.class))).thenReturn(true);
+        when(dissector.getDissectedFields()).thenReturn(List.of(field1, field2));
+
+        Map<String, TargetType> targetsMap = Map.of("field1", TargetType.INTEGER);
+        when(dissectConfig.getTargetTypes()).thenReturn(targetsMap);
+
+        final DissectProcessor processor = createObjectUnderTest();
+        reflectivelySetDissectorMap(processor);
+        final Record<Event> record = getEvent("");
+        final List<Record<Event>> dissectedRecords = (List<Record<Event>>) processor.doExecute(Collections.singletonList(record));
+
+        assertTrue(dissectedRecords.get(0).getData().containsKey("field1"));
+        assertTrue(dissectedRecords.get(0).getData().get("field1", Object.class) instanceof Integer);
+        assertThat(dissectedRecords.get(0).getData().get("field1", Object.class), is(20));
+
+        assertTrue(dissectedRecords.get(0).getData().containsKey("field2"));
+        assertTrue(dissectedRecords.get(0).getData().get("field2", Object.class) instanceof String);
+        assertThat(dissectedRecords.get(0).getData().get("field2", Object.class), is("30"));
+    }
+
+    @Test
+    void test_target_type_bool() throws NoSuchFieldException, IllegalAccessException {
+        Field field1 = new IndirectField("field1");
+        Field field2 = new IndirectField("field2");
+        field1.setValue("true");
+        field2.setValue("30");
+        when(dissector.dissectText(any(String.class))).thenReturn(true);
+        when(dissector.getDissectedFields()).thenReturn(List.of(field1, field2));
+
+        Map<String, TargetType> targetsMap = Map.of("field1", TargetType.BOOLEAN);
+        when(dissectConfig.getTargetTypes()).thenReturn(targetsMap);
+
+        final DissectProcessor processor = createObjectUnderTest();
+        reflectivelySetDissectorMap(processor);
+        final Record<Event> record = getEvent("");
+        final List<Record<Event>> dissectedRecords = (List<Record<Event>>) processor.doExecute(Collections.singletonList(record));
+
+        assertTrue(dissectedRecords.get(0).getData().containsKey("field1"));
+        assertTrue(dissectedRecords.get(0).getData().get("field1", Object.class) instanceof Boolean);
+        assertThat(dissectedRecords.get(0).getData().get("field1", Object.class), is(true));
+
+        assertTrue(dissectedRecords.get(0).getData().containsKey("field2"));
+        assertTrue(dissectedRecords.get(0).getData().get("field2", Object.class) instanceof String);
+        assertThat(dissectedRecords.get(0).getData().get("field2", Object.class), is("30"));
+    }
+
+    @Test
+    void test_target_type_double() throws NoSuchFieldException, IllegalAccessException {
+        Field field1 = new IndirectField("field1");
+        Field field2 = new IndirectField("field2");
+        field1.setValue("20.0");
+        field2.setValue("30");
+        when(dissector.dissectText(any(String.class))).thenReturn(true);
+        when(dissector.getDissectedFields()).thenReturn(List.of(field1, field2));
+
+        Map<String, TargetType> targetsMap = Map.of("field1", TargetType.DOUBLE);
+        when(dissectConfig.getTargetTypes()).thenReturn(targetsMap);
+
+        final DissectProcessor processor = createObjectUnderTest();
+        reflectivelySetDissectorMap(processor);
+        final Record<Event> record = getEvent("");
+        final List<Record<Event>> dissectedRecords = (List<Record<Event>>) processor.doExecute(Collections.singletonList(record));
+
+        assertTrue(dissectedRecords.get(0).getData().containsKey("field1"));
+        assertTrue(dissectedRecords.get(0).getData().get("field1", Object.class) instanceof Double);
+        assertThat(dissectedRecords.get(0).getData().get("field1", Object.class), is(20.0d));
+
+        assertTrue(dissectedRecords.get(0).getData().containsKey("field2"));
+        assertTrue(dissectedRecords.get(0).getData().get("field2", Object.class) instanceof String);
+        assertThat(dissectedRecords.get(0).getData().get("field2", Object.class), is("30"));
+    }
+
+    private DissectProcessor createObjectUnderTest() {
+        return new DissectProcessor(pluginMetrics, dissectConfig, expressionEvaluator);
+    }
+
+    private Record<Event> getEvent(String dissectText) {
+        final Map<String, Object> testData = new HashMap<>();
+        testData.put("test", dissectText);
+        return buildRecordWithEvent(testData);
+    }
+
+    private static Record<Event> buildRecordWithEvent(final Map<String, Object> data) {
+        return new Record<>(JacksonEvent
+                                    .builder()
+                                    .withData(data)
+                                    .withEventType("event")
+                                    .build());
+    }
+
+    private void reflectivelySetDissectorMap(DissectProcessor processor) throws NoSuchFieldException, IllegalAccessException {
+        Map<String, Dissector> dissectorMap = Map.of("test", dissector);
+        java.lang.reflect.Field reflectField = DissectProcessor.class.getDeclaredField("dissectorMap");
+
+        try {
+            reflectField.setAccessible(true);
+            reflectField.set(processor, dissectorMap);
+        } finally {
+            reflectField.setAccessible(false);
+        }
+    }
+}
\ No newline at end of file
diff --git a/data-prepper-plugins/dissect-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/dissect/DissectorTest.java b/data-prepper-plugins/dissect-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/dissect/DissectorTest.java
new file mode 100644
index 0000000000..418e888e52
--- /dev/null
+++ b/data-prepper-plugins/dissect-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/dissect/DissectorTest.java
@@ -0,0 +1,226 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.processor.dissect;
+
+import org.junit.jupiter.api.Test;
+import org.opensearch.dataprepper.plugins.processor.dissect.Fields.Field;
+
+import java.util.List;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class DissectorTest {
+
+    @Test
+    void test_normal_field_trailing_spaces(){
+        Dissector dissector = createObjectUnderTest(" %{field1} %{field2} ");
+        boolean result = dissector.dissectText(" foo bar ");
+
+        assertTrue(result);
+
+        List<Field> fields = dissector.getDissectedFields();
+
+        assertThat(fields.get(0).getKey(), is("field1"));
+        assertThat(fields.get(0).getValue(), is("foo"));
+        assertThat(fields.get(1).getKey(), is("field2"));
+        assertThat(fields.get(1).getValue(), is("bar"));
+    }
+
+    @Test
+    void test_normal_field_without_trailing_spaces(){
+        Dissector dissector = createObjectUnderTest("dm1 %{field1} %{field2} dm2");
+        boolean result = dissector.dissectText("dm1 foo bar dm2");
+        assertTrue(result);
+
+        List<Field> fields = dissector.getDissectedFields();
+
+        assertThat(fields.get(0).getKey(), is("field1"));
+        assertThat(fields.get(0).getValue(), is("foo"));
+        assertThat(fields.get(1).getKey(), is("field2"));
+        assertThat(fields.get(1).getValue(), is("bar"));
+    }
+
+    @Test
+    void test_normal_field_failure_without_delimiters(){
+        Dissector dissector = createObjectUnderTest("dm1 %{field1} %{field2} dm2");
+
+        boolean result = dissector.dissectText("dm1 foo bar");
+        assertFalse(result);
+    }
+
+    @Test
+    void test_normal_field_failure_with_extra_whitespaces(){
+        Dissector dissector = createObjectUnderTest("dm1 %{field1} %{field2} dm2");
+
+        boolean result = dissector.dissectText(" dm1 foo bar dm2");
+        assertFalse(result);
+    }
+
+    @Test
+    void test_named_skip_field(){
+        Dissector dissector = createObjectUnderTest("dm1 %{?field1} %{field2} dm2");
+
+
+        boolean result = dissector.dissectText("dm1 foo bar dm2");
+        assertTrue(result);
+        List<Field> fields = dissector.getDissectedFields();
+
+        assertThat(fields.size(), is(1));
+        assertThat(fields.get(0).getKey(), is("field2"));
+        assertThat(fields.get(0).getValue(), is("bar"));
+    }
+
+    @Test
+    void test_unnamed_skip_field(){
+        Dissector dissector = createObjectUnderTest("dm1 %{} %{field2} dm2");
+
+
+        boolean result = dissector.dissectText("dm1 foo bar dm2");
+        assertTrue(result);
+        List<Field> fields = dissector.getDissectedFields();
+
+        assertThat(fields.size(), is(1));
+        assertThat(fields.get(0).getKey(), is("field2"));
+        assertThat(fields.get(0).getValue(), is("bar"));
+    }
+
+    @Test
+    void test_indirect_field_with_skip_field(){
+        Dissector dissector = createObjectUnderTest("dm1 %{?field1} %{&field1} dm2");
+
+
+        boolean result = dissector.dissectText("dm1 foo bar dm2");
+        assertTrue(result);
+        List<Field> fields = dissector.getDissectedFields();
+
+        assertThat(fields.size(), is(1));
+        assertThat(fields.get(0).getKey(), is("foo"));
+        assertThat(fields.get(0).getValue(), is("bar"));
+    }
+
+    @Test
+    void test_indirect_field_with_normal_field(){
+        Dissector dissector = createObjectUnderTest("dm1 %{field1} %{&field1} dm2");
+
+
+        boolean result = dissector.dissectText("dm1 foo bar dm2");
+        assertTrue(result);
+        List<Field> fields = dissector.getDissectedFields();
+
+        assertThat(fields.get(0).getKey(), is("field1"));
+        assertThat(fields.get(0).getValue(), is("foo"));
+        assertThat(fields.get(1).getKey(), is("foo"));
+        assertThat(fields.get(1).getValue(), is("bar"));
+    }
+
+    @Test
+    void test_append_field_without_index(){
+        Dissector dissector = createObjectUnderTest("dm1 %{+field1} %{+field1} dm2");
+
+
+        boolean result = dissector.dissectText("dm1 foo bar dm2");
+        assertTrue(result);
+        List<Field> fields = dissector.getDissectedFields();
+
+        assertThat(fields.size(), is(1));
+        assertThat(fields.get(0).getKey(), is("field1"));
+        assertThat(fields.get(0).getValue(), is("foobar"));
+    }
+
+    @Test
+    void test_append_field_with_index(){
+        Dissector dissector = createObjectUnderTest("dm1 %{+field1/2} %{+field1/1} dm2");
+
+
+        boolean result = dissector.dissectText("dm1 foo bar dm2");
+        assertTrue(result);
+        List<Field> fields = dissector.getDissectedFields();
+
+        assertThat(fields.size(), is(1));
+        assertThat(fields.get(0).getKey(), is("field1"));
+        assertThat(fields.get(0).getValue(), is("barfoo"));
+    }
+
+    @Test
+    void test_append_whitespace_normal_field(){
+        Dissector dissector = createObjectUnderTest("dm1 %{field1->} %{field2} dm2");
+
+
+        boolean result = dissector.dissectText("dm1 foo      bar dm2");
+        assertTrue(result);
+        List<Field> fields = dissector.getDissectedFields();
+
+        assertThat(fields.get(0).getKey(), is("field1"));
+        assertThat(fields.get(0).getValue(), is("foo"));
+        assertThat(fields.get(1).getKey(), is("field2"));
+        assertThat(fields.get(1).getValue(), is("bar"));
+    }
+
+    @Test
+    void test_append_whitespace_append_field(){
+        Dissector dissector = createObjectUnderTest("dm1 %{+field1->} %{+field1} dm2");
+
+
+        boolean result = dissector.dissectText("dm1 foo      bar dm2");
+        assertTrue(result);
+        List<Field> fields = dissector.getDissectedFields();
+
+        assertThat(fields.size(), is(1));
+        assertThat(fields.get(0).getKey(), is("field1"));
+        assertThat(fields.get(0).getValue(), is("foobar"));
+    }
+
+    @Test
+    void test_append_whitespace_indirect_field(){
+        Dissector dissector = createObjectUnderTest("dm1 %{?field1->} %{&field1} dm2");
+
+
+        boolean result = dissector.dissectText("dm1 foo      bar dm2");
+        assertTrue(result);
+        List<Field> fields = dissector.getDissectedFields();
+
+        assertThat(fields.size(), is(1));
+        assertThat(fields.get(0).getKey(), is("foo"));
+        assertThat(fields.get(0).getValue(), is("bar"));
+    }
+
+    @Test
+    void test_skip_fields_with_padding(){
+        Dissector dissector = createObjectUnderTest("dm1 %{?field1->} %{?field3} %{field2} dm2");
+
+
+        boolean result = dissector.dissectText("dm1 foo     skip   bar dm2");
+        assertTrue(result);
+        List<Field> fields = dissector.getDissectedFields();
+
+        assertThat(fields.size(), is(1));
+        assertThat(fields.get(0).getKey(), is("field2"));
+        assertThat(fields.get(0).getValue(), is("bar"));
+    }
+
+    @Test
+    void test_indirect_field_with_append(){
+        Dissector dissector = createObjectUnderTest("%{+field1->} %{+field1} %{&field1->}");
+
+
+        boolean result = dissector.dissectText("foo     bar result     ");
+        assertTrue(result);
+        List<Field> fields = dissector.getDissectedFields();
+
+        assertThat(fields.size(), is(2));
+        assertThat(fields.get(0).getKey(), is("field1"));
+        assertThat(fields.get(0).getValue(), is("foobar"));
+        assertThat(fields.get(1).getKey(), is("foobar"));
+        assertThat(fields.get(1).getValue(), is("result"));
+    }
+
+    private Dissector createObjectUnderTest(String dissectPatternString) {
+        return new Dissector(dissectPatternString);
+    }
+}
\ No newline at end of file
diff --git a/data-prepper-plugins/dissect-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/dissect/Fields/FieldHelperTest.java b/data-prepper-plugins/dissect-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/dissect/Fields/FieldHelperTest.java
new file mode 100644
index 0000000000..2ec36b7ba2
--- /dev/null
+++ b/data-prepper-plugins/dissect-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/dissect/Fields/FieldHelperTest.java
@@ -0,0 +1,80 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.processor.dissect.Fields;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.hamcrest.CoreMatchers.nullValue;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+
+class FieldHelperTest {
+    private FieldHelper fieldHelper;
+
+    @BeforeEach
+    void setUp() {
+        fieldHelper = new FieldHelper();
+    }
+
+    @Test
+    void testNullFieldString() {
+        assertThat(fieldHelper.getField(null), is(nullValue()));
+    }
+
+    @Test
+    void testWhitespaces() {
+        Field field = fieldHelper.getField("  ");
+        assertThat(field, is(instanceOf(AppendField.class)));
+        assertThat(field.getKey(), is(""));
+    }
+
+    @Test
+    void testNormalFieldString() {
+        Field field = fieldHelper.getField("field1");
+        assertThat(field, is(instanceOf(NormalField.class)));
+        assertThat(field.getKey(), is("field1"));
+        assertThat(field.stripTrailing, is(false));
+    }
+
+    @Test
+    void testNormalFieldWithSuffixString() {
+        Field field = fieldHelper.getField("field1->");
+        assertThat(field, is(instanceOf(NormalField.class)));
+        assertThat(field.getKey(), is("field1"));
+        assertThat(field.stripTrailing, is(true));
+    }
+
+    @Test
+    void testNamedSkipFieldString() {
+        Field field = fieldHelper.getField("?field1");
+        assertThat(field, is(instanceOf(SkipField.class)));
+        assertThat(field.getKey(), is("field1"));
+    }
+
+    @Test
+    void testAppendFieldString() {
+        Field field = fieldHelper.getField("+field1");
+        assertThat(field, is(instanceOf(AppendField.class)));
+        assertThat(field.getKey(), is("field1"));
+    }
+
+    @Test
+    void testAppendFieldWithIndexString() {
+        Field field = fieldHelper.getField("+field1/1");
+        assertThat(field, is(instanceOf(AppendField.class)));
+        assertThat(field.getKey(), is("field1"));
+        assertThat(((AppendField)field).getIndex(), is(1));
+    }
+
+    @Test
+    void testIndirectFieldString() {
+        Field field = fieldHelper.getField("&field1");
+        assertThat(field, is(instanceOf(IndirectField.class)));
+        assertThat(field.getKey(), is("field1"));
+    }
+}
\ No newline at end of file
diff --git a/settings.gradle b/settings.gradle
index ae1dfc9e02..059aca1505 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -136,3 +136,5 @@ include 'data-prepper-plugins:cloudwatch-logs'
 include 'data-prepper-plugins:http-sink'
 include 'data-prepper-plugins:sns-sink'
 include 'data-prepper-plugins:prometheus-sink'
+include 'data-prepper-plugins:dissect-processor'
+