FIX: NPE in JsonCodec with keyName (#5659)

Signed-off-by: George Chen <qchea@amazon.com>

Author: chenqi0805

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/codec/JsonDecoder.java

@@ -72,7 +72,7 @@ public void parse(InputStream inputStream, Instant timeReceived, Consumer<Record
             }
 
             if (jsonParser.getCurrentToken() == JsonToken.START_ARRAY) {
-                if (keyName != null && !nodeName.equals(keyName)) {
+                if (keyName != null && !keyName.equals(nodeName)) {
                     continue;
                 }
                 parseRecordsArray(jsonParser, timeReceived, eventConsumer, includeKeysMap, includeMetadataKeysMap);

data-prepper-api/src/test/java/org/opensearch/dataprepper/model/codec/JsonDecoderTest.java

@@ -171,6 +171,20 @@ void test_basicJsonDecoder_withInputConfig() throws IOException {
             assertThat(receivedTime, equalTo(now));
         }
 
+        @Test
+        void test_JsonDecoder_withKeyName_when_parsing_json_array_should_skip() throws IOException {
+            final Instant now = Instant.now();
+            String inputString = "[]";
+            List<Record<Event>> records = new ArrayList<>();
+            jsonDecoder = new JsonDecoder(key_name, null, null, maxEventLength);
+            jsonDecoder.parse(new ByteArrayInputStream(inputString.getBytes()), now, (record) -> {
+                records.add(record);
+                receivedTime = record.getData().getEventHandle().getInternalOriginationTime();
+            });
+
+            assertTrue(records.isEmpty());
+        }
+
         @Test
         void test_basicJsonDecoder_withInputConfig_withoutEvents_empty_metadata_keys() throws IOException {
             final Instant now = Instant.now();