Merge branch 'main' into fix-dyn-crashes

Signed-off-by: Chase <[email protected]>

.github/workflows/data-prepper-log-analytics-basic-grok-e2e-tests.yml

@@ -13,7 +13,7 @@ jobs:
   build:
     strategy:
       matrix:
-        java: [11, 17]
+        java: [11, 17, 21, docker]
         test: ['basicLogEndToEndTest', 'parallelGrokStringSubstituteTest']
       fail-fast: false
 

.github/workflows/data-prepper-peer-forwarder-local-node-e2e-tests.yml

@@ -13,7 +13,7 @@ jobs:
   build:
     strategy:
       matrix:
-        java: [11, 17]
+        java: [11, 17, 21, docker]
       fail-fast: false
 
     runs-on: ubuntu-latest

.github/workflows/data-prepper-peer-forwarder-static-e2e-tests.yml

@@ -13,7 +13,7 @@ jobs:
   build:
     strategy:
       matrix:
-        java: [11, 17]
+        java: [11, 17, 21, docker]
         test: ['staticAggregateEndToEndTest', 'staticLogMetricsEndToEndTest']
       fail-fast: false
 

.github/workflows/data-prepper-trace-analytics-raw-span-compatibility-e2e-tests.yml

@@ -13,7 +13,7 @@ jobs:
   build:
     strategy:
       matrix:
-        java: [11, 17]
+        java: [11, 17, 21, docker]
       fail-fast: false
 
     runs-on: ubuntu-latest

.github/workflows/data-prepper-trace-analytics-raw-span-e2e-tests.yml

@@ -13,7 +13,7 @@ jobs:
   build:
     strategy:
       matrix:
-        java: [11, 17]
+        java: [11, 17, 21, docker]
         otelVersion: ['0.9.0-alpha', '0.16.0-alpha']
       fail-fast: false
 

.github/workflows/data-prepper-trace-analytics-raw-span-peer-forwarder-e2e-tests.yml

@@ -13,7 +13,7 @@ jobs:
   build:
     strategy:
       matrix:
-        java: [11, 17]
+        java: [11, 17, 21, docker]
       fail-fast: false
 
     runs-on: ubuntu-latest

.github/workflows/data-prepper-trace-analytics-service-map-e2e-tests.yml

@@ -13,7 +13,7 @@ jobs:
   build:
     strategy:
       matrix:
-        java: [11, 17]
+        java: [11, 17, 21, docker]
       fail-fast: false
 
     runs-on: ubuntu-latest

.github/workflows/gradle.yml

@@ -12,7 +12,7 @@ jobs:
   build:
     strategy:
       matrix:
-        java: [11, 17]
+        java: [11, 17, 21]
       fail-fast: false
 
     runs-on: ubuntu-latest

.github/workflows/opensearch-sink-opensearch-integration-tests.yml

@@ -20,7 +20,7 @@ jobs:
     strategy:
       matrix:
         java: [11]
-        opensearch: [1.0.1, 1.1.0, 1.2.4, 1.3.9, 2.0.1, 2.1.0, 2.2.1, 2.3.0, 2.4.0, 2.5.0, 2.6.0]
+        opensearch: [1.0.1, 1.1.0, 1.2.4, 1.3.13, 2.0.1, 2.1.0, 2.3.0, 2.5.0, 2.7.0, 2.9.0, 2.11.0]
       fail-fast: false
 
     runs-on: ubuntu-latest

.github/workflows/performance-test-compile.yml

@@ -19,7 +19,7 @@ jobs:
   build:
     strategy:
       matrix:
-        java: [11, 17]
+        java: [11, 17, 21]
 
     runs-on: ubuntu-latest
 

build.gradle

@@ -26,7 +26,7 @@ buildscript {
 }
 
 plugins {
-    id 'com.diffplug.spotless' version '6.11.0'
+    id 'com.diffplug.spotless' version '6.22.0'
     id 'io.spring.dependency-management' version '1.1.0'
 }
 

buildSrc/build.gradle

@@ -0,0 +1,9 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+plugins {
+    id 'java-gradle-plugin'
+    id 'java'
+}

buildSrc/src/main/java/org/opensearch/dataprepper/gradle/end_to_end/DockerProviderTask.java

@@ -0,0 +1,24 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.gradle.end_to_end;
+
+import org.gradle.api.DefaultTask;
+import org.gradle.api.provider.Property;
+import org.gradle.api.tasks.Input;
+
+/**
+ * A task which can provide a Docker image to use for an end-to-end test.
+ */
+public abstract class DockerProviderTask extends DefaultTask {
+    /**
+     * The Docker image with both the name and tag in the standard string
+     * format - <i>my-image:mytag</i>
+     *
+     * @return The Docker image
+     */
+    @Input
+    abstract Property<String> getImageId();
+}

data-prepper-api/src/main/java/org/opensearch/dataprepper/expression/ExpressionEvaluator.java

@@ -34,4 +34,6 @@ default Boolean evaluateConditional(final String statement, final Event context)
     }
 
     Boolean isValidExpressionStatement(final String statement);
-}
+
+    Boolean isValidFormatExpression(final String format);
+}

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/acknowledgements/AcknowledgementSet.java

@@ -8,6 +8,9 @@
 import org.opensearch.dataprepper.model.event.Event;
 import org.opensearch.dataprepper.model.event.EventHandle;
 
+import java.time.Duration;
+import java.util.function.Consumer;
+
 /**
  * AcknowledgmentSet keeps track of set of events that
  * belong to the batch of events that a source creates.
@@ -58,4 +61,16 @@ public interface AcknowledgementSet {
       * initial events are going through the pipeline line.
      */
     public void complete();
