-Support for Sink Codecs (#3030)

* -Support for Sink Codecs Signed-off-by: omkarmmore95 <[email protected]> * -Support for Sink Codecs Signed-off-by: omkarmmore95 <[email protected]> * -Support for Sink Codecs Signed-off-by: omkarmmore95 <[email protected]>
opensearch-project · Jul 17, 2023 · 37d05bb · 37d05bb
1 parent 2476813
commit 37d05bb
Show file tree

Hide file tree

Showing 6 changed files with 61 additions and 22 deletions.
diff --git a/.../newline/NewlineDelimitedOutputCodec.java → ...plugins/codec/json/NdjsonOutputCodec.java b/.../newline/NewlineDelimitedOutputCodec.java → ...plugins/codec/json/NdjsonOutputCodec.java
@@ -2,7 +2,7 @@
  * Copyright OpenSearch Contributors
  * SPDX-License-Identifier: Apache-2.0
  */
-package org.opensearch.dataprepper.plugins.codec.newline;
+package org.opensearch.dataprepper.plugins.codec.json;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin;
@@ -19,14 +19,14 @@
  * An implementation of {@link OutputCodec} which deserializes Data-Prepper events
  * and writes them to Output Stream as ND-JSON data
  */
-@DataPrepperPlugin(name = "newline", pluginType = OutputCodec.class, pluginConfigurationType = NewlineDelimitedOutputConfig.class)
-public class NewlineDelimitedOutputCodec implements OutputCodec {
+@DataPrepperPlugin(name = "ndjson", pluginType = OutputCodec.class, pluginConfigurationType = NdjsonOutputConfig.class)
+public class NdjsonOutputCodec implements OutputCodec {
     private static final String NDJSON = "ndjson";
     private static final ObjectMapper objectMapper = new ObjectMapper();
-    private final NewlineDelimitedOutputConfig config;
+    private final NdjsonOutputConfig config;
 
     @DataPrepperPluginConstructor
-    public NewlineDelimitedOutputCodec(final NewlineDelimitedOutputConfig config) {
+    public NdjsonOutputCodec(final NdjsonOutputConfig config) {
         Objects.requireNonNull(config);
         this.config = config;
     }

diff --git a/...newline/NewlineDelimitedOutputConfig.java → ...lugins/codec/json/NdjsonOutputConfig.java b/...newline/NewlineDelimitedOutputConfig.java → ...lugins/codec/json/NdjsonOutputConfig.java
@@ -3,7 +3,7 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-package org.opensearch.dataprepper.plugins.codec.newline;
+package org.opensearch.dataprepper.plugins.codec.json;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 
@@ -13,7 +13,7 @@
 /**
  * Configuration class for the newline delimited codec.
  */
-public class NewlineDelimitedOutputConfig {
+public class NdjsonOutputConfig {
     private static final List<String> DEFAULT_EXCLUDE_KEYS = new ArrayList<>();
 
     @JsonProperty("exclude_keys")
@@ -22,4 +22,8 @@ public class NewlineDelimitedOutputConfig {
     public List<String> getExcludeKeys() {
         return excludeKeys;
     }
+
+    public void setExcludeKeys(List<String> excludeKeys) {
+        this.excludeKeys = excludeKeys;
+    }
 }
diff --git a/...ssor/src/test/java/org/opensearch/dataprepper/plugins/codec/json/JsonOutputCodecTest.java b/...ssor/src/test/java/org/opensearch/dataprepper/plugins/codec/json/JsonOutputCodecTest.java
@@ -0,0 +1,24 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.opensearch.dataprepper.plugins.codec.json;
+
+import org.junit.jupiter.api.Test;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+
+public class JsonOutputCodecTest {
+
+    private JsonOutputCodec createObjectUnderTest() {
+        return new JsonOutputCodec();
+    }
+
+    @Test
+    void testGetExtension() {
+        JsonOutputCodec jsonOutputCodec = createObjectUnderTest();
+        assertThat(null, equalTo(jsonOutputCodec.getExtension()));
+    }
+}
diff --git a/...line/NewlineDelimitedOutputCodecTest.java → ...json/NewlineDelimitedOutputCodecTest.java b/...line/NewlineDelimitedOutputCodecTest.java → ...json/NewlineDelimitedOutputCodecTest.java
@@ -2,10 +2,11 @@
  * Copyright OpenSearch Contributors
  * SPDX-License-Identifier: Apache-2.0
  */
-package org.opensearch.dataprepper.plugins.codec.newline;
+package org.opensearch.dataprepper.plugins.codec.json;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.hamcrest.Matchers;
+import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
 import org.opensearch.dataprepper.model.event.Event;
@@ -16,39 +17,42 @@
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.MatcherAssert.assertThat;
 
 
 public class NewlineDelimitedOutputCodecTest {
     private ByteArrayOutputStream outputStream;
 
-    private static NewlineDelimitedOutputConfig config;
+    private static NdjsonOutputConfig config;
 
     private static int numberOfRecords;
     private static final String REGEX = "\\r?\\n";
     private static ObjectMapper objectMapper = new ObjectMapper();
 
-    private NewlineDelimitedOutputCodec createObjectUnderTest() {
-        config = new NewlineDelimitedOutputConfig();
-        return new NewlineDelimitedOutputCodec(config);
+    private NdjsonOutputCodec createObjectUnderTest() {
+        config = new NdjsonOutputConfig();
+        config.setExcludeKeys(Arrays.asList("S3"));
+        return new NdjsonOutputCodec(config);
     }
 
     @ParameterizedTest
     @ValueSource(ints = {1, 3, 10, 100})
     void test_happy_case(final int numberOfRecords) throws IOException {
         this.numberOfRecords = numberOfRecords;
-        NewlineDelimitedOutputCodec newlineDelimitedOutputCodec = createObjectUnderTest();
+        NdjsonOutputCodec ndjsonOutputCodec = createObjectUnderTest();
         outputStream = new ByteArrayOutputStream();
-        newlineDelimitedOutputCodec.start(outputStream);
+        ndjsonOutputCodec.start(outputStream);
         for (int index = 0; index < numberOfRecords; index++) {
             final Event event = (Event) getRecord(index).getData();
-            newlineDelimitedOutputCodec.writeEvent(event, outputStream, null);
+            ndjsonOutputCodec.writeEvent(event, outputStream, null);
         }
-        newlineDelimitedOutputCodec.complete(outputStream);
+        ndjsonOutputCodec.complete(outputStream);
         byte[] byteArray = outputStream.toByteArray();
         String jsonString = null;
         try {
@@ -84,4 +88,11 @@ private static List<HashMap> generateRecords(int numberOfRecords) {
         }
         return recordList;
     }
+
+    @Test
+    void testGetExtension() {
+        NdjsonOutputCodec ndjsonOutputCodec = createObjectUnderTest();
+
+        assertThat("ndjson", equalTo(ndjsonOutputCodec.getExtension()));
+    }
 }
diff --git a/data-prepper-plugins/s3-sink/build.gradle b/data-prepper-plugins/s3-sink/build.gradle
@@ -21,7 +21,7 @@ dependencies {
     implementation 'org.apache.commons:commons-lang3:3.12.0'
     testImplementation project(':data-prepper-test-common')
 
-    implementation project(':data-prepper-plugins:newline-codecs')
+    implementation project(':data-prepper-plugins:parse-json-processor')
 }
 
 test {

diff --git a/...ink/src/integrationTest/java/org/opensearch/dataprepper/plugins/sink/S3SinkServiceIT.java b/...ink/src/integrationTest/java/org/opensearch/dataprepper/plugins/sink/S3SinkServiceIT.java
@@ -24,8 +24,8 @@
 import org.opensearch.dataprepper.model.log.JacksonLog;
 import org.opensearch.dataprepper.model.record.Record;
 import org.opensearch.dataprepper.model.types.ByteCount;
-import org.opensearch.dataprepper.plugins.codec.newline.NewlineDelimitedOutputCodec;
-import org.opensearch.dataprepper.plugins.codec.newline.NewlineDelimitedOutputConfig;
+import org.opensearch.dataprepper.plugins.codec.json.NdjsonOutputCodec;
+import org.opensearch.dataprepper.plugins.codec.json.NdjsonOutputConfig;
 import org.opensearch.dataprepper.plugins.sink.accumulator.BufferFactory;
 import org.opensearch.dataprepper.plugins.sink.accumulator.InMemoryBufferFactory;
 import org.opensearch.dataprepper.plugins.sink.accumulator.ObjectKey;
@@ -90,7 +90,7 @@ class S3SinkServiceIT {
     private OutputCodec codec;
 
     @Mock
-    NewlineDelimitedOutputConfig newlineDelimitedOutputConfig;
+    NdjsonOutputConfig ndjsonOutputConfig;
 
 
     @BeforeEach
@@ -131,8 +131,8 @@ void verify_flushed_object_count_into_s3_bucket() {
     }
 
     void configureNewLineCodec() {
-        codec = new NewlineDelimitedOutputCodec(newlineDelimitedOutputConfig);
-        when(newlineDelimitedOutputConfig.getExcludeKeys()).thenReturn(new ArrayList<>());
+        codec = new NdjsonOutputCodec(ndjsonOutputConfig);
+        when(ndjsonOutputConfig.getExcludeKeys()).thenReturn(new ArrayList<>());
     }
 
     @Test