Add DLQ support to CloudWatchLogs sink

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/failures/DlqObject.java

@@ -7,9 +7,9 @@
 
 
 import com.fasterxml.jackson.annotation.JsonIgnore;
-import org.opensearch.dataprepper.model.event.EventHandle;
 import org.apache.commons.lang3.StringUtils;
 import org.opensearch.dataprepper.model.event.EventHandle;
+import org.opensearch.dataprepper.model.configuration.PluginSetting;
 
 import java.time.Instant;
 import java.time.ZoneId;
@@ -122,6 +122,16 @@ public String toString() {
             '}';
     }
 
+    public static DlqObject createDlqObject(PluginSetting pluginSetting, EventHandle eventHandle, Object failedData) {
+        return DlqObject.builder()
+                .withEventHandle(eventHandle)
+                .withFailedData(failedData)
+                .withPluginName(pluginSetting.getName())
+                .withPipelineName(pluginSetting.getPipelineName())
+                .withPluginId(pluginSetting.getName())
+                .build();
+    }
+
     public static Builder builder() {
         return new Builder();
     }

data-prepper-api/src/test/java/org/opensearch/dataprepper/model/failures/DlqObjectTest.java

@@ -5,6 +5,7 @@
 
 package org.opensearch.dataprepper.model.failures;
 
+import org.opensearch.dataprepper.model.configuration.PluginSetting;
 import org.opensearch.dataprepper.model.event.EventHandle;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Nested;
@@ -18,6 +19,8 @@
 import java.time.ZonedDateTime;
 import java.time.format.DateTimeFormatter;
 
+import java.util.HashMap;
+import java.util.Map;
 import static java.util.UUID.randomUUID;
 import static org.hamcrest.CoreMatchers.allOf;
 import static org.hamcrest.CoreMatchers.containsString;
@@ -31,6 +34,7 @@
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.when;
 import static org.mockito.ArgumentMatchers.any;
 
 public class DlqObjectTest {
@@ -119,6 +123,24 @@ public void test_invalid_failedData() {
             failedData = null;
             assertThrows(NullPointerException.class, this::createTestObject);
         }
+
+        @Test
+        public void test_createDlqObject() {
+            final String testName = randomUUID().toString();
+            final String testPipelineName = randomUUID().toString();
+            PluginSetting pluginSetting = mock(PluginSetting.class);
+            when(pluginSetting.getName()).thenReturn(testName);
+            when(pluginSetting.getPipelineName()).thenReturn(testPipelineName);
+            eventHandle = mock(EventHandle.class);
+            Map<String, Object> data = new HashMap<>();
+            DlqObject dlqObject = DlqObject.createDlqObject(pluginSetting, eventHandle, data);
+            assertThat(dlqObject, is(notNullValue()));
+            assertThat(dlqObject.getEventHandle(), is(eventHandle));
+            assertThat(dlqObject.getFailedData(), is(data));
+            assertThat(dlqObject.getPluginName(), is(testName));
+            assertThat(dlqObject.getPipelineName(), is(testPipelineName));
+
+        }
     }
 
     @Nested

data-prepper-plugins/cloudwatch-logs/build.gradle

@@ -26,12 +26,14 @@ configurations {
 dependencies {
     implementation project(':data-prepper-plugins:aws-plugin-api')
     implementation project(path: ':data-prepper-plugins:common')
+    implementation project(path: ':data-prepper-plugins:failures-common')
     implementation 'io.micrometer:micrometer-core'
     implementation 'com.fasterxml.jackson.core:jackson-core'
     implementation 'com.fasterxml.jackson.core:jackson-databind'
     implementation 'com.fasterxml.jackson.dataformat:jackson-dataformat-yaml'
     implementation 'software.amazon.awssdk:cloudwatch'
     implementation 'software.amazon.awssdk:cloudwatchlogs'
+    implementation 'software.amazon.awssdk:s3'
     implementation libs.commons.lang3
     implementation 'org.projectlombok:lombok:1.18.26'
     implementation 'org.hibernate.validator:hibernate-validator:8.0.0.Final'
@@ -66,6 +68,7 @@ task integrationTest(type: Test) {
     systemProperty 'tests.cloudwatch.log_stream', System.getProperty('tests.cloudwatch.log_stream')
     systemProperty 'tests.aws.region', System.getProperty('tests.aws.region')
     systemProperty 'tests.aws.role', System.getProperty('tests.aws.role')
+    systemProperty 'tests.s3.bucket', System.getProperty('tests.s3.bucket')
     filter {
         includeTestsMatching '*IT'
     }