[GitHub] [flink-connector-aws] hlteoh37 commented on a diff in pull request #65: [FLINK-31108][Connectors/Kinesis] Use Stream ARN instead of Stream Na…

["hlteoh37 (via GitHub)" <gi...@apache.org>]

hlteoh37 commented on code in PR #65:
URL: https://github.com/apache/flink-connector-aws/pull/65#discussion_r1182318616


##########
flink-connector-kinesis/src/main/resources/META-INF/NOTICE:
##########
@@ -45,15 +45,15 @@ This project bundles the following dependencies under the Apache Software Licens
 - com.fasterxml.jackson.core:jackson-databind:2.13.4.2
 - com.fasterxml.jackson.core:jackson-core:2.13.4
 - com.fasterxml.jackson.core:jackson-annotations:2.13.4
-- com.amazonaws:jmespath-java:1.12.276
+- com.amazonaws:jmespath-java:1.12.439
 - com.amazonaws:dynamodb-streams-kinesis-adapter:1.5.3
-- com.amazonaws:aws-java-sdk-sts:1.12.276
-- com.amazonaws:aws-java-sdk-s3:1.12.276
-- com.amazonaws:aws-java-sdk-kms:1.12.276
-- com.amazonaws:aws-java-sdk-kinesis:1.12.276
-- com.amazonaws:aws-java-sdk-dynamodb:1.12.276
-- com.amazonaws:aws-java-sdk-core:1.12.276
-- com.amazonaws:aws-java-sdk-cloudwatch:1.12.276
+- com.amazonaws:aws-java-sdk-sts:1.12.439

Review Comment:
   good shout - updated



##########
flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxy.java:
##########
@@ -140,6 +141,9 @@ public class KinesisProxy implements KinesisProxyInterface {
     /** Exponential backoff power constant for the describe stream operation. */
     private final double describeStreamExpConstant;
 
+    /** Caches retrieved stream ARNs for give stream names. */
+    private Map<String, String> streamNameToArnLookup = new ConcurrentHashMap<>();

Review Comment:
   done



##########
flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/KinesisProxy.java:
##########
@@ -488,6 +495,15 @@ private List<StreamShardHandle> getShardsOfStream(
         return shardsOfStream;
     }
 
+    private synchronized String lookupStreamArn(String streamName) throws InterruptedException {
+        if (streamNameToArnLookup.containsKey(streamName)) {
+            return streamNameToArnLookup.get(streamName);
+        }
+        String streamArn = describeStream(streamName, null).getStreamDescription().getStreamARN();

Review Comment:
   It would first respect the `DESCRIBE_STREAM` restart configurations set by customer (defaults to 3 max retries). And if it fails after that, it would restart the job. Given we are caching the result, we should only see this on job startup.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org