Removed unnecessary changes

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

Author: kkondaka

data-prepper-plugins/common/src/main/java/org/opensearch/dataprepper/common/sink/DefaultSinkMetrics.java

@@ -14,7 +14,7 @@ public class DefaultSinkMetrics implements SinkMetrics {
     public static final String SINK_REQUESTS_SUCCEEDED = "SinkRequestsSucceeded";
     public static final String SINK_EVENTS_SUCCEEDED = "SinkEventsSucceeded";
     public static final String SINK_EVENTS_FAILED = "SinkEventsFailed";
-    public static final String SINK_EVENTS_FAILED = "SinkEventsDropped";
+    public static final String SINK_EVENTS_DROPPED = "SinkEventsDropped";
     public static final String SINK_REQUESTS_FAILED = "SinkRequestsFailed";
     public static final String SINK_REQUEST_LATENCY = "SinkRequestLatency";
     public static final String SINK_RETRIES = "SinkRetries";
@@ -33,7 +33,7 @@ public DefaultSinkMetrics(final PluginMetrics pluginMetrics, final String sinkPr
         this.sinkEventsSucceeded = pluginMetrics.counter(sinkPrefix + "Sink"+eventName+"sSucceeded");
         this.sinkRequestsFailed = pluginMetrics.counter(sinkPrefix + SINK_REQUESTS_FAILED);
         this.sinkEventsFailed = pluginMetrics.counter(sinkPrefix + "Sink"+eventName+"sFailed");
-        this.sinkEventsFailed = pluginMetrics.counter(sinkPrefix + "Sink"+eventName+"sDropped");
+        this.sinkEventsDropped = pluginMetrics.counter(sinkPrefix + "Sink"+eventName+"sDropped");
         this.sinkRetries = pluginMetrics.counter(sinkPrefix + SINK_RETRIES);
         this.sinkRequestLatency = pluginMetrics.summary(sinkPrefix + SINK_REQUEST_LATENCY);
         this.sinkRequestSize = pluginMetrics.summary(sinkPrefix + SINK_REQUEST_SIZE);

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

@@ -6,6 +6,8 @@
 package org.opensearch.dataprepper.plugins.sink.sqs;
 
 import org.opensearch.dataprepper.model.event.Event;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import software.amazon.awssdk.services.sqs.model.BatchResultErrorEntry;
 import org.opensearch.dataprepper.model.codec.OutputCodec;
 import org.opensearch.dataprepper.model.sink.OutputCodecContext;
@@ -30,7 +32,10 @@
 import java.util.UUID;
 import java.util.function.BiConsumer;
 
+import static org.opensearch.dataprepper.logging.DataPrepperMarkers.NOISY;
+
 public class SqsSinkBatch {
+    private static final Logger LOG = LoggerFactory.getLogger(SqsSinkBatch.class);
     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";
@@ -227,6 +232,7 @@ public boolean flushOnce() {
             sinkMetrics.recordRequestSize((double)requestSize);
 
         } catch (SqsException e) {
+            LOG.error(NOISY, "Failed to send messages to SQS: {}", e.getMessage());
             sinkMetrics.incrementRequestsFailedCounter(1);
             sinkMetrics.incrementEventsFailedCounter(entries.size());
             if (!isRetryableException(e)) {

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

@@ -31,6 +31,7 @@ public class SqsSinkConfig {
     private PluginModel codec;
 
     @JsonProperty("threshold")
+    @Valid
     private SqsThresholdConfig thresholdConfig = new SqsThresholdConfig();
 
     @JsonProperty("max_retries")

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

@@ -30,11 +30,11 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
-import org.opensearch.dataprepper.common.sink.RetrySinkOutputStrategy;
+import java.util.concurrent.locks.ReentrantLock;
 
 import static org.opensearch.dataprepper.logging.DataPrepperMarkers.NOISY;
 
-public class SqsSinkService extends RetrySinkOutputStrategy {
+public class SqsSinkService extends SqsSinkExecutor {
     private static final Logger LOG = LoggerFactory.getLogger(SqsSinkService.class);
     public static final int MAX_BYTES_IN_BATCH = 256*1024;
     public static final int MAX_EVENT_SIZE = 256*1024;
@@ -48,6 +48,7 @@ public class SqsSinkService extends RetrySinkOutputStrategy {
     private final boolean isDynamicDeDupId;
     private final boolean isDynamicQueueUrl;
     private final ExpressionEvaluator expressionEvaluator;
+    private final ReentrantLock reentrantLock;
     private final SqsThresholdConfig thresholdConfig;
     private final SqsSinkConfig sqsSinkConfig;
     private final SinkContext sinkContext;
@@ -74,6 +75,7 @@ public SqsSinkService(final SqsSinkConfig sqsSinkConfig,
         this.thresholdConfig = sqsSinkConfig.getThresholdConfig();
         this.codec = codec;
         this.sqsSinkConfig = sqsSinkConfig;
+        reentrantLock = new ReentrantLock();
         this.sinkMetrics = new SqsSinkMetrics(pluginMetrics);
 
         queueUrl = sqsSinkConfig.getQueueUrl();
@@ -122,7 +124,7 @@ public void pushDLQList() {
     }
 
     @Override
-    public void addFailedObjectsToDlqList(Object object) {
+    public void pushFailedObjectsToDlq(Object object) {
         List<SqsSinkBatch> failedBatches = (List<SqsSinkBatch>) object;
         for (SqsSinkBatch failedBatch: failedBatches) {
             for (Map.Entry<String, SqsSinkBatchEntry> entry: failedBatch.getEntries().entrySet()) {
@@ -138,7 +140,7 @@ public long getEstimatedSize(final Event event) throws Exception {
     }
 
     @Override
-    public boolean willExceedMaxRequestSizeBytes(final Event event, final long estimatedSize) throws Exception {
+    public boolean willExceedMaxBatchSize(final Event event, final long estimatedSize) throws Exception {
         String qUrl = getQueueUrl(event, false);
         if (qUrl == null)
             return false;
@@ -306,6 +308,16 @@ public void addEventToDLQList(final Event event, Throwable ex) {
         addMessageToDLQ(event.toJsonString(), eventHandles, ex.getMessage());
     }
 
+    @Override
+    public void lock() {
+        reentrantLock.lock();
+    }
+
+    @Override
+    public void unlock() {
+        reentrantLock.unlock();
+    }
+
     void output(Collection<Record<Event>> records) {
         execute(records);
     }

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

@@ -10,6 +10,8 @@
 import jakarta.validation.constraints.Max;
 import org.hibernate.validator.constraints.time.DurationMax;
 import org.hibernate.validator.constraints.time.DurationMin;
+//import org.opensearch.dataprepper.model.constraints.ByteCountMax;
+//import org.opensearch.dataprepper.model.constraints.ByteCountMin;
 import org.opensearch.dataprepper.model.types.ByteCount;
 
 import java.time.Duration;
@@ -25,6 +27,8 @@ public class SqsThresholdConfig {
     private int maxEventsPerMessage = DEFAULT_MESSAGES_PER_EVENT;
 
     @JsonProperty("max_message_size")
+    //@ByteCountMin("1b")
+    //@ByteCountMax("1mb")
     private ByteCount maxMessageSize = DEFAULT_MAX_MESSAGE_SIZE;
 
     @JsonProperty("flush_interval")