Addressed review comments

kkondaka · kkondaka · commit f1182d6bcb8d · 2025-04-24T23:25:01.000Z
Signed-off-by: Krishna Kondaka &lt;krishkdk@amazon.com&gt;
diff --git a/data-prepper-plugins/aws-plugin-api/build.gradle b/data-prepper-plugins/aws-plugin-api/build.gradle
@@ -4,6 +4,7 @@ dependencies {
     implementation 'software.amazon.awssdk:apache-client'
     implementation 'org.apache.httpcomponents.client5:httpclient5:5.3.1'
     implementation 'com.fasterxml.jackson.core:jackson-annotations' 
+    testImplementation libs.commons.lang3
     testImplementation 'org.hibernate.validator:hibernate-validator:8.0.2.Final'
 }
 
diff --git a/data-prepper-plugins/aws-plugin-api/src/test/java/org/opensearch/dataprepper/aws/api/AwsConfigTest.java b/data-prepper-plugins/aws-plugin-api/src/test/java/org/opensearch/dataprepper/aws/api/AwsConfigTest.java
@@ -3,7 +3,7 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-package org.opensearch.dataprepper.plugins.common.aws;
+package org.opensearch.dataprepper.aws.api;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
diff --git a/data-prepper-plugins/parse-json-processor/src/main/java/org/opensearch/dataprepper/plugins/codec/json/NdjsonOutputCodec.java b/data-prepper-plugins/parse-json-processor/src/main/java/org/opensearch/dataprepper/plugins/codec/json/NdjsonOutputCodec.java
@@ -39,15 +39,24 @@ public void start(final OutputStream outputStream, Event event, final OutputCode
         this.codecContext = codecContext;
     }
 
+    private String getFullEventJson(final Event event, final OutputCodecContext context) {
+        return event.jsonBuilder()
+                .includeKeys(context.getIncludeKeys())
+                .excludeKeys(context.getExcludeKeys())
+                .includeTags(context.getTagsTargetKey())
+                .toJsonString();
+    }
+
+    @Override
+    public long getEstimatedSize(final Event event, final OutputCodecContext context) throws IOException {
+        return getFullEventJson(event, context).length();
+    }
+
     @Override
     public void writeEvent(final Event event, final OutputStream outputStream) throws IOException {
         Objects.requireNonNull(event);
 
-        String json = event.jsonBuilder()
-                .includeKeys(codecContext.getIncludeKeys())
-                .excludeKeys(codecContext.getExcludeKeys())
-                .includeTags(codecContext.getTagsTargetKey())
-                .toJsonString();
+        String json = getFullEventJson(event, codecContext);
         outputStream.write(json.getBytes());
         outputStream.write(System.lineSeparator().getBytes());
     }
diff --git a/data-prepper-plugins/parse-json-processor/src/test/java/org/opensearch/dataprepper/plugins/codec/json/NewlineDelimitedOutputCodecTest.java b/data-prepper-plugins/parse-json-processor/src/test/java/org/opensearch/dataprepper/plugins/codec/json/NewlineDelimitedOutputCodecTest.java
@@ -99,7 +99,7 @@ void testGetEstimatedSize() throws Exception {
         ndjsonOutputCodec.start(outputStream, null, codecContext);
 
         Record<Event> record = getRecord(0);
-        String expectedEventString = "{\"name\":\"Person0\",\"age\":0}\n";
+        String expectedEventString = "{\"name\":\"Person0\",\"age\":0}";
         assertThat(ndjsonOutputCodec.getEstimatedSize(record.getData(), codecContext), equalTo((long)expectedEventString.length()));
     }
 }
diff --git a/data-prepper-plugins/sqs-sink/src/integrationTest/java/org/opensearch/dataprepper/plugins/sink/sqs/SqsSinkIT.java b/data-prepper-plugins/sqs-sink/src/integrationTest/java/org/opensearch/dataprepper/plugins/sink/sqs/SqsSinkIT.java
@@ -17,7 +17,6 @@
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
-import static org.mockito.Mockito.doAnswer;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
 import org.opensearch.dataprepper.metrics.PluginMetrics;
