Addressed review comments

Signed-off-by: Krishna Kondaka <[email protected]>
opensearch-project · Apr 22, 2024 · cf1ae56 · cf1ae56
1 parent 4f3b17e
commit cf1ae56
Show file tree

Hide file tree

Showing 2 changed files with 46 additions and 7 deletions.
diff --git a/...rc/main/java/org/opensearch/dataprepper/plugins/codec/event_json/EventJsonInputCodec.java b/...rc/main/java/org/opensearch/dataprepper/plugins/codec/event_json/EventJsonInputCodec.java
@@ -48,6 +48,22 @@ public EventJsonInputCodec(final EventJsonInputCodecConfig config) {
         this.overrideTimeReceived = config.getOverrideTimeReceived();
     }
 
+    private boolean isCompatibleVersion(Map<String, Object> json) {
+        final String versionStr = (String)json.get(EventJsonDefines.VERSION);
+        final String[] version = versionStr.split("[.]");
+        final String currentVersionStr = DataPrepperVersion.getCurrentVersion().toString();
+        final String[] currentVersion = currentVersionStr.split("[.]");
+        if (version.length < 2 || currentVersion.length < 2) {
+            LOG.error("Invalid Version! Current version {} Received data version {}", currentVersionStr, versionStr);
+            return false;
+        }
+        if (!version[0].equals(currentVersion[0])) {
+            LOG.error("Version mismatch! Current version {} Received data version {}", currentVersionStr, versionStr);
+            return false;
+        }
+        return true;
+    }
+
     public void parse(InputStream inputStream, Consumer<Record<Event>> eventConsumer) throws IOException {
         Objects.requireNonNull(inputStream);
         Objects.requireNonNull(eventConsumer);
@@ -57,14 +73,12 @@ public void parse(InputStream inputStream, Consumer<Record<Event>> eventConsumer
         while (!jsonParser.isClosed() && jsonParser.nextToken() != JsonToken.END_OBJECT) {
             if (jsonParser.getCurrentToken() == JsonToken.START_OBJECT) {
                 final Map<String, Object> innerJson = objectMapper.readValue(jsonParser, Map.class);
-                final List<Map<String, Object>> maps = (List<Map<String, Object>>)innerJson.get(EventJsonDefines.EVENTS);
-                final String version = (String)innerJson.get(EventJsonDefines.VERSION);
-                if (!version.equals(DataPrepperVersion.getCurrentVersion().toString())) {
-                    LOG.error("Version mismatch! Current version {} Received data version {}", DataPrepperVersion.getCurrentVersion().toString(), version);
+                if (!isCompatibleVersion(innerJson)) {
                     return;
                 }
-                for (Map<String, Object> map: maps) {
-                    final Record<Event> record = createRecord(map);
+                final List<Map<String, Object>> events = (List<Map<String, Object>>)innerJson.get(EventJsonDefines.EVENTS);
+                for (Map<String, Object> eventMap: events) {
+                    final Record<Event> record = createRecord(eventMap);
                     if (record != null) {
                         eventConsumer.accept(record);
                     }

diff --git a/...est/java/org/opensearch/dataprepper/plugins/codec/event_json/EventJsonInputCodecTest.java b/...est/java/org/opensearch/dataprepper/plugins/codec/event_json/EventJsonInputCodecTest.java
@@ -75,7 +75,32 @@ public void invalidVersionTest() throws Exception {
 
         Map<String, Object> dataMap = event.toMap();
         Map<String, Object> metadataMap = objectMapper.convertValue(event.getMetadata(), Map.class);
-        String input = "{\""+EventJsonDefines.VERSION+"\":\"2.0\", \""+EventJsonDefines.EVENTS+"\":[";
+        String input = "{\""+EventJsonDefines.VERSION+"\":\"2\", \""+EventJsonDefines.EVENTS+"\":[";
+        String comma = "";
+        for (int i = 0; i < 2; i++) {
+            input += comma+"{\"data\":"+objectMapper.writeValueAsString(dataMap)+","+"\"metadata\":"+objectMapper.writeValueAsString(metadataMap)+"}";
+            comma = ",";
+        }
+        input += "]}";
+        inputStream = new ByteArrayInputStream(input.getBytes());
+        List<Record<Event>> records = new LinkedList<>();
+        inputCodec.parse(inputStream, records::add);
+        assertThat(records.size(), equalTo(0));
+    }
+
+
+    @Test
+    public void inCompatibleVersionTest() throws Exception {
+        inputCodec = createInputCodec();
+        final String key = UUID.randomUUID().toString();
+        final String value = UUID.randomUUID().toString();
+        Map<String, Object> data = Map.of(key, value);
+        Instant startTime = Instant.now();
+        Event event = createEvent(data, startTime);
+
+        Map<String, Object> dataMap = event.toMap();
+        Map<String, Object> metadataMap = objectMapper.convertValue(event.getMetadata(), Map.class);
+        String input = "{\""+EventJsonDefines.VERSION+"\":\"1.0\", \""+EventJsonDefines.EVENTS+"\":[";
         String comma = "";
         for (int i = 0; i < 2; i++) {
             input += comma+"{\"data\":"+objectMapper.writeValueAsString(dataMap)+","+"\"metadata\":"+objectMapper.writeValueAsString(metadataMap)+"}";