Added resetBuffer method, removed unnecessary RetransmissionException…

…, and added logString pass in parameter for staging log events.

Signed-off-by: Marcos Gonzalez Mayedo <[email protected]>

data-prepper-plugins/cloudwatch-logs/src/main/java/org/opensearch/dataprepper/plugins/sink/buffer/Buffer.java

@@ -34,4 +34,6 @@ public interface Buffer {
     List<byte[]> getBufferedData();
 
     void clearBuffer();
+
+    void resetBuffer();
 }

data-prepper-plugins/cloudwatch-logs/src/main/java/org/opensearch/dataprepper/plugins/sink/buffer/InMemoryBuffer.java

@@ -10,7 +10,7 @@
 import java.util.List;
 
 public class InMemoryBuffer implements Buffer {
-    private final List<byte[]> eventsBuffered;
+    private List<byte[]> eventsBuffered;
     private int bufferSize = 0;
 
     InMemoryBuffer() {
@@ -49,4 +49,10 @@ public void clearBuffer() {
         bufferSize = 0;
         eventsBuffered.clear();
     }
+
+    @Override
+    public void resetBuffer() {
+        bufferSize = 0;
+        eventsBuffered = new ArrayList<>();
+    }
 }

data-prepper-plugins/cloudwatch-logs/src/main/java/org/opensearch/dataprepper/plugins/sink/client/CloudWatchLogsService.java

@@ -82,7 +82,8 @@ public CloudWatchLogsService(final Buffer buffer,
     public void processLogEvents(final Collection<Record<Event>> logs) {
         sinkStopWatch.startIfNotRunning();
         for (Record<Event> log: logs) {
-            int logLength = log.getData().toJsonString().length();
+            String logString = log.getData().toJsonString();
+            int logLength = logString.length();
 
             if (cloudWatchLogsLimits.isGreaterThanMaxEventSize(logLength)) {
                 LOG.warn("Event blocked due to Max Size restriction! {Event Size: {} bytes}", (logLength + CloudWatchLogsLimits.APPROXIMATE_LOG_EVENT_OVERHEAD_SIZE));
@@ -100,12 +101,12 @@ public void processLogEvents(final Collection<Record<Event>> logs) {
 
             if ((cloudWatchLogsLimits.isGreaterThanLimitReached(time, bufferSizeWithAddedEvent, bufferEventCountWithEvent) && (bufferEventCount > 0))) {
                 stageLogEvents();
-                addToBuffer(log);
+                addToBuffer(log, logString);
             } else if (cloudWatchLogsLimits.isEqualToLimitReached(bufferSizeWithAddedEvent, bufferEventCountWithEvent)) {
-                addToBuffer(log);
+                addToBuffer(log, logString);
                 stageLogEvents();
             } else {
-                addToBuffer(log);
+                addToBuffer(log, logString);
             }
 
             bufferLock.unlock();
@@ -115,13 +116,12 @@ public void processLogEvents(final Collection<Record<Event>> logs) {
     private void stageLogEvents() {
         sinkStopWatch.stopAndResetStopWatch();
 
-        ArrayList<byte[]> eventMessageClone = new ArrayList<>();
-        cloneLists(buffer.getBufferedData(), eventMessageClone);
+        List<byte[]> eventMessageClone = buffer.getBufferedData();
 
         ThreadTaskEvents dataToPush = new ThreadTaskEvents(eventMessageClone, bufferedEventHandles);
         taskQueue.add(dataToPush);
 
-        buffer.clearBuffer();
+        buffer.resetBuffer();
         bufferedEventHandles = new ArrayList<>();
 
         CloudWatchLogsDispatcher newTaskDispatcher = CloudWatchLogsDispatcher.builder()
@@ -137,11 +137,11 @@ private void stageLogEvents() {
         asyncExecutor.execute(newTaskDispatcher);
     }
 
-    private void addToBuffer(final Record<Event> log) {
+    private void addToBuffer(final Record<Event> log, final String logString) {
         if (log.getData().getEventHandle() != null) {
             bufferedEventHandles.add(log.getData().getEventHandle());
         }
-        buffer.writeEvent(log.getData().toString().getBytes());
+        buffer.writeEvent(logString.getBytes());
     }
 
     private void cloneLists(List<byte[]> listToCopy, List<byte[]> listToCopyInto) {