Updates to the OutputCodec Javadocs and writer-based implementation for NdjsonOutputCodec.

Signed-off-by: David Venable <dlv@amazon.com>

Author: dlvenable

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

@@ -22,11 +22,45 @@ public interface OutputCodec {
 
     static final ObjectMapper objectMapper = new ObjectMapper();
 
+    /**
+     * A writer specific to a single buffer.
+     *
+     * @since 2.12
+     */
     interface Writer {
+        /**
+         * Writes a single event to the {@link OutputStream}.
+         *
+         * @param event A Data Prepper {@link Event}
+         * @throws IOException An IO exception writing to the stream
+         *
+         * @since 2.12
+         */
         void writeEvent(Event event) throws IOException;
+
+        /**
+         * Completes a writer.
+         *
+         * @throws IOException An IO exception completing the stream
+         *
+         * @since 2.12
+         */
         void complete() throws IOException;
     }
 
+    /**
+     * Creates a new {@link Writer} for a given {@link OutputStream}.
+     * Typically, you create one per buffer.
+     *
+     * @param outputStream The {@link OutputStream} to write to
+     * @param sampleEvent A sample Data Prepper {@link Event}.
+     *                    It is not written to the stream, but may be used for metadata.
+     * @param codecContext The {@link OutputCodecContext}
+     * @return A {@link Writer} to use for this buffer.
+     * @throws IOException An IO exception occurs initializing the writer or stream
+     *
+     * @since 2.12
+     */
     default Writer createWriter(final OutputStream outputStream, final Event sampleEvent, final OutputCodecContext codecContext) throws IOException {
         final OutputCodec codec = this;
         codec.start(outputStream, sampleEvent, codecContext);

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

@@ -71,7 +71,6 @@ public void complete() throws IOException {
             outputStream.close();
         }
 
-
         private Map<String, Object> getDataMapToSerialize(final Event event) throws JsonProcessingException {
             final Event modifiedEvent;
             if (codecContext.getTagsTargetKey() != null) {

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

@@ -4,7 +4,6 @@
  */
 package org.opensearch.dataprepper.plugins.codec.json;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
 import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin;
 import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor;
 import org.opensearch.dataprepper.model.codec.OutputCodec;
@@ -22,31 +21,56 @@
 @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 NdjsonOutputConfig config;
-    private OutputCodecContext codecContext;
+    private OutputCodecContext deprecatedSupportCodecContext;
 
     @DataPrepperPluginConstructor
     public NdjsonOutputCodec(final NdjsonOutputConfig config) {
         Objects.requireNonNull(config);
-        this.config = config;
+    }
+
+    private static class NdjsonWriter implements Writer {
+        private final OutputStream outputStream;
+        private final OutputCodecContext codecContext;
+
+        private NdjsonWriter(final OutputStream outputStream, final OutputCodecContext codecContext) {
+            this.outputStream = outputStream;
+            this.codecContext = codecContext;
+        }
+
+        @Override
+        public void writeEvent(final Event event) throws IOException {
+            doWriteEvent(outputStream, event, codecContext);
+        }
+
+        @Override
+        public void complete() throws IOException {
+            outputStream.close();
+        }
+    }
+
+    @Override
+    public Writer createWriter(final OutputStream outputStream, final Event sampleEvent, final OutputCodecContext codecContext) {
+        Objects.requireNonNull(outputStream);
+        Objects.requireNonNull(codecContext);
+
+        return new NdjsonWriter(outputStream, codecContext);
     }
 
     @Override
     public void start(final OutputStream outputStream, Event event, final OutputCodecContext codecContext) throws IOException {
         Objects.requireNonNull(outputStream);
         Objects.requireNonNull(codecContext);
-        this.codecContext = codecContext;
+        this.deprecatedSupportCodecContext = codecContext;
     }
 
     @Override
     public void writeEvent(final Event event, final OutputStream outputStream) throws IOException {
         Objects.requireNonNull(event);
 
         String json = event.jsonBuilder()
-                .includeKeys(codecContext.getIncludeKeys())
-                .excludeKeys(codecContext.getExcludeKeys())
-                .includeTags(codecContext.getTagsTargetKey())
+                .includeKeys(deprecatedSupportCodecContext.getIncludeKeys())
+                .excludeKeys(deprecatedSupportCodecContext.getExcludeKeys())
+                .includeTags(deprecatedSupportCodecContext.getTagsTargetKey())
                 .toJsonString();
         outputStream.write(json.getBytes());
         outputStream.write(System.lineSeparator().getBytes());
@@ -61,4 +85,16 @@ public void complete(final OutputStream outputStream) throws IOException {
     public String getExtension() {
         return NDJSON;
     }
+
+    private static void doWriteEvent(final OutputStream outputStream, final Event event, final OutputCodecContext codecContext) throws IOException {
+        Objects.requireNonNull(event);
+
+        String json = event.jsonBuilder()
+                .includeKeys(codecContext.getIncludeKeys())
+                .excludeKeys(codecContext.getExcludeKeys())
+                .includeTags(codecContext.getTagsTargetKey())
+                .toJsonString();
+        outputStream.write(json.getBytes());
+        outputStream.write(System.lineSeparator().getBytes());
+    }
 }