Update the version of KCL from 2.6.0 to 2.7.2

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

Author: sbose2k21

.github/workflows/kinesis-source-integration-tests.yml

@@ -5,13 +5,18 @@ on:
     paths:
       - 'data-prepper-plugins/kinesis-source/**'
       - '*gradle*'
-  pull_request:
+  pull_request_target:
+    types: [ opened, synchronize, reopened ]
     paths:
       - 'data-prepper-plugins/kinesis-source/**'
       - '*gradle*'
 
   workflow_dispatch:
 
+permissions:
+  id-token: write
+  contents: read
+
 jobs:
   build:
     strategy:

data-prepper-plugins/kinesis-source/build.gradle

@@ -34,7 +34,7 @@ dependencies {
     implementation libs.armeria.core
     implementation 'com.fasterxml.jackson.core:jackson-core'
     implementation 'io.micrometer:micrometer-core'
-    implementation 'software.amazon.kinesis:amazon-kinesis-client:2.6.0'
+    implementation 'software.amazon.kinesis:amazon-kinesis-client:2.7.2'
     compileOnly 'org.projectlombok:lombok:1.18.20'
     annotationProcessor 'org.projectlombok:lombok:1.18.20'
 

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

@@ -249,7 +249,7 @@ public void testKinesisService() throws Exception {
         when(pluginFactory.loadPlugin(eq(InputCodec.class), any()))
                 .thenReturn(new NdjsonInputCodec(new NdjsonInputConfig(), TestEventFactory.getTestEventFactory()));
 
-        final List<Record<?>> actualRecordsWritten = new ArrayList<>();
+        final List<Record<?>> actualRecordsWritten = Collections.synchronizedList(new ArrayList<>());
         doAnswer(a -> actualRecordsWritten.addAll(a.getArgument(0, Collection.class)))
                 .when(buffer).writeAll(anyCollection(), anyInt());
 

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

@@ -10,7 +10,6 @@
 
 package org.opensearch.dataprepper.plugins.kinesis.source;
 
-import com.amazonaws.SdkClientException;
 import com.linecorp.armeria.client.retry.Backoff;
 import lombok.Getter;
 import lombok.Setter;
@@ -34,6 +33,7 @@
 import org.opensearch.dataprepper.plugins.kinesis.source.processor.KinesisShardRecordProcessorFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.core.exception.SdkClientException;
 import software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClient;
 import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient;
 import software.amazon.awssdk.services.dynamodb.model.BillingMode;

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

@@ -14,6 +14,7 @@
 import lombok.extern.slf4j.Slf4j;
 import org.opensearch.dataprepper.plugins.kinesis.source.exceptions.KinesisConsumerNotFoundException;
 import software.amazon.awssdk.arns.Arn;
+import software.amazon.awssdk.core.exception.SdkClientException;
 import software.amazon.awssdk.services.kinesis.KinesisAsyncClient;
 import software.amazon.awssdk.services.kinesis.model.DescribeStreamConsumerRequest;
 import software.amazon.awssdk.services.kinesis.model.DescribeStreamConsumerResponse;
@@ -111,7 +112,7 @@ private DescribeStreamSummaryResponse getStreamDescriptionSummary(
 
     private void handleStreamSummaryException(CompletionException ex, String streamName) {
         Throwable cause = ex.getCause();
-        if (cause instanceof KinesisException || cause instanceof com.amazonaws.SdkClientException) {
+        if (cause instanceof KinesisException || cause instanceof SdkClientException) {
             log.error("AWS error while describing stream summary for stream {}: {}",
                     streamName, ex.getMessage());
         } else {
@@ -136,7 +137,7 @@ private DescribeStreamConsumerResponse describeStreamConsumer(
 
     private void handleConsumerException(CompletionException ex, String streamArn, int attempt) {
         Throwable cause = ex.getCause();
-        if (cause instanceof KinesisException || cause instanceof com.amazonaws.SdkClientException) {
+        if (cause instanceof KinesisException || cause instanceof SdkClientException) {
             log.error("AWS error while describing stream consumer for stream {}: {}. Attempt {}.",
                     streamArn, ex.getMessage(), attempt + 1);
         } else {