Add ByteDecoder for opensearch_api source to support Kafka buffer

The opensearch_api source did not work with Kafka buffer because no
ByteDecoder was registered. When buffer.isByteBuffer()=true, raw bytes
are written but the consumer side had no way to reconstruct events from
the NDJSON bulk format.

This adds OpenSearchBulkByteDecoder which parses NDJSON bulk format
(action/metadata + document line pairs) back into Data Prepper events
with correct metadata attributes.

Resolves #6876

Signed-off-by: Divakar Pratap Singh <divakar.p.singh@gmail.com>

Author: divakarsingh

data-prepper-plugins/opensearch-api-source/src/main/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchAPISource.java

@@ -11,6 +11,7 @@
 import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin;
 import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor;
 import org.opensearch.dataprepper.model.buffer.Buffer;
+import org.opensearch.dataprepper.model.codec.ByteDecoder;
 import org.opensearch.dataprepper.model.configuration.PipelineDescription;
 import org.opensearch.dataprepper.model.event.Event;
 import org.opensearch.dataprepper.model.plugin.PluginFactory;
@@ -38,4 +39,9 @@ public BaseHttpService getHttpService(final int bufferWriteTimeoutInMillis, fina
     public String getHttpHealthCheckPath() {
         return HTTP_HEALTH_CHECK_PATH;
     }
+
+    @Override
+    public ByteDecoder getDecoder() {
+        return new OpenSearchBulkByteDecoder();
+    }
 }

data-prepper-plugins/opensearch-api-source/src/main/java/org/opensearch/dataprepper/plugins/source/opensearchapi/OpenSearchBulkByteDecoder.java

@@ -0,0 +1,101 @@
+/*
+ * 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.source.opensearchapi;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.opensearch.dataprepper.model.codec.ByteDecoder;
+import org.opensearch.dataprepper.model.event.Event;
+import org.opensearch.dataprepper.model.event.EventType;
+import org.opensearch.dataprepper.model.event.JacksonEvent;
+import org.opensearch.dataprepper.model.opensearch.OpenSearchBulkActions;
+import org.opensearch.dataprepper.model.record.Record;
+import org.opensearch.dataprepper.plugins.source.opensearchapi.model.BulkAPIEventMetadataKeyAttributes;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.time.Instant;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.function.Consumer;
+
+public class OpenSearchBulkByteDecoder implements ByteDecoder {
+
+    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+    private static final TypeReference<Map<String, Object>> MAP_TYPE = new TypeReference<>() {};
+
+    @Override
+    public void parse(InputStream inputStream, Instant timeReceived, Consumer<Record<Event>> eventConsumer) throws IOException {
+        final BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8));
+        String line;
+        while ((line = reader.readLine()) != null) {
+            if (line.isBlank()) continue;
+
+            Map<String, Object> actionLine = OBJECT_MAPPER.readValue(line, MAP_TYPE);
+            String action = extractAction(actionLine);
+            if (action == null) continue;
+
+            @SuppressWarnings("unchecked")
+            Map<String, Object> metadata = (Map<String, Object>) actionLine.get(action);
+            boolean isDelete = OpenSearchBulkActions.DELETE.toString().equals(action);
+
+            Map<String, Object> documentData = null;
+            if (!isDelete) {
+                String docLine = reader.readLine();
+                if (docLine == null || docLine.isBlank()) continue;
+                documentData = OBJECT_MAPPER.readValue(docLine, MAP_TYPE);
+            }
+
+            JacksonEvent.Builder builder = JacksonEvent.builder()
+                    .withEventType(EventType.DOCUMENT.toString());
+            if (documentData != null) {
+                builder.withData(documentData);
+            }
+            if (timeReceived != null) {
+                builder.withTimeReceived(timeReceived);
+            }
+            JacksonEvent event = builder.build();
+
+            event.getMetadata().setAttribute(
+                    BulkAPIEventMetadataKeyAttributes.BULK_API_EVENT_METADATA_ATTRIBUTE_ACTION, action);
+            if (metadata != null) {
+                setIfPresent(event, metadata, "_index",
+                        BulkAPIEventMetadataKeyAttributes.BULK_API_EVENT_METADATA_ATTRIBUTE_INDEX);
+                setIfPresent(event, metadata, "_id",
+                        BulkAPIEventMetadataKeyAttributes.BULK_API_EVENT_METADATA_ATTRIBUTE_ID);
+                setIfPresent(event, metadata, "routing",
+                        BulkAPIEventMetadataKeyAttributes.BULK_API_EVENT_METADATA_ATTRIBUTE_ROUTING);
+                setIfPresent(event, metadata, "pipeline",
+                        BulkAPIEventMetadataKeyAttributes.BULK_API_EVENT_METADATA_ATTRIBUTE_PIPELINE);
+            }
+
+            eventConsumer.accept(new Record<>(event));
+        }
+    }
+
+    private String extractAction(Map<String, Object> actionLine) {
+        return Arrays.stream(OpenSearchBulkActions.values())
+                .map(OpenSearchBulkActions::toString)
+                .filter(actionLine::containsKey)
+                .findFirst()
+                .orElse(null);
+    }
+
+    private void setIfPresent(JacksonEvent event, Map<String, Object> metadata, String key, String attribute) {
+        Object value = metadata.get(key);
+        if (value != null) {
+            event.getMetadata().setAttribute(attribute, value.toString());
+        }
+    }
+}