Attempt to fix the integration test for Kinesis source (opensearch-project#6886)

Signed-off-by: Souvik Bose <souvbose@amazon.com>
Co-authored-by: Souvik Bose <souvbose@amazon.com>

Author: sb2k16

data-prepper-plugins/kinesis-source/src/integrationTest/java/org/opensearch/dataprepper/plugins/kinesis/source/KinesisSourceIT.java

@@ -194,6 +194,7 @@ void setup() {
         when(kinesisSourceConfig.getNumberOfRecordsToAccumulate()).thenReturn(NUMBER_OF_RECORDS_TO_ACCUMULATE);
         when(kinesisSourceConfig.getBufferTimeout()).thenReturn(BUFFER_TIMEOUT);
         when(kinesisSourceConfig.getMaxInitializationAttempts()).thenReturn(MAX_INITIALIZATION_ATTEMPTS);
+        when(kinesisSourceConfig.getInitializationBackoffTime()).thenReturn(Duration.ofMillis(1000));
 
         kinesisClientFactory = mock(KinesisClientFactory.class);
         when(kinesisClientFactory.buildDynamoDBClient(kinesisLeaseCoordinationTableConfig.getAwsRegion())).thenReturn(DynamoDbAsyncClient.builder()

data-prepper-plugins/kinesis-source/src/main/java/org/opensearch/dataprepper/plugins/kinesis/source/apihandler/KinesisClientApiRetryHandler.java

@@ -51,7 +51,7 @@ public <T> T executeWithRetry(String operationName, Supplier<T> operation,
     }
 
     private void applyBackoff(int attempt) {
-        final long delayMillis = backoff.nextDelayMillis(attempt);
+        final long delayMillis = backoff.nextDelayMillis(attempt + 1);
         if (delayMillis < 0) {
             throw new KinesisRetriesExhaustedException(
                     "Retries exhausted. Make sure that configuration is valid and required permissions are present.");

data-prepper-plugins/kinesis-source/src/test/java/org/opensearch/dataprepper/plugins/kinesis/source/apihandler/KinesisClientApiRetryHandlerTest.java

@@ -81,7 +81,7 @@ void executeWithRetry_successOnFirstAttempt_returnsResult() {
 
     @Test
     void executeWithRetry_successOnSecondAttempt_returnsResult() {
-        when(backoff.nextDelayMillis(0)).thenReturn(DELAY_MILLIS);
+        when(backoff.nextDelayMillis(1)).thenReturn(DELAY_MILLIS);
         AtomicInteger attempts = new AtomicInteger(0);
         String expectedResult = "success";
 
@@ -97,7 +97,7 @@ void executeWithRetry_successOnSecondAttempt_returnsResult() {
         );
 
         assertEquals(expectedResult, result);
-        verify(backoff, times(1)).nextDelayMillis(0);
+        verify(backoff, times(1)).nextDelayMillis(1);
         verify(exceptionHandler, times(1)).handle(any(), eq(0));
     }
 
@@ -125,7 +125,7 @@ void executeWithRetry_allAttemptsFail_throwsException() {
 
     @Test
     void executeWithRetry_negativeBackoffDelay_throwsException() {
-        when(backoff.nextDelayMillis(0)).thenReturn(-1L);
+        when(backoff.nextDelayMillis(1)).thenReturn(-1L);
         RuntimeException testException = new RuntimeException("Test failure");
 
         KinesisRetriesExhaustedException exception = assertThrows(
@@ -141,13 +141,13 @@ void executeWithRetry_negativeBackoffDelay_throwsException() {
                 "Retries exhausted. Make sure that configuration is valid and required permissions are present.",
                 exception.getMessage()
         );
-        verify(backoff, times(1)).nextDelayMillis(0);
+        verify(backoff, times(1)).nextDelayMillis(1);
         verify(exceptionHandler, times(1)).handle(any(), eq(0));
     }
 
     @Test
     void executeWithRetry_interruptedDuringSleep_throwsException() {
-        when(backoff.nextDelayMillis(0)).thenReturn(DELAY_MILLIS);
+        when(backoff.nextDelayMillis(1)).thenReturn(DELAY_MILLIS);
         RuntimeException testException = new RuntimeException("Test failure");
         Thread.currentThread().interrupt();
 
@@ -160,7 +160,7 @@ void executeWithRetry_interruptedDuringSleep_throwsException() {
                 )
         );
 
-        verify(backoff, times(1)).nextDelayMillis(0);
+        verify(backoff, times(1)).nextDelayMillis(1);
         verify(exceptionHandler, times(1)).handle(any(), eq(0));
     }
 

testing/aws-testing-cdk/lib/kinesis/KinesisSourceStack.ts

@@ -27,6 +27,7 @@ export class KinesisSourceStack extends Stack {
             'kinesis:ListStreamConsumers',
             'kinesis:ListShards',
             'kinesis:DescribeStream',
+            'kinesis:DescribeStreamSummary',
             'kinesis:GetRecords',
             'kinesis:GetResourcePolicy',
             'kinesis:SubscribeToShard',
@@ -53,6 +54,7 @@ export class KinesisSourceStack extends Stack {
             'dynamodb:DescribeTimeToLive',
             'dynamodb:CreateTable',
             'dynamodb:DescribeTable',
+            'dynamodb:DeleteTable',
             'dynamodb:GetItem',
             'dynamodb:PutItem',
             'dynamodb:UpdateItem',