Add negative tests to sqs and cloudwatch logs sink

data-prepper-plugins/cloudwatch-logs/src/integrationTest/java/org/opensearch/dataprepper/plugins/sink/cloudwatch_logs/CloudWatchLogsIT.java

@@ -60,6 +60,7 @@
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.CoreMatchers.notNullValue;
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 
 import java.time.Duration;
 import java.time.Instant;
@@ -542,6 +543,37 @@ void testWithBadCredentials_AllEventsToDLQ() {
 
     }
 
+    @Test
+    public void testToVerifyLackOfCredentialsResultInFailure() throws Exception {
+        long startTime = Instant.now().toEpochMilli();
+        when(thresholdConfig.getBatchSize()).thenReturn(1);
+        when(thresholdConfig.getMaxEventSizeBytes()).thenReturn(1000L);
+        when(thresholdConfig.getMaxRequestSizeBytes()).thenReturn(1000L);
+        lenient().when(thresholdConfig.getFlushInterval()).thenReturn(1L);
+        AwsCredentialsProvider provider = mock(AwsCredentialsProvider.class);
+        when(awsCredentialsSupplier.getProvider(any())).thenReturn(provider);
+
+        sink = createObjectUnderTest();
+        Collection<Record<Event>> records = getRecordList(NUM_RECORDS);
+        sink.doOutput(records);
+
+        assertThrows( org.awaitility.core.ConditionTimeoutException.class, () ->  await().atMost(Duration.ofSeconds(2))
+                .untilAsserted(() -> {
+                    long endTime = Instant.now().toEpochMilli();
+                    GetLogEventsRequest getRequest = GetLogEventsRequest
+                                       .builder()
+                                       .logGroupName(logGroupName)
+                                       .logStreamName(logStreamName)
+                                       .startTime(startTime)
+                                       .endTime(endTime)
+                                       .build();
+                    GetLogEventsResponse response = cloudWatchLogsClient.getLogEvents(getRequest);
+                    List<OutputLogEvent> events = response.events();
+                    assertThat(events.size(), equalTo(NUM_RECORDS));
+        }));
+        assertThat(eventsSuccessCount.get(), equalTo(0));
+    }
+
     private Collection<Record<Event>> getRecordList(int numberOfRecords) {
         final Collection<Record<Event>> recordList = new ArrayList<>();
         List<HashMap> records = generateRecords(numberOfRecords);

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

@@ -66,11 +66,11 @@
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 import static org.mockito.Mockito.lenient;
-import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.times;
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.time.Duration;
@@ -170,7 +170,7 @@ void setUp() {
         requestsFailedCounter = mock(Counter.class);
         dlqSuccessCounter = mock(Counter.class);
         summary = mock(DistributionSummary.class);
-        doNothing().when(summary).record(any(Double.class));
+        lenient().doNothing().when(summary).record(any(Double.class));
         lenient().doAnswer((a)-> {
             int v = (int)(double)(a.getArgument(0));
             eventsSuccessCount.addAndGet(v);
@@ -258,7 +258,7 @@ void setUp() {
         when(sqsSinkConfig.getDlq()).thenReturn(null);
 
         thresholdConfig = mock(SqsThresholdConfig.class);
-        when(sqsSinkConfig.getMaxRetries()).thenReturn(3);
+        lenient().when(sqsSinkConfig.getMaxRetries()).thenReturn(3);
         when(thresholdConfig.getMaxEventsPerMessage()).thenReturn(1);
         when(thresholdConfig.getMaxMessageSizeBytes()).thenReturn(250*1024L);
         when(sqsSinkConfig.getThresholdConfig()).thenReturn(thresholdConfig);
@@ -672,6 +672,29 @@ void TestWithManyRecordsWithRandomSizes() throws Exception {
         verify(eventHandle, times(numRecords)).release(true);
     }
 
+    @Test
+    public void testToVerifyLackOfCredentialsResultInFailure() throws Exception {
+        AwsCredentialsProvider provider = mock(AwsCredentialsProvider.class);
+        when(awsCredentialsSupplier.getProvider(any())).thenReturn(provider);
+        when(thresholdConfig.getMaxEventsPerMessage()).thenReturn(1);
+        lenient().when(thresholdConfig.getFlushInterval()).thenReturn(1L);
+        sink = createObjectUnderTest();
+        int numRecords = 2;
+        Collection<Record<Event>> records = getRecordList(numRecords, false);
+        sink.doOutput(records);
+        assertThrows( org.awaitility.core.ConditionTimeoutException.class, () ->  await().atMost(Duration.ofSeconds(2))
+                .untilAsserted(() -> {
+                    final Map<String, Object> expectedMap = new HashMap<>();
+                    for (int i = 0; i < numRecords; i++) {
+                        expectedMap.put("Person"+i, Integer.toString(i));
+                    }
+                    List<Message> msgs = getMessages(queueUrl);
+                    messages.addAll(msgs);
+                    assertThat(messages.size(), equalTo(numRecords));
+        }));
+        assertThat(eventsSuccessCount.get(), equalTo(0));
+    }
+
 
     private Collection<Record<Event>> getRecordList(int numberOfRecords, boolean randomSize) throws Exception {
         final Collection<Record<Event>> recordList = new ArrayList<>();