config to write empty events

Signed-off-by: Subrahmanyam-Gollapalli <subrahmanyam.gollapalli@freshworks.com>

Author: Subrahmanyam-Gollapalli

data-prepper-plugins/newline-codecs/src/main/java/org/opensearch/dataprepper/plugins/codec/newline/NewlineDelimitedOutputCodec.java

@@ -1,6 +1,11 @@
 /*
  * Copyright OpenSearch Contributors
  * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ *
  */
 package org.opensearch.dataprepper.plugins.codec.newline;
 
@@ -25,24 +30,28 @@ public class NewlineDelimitedOutputCodec implements OutputCodec {
     private static final String MESSAGE_FIELD = "message";
     @SuppressWarnings("unused")
     private OutputCodecContext deprecatedSupportCodecContext;
+    private final boolean includeEmptyObjects;
 
     @DataPrepperPluginConstructor
     public NewlineDelimitedOutputCodec(final NewlineDelimitedOutputConfig config) {
         Objects.requireNonNull(config);
+        this.includeEmptyObjects = config.isIncludeEmptyObjects();
     }
 
     private static class NewlineWriter implements Writer {
         private final OutputStream outputStream;
         private final OutputCodecContext codecContext;
+        private final boolean includeEmptyObjects;
 
-        private NewlineWriter(final OutputStream outputStream, final OutputCodecContext codecContext) {
+        private NewlineWriter(final OutputStream outputStream, final OutputCodecContext codecContext, final boolean includeEmptyObjects) {
             this.outputStream = outputStream;
             this.codecContext = codecContext;
+            this.includeEmptyObjects = includeEmptyObjects;
         }
 
         @Override
         public void writeEvent(final Event event) throws IOException {
-            doWriteEvent(outputStream, event, codecContext);
+            doWriteEvent(outputStream, event, codecContext, includeEmptyObjects);
         }
 
         @Override
@@ -56,7 +65,7 @@ public Writer createWriter(final OutputStream outputStream, final Event sampleEv
         Objects.requireNonNull(outputStream);
         Objects.requireNonNull(codecContext);
 
-        return new NewlineWriter(outputStream, codecContext);
+        return new NewlineWriter(outputStream, codecContext, includeEmptyObjects);
     }
 
     @Override
@@ -68,25 +77,7 @@ public void start(final OutputStream outputStream, Event event, final OutputCode
 
     @Override
     public void writeEvent(final Event event, final OutputStream outputStream) throws IOException {
-        Objects.requireNonNull(event);
-
-        // Extract the message field and write it as plain text
-        String message = null;
-        if (event.containsKey(MESSAGE_FIELD)) {
-            Object messageObj = event.get(MESSAGE_FIELD, Object.class);
-            if (messageObj != null) {
-                message = messageObj.toString();
-            }
-        }
-
-        // If message is null or empty, write empty string
-        if (message == null) {
-            message = "";
-        }
-
-        // Write the message as plain text followed by a newline
-        outputStream.write(message.getBytes());
-        outputStream.write(System.lineSeparator().getBytes());
+        doWriteEvent(outputStream, event, deprecatedSupportCodecContext, includeEmptyObjects);
     }
 
     @Override
@@ -99,7 +90,7 @@ public String getExtension() {
         return NEWLINE;
     }
 
-    private static void doWriteEvent(final OutputStream outputStream, final Event event, final OutputCodecContext codecContext) throws IOException {
+    private static void doWriteEvent(final OutputStream outputStream, final Event event, final OutputCodecContext codecContext, final boolean includeEmptyObjects) throws IOException {
         Objects.requireNonNull(event);
 
         // Extract the message field and write it as plain text
@@ -111,8 +102,11 @@ private static void doWriteEvent(final OutputStream outputStream, final Event ev
             }
         }
 
-        // If message is null or empty, write empty string
-        if (message == null) {
+        // Default: do not write anything if message is null or empty
+        if (message == null || message.isEmpty()) {
+            if (!includeEmptyObjects) {
+                return;
+            }
             message = "";
         }
 

data-prepper-plugins/newline-codecs/src/main/java/org/opensearch/dataprepper/plugins/codec/newline/NewlineDelimitedOutputConfig.java

@@ -1,12 +1,34 @@
 /*
  * Copyright OpenSearch Contributors
  * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ *
  */
 
 package org.opensearch.dataprepper.plugins.codec.newline;
 
+import com.fasterxml.jackson.annotation.JsonProperty;
+
 /**
  * Configuration class for the newline delimited output codec.
  */
 public class NewlineDelimitedOutputConfig {
+
+    /**
+     * When false (default), events with no message or an empty message are not written.
+     * When true, empty messages are written as blank lines, similar to ndjson input codec.
+     */
+    @JsonProperty("include_empty_objects")
+    private boolean includeEmptyObjects = false;
+
+    public boolean isIncludeEmptyObjects() {
+        return includeEmptyObjects;
+    }
+
+    public void setIncludeEmptyObjects(final boolean includeEmptyObjects) {
+        this.includeEmptyObjects = includeEmptyObjects;
+    }
 }

data-prepper-plugins/newline-codecs/src/test/java/org/opensearch/dataprepper/plugins/codec/newline/NewlineDelimitedOutputCodecTest.java

@@ -19,6 +19,7 @@
 import java.nio.charset.StandardCharsets;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.UUID;
 
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.MatcherAssert.assertThat;
@@ -44,23 +45,61 @@ void constructor_throws_if_config_is_null() {
 
     @Test
     void writeEvent_writes_message_field_as_plain_text() throws IOException {
+        final String message = UUID.randomUUID().toString();
         final ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
         final OutputCodecContext codecContext = new OutputCodecContext();
 
         final Map<String, Object> eventData = new HashMap<>();
-        eventData.put("message", "r_id=2bb2bd0aeece11f0bea286fb87d48915-ticket_update,tp=00-13a3bb055e6b589dbc0f952e0d75020a-1f2e986790c19742-01");
+        eventData.put("message", message);
         final Event event = eventFactory.eventBuilder(EventBuilder.class).withData(eventData).build();
 
         codec.start(outputStream, event, codecContext);
         codec.writeEvent(event, outputStream);
         codec.complete(outputStream);
 
         final String output = outputStream.toString(StandardCharsets.UTF_8);
-        assertThat(output, equalTo("r_id=2bb2bd0aeece11f0bea286fb87d48915-ticket_update,tp=00-13a3bb055e6b589dbc0f952e0d75020a-1f2e986790c19742-01" + System.lineSeparator()));
+        assertThat(output, equalTo(message + System.lineSeparator()));
     }
 
     @Test
-    void writeEvent_writes_empty_string_when_message_is_missing() throws IOException {
+    void writeEvent_writes_nothing_when_message_is_missing_and_include_empty_objects_false() throws IOException {
+        final ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+        final OutputCodecContext codecContext = new OutputCodecContext();
+
+        final Map<String, Object> eventData = new HashMap<>();
+        eventData.put("other_field", "some value");
+        final Event event = eventFactory.eventBuilder(EventBuilder.class).withData(eventData).build();
+
+        codec.start(outputStream, event, codecContext);
+        codec.writeEvent(event, outputStream);
+        codec.complete(outputStream);
+
+        final String output = outputStream.toString(StandardCharsets.UTF_8);
+        assertThat(output, equalTo(""));
+    }
+
+    @Test
+    void writeEvent_writes_nothing_when_message_is_null_and_include_empty_objects_false() throws IOException {
+        final ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+        final OutputCodecContext codecContext = new OutputCodecContext();
+
+        final Map<String, Object> eventData = new HashMap<>();
+        eventData.put("message", null);
+        final Event event = eventFactory.eventBuilder(EventBuilder.class).withData(eventData).build();
+
+        codec.start(outputStream, event, codecContext);
+        codec.writeEvent(event, outputStream);
+        codec.complete(outputStream);
+
+        final String output = outputStream.toString(StandardCharsets.UTF_8);
+        assertThat(output, equalTo(""));
+    }
+
+    @Test
+    void writeEvent_writes_empty_line_when_message_is_missing_and_include_empty_objects_true() throws IOException {
+        config.setIncludeEmptyObjects(true);
+        codec = new NewlineDelimitedOutputCodec(config);
+
         final ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
         final OutputCodecContext codecContext = new OutputCodecContext();
 
@@ -77,7 +116,10 @@ void writeEvent_writes_empty_string_when_message_is_missing() throws IOException
     }
 
     @Test
-    void writeEvent_writes_empty_string_when_message_is_null() throws IOException {
+    void writeEvent_writes_empty_line_when_message_is_null_and_include_empty_objects_true() throws IOException {
+        config.setIncludeEmptyObjects(true);
+        codec = new NewlineDelimitedOutputCodec(config);
+
         final ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
         final OutputCodecContext codecContext = new OutputCodecContext();