[GitHub] [flink] elphastori commented on a diff in pull request #21102: [FLINK-26890][Connector/Kinesis] Handle invalid shards in DynamoDB

[GitBox <gi...@apache.org>]

elphastori commented on code in PR #21102:
URL: https://github.com/apache/flink/pull/21102#discussion_r1009635044


##########
flink-connectors/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/proxy/DynamoDBStreamsProxy.java:
##########
@@ -110,6 +111,24 @@ public GetShardListResult getShardList(Map<String, String> streamNamesWithLastSe
         return result;
     }
 
+    @Override
+    public String getShardIterator(
+            StreamShardHandle shard, String shardIteratorType, @Nullable Object startingMarker)
+            throws InterruptedException {
+        try {
+            return super.getShardIterator(shard, shardIteratorType, startingMarker);
+        } catch (ResourceNotFoundException re) {
+            if (LOG.isInfoEnabled()) {
+                LOG.info(
+                        "Received ResourceNotFoundException. "
+                                + "Shard {} of stream {} is no longer valid, marking it as complete.",
+                        shard.getShard().getShardId(),
+                        shard.getStreamName());
+            }
+            return null;
+        }
+    }
+

Review Comment:
   Thanks, that's a good catch. From analysing the code, `getShardIterator` is only called after `getShardList`, particularly in the `DynamoDBStreamsProxy` which guarantees that deleted streams/tables throw an unhandled `NoResourceFoundException`.
   
   This is for both when the job is
   - starting (with and without a savepoint)
   - running - the consumer is discovering new shards.
   
   I've now the tested the following scenarios to verify the analysis - the failure modes have not changed
   - Starting the Flink job with or without a savepoint for a recently deleted stream/table (< 24 hrs) results in all shards getting closed - the DDB streams service returns a stream with `null` shard iterators
   - Deleting a table/stream on the fly also results in all shards getting closed
   - Using a stream/table that has been deleted for more than 24 hrs (or providing an altogether incorrect stream/table name) results in the job failing with a `NoResourceFoundException` thrown in `getShardList`



-- 
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