Addressed review comments

Signed-off-by: Krishna Kondaka <krishkdk@amazon.com>

Author: kkondaka

data-prepper-plugins/aws-common/build.gradle

@@ -3,6 +3,7 @@ dependencies {
     implementation 'software.amazon.awssdk:auth'
     implementation 'software.amazon.awssdk:apache-client'
     implementation 'org.apache.httpcomponents.client5:httpclient5:5.3.1'
+    implementation 'com.fasterxml.jackson.core:jackson-annotations' 
     testImplementation 'org.hibernate.validator:hibernate-validator:8.0.2.Final'
 }
 

data-prepper-plugins/aws-plugin-api/build.gradle

@@ -3,7 +3,7 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-package org.opensearch.dataprepper.plugins.common.aws;
+package org.opensearch.dataprepper.aws.api;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 import jakarta.validation.constraints.Size;

…repper/plugins/common/aws/AwsConfig.java …earch/dataprepper/aws/api/AwsConfig.javadata-prepper-plugins/aws-common/src/main/java/org/opensearch/dataprepper/plugins/common/aws/AwsConfig.java renamed to data-prepper-plugins/aws-plugin-api/src/main/java/org/opensearch/dataprepper/aws/api/AwsConfig.java data-prepper-plugins/aws-common/src/main/java/org/opensearch/dataprepper/plugins/common/aws/AwsConfig.java renamed to data-prepper-plugins/aws-plugin-api/src/main/java/org/opensearch/dataprepper/aws/api/AwsConfig.java

