Addressed review comments

Signed-off-by: Kondaka <krishkdk@amazon.com>

Author: kkondaka

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/event/Event.java

@@ -35,6 +35,16 @@ public interface Event extends Serializable {
      */
     void put(EventKey key, Object value);
 
+    /**
+     * Adds or updates the key with a given value in the Event
+     *
+     * @param key where the value will be set
+     * @param value value to set the key to
+     * @param repalceInvalidCharacters flag indicating if invalid characters should be replaced or not
+     * @since 2.13
+     */
+    void put(String key, Object value, boolean replaceInvalidCharacters);
+
     /**
      * Adds or updates the key with a given value in the Event
      *

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/event/JacksonEvent.java

@@ -166,6 +166,14 @@ public void put(EventKey key, Object value) {
         }
     }
 
+    @Override
+    public void put(String key, final Object value, final boolean replaceInvalidCharacters) {
+        if (replaceInvalidCharacters) {
+            key = JacksonEventKey.replaceInvalidCharacters(key);
+        }
+        put(key, value);
+    }
+
     /**
      * Adds or updates the key with a given value in the Event.
      *
@@ -261,7 +269,6 @@ public <T> T get(final String key, final Class<T> clazz) {
         return get(jacksonEventKey, clazz);
     }
 
-
     public static String replaceInvalidKeyChars(final String key) {
         return JacksonEventKey.replaceInvalidCharacters(key);
     }

data-prepper-api/src/test/java/org/opensearch/dataprepper/model/event/JacksonEventKeyTest.java

@@ -196,6 +196,13 @@ public Stream<? extends Arguments> provideArguments(final ExtensionContext exten
         }
     }
 
+    @ParameterizedTest
+    @ValueSource(strings = {"key 1", "key$1", "key&1", "key^1", "key%1", "key_1"})
+    public void testReplaceInvalidKeyChars(final String key) {
+        assertThat(JacksonEventKey.replaceInvalidCharacters(key), equalTo("key_1"));
+        assertThat(JacksonEventKey.replaceInvalidCharacters(null), equalTo(null));
+    }
+
     @ParameterizedTest
     @EnumSource(EventKeyFactory.EventAction.class)
     void equals_returns_true_for_same_key_and_actions(final EventKeyFactory.EventAction eventAction) {

data-prepper-api/src/test/java/org/opensearch/dataprepper/model/event/JacksonEventTest.java

@@ -230,13 +230,30 @@ public void testPutKeyCannotBeEmptyString() {
     }
 
     @ParameterizedTest
-    @ValueSource(strings = {"key 1", "key$1", "key&1", "key^1", "key%1"})
+    @ValueSource(strings = {"key 1", "key$1", "key&1", "key^1", "key%1", "key_1"})
     public void testReplaceInvalidKeyChars(final String key) {
         assertThat(JacksonEvent.replaceInvalidKeyChars(key), equalTo("key_1"));
         assertThat(JacksonEvent.replaceInvalidKeyChars(key.substring(0,3)), equalTo("key"));
         assertThat(JacksonEvent.replaceInvalidKeyChars(null), equalTo(null));
     }
 
+    @ParameterizedTest
+    @ValueSource(strings = {"key 1", "key$1", "key&1", "key^1", "key%1", "key_1"})
+    public void testPutWithReplaceInvalidKeyChars(final String key) {
+        final String value = UUID.randomUUID().toString();
+
+        event.put(key, value, true);
+        assertThat(event.get("key_1", String.class), equalTo(value));
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = {"key 1", "key$1", "key&1", "key^1", "key%1"})
+    public void testPutWithoutReplaceInvalidKeyChars(final String key) {
+        final String value = UUID.randomUUID().toString();
+
+        assertThrows(IllegalArgumentException.class, () -> event.put(key, value, false));
+    }
+
     @Test
     public void testPutAndGet_withMultiLevelKey() {
         final String key = "foo/bar";
@@ -1104,7 +1121,7 @@ void testJsonStringBuilderWithIncludeKeys() {
 
 
     }
-    
+
     @Test
     void testJsonStringBuilderWithExcludeKeys() {
         final String jsonString = "{\"id\":1,\"foo\":\"bar\",\"info\":{\"name\":\"hello\",\"foo\":\"bar\"},\"tags\":[{\"key\":\"a\",\"value\":\"b\"},{\"key\":\"c\",\"value\":\"d\"}]}";

data-prepper-plugins/csv-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/csv/CsvProcessor.java

@@ -214,10 +214,7 @@ private void putDataInEvent(final Event event, final List<String> header, final
         int providedHeaderColIdx = 0;
         for (; providedHeaderColIdx < header.size() && providedHeaderColIdx < data.size(); providedHeaderColIdx++) {
             String key = header.get(providedHeaderColIdx);
-            if (normalizeKeys) {
-                key = JacksonEvent.replaceInvalidKeyChars(key);
-            }
-            event.put(key, data.get(providedHeaderColIdx));
+            event.put(key, data.get(providedHeaderColIdx), normalizeKeys);
         }
         for (int remainingColIdx = providedHeaderColIdx; remainingColIdx < data.size(); remainingColIdx++) {
             event.put(generateColumnHeader(remainingColIdx), data.get(remainingColIdx));

data-prepper-plugins/csv-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/csv/CsvProcessorConfig.java

@@ -43,7 +43,7 @@ public class CsvProcessorConfig {
     private Boolean deleteHeader = DEFAULT_DELETE_HEADERS;
 
     @JsonProperty(value = "normalize_keys", defaultValue = "false")
-    @JsonPropertyDescription("If specified, replaces invalid characters with underscore character")
+    @JsonPropertyDescription("If set to true, replaces invalid characters with underscore character")
     private Boolean normalizeKeys = false;
 
 

data-prepper-plugins/parse-json-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/parse/AbstractParseProcessor.java

@@ -127,7 +127,7 @@ protected void replaceInvalidChars(Map<String, Object> map) {
             if (value instanceof Map) {
                 replaceInvalidChars((Map<String, Object>)value);
             }
-            if (newKey != key) {
+            if (!newKey.equals(key)) {
                 toAdd.put(newKey, value);
                 iterator.remove();
             }