+
+    /**
+     * adds progress check callback to the acknowledgement set. When added
+     * the callback is called every progressCheckInterval time with the 
+     * indication of current progress as a ratio of pending number of
+     * acknowledgements over total acknowledgements
+     *
+     * @param progressCheckCallback progress check callback to be called
+     * @param progressCheckInterval frequency of invocation of progress check callback
+     * @since 2.6
+     */
+    public void addProgressCheck(final Consumer<ProgressCheck> progressCheckCallback, final Duration progressCheckInterval);
 }

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/acknowledgements/ProgressCheck.java

@@ -0,0 +1,17 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.model.acknowledgements;
+
+public interface ProgressCheck {
+    /**
+     * Returns the pending ratio
+     *
+     * @return returns the ratio of pending to the total acknowledgements
+     * @since 2.6
+     */
+    Double getRatio();
+}
+

data-prepper-core/src/main/java/org/opensearch/dataprepper/breaker/CircuitBreaker.java → data-prepper-api/src/main/java/org/opensearch/dataprepper/model/breaker/CircuitBreaker.java

@@ -3,20 +3,20 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-package org.opensearch.dataprepper.breaker;
+package org.opensearch.dataprepper.model.breaker;
 
 /**
  * Represents a circuit breaker in Data Prepper.
  *
- * @since 2.1
+ * @since 2.6
  */
 public interface CircuitBreaker {
     /**
      * Checks a circuit breaker. If open, then the circuit breaker has
      * been tripped.
      *
      * @return true if open; false if closed.
-     * @since 2.1
+     * @since 2.6
      */
     boolean isOpen();
 }

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

@@ -49,7 +49,7 @@ public interface Buffer<T extends Record<?>> {
      * @throws RuntimeException Other exceptions
      */
     default void writeBytes(final byte[] bytes, final String key, int timeoutInMillis) throws Exception {
-        throw new RuntimeException("Not supported");
+        throw new UnsupportedOperationException("This buffer type does not support bytes.");
     }
 
     /**
@@ -92,6 +92,17 @@ default Duration getDrainTimeout() {
         return Duration.ZERO;
     }
 
+    /**
+     * Indicates if writes to this buffer are also in some way written
+     * onto the JVM heap. If writes do go on heap, this should <b>false</b>
+     * which is the default.
+     *
+     * @return True if this buffer does not write to the JVM heap.
+     */
+    default boolean isWrittenOffHeapOnly() {
+        return false;
+    }
+
     /**
      * shuts down the buffer
      */

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

@@ -0,0 +1,90 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.model.buffer;
+
+import org.opensearch.dataprepper.model.CheckpointState;
+import org.opensearch.dataprepper.model.record.Record;
+
+import java.time.Duration;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.TimeoutException;
+
+/**
+ * An implementation of {@link Buffer} which delegates all calls to a delgate
+ * (or inner) buffer.
+ * <p>
+ * This class exists to help with writing decorators of the {@link Buffer} interface.
+ *
+ * @param <T> The type of data in the buffer
+ *
+ * @since 2.6
+ */
+public abstract class DelegatingBuffer<T extends Record<?>> implements Buffer<T> {
+    private final Buffer<T> delegateBuffer;
+
+    /**
+     * Constructor for subclasses to use.
+     *
+     * @param delegateBuffer The delegate (or inner) buffer.
+     *
+     * @since 2.6
+     */
+    protected DelegatingBuffer(final Buffer<T> delegateBuffer) {
+        this.delegateBuffer = Objects.requireNonNull(delegateBuffer);
+    }
+
+    @Override
+    public void write(final T record, final int timeoutInMillis) throws TimeoutException {
+        delegateBuffer.write(record, timeoutInMillis);
+    }
+
+    @Override
+    public void writeAll(final Collection<T> records, final int timeoutInMillis) throws Exception {
+        delegateBuffer.writeAll(records, timeoutInMillis);
+    }
+
+    @Override
+    public void writeBytes(final byte[] bytes, final String key, final int timeoutInMillis) throws Exception {
+        delegateBuffer.writeBytes(bytes, key, timeoutInMillis);
+    }
+
+    @Override
+    public Map.Entry<Collection<T>, CheckpointState> read(final int timeoutInMillis) {
+        return delegateBuffer.read(timeoutInMillis);
+    }
+
+    @Override
+    public void checkpoint(final CheckpointState checkpointState) {
+        delegateBuffer.checkpoint(checkpointState);
+    }
+
+    @Override
+    public boolean isEmpty() {
+        return delegateBuffer.isEmpty();
+    }
+
+    @Override
+    public boolean isByteBuffer() {
+        return delegateBuffer.isByteBuffer();
+    }
+
+    @Override
+    public Duration getDrainTimeout() {
+        return delegateBuffer.getDrainTimeout();
+    }
+
+    @Override
+    public boolean isWrittenOffHeapOnly() {
+        return delegateBuffer.isWrittenOffHeapOnly();
+    }
+
+    @Override
+    public void shutdown() {
+        delegateBuffer.shutdown();
+    }
+}

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

@@ -161,6 +161,7 @@ public <T> List<T> getTypedList(final String attribute, final Class<T> type) {
      * Returns the value of the specified {@literal List<Map<String, String>>}, or {@code defaultValue} if this settings contains no value for
      * the attribute.
      *
+     * @param attribute    attribute to be looked up
      * @param keyType      key type of the Map
      * @param valueType    value type stored in the Map
      * @param <K> The key type