@@ -77,8 +77,6 @@ public double getDlqFailedCounter() {
     public boolean perform(final List<DlqObject> dlqObjects) {
         try {
             if (dlqWriter != null && dlqObjects != null && dlqObjects.size() > 0) {
-                for (DlqObject dlqObject: dlqObjects) {
-                }
                 dlqWriter.write(dlqObjects, dlqPluginSetting.getPipelineName(), dlqPluginSetting.getName());
                 dlqSuccessCounter.increment(dlqObjects.size());
                 return true;

data-prepper-plugins/failures-common/src/main/java/org/opensearch/dataprepper/plugins/dlq/DlqPushHandler.java

@@ -3,11 +3,6 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-plugins {
-    id 'java'
-    id 'java-library'
-}
-
 sourceSets {
     integrationTest {
         java {
@@ -25,12 +20,11 @@ configurations {
 
 dependencies {
     implementation project(':data-prepper-plugins:aws-plugin-api')
-    implementation project(path: ':data-prepper-api')
-    implementation project(path: ':data-prepper-plugins:common')
-    implementation project(path: ':data-prepper-plugins:aws-common')
-    implementation project(path: ':data-prepper-plugins:failures-common')
-    implementation project(path: ':data-prepper-plugins:sqs-common')
-    implementation project(path: ':data-prepper-plugins:parse-json-processor:')
+    implementation project(':data-prepper-api')
+    implementation project(':data-prepper-plugins:common')
+    implementation project(':data-prepper-plugins:failures-common')
+    implementation project(':data-prepper-plugins:sqs-common')
+    implementation project(':data-prepper-plugins:parse-json-processor:')
     implementation 'io.micrometer:micrometer-core'
     implementation 'com.fasterxml.jackson.core:jackson-core'
     implementation 'com.fasterxml.jackson.core:jackson-databind'
@@ -40,9 +34,10 @@ dependencies {
     implementation 'software.amazon.awssdk:s3'
     implementation 'software.amazon.awssdk:sts'
     implementation libs.commons.lang3
+    implementation libs.armeria.core
     implementation 'org.projectlombok:lombok:1.18.26'
     implementation 'org.hibernate.validator:hibernate-validator:8.0.0.Final'
-    testImplementation project(path: ':data-prepper-test-common')
+    testImplementation project(':data-prepper-test-common')
     annotationProcessor 'org.projectlombok:lombok:1.18.24'
 }
 
@@ -57,10 +52,6 @@ jacocoTestCoverageVerification {
     }
 }
 
-test {
-    useJUnitPlatform()
-}
-
 task integrationTest(type: Test) {
     group = 'verification'
     testClassesDirs = sourceSets.integrationTest.output.classesDirs

data-prepper-plugins/sqs-sink/build.gradle

@@ -9,7 +9,7 @@
 import org.opensearch.dataprepper.model.configuration.PluginModel;
 import org.opensearch.dataprepper.model.plugin.PluginFactory;
 import org.opensearch.dataprepper.aws.api.AwsCredentialsSupplier;
-import org.opensearch.dataprepper.plugins.common.aws.AwsConfig;
+import org.opensearch.dataprepper.aws.api.AwsConfig;
 import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
 import io.micrometer.core.instrument.Counter;
 import com.fasterxml.jackson.databind.ObjectMapper;

data-prepper-plugins/sqs-sink/src/integrationTest/java/org/opensearch/dataprepper/plugins/sink/sqs/SqsSinkIT.java

@@ -9,9 +9,10 @@
 import org.opensearch.dataprepper.metrics.PluginMetrics;
 import org.opensearch.dataprepper.model.configuration.PluginModel;
 import org.opensearch.dataprepper.model.plugin.PluginFactory;
+import org.opensearch.dataprepper.model.annotations.Experimental;
 
 import org.opensearch.dataprepper.aws.api.AwsCredentialsSupplier;
-import org.opensearch.dataprepper.plugins.common.aws.AwsConfig;
+import org.opensearch.dataprepper.aws.api.AwsConfig;
 import org.opensearch.dataprepper.aws.api.AwsCredentialsOptions;
 import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
 import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin;
@@ -36,6 +37,7 @@
 import java.time.Duration;
 import java.util.Collection;
 
+@Experimental
 @DataPrepperPlugin(name = "sqs", pluginType = Sink.class, pluginConfigurationType = SqsSinkConfig.class)
 public class SqsSink extends AbstractSink<Record<Event>> {
 
@@ -55,7 +57,7 @@ public SqsSink(final PluginSetting pluginSetting,
                    final AwsCredentialsSupplier awsCredentialsSupplier) {
         super(pluginSetting);
         this.sqsSinkConfig = sqsSinkConfig;
-        sinkInitialized = Boolean.FALSE;
+        sinkInitialized = false;
         final PluginModel codecConfiguration = sqsSinkConfig.getCodec();
         final PluginSetting codecPluginSettings = new PluginSetting(codecConfiguration.getPluginName(),
                 codecConfiguration.getPluginSettings());
@@ -97,7 +99,7 @@ public boolean isReady() {
 
     @Override
     public void doInitialize() {
-        sinkInitialized = Boolean.TRUE;
+        sinkInitialized = true;
     }
 
     /**

data-prepper-plugins/sqs-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/sqs/SqsSink.java

@@ -28,7 +28,7 @@
 import java.util.function.BiConsumer;
 
 public class SqsSinkBatch {
-    public static final int MAX_ENTRIES_PER_BATCH = 10;
+    public static final int MAX_MESSAGES_PER_BATCH = 10;
     public static final int MAX_BATCH_SIZE_BYTES = 256*1024;
     private static final String SQS_FIFO_SUFFIX = ".fifo";
     private long lastFlushedTime;
@@ -73,7 +73,7 @@ public String getQueueUrl() {
     }
 
     private boolean isFull() {
-        return entries.size() == MAX_ENTRIES_PER_BATCH && (currentBatchEntry.getEventCount() == maxEvents || currentBatchEntry.getSize() == maxMessageSize);
+        return entries.size() == MAX_MESSAGES_PER_BATCH && (currentBatchEntry.getEventCount() == maxEvents || currentBatchEntry.getSize() == maxMessageSize);
     }
 
     public boolean willExceedLimits(long estimatedSize) {
@@ -86,7 +86,7 @@ public boolean willExceedLimits(long estimatedSize) {
                 return false;
             }
         }
-        if (entries.size() == MAX_ENTRIES_PER_BATCH) {
+        if (entries.size() == MAX_MESSAGES_PER_BATCH) {
             return true;
         }
         return false;
@@ -102,8 +102,8 @@ public boolean addEntry(final Event event, String groupId, String deDupId, final
                 currentBatchEntry.complete();
             }
         }
-        if (entries.size() == MAX_ENTRIES_PER_BATCH) {
-            throw new RuntimeException("Exceeds max entries per batch");
+        if (entries.size() == MAX_MESSAGES_PER_BATCH) {
+            throw new RuntimeException("Exceeds max messages per batch");
         }
         if (groupId == null) {
             groupId = UUID.randomUUID().toString();

data-prepper-plugins/sqs-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/sqs/SqsSinkBatch.java

@@ -15,10 +15,10 @@
 import java.util.List;
 
 public class SqsSinkBatchEntry {
-    private List<EventHandle> eventHandles;
+    private final List<EventHandle> eventHandles;
     private final String groupId;
     private final String deDupId;
-    private Buffer buffer;
+    private final Buffer buffer;
     private final OutputCodec codec;
     private final OutputCodecContext codecContext;
     private int eventCount;

data-prepper-plugins/sqs-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/sqs/SqsSinkBatchEntry.java

@@ -5,7 +5,7 @@
 
 package org.opensearch.dataprepper.plugins.sink.sqs;
 
-import org.opensearch.dataprepper.plugins.common.aws.AwsConfig;
+import org.opensearch.dataprepper.aws.api.AwsConfig;
 import org.opensearch.dataprepper.model.configuration.PluginModel;
 
 import com.fasterxml.jackson.annotation.JsonProperty;