@@ -574,12 +573,6 @@ void TestSinkOperationWithQueuesAsExpression() throws Exception {
         when(thresholdConfig.getMaxEventsPerMessage()).thenReturn(1);
         when(sqsSinkConfig.getDlq()).thenReturn(dlqConfig);
         when(expressionEvaluator.isValidFormatExpression(anyString())).thenReturn(true);
-        doAnswer( (a) -> {
-            String qUrl = (String)a.getArgument(0);
-            Event event = (Event)a.getArgument(1);
-            int idx = qUrl.indexOf("${");
-            return qUrl.substring(0, idx)+event.get("id", String.class);
-        }).when(expressionEvaluator).evaluate(anyString(), any(Event.class));
 
         when(sqsSinkConfig.getQueueUrl()).thenReturn(queueUrl+"${/id}");
 
diff --git a/data-prepper-plugins/sqs-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/sqs/SqsSink.java b/data-prepper-plugins/sqs-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/sqs/SqsSink.java
@@ -35,6 +35,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.time.Duration;
+import java.util.Map;
 import java.util.Collection;
 
 @Experimental
@@ -59,8 +60,18 @@ public SqsSink(final PluginSetting pluginSetting,
         this.sqsSinkConfig = sqsSinkConfig;
         sinkInitialized = false;
         final PluginModel codecConfiguration = sqsSinkConfig.getCodec();
-        final PluginSetting codecPluginSettings = new PluginSetting(codecConfiguration.getPluginName(),
+        final PluginSetting codecPluginSettings;
+        if (codecConfiguration != null) {
+            String codecPluginName = codecConfiguration.getPluginName();
+            if (!codecPluginName.equals("json") && !codecPluginName.equals("ndjson")) {
+                throw new RuntimeException(String.format("Codec {} not supported.", codecPluginName));
+            }
+            codecPluginSettings = new PluginSetting(codecConfiguration.getPluginName(),
                 codecConfiguration.getPluginSettings());
+        } else {
+            codecPluginSettings = new PluginSetting("ndjson", Map.of());
+        }
+        
         final OutputCodec outputCodec = pluginFactory.loadPlugin(OutputCodec.class, codecPluginSettings);
         AwsConfig awsConfig = sqsSinkConfig.getAwsConfig();
         if (awsConfig == null && awsCredentialsSupplier == null) {
diff --git a/data-prepper-plugins/sqs-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/sqs/SqsSinkConfig.java b/data-prepper-plugins/sqs-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/sqs/SqsSinkConfig.java
@@ -27,7 +27,6 @@ public class SqsSinkConfig {
     private String queueUrl;
 
     @JsonProperty("codec")
-    @NotNull
     private PluginModel codec;
 
     @JsonProperty("threshold")
@@ -45,21 +44,33 @@ public class SqsSinkConfig {
     @JsonProperty("dlq")
     private PluginModel dlq;
 
-    @AssertTrue(message = "FIFO queues wth dynamic group id or dynamic deduplication id and more than one events per message is not valid")
+    @AssertTrue(message = "FIFO queues wth dynamic group id or dynamic deduplication id and more than one events per message is not valid OR standard queues do not support groupId or deduplication configuration")
     boolean isValidConfig() {
         String deDupId = getDeDuplicationId();
         String groupId = getGroupId();
+        String queueUrl = getQueueUrl();
         boolean isDynamicDeDupId = deDupId != null && deDupId.contains("${");
         boolean isDynamicGroupId = groupId != null && groupId.contains("${");
+        boolean isDynamicQueueUrl = queueUrl != null && queueUrl.contains("${");
+        if (isDynamicQueueUrl) {
+            return true;
+        }
         if (getQueueUrl().endsWith(".fifo")) {
             if ((isDynamicGroupId  || isDynamicDeDupId) && thresholdConfig.getMaxEventsPerMessage() > 1) {
                 return false;
             } else{
                 return true;
             }
+        } else {
+             return (groupId == null && deDupId == null);
         }
-        return true;
     }
 
+    @AssertTrue(message = "ndjson codec (default codec) doesn't support max events per message greater than 1")
+    boolean isValidCondecConfig() {
+        if ((codec == null || codec.getPluginName().equals("ndjson")) && thresholdConfig.getMaxEventsPerMessage() > 1)
+            return false;
+        return true;
+    }
 }
 
diff --git a/data-prepper-plugins/sqs-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/sqs/SqsSinkService.java b/data-prepper-plugins/sqs-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/sqs/SqsSinkService.java
@@ -97,7 +97,7 @@ public SqsSinkService(final SqsSinkConfig sqsSinkConfig,
         isDynamicDeDupId = deDupId != null && deDupId.contains("${");
         if (isDynamicDeDupId) {
             if (!expressionEvaluator.isValidFormatExpression(deDupId)) {
-                throw new IllegalArgumentException("Invalid groupId expression");
+                throw new IllegalArgumentException("Invalid deduplicationId expression");
             }
         }
 
@@ -199,7 +199,7 @@ private String getQueueUrl(final Event event, boolean logError) {
         String qUrl = queueUrl;
         if (isDynamicQueueUrl) {
             try {
-                Object obj = expressionEvaluator.evaluate(queueUrl, event);
+                Object obj = event.formatString(queueUrl, expressionEvaluator);
                 if (obj instanceof String) {
                     qUrl = (String) obj;
                 } else {
@@ -220,7 +220,7 @@ private String getGroupId(final Event event) {
         String gId = groupId;
         if (isDynamicGroupId) {
             try {
-                Object obj = expressionEvaluator.evaluate(groupId, event);
+                Object obj = event.formatString(groupId, expressionEvaluator);
                 if (obj instanceof String) {
                     gId = (String) obj;
                 } else {
@@ -237,7 +237,7 @@ private String getDeDupId(final Event event) {
         String ddId = deDupId;
         if (isDynamicDeDupId) {
             try {
-                Object obj = expressionEvaluator.evaluate(deDupId, event);
+                Object obj = event.formatString(deDupId, expressionEvaluator);
                 if (obj instanceof String) {
                     ddId = (String) obj;
                 } else {
diff --git a/data-prepper-plugins/sqs-sink/src/test/java/org/opensearch/dataprepper/plugins/sink/sqs/SqsSinkTest.java b/data-prepper-plugins/sqs-sink/src/test/java/org/opensearch/dataprepper/plugins/sink/sqs/SqsSinkTest.java
@@ -49,6 +49,7 @@
 import java.util.UUID;
 
 public class SqsSinkTest {
+    private static final String TEST_CODEC_PLUGIN_NAME = "json";
     private static final String TEST_PLUGIN_NAME = "testPluginName";
     private static final String TEST_PIPELINE_NAME = "testPipelineName";
     @Mock
@@ -90,7 +91,7 @@ void setup() {
         awsCredentialsProvider = mock(AwsCredentialsProvider.class);
         when(sqsSinkConfig.getDlq()).thenReturn(null);
         codecConfig = mock(PluginModel.class);
-        when(codecConfig.getPluginName()).thenReturn(TEST_PLUGIN_NAME);
+        when(codecConfig.getPluginName()).thenReturn(TEST_CODEC_PLUGIN_NAME);
         when(codecConfig.getPluginSettings()).thenReturn(Map.of());
         when(sqsSinkConfig.getCodec()).thenReturn(codecConfig);
         queueUrl = UUID.randomUUID().toString();

Original file line number	Diff line number	Diff line change
`@@ -4,6 +4,7 @@ dependencies {`
`4`	`4`	`implementation 'software.amazon.awssdk:apache-client'`
`5`	`5`	`implementation 'org.apache.httpcomponents.client5:httpclient5:5.3.1'`
`6`	`6`	`implementation 'com.fasterxml.jackson.core:jackson-annotations'`
	`7`	`+ testImplementation libs.commons.lang3`
`7`	`8`	`testImplementation 'org.hibernate.validator:hibernate-validator:8.0.2.Final'`
`8`	`9`	`}`
`9`	`10`
Original file line number	Diff line number	Diff line change
`@@ -99,7 +99,7 @@ void testGetEstimatedSize() throws Exception {`
`99`	`99`	`ndjsonOutputCodec.start(outputStream, null, codecContext);`
`100`	`100`
`101`	`101`	`Record<Event> record = getRecord(0);`
`102`		`- String expectedEventString = "{\"name\":\"Person0\",\"age\":0}\n";`
	`102`	`+ String expectedEventString = "{\"name\":\"Person0\",\"age\":0}";`
`103`	`103`	`assertThat(ndjsonOutputCodec.getEstimatedSize(record.getData(), codecContext), equalTo((long)expectedEventString.length()));`
`104`	`104`	`}`
`105`	`105`	`}`