Merge branch 'main' into fix-issue-5530

Signed-off-by: Saketh Pallempati <pallempati.saketh@fmr.com>

Author: saketh-pallempati

RELEASING.md

@@ -170,6 +170,12 @@ You can also deny a release by using _deny_ or _denied_ in the comment.
 
 **NOTE** The smoke tests currently report a build failure, even when they succeed. Thus, you need to manually verify the output.
 
+### OpenSearch CI server build
+
+After two maintainers approve the GitHub issue, the [OpenSearch CI server](https://build.ci.opensearch.org/) will start a build.
+This is a Jenkins server that promotes our releases. 
+You can check the promotion status by checking the [release-data-prepper](https://build.ci.opensearch.org/job/release-data-prepper/) job.
+
 ### Further details
 
 For more details on the release build, or to setup your own GitHub repository, see [release/README.md](release/README.md).

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/buffer/Buffer.java

@@ -6,6 +6,7 @@
 package org.opensearch.dataprepper.model.buffer;
 
 import org.opensearch.dataprepper.model.CheckpointState;
+import org.opensearch.dataprepper.model.plugin.PluginComponentType;
 import org.opensearch.dataprepper.model.record.Record;
 
 import java.time.Duration;
@@ -18,6 +19,7 @@
  * Buffer queues the records between TI components and acts as a layer between source and processor/sink. Buffer can
  * be in-memory, disk based or other a standalone implementation.
  */
+@PluginComponentType("buffer")
 public interface Buffer<T extends Record<?>> {
     /**
      * writes the record to the buffer

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

@@ -8,6 +8,7 @@
 import com.fasterxml.jackson.core.JsonFactory;
 import com.fasterxml.jackson.core.JsonParser;
 import com.fasterxml.jackson.core.JsonToken;
+import com.fasterxml.jackson.core.StreamReadConstraints;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
 import org.opensearch.dataprepper.model.event.Event;
@@ -30,10 +31,15 @@ public class JsonDecoder implements ByteDecoder {
     private Collection<String> includeKeys;
     private Collection<String> includeKeysMetadata;
 
-    public JsonDecoder(String keyName, Collection<String> includeKeys, Collection<String> includeKeysMetadata) {
+    public JsonDecoder(String keyName, Collection<String> includeKeys, Collection<String> includeKeysMetadata, Integer maxEventLength) {
         this.keyName = keyName;
         this.includeKeys = includeKeys;
         this.includeKeysMetadata = includeKeysMetadata;
+        if (maxEventLength != null) {
+        jsonFactory.setStreamReadConstraints(StreamReadConstraints.builder()
+                .maxStringLength(maxEventLength)
+                .build());
+        }
     }
 
     public JsonDecoder() {
@@ -66,7 +72,7 @@ public void parse(InputStream inputStream, Instant timeReceived, Consumer<Record
             }
 
             if (jsonParser.getCurrentToken() == JsonToken.START_ARRAY) {
-                if (keyName != null && !nodeName.equals(keyName)) {
+                if (keyName != null && !keyName.equals(nodeName)) {
                     continue;
                 }
                 parseRecordsArray(jsonParser, timeReceived, eventConsumer, includeKeysMap, includeMetadataKeysMap);

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

@@ -15,21 +15,79 @@
 
 import java.io.IOException;
 import java.io.OutputStream;
+import java.io.ByteArrayOutputStream;
 import java.util.Map;
 
 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);
+        return new Writer() {
+            @Override
+            public void writeEvent(final Event event) throws IOException {
+                codec.writeEvent(event, outputStream);
+            }
+
+            @Override
+            public void complete() throws IOException {
+                codec.complete(outputStream);
+            }
+        };
+    }
+
     /**
      * this method get called from {@link Sink} to do initial wrapping in {@link OutputStream}
      * Implementors should do initial wrapping according to the implementation
      *
      * @param outputStream  outputStream param for wrapping
      * @param event         Event to auto-generate schema
      * @param context       Extra Context used in Codec.
-     * @throws IOException throws IOException when invalid input is received or not able to create wrapping
+     * @throws IOException  throws IOException when invalid input is received or not able to create wrapping
+     * @deprecated Use {@link OutputCodec#createWriter(OutputStream, Event, OutputCodecContext)} instead.
      */
+    @Deprecated
     void start(OutputStream outputStream, Event event, OutputCodecContext context) throws IOException;
 
     /**
@@ -39,7 +97,9 @@ public interface OutputCodec {
      * @param event         event Record event
      * @param outputStream  outputStream param to hold the event data
      * @throws IOException throws IOException when not able to write data to {@link OutputStream}
+     * @deprecated @deprecated Use {@link OutputCodec.Writer#writeEvent(Event)} instead.
      */
+    @Deprecated
     void writeEvent(Event event, OutputStream outputStream) throws IOException;
 
     /**
@@ -48,9 +108,26 @@ public interface OutputCodec {
      *
      * @param outputStream outputStream param for wrapping
      * @throws IOException throws IOException when invalid input is received or not able to create wrapping
+     * @deprecated @deprecated Use {@link Writer#complete()} instead.
      */
+    @Deprecated
     void complete(OutputStream outputStream) throws IOException;
 
+    /**
+     * this method get called from {@link Sink} to estimate size of event in {@link OutputStream}
+     *
+     * @param event        event Record event
+     * @return long        size of the serialized event
+     * @throws IOException throws IOException when invalid input is received or not able to create wrapping
+     */
+    default long getEstimatedSize(Event event, OutputCodecContext codecContext) throws IOException {
+        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+        start(outputStream, event, codecContext);
+        writeEvent(event, outputStream);
+        complete(outputStream);
+        return outputStream.toByteArray().length;
+    }
+
     /**
      * used to get extension of file
      *

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/configuration/DataPrepperVersion.java

@@ -6,7 +6,7 @@
 import java.util.regex.Pattern;
 
 public class DataPrepperVersion {
-    private static final String CURRENT_VERSION = "2.11";
+    private static final String CURRENT_VERSION = "2.12";
 
     private static final String FULL_FORMAT = "%d.%d";
     private static final String SHORTHAND_FORMAT = "%d";

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/encryption/EncryptionEngine.java

@@ -0,0 +1,24 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.model.encryption;
+
+public interface EncryptionEngine {
+    /**
+     * Encrypts raw data into {@link EncryptionEnvelope}.
+     *
+     * @param data the raw data in bytes
+     * @return returns the encryption envelope
+     */
+    EncryptionEnvelope encrypt(byte[] data);
+
+    /**
+     * Decrypts the encryption envelope into raw data.
+     *
+     * @param encryptionEnvelope the encryption envelope
+     * @return returns the raw data in bytes
+     */
+    byte[] decrypt(EncryptionEnvelope encryptionEnvelope);
+}

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/encryption/EncryptionEnvelope.java

@@ -0,0 +1,18 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.model.encryption;
+
+public interface EncryptionEnvelope {
+    /**
+     * The encrypted data.
+     */
+    byte[] getEncryptedData();
+
+    /**
+     * The encrypted data key.
+     */
+    String getEncryptedDataKey();
+}

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/encryption/KeyProvider.java

@@ -0,0 +1,11 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.model.encryption;
+
+@FunctionalInterface
+public interface KeyProvider {
+    byte[] decryptKey(byte[] encryptedKey);
+}

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/event/JacksonEvent.java

@@ -61,6 +61,8 @@ public class JacksonEvent implements Event {
 
     private static final Logger LOG = LoggerFactory.getLogger(JacksonEvent.class);
 
+    private static final int FILL_OUT_OF_BOUNDS_ELEMENTS_LIMIT = 0;
+
     private static final String SEPARATOR = "/";
 
     private static final ObjectMapper mapper = JsonMapper.builder()
@@ -195,8 +197,31 @@ private JsonNode getOrCreateNode(final JsonNode node, final String key) {
         JsonNode childNode = node.get(key);
         if (childNode == null) {
             childNode = mapper.createObjectNode();
-            ((ObjectNode) node).set(key, childNode);
+            if (node.isArray()) {
+                int index = Integer.parseInt(key);
+                ArrayNode arrayNode = (ArrayNode) node;
+
+                int distanceFromArrayEnd = index - arrayNode.size();
+                if (distanceFromArrayEnd >= FILL_OUT_OF_BOUNDS_ELEMENTS_LIMIT + 1) {
+                    throw new IndexOutOfBoundsException(
+                            String.format("Cannot expand array past the limit of size %s to reach index %s", arrayNode.size(), index));
+                }
+                while (arrayNode.size() <= index) {
+                    arrayNode.addNull();
+                }
+
+                JsonNode existing = arrayNode.get(index);
+                if (existing == null || !existing.isObject()) {
+                    childNode = mapper.createObjectNode();
+                    arrayNode.set(index, childNode);
+                } else {
+                    childNode = existing;
+                }
+            } else {
+                ((ObjectNode) node).set(key, childNode);
+            }
         }
+
         return childNode;
     }
 

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/failures/DlqObject.java

@@ -15,6 +15,8 @@
 import java.time.ZoneId;
 import java.time.format.DateTimeFormatter;
 import java.util.Objects;
+import java.util.List;
+import java.util.ArrayList;
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
@@ -43,8 +45,11 @@ public class DlqObject {
     @JsonIgnore
     private final EventHandle eventHandle;
 
+    @JsonIgnore
+    private final List<EventHandle> eventHandles;
+
     private DlqObject(final String pluginId, final String pluginName, final String pipelineName,
-                      final String timestamp, final Object failedData, final EventHandle eventHandle) {
+                      final String timestamp, final Object failedData, final List<EventHandle> eventHandles) {
 
         checkNotNull(pluginId, "pluginId cannot be null");
         checkArgument(!pluginId.isEmpty(), "pluginId cannot be an empty string");
@@ -58,7 +63,8 @@ private DlqObject(final String pluginId, final String pluginName, final String p
         this.pluginName = pluginName;
         this.pipelineName = pipelineName;
         this.failedData = failedData;
-        this.eventHandle = eventHandle;
+        this.eventHandles = eventHandles;
+        this.eventHandle = null;
 
         this.timestamp = StringUtils.isEmpty(timestamp) ? FORMATTER.format(Instant.now()) : timestamp;
     }
@@ -83,12 +89,18 @@ public String getTimestamp() {
         return timestamp;
     }
 
-    public EventHandle getEventHandle() {
-        return eventHandle;
+    public List<EventHandle> getEventHandles() {
+        return eventHandles;
     }
 
     public void releaseEventHandle(boolean result) {
-        if (eventHandle != null) {
+        if (eventHandles != null && eventHandles.size() == 1) {
+            eventHandles.get(0).release(result);
+        }
+    }
+
+    public void releaseEventHandles(boolean result) {
+        for (final EventHandle eventHandle: eventHandles) {
             eventHandle.release(result);
         }
     }
@@ -102,7 +114,7 @@ public boolean equals(final Object o) {
             && Objects.equals(pluginId, that.pluginId)
             && Objects.equals(pluginName, that.pluginName)
             && Objects.equals(pipelineName, that.pipelineName)
-            && Objects.equals(eventHandle, that.eventHandle)
+            && Objects.equals(eventHandles, that.eventHandles)
             && Objects.equals(timestamp, that.getTimestamp());
     }
 
@@ -122,9 +134,9 @@ public String toString() {
             '}';
     }
 
-    public static DlqObject createDlqObject(PluginSetting pluginSetting, EventHandle eventHandle, Object failedData) {
+    public static DlqObject createDlqObject(PluginSetting pluginSetting, List<EventHandle> eventHandles, Object failedData) {
         return DlqObject.builder()
-                .withEventHandle(eventHandle)
+                .withEventHandles(eventHandles)
                 .withFailedData(failedData)
                 .withPluginName(pluginSetting.getName())
                 .withPipelineName(pluginSetting.getPipelineName())
@@ -142,7 +154,7 @@ public static class Builder {
         private String pluginName;
         private String pipelineName;
         private Object failedData;
-        private EventHandle eventHandle;
+        private List<EventHandle> eventHandles;
 
         private String timestamp;
 
@@ -171,8 +183,14 @@ public Builder withTimestamp(final String timestamp) {
             return this;
         }
 
+        public Builder withEventHandles(final List<EventHandle> eventHandles) {
+            this.eventHandles = eventHandles;
+            return this;
+        }
+
         public Builder withEventHandle(final EventHandle eventHandle) {
-            this.eventHandle = eventHandle;
+            this.eventHandles = new ArrayList<>();
+            this.eventHandles.add(eventHandle);
             return this;
         }
 
@@ -182,7 +200,7 @@ public Builder withTimestamp(final Instant instant) {
         }
 
         public DlqObject build() {
-            return new DlqObject(this.pluginId, this.pluginName, this.pipelineName, this.timestamp, this.failedData, this.eventHandle);
+            return new DlqObject(this.pluginId, this.pluginName, this.pipelineName, this.timestamp, this.failedData, this.eventHandles);
         }
 
     }