Avoid breaking change by making the aws config block required.

- Updated README
- Fixed init bug
- Performed e2e and perf tests

Signed-off-by: huy pham <huyp@amazon.com>

Author: huypham612

data-prepper-plugins/otlp-sink/README.md

@@ -17,6 +17,25 @@ to any OTLP Protobuf-compatible endpoint.
 
 ## Sample Pipeline Configuration
 
+### Minimal Configuration (No STS)
+
+Use this when Data Prepper has permission to write to AWS X-Ray directly.
+
+```yaml
+otlp_pipeline:
+  source:
+    otel_trace_source:
+
+  sink:
+    - otlp:
+        endpoint: "https://xray.us-west-2.amazonaws.com/v1/traces"
+        aws: { }
+```
+
+### Full Configuration with STS
+
+Use this when assuming a cross-account role is required.
+
 ```yaml
 otlp_pipeline:
   workers: 2
@@ -56,7 +75,7 @@ otlp_pipeline:
 | `threshold.max_events`     | `int`    | No       | `512` (recommended)   | Maximum number of spans per batch. Use `0` to disable count-based flushing. Must be ≥ 0.                 |
 | `threshold.max_batch_size` | `String` | No       | `1mb` (recommended)   | Maximum total payload bytes per batch. Supports human-readable suffixes (`kb`, `mb`).                    |   
 | `threshold.flush_timeout`  | `String` | No       | `200ms` (recommended) | Maximum time to wait before flushing a non-empty batch. Minimum: 1ms (e.g., `200ms`, `1s`)               |
-| **aws**                    | `Object` | No       | —                     | AWS authentication settings. See below.                                                                  |
+| **aws**                    | `Object` | Yes      | —                     | AWS authentication settings. Use `{}` if no STS role is needed. See below.                               |
 | `aws.sts_role_arn`         | `String` | No       | —                     | IAM Role ARN that Data Prepper (or OSI) assumes to send spans to X-Ray on behalf of a customer account.  |
 | `aws.sts_external_id`      | `String` | No       | —                     | External ID to use when assuming the role. Required only if the target IAM role enforces sts:ExternalId. |
 

data-prepper-plugins/otlp-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/otlp/OtlpSink.java

@@ -32,6 +32,7 @@
         pluginConfigurationType = OtlpSinkConfig.class
 )
 public class OtlpSink extends AbstractSink<Record<Span>> {
+    private volatile boolean initialized = false;
 
     private final OtlpSinkBuffer buffer;
     private final OtlpSinkMetrics sinkMetrics;
@@ -48,6 +49,8 @@ public class OtlpSink extends AbstractSink<Record<Span>> {
     public OtlpSink(@Nonnull final AwsCredentialsSupplier awsCredentialsSupplier, @Nonnull final OtlpSinkConfig config, @Nonnull final PluginMetrics pluginMetrics, @Nonnull final PluginSetting pluginSetting) {
         super(pluginSetting);
 
+        config.validate();
+
         this.sinkMetrics = new OtlpSinkMetrics(pluginMetrics, pluginSetting);
         this.buffer = new OtlpSinkBuffer(awsCredentialsSupplier, config, sinkMetrics);
     }
@@ -58,6 +61,7 @@ public OtlpSink(@Nonnull final AwsCredentialsSupplier awsCredentialsSupplier, @N
     @Override
     public void doInitialize() {
         buffer.start();
+        initialized = true;
     }
 
     /**
@@ -79,7 +83,7 @@ public void doOutput(@Nonnull final Collection<Record<Span>> records) {
      */
     @Override
     public boolean isReady() {
-        return buffer.isRunning();
+        return initialized && buffer.isRunning();
     }
 
     /**

data-prepper-plugins/otlp-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/otlp/configuration/OtlpSinkConfig.java

@@ -68,7 +68,7 @@ public long getFlushTimeoutMillis() {
      */
     @JsonProperty("aws")
     @Valid
-    private AwsConfig awsAuthenticationConfig;
+    private AwsConfig awsConfig;
 
     /**
      * Get AWS region from the provided endpoint.
@@ -97,18 +97,30 @@ public Region getAwsRegion() {
     }
 
     public String getStsRoleArn() {
-        if (awsAuthenticationConfig == null) {
+        if (awsConfig == null || awsConfig.getAwsStsRoleArn() == null) {
             return null;
         }
 
-        return awsAuthenticationConfig.getAwsStsRoleArn();
+        return awsConfig.getAwsStsRoleArn();
     }
 
     public String getStsExternalId() {
-        if (awsAuthenticationConfig == null) {
+        if (awsConfig == null || awsConfig.getAwsStsExternalId() == null) {
             return null;
         }
 
-        return awsAuthenticationConfig.getAwsStsExternalId();
+        return awsConfig.getAwsStsExternalId();
+    }
+
+    /**
+     * Validate the AWS configuration.
+     * This method ensures breaking change in future release where non-AWS OTLP endpoints are supported.
+     *
+     * @throws IllegalArgumentException if the AWS configuration is invalid
+     */
+    public void validate() {
+        if (awsConfig == null) {
+            throw new IllegalArgumentException("aws configuration is required");
+        }
     }
 }

data-prepper-plugins/otlp-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/otlp/http/OtlpHttpSender.java

@@ -14,7 +14,6 @@
 import com.linecorp.armeria.common.HttpMethod;
 import com.linecorp.armeria.common.HttpRequest;
 import com.linecorp.armeria.common.HttpResponse;
-import com.linecorp.armeria.common.MediaType;
 import com.linecorp.armeria.common.RequestHeaders;
 import com.linecorp.armeria.common.RequestHeadersBuilder;
 import io.opentelemetry.proto.collector.trace.v1.ExportTraceServiceRequest;
@@ -42,7 +41,6 @@
 public class OtlpHttpSender {
     private static final Logger LOG = LoggerFactory.getLogger(OtlpHttpSender.class);
     private static final Set<Integer> RETRYABLE_STATUS_CODES = Set.of(429, 502, 503, 504);
-    private static final MediaType PROTOBUF = MediaType.parse("application/x-protobuf");
 
     private final SigV4Signer signer;
     private final WebClient webClient;
@@ -85,7 +83,6 @@ public OtlpHttpSender(@Nonnull final AwsCredentialsSupplier awsCredentialsSuppli
      * adding complexity with minimal benefit for most OTLP endpoints.
      * - Our exponential backoff already handles typical retry intervals gracefully.
      */
-
     private static WebClient buildWebClient(final OtlpSinkConfig config) {
         final RetryRuleWithContent<HttpResponse> retryRule = RetryRuleWithContent.<HttpResponse>builder()
                 .onStatus((ctx, status) -> RETRYABLE_STATUS_CODES.contains(status.code()))
@@ -118,11 +115,14 @@ public void send(@Nonnull final List<Pair<ResourceSpans, EventHandle>> batch) {
             return;
         }
 
-        final Pair<byte[], byte[]> payloadAndCompressedPayload = getPayloadAndCompressedPayload(batch);
-        final int spans = batch.size();
+        // Defensive copy to avoid ConcurrentModificationException
+        final List<Pair<ResourceSpans, EventHandle>> immutableBatch = List.copyOf(batch);
+
+        final Pair<byte[], byte[]> payloadAndCompressedPayload = getPayloadAndCompressedPayload(immutableBatch);
+        final int spans = immutableBatch.size();
         if (payloadAndCompressedPayload.right().length == 0) {
             sinkMetrics.incrementFailedSpansCount(spans);
-            releaseAllEventHandle(batch, false);
+            releaseAllEventHandle(immutableBatch, false);
             return;
         }
 
@@ -139,12 +139,12 @@ public void send(@Nonnull final List<Pair<ResourceSpans, EventHandle>> batch) {
 
                     final int statusCode = response.status().code();
                     final byte[] responseBytes = response.content().array();
-                    handleResponse(statusCode, responseBytes, batch);
+                    handleResponse(statusCode, responseBytes, immutableBatch);
                 })
                 .exceptionally(e -> {
                     LOG.error("Failed to send {} spans.", spans, e);
                     sinkMetrics.incrementRejectedSpansCount(spans);
-                    releaseAllEventHandle(batch, false);
+                    releaseAllEventHandle(immutableBatch, false);
                     return null;
                 });
     }
@@ -161,12 +161,14 @@ private Pair<byte[], byte[]> getPayloadAndCompressedPayload(final List<Pair<Reso
 
     private HttpRequest buildHttpRequest(final byte[] compressedPayload) {
         final SdkHttpFullRequest signedRequest = signer.signRequest(compressedPayload);
+
         final RequestHeadersBuilder headersBuilder = RequestHeaders.builder()
                 .method(HttpMethod.POST)
-                .path(signedRequest.getUri().getPath())
-                .contentType(PROTOBUF)
-                .add("Content-Encoding", "gzip");
+                .scheme(signedRequest.getUri().getScheme())
+                .path(signedRequest.getUri().getRawPath())
+                .authority(signedRequest.getUri().getAuthority());
 
+        // ONLY use the signed headers
         signedRequest.headers().forEach((k, vList) -> vList.forEach(v -> headersBuilder.add(k, v)));
         return HttpRequest.of(headersBuilder.build(), HttpData.wrap(compressedPayload));
     }

data-prepper-plugins/otlp-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/otlp/http/SigV4Signer.java

@@ -63,6 +63,7 @@ SdkHttpFullRequest signRequest(@Nonnull final byte[] payload) {
                 .method(SdkHttpMethod.POST)
                 .uri(endpointUri)
                 .putHeader("Content-Type", "application/x-protobuf")
+                .putHeader("Content-Encoding", "gzip")
                 .contentStreamProvider(() -> SdkBytes.fromByteArray(payload).asInputStream())
                 .build();
 

data-prepper-plugins/otlp-sink/src/test/java/org/opensearch/dataprepper/plugins/sink/otlp/OtlpSinkTest.java

@@ -4,8 +4,10 @@
  */
 package org.opensearch.dataprepper.plugins.sink.otlp;
 
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.function.Executable;
 import org.opensearch.dataprepper.aws.api.AwsCredentialsSupplier;
 import org.opensearch.dataprepper.metrics.PluginMetrics;
 import org.opensearch.dataprepper.model.configuration.PluginSetting;
@@ -21,6 +23,7 @@
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
@@ -40,6 +43,7 @@ void setUp() throws Exception {
         mockAwsCredSupplier = mock(AwsCredentialsSupplier.class);
         mockConfig = mock(OtlpSinkConfig.class);
         when(mockConfig.getAwsRegion()).thenReturn(Region.of("us-west-2"));
+        when(mockConfig.getEndpoint()).thenReturn("https://localhost/v1/traces");
 
         mockMetrics = mock(PluginMetrics.class);
 
@@ -66,6 +70,23 @@ void testInitialize_startsBuffer() {
         verify(mockBuffer).start();
     }
 
+    @Test
+    void testConstructor_throwsWhenAwsConfigIsMissing() {
+        doThrow(new IllegalArgumentException("aws configuration is required"))
+                .when(mockConfig).validate();
+
+        // Act & Assert
+        final Executable constructorCall = () ->
+                new OtlpSink(mockAwsCredSupplier, mockConfig, mockMetrics, mockSetting);
+
+        final IllegalArgumentException thrown = Assertions.assertThrows(
+                IllegalArgumentException.class,
+                constructorCall
+        );
+
+        Assertions.assertEquals("aws configuration is required", thrown.getMessage());
+    }
+
     @Test
     void testOutput_addsEveryRecordToBuffer() {
         // Arrange
@@ -82,12 +103,17 @@ void testOutput_addsEveryRecordToBuffer() {
     }
 
     @Test
-    void testIsReady_delegatesToBuffer() {
-        // true case
+    void testIsReady_returnsTrueOnlyAfterInitialization() {
         when(mockBuffer.isRunning()).thenReturn(true);
+
+        // Not initialized yet
+        assertFalse(target.isReady());
+
+        // Initialize, which sets 'initialized = true' and starts the buffer
+        target.initialize();
         assertTrue(target.isReady());
 
-        // false case
+        // Now simulate buffer being not running
         when(mockBuffer.isRunning()).thenReturn(false);
         assertFalse(target.isReady());
     }

data-prepper-plugins/otlp-sink/src/test/java/org/opensearch/dataprepper/plugins/sink/otlp/buffer/OtlpSinkBufferTest.java

@@ -53,6 +53,7 @@ class OtlpSinkBufferTest {
     @BeforeEach
     void setUp() {
         config = mock(OtlpSinkConfig.class);
+        when(config.getEndpoint()).thenReturn("https://localhost/v1/traces");
         when(config.getMaxEvents()).thenReturn(2);
         when(config.getMaxRetries()).thenReturn(2);
         when(config.getMaxBatchSize()).thenReturn(1_000_000L);
@@ -456,7 +457,7 @@ void testMaxEventsZeroDoesNotTriggerFlush() throws Exception {
     }
 
     @Test
-    void testDaemonThreadConfiguration() throws Exception {
+    void testDaemonThreadConfiguration() {
         // This test verifies that the thread is created as non-daemon
         // We can't directly test this, but we can verify the thread factory is called
         buffer.start();