Addressed review comments

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

Author: kkondaka

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

@@ -52,7 +52,7 @@ public class CsvProcessor extends AbstractProcessor<Record<Event>, Record<Event>
     private final CsvMapper mapper;
     private final CsvSchema schema;
 
-    private final Boolean normalizeKeys;
+    private final boolean normalizeKeys;
 
     @DataPrepperPluginConstructor
     public CsvProcessor(final PluginMetrics pluginMetrics,

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

@@ -74,7 +74,7 @@ public class ParseJsonProcessorConfig implements CommonParseConfig {
 
     @JsonProperty("normalize_keys")
     @JsonPropertyDescription("If specified, replaces invalid characters with underscore character")
-    private Boolean normalizeKeys;
+    private Boolean normalizeKeys = false;
 
     @JsonProperty("parse_when")
     @JsonPropertyDescription("A <a href=\"https://opensearch.org/docs/latest/data-prepper/pipelines/expression-syntax/\">conditional expression</a> such as <code>/some_key == \"test\"</code>. " +

data-prepper-plugins/parse-json-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/parse/ion/ParseIonProcessorConfigTest.java

@@ -28,6 +28,7 @@ public void test_when_defaultParseIonProcessorConfig_then_returns_default_values
         assertThat(objectUnderTest.getSource(), equalTo(ParseIonProcessorConfig.DEFAULT_SOURCE));
         assertThat(objectUnderTest.getDestination(), equalTo(null));
         assertThat(objectUnderTest.getPointer(), equalTo(null));
+        assertThat(objectUnderTest.getNormalizeKeys(), equalTo(false));
         assertThat(objectUnderTest.getTagsOnFailure(), equalTo(null));
         assertThat(objectUnderTest.getOverwriteIfDestinationExists(), equalTo(true));
         assertThat(objectUnderTest.getHandleFailedEventsOption(), equalTo(HandleFailedEventsOption.SKIP));

data-prepper-plugins/parse-json-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/parse/ion/ParseIonProcessorTest.java

@@ -20,6 +20,7 @@
 
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.Mockito.lenient;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verifyNoInteractions;
@@ -64,6 +65,21 @@ void invalid_parse_when_throws_InvalidPluginConfigurationException() {
         assertThrows(InvalidPluginConfigurationException.class, this::createObjectUnderTest);
     }
 
+    @Test
+    void test_replace_invalid_key_characters() throws Exception {
+        final String source = "root_source";
+        when(processorConfig.getSource()).thenReturn(source);
+        when(processorConfig.getDestination()).thenReturn(source);
+        when(processorConfig.getDepth()).thenReturn(0);
+        parseJsonProcessor = createObjectUnderTest(); // need to recreate so that new config options are used
+        final String serializedMessage = "{bare Key: 1, sym$bol: SYMBOL, time%stamp: 2023-11-30T21:05:23.383Z, attr^ibute: dollars::100.0 }";
+        final Event parsedEvent = createAndParseMessageEvent(serializedMessage);
+        // Currently doesn't support invalid keys replacement
+        Object rootSourceObject = parsedEvent.get(source, Object.class);
+        assertTrue(rootSourceObject instanceof String);
+
+    }
+
     @Test
     void test_when_using_ion_features_then_processorParsesCorrectly() {
         parseJsonProcessor = createObjectUnderTest();

data-prepper-plugins/parse-json-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/parse/json/ParseJsonProcessorConfigTest.java

@@ -28,6 +28,7 @@ public void test_when_defaultParseJsonProcessorConfig_then_returns_default_value
         assertThat(objectUnderTest.getSource(), equalTo(ParseJsonProcessorConfig.DEFAULT_SOURCE));
         assertThat(objectUnderTest.getDestination(), equalTo(null));
         assertThat(objectUnderTest.getPointer(), equalTo(null));
+        assertThat(objectUnderTest.getNormalizeKeys(), equalTo(false));
         assertThat(objectUnderTest.getTagsOnFailure(), equalTo(null));
         assertThat(objectUnderTest.getOverwriteIfDestinationExists(), equalTo(true));
         assertThat(objectUnderTest.isDeleteSourceRequested(), equalTo(false));
@@ -64,6 +65,8 @@ void test_when_destinationIsWhiteSpaceOrFrontSlash_then_isValidDestinationFalse(
 
             setField(ParseJsonProcessorConfig.class, config, "deleteSource", true);
             assertThat(config.isDeleteSourceRequested(), equalTo(true));
+            setField(ParseJsonProcessorConfig.class, config, "normalizeKeys", true);
+            assertThat(config.getNormalizeKeys(), equalTo(true));
         }
 
         @Test

data-prepper-plugins/parse-json-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/parse/xml/ParseXmlProcessorConfigTest.java

@@ -23,6 +23,7 @@ public void test_when_defaultParseXmlProcessorConfig_then_returns_default_values
         assertThat(objectUnderTest.getSource(), equalTo(ParseXmlProcessorConfig.DEFAULT_SOURCE));
         assertThat(objectUnderTest.getDestination(), equalTo(null));
         assertThat(objectUnderTest.getPointer(), equalTo(null));
+        assertThat(objectUnderTest.getNormalizeKeys(), equalTo(false));
         assertThat(objectUnderTest.getTagsOnFailure(), equalTo(null));
         assertThat(objectUnderTest.getOverwriteIfDestinationExists(), equalTo(true));
         assertThat(objectUnderTest.getHandleFailedEventsOption(), equalTo(HandleFailedEventsOption.SKIP));