[jira] [Comment Edited] (FLINK-35396) DynamoDB Streams Consumer consumption data is out of order

["Suxing Lee (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-35396?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17847688#comment-17847688 ] 

Suxing Lee edited comment on FLINK-35396 at 5/19/24 2:48 PM:
-------------------------------------------------------------

[~chalixar] Thanks for your reply.
`dynamodb-streams-kinesis-adapter` SDK use [ShardSyncer/ Leases|https://github.com/awslabs/dynamodb-streams-kinesis-adapter/blob/4fc3d79a0a0ea51b05175510b4cbd7daaa286587/src/main/java/com/amazonaws/services/dynamodbv2/streamsadapter/DynamoDBStreamsShardSyncer.java] to ensure that 
child shard does not start processing data until data from all its parents has been processed.
But this logic does not seem to be seen in `FlinkDynamoDBStreamsConsumer`. This may be the cause of the problem.


was (Author: suxing lee):
[~chalixar] Thanks for your reply.
`dynamodb-streams-kinesis-adapter` SDK uses  [ShardSyncer/ Leases|https://github.com/awslabs/dynamodb-streams-kinesis-adapter/blob/4fc3d79a0a0ea51b05175510b4cbd7daaa286587/src/main/java/com/amazonaws/services/dynamodbv2/streamsadapter/DynamoDBStreamsShardSyncer.java] to ensure that 
child shard does not start processing data until data from all its parents has been processed.
But this logic does not seem to be seen in `FlinkDynamoDBStreamsConsumer`. This may be the cause of the problem.

> DynamoDB Streams Consumer consumption data is out of order
> ----------------------------------------------------------
>
>                 Key: FLINK-35396
>                 URL: https://issues.apache.org/jira/browse/FLINK-35396
>             Project: Flink
>          Issue Type: Bug
>          Components: Connectors / AWS, Connectors / Kinesis
>    Affects Versions: 1.16.2, aws-connector-4.2.0
>            Reporter: Suxing Lee
>            Priority: Major
>              Labels: AWS
>
> When we use `FlinkDynamoDBStreamsConsumer` in `flink-connector-aws/flink-connector-kinesis` to consume dynamodb stream data, there is an out-of-order problem.
> The service exception log is as follows:
> {noformat}
> 2024-05-06 00:00:40,639 INFO  org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher [] - Subtask 0 has discovered a new shard StreamShardHandle{streamName='arn:aws:dynamodb:ap-southeast-1:***', shard='{ShardId: shardId-00000001714924828427-d73b6b68,
>   ParentShardId: shardId-00000001714910797443-fb1d3b22,HashKeyRange: {StartingHashKey: 0,EndingHashKey: 1},SequenceNumberRange: {StartingSequenceNumber: 2958376400000000058201168012,}}'} due to resharding, and will start consuming the shard from sequence number EARLIEST_SEQUENCE_NUM with ShardConsumer 2807
> ......
> ......
> ......
> 2024-05-06 00:00:46,729 INFO  org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher [] - Subtask 0 has reached the end of subscribed shard: StreamShardHandle{streamName='arn:aws:dynamodb:ap-southeast-1:***', shard='{ShardId: shardId-00000001714910797443-fb1d3b22,ParentShardId: shardId-00000001714897099372-17932b9a,HashKeyRange: {StartingHashKey: 0,EndingHashKey: 1},SequenceNumberRange: {StartingSequenceNumber: 2955440900000000051102788386,}}'}
> {noformat}
> It looks like the failure process is:
> `2024-05-06 00:00:40,639` A new shard is discovered and new sub-shards are consumed immediately.（ShardId: shardId-00000001714924828427-d73b6b68）.
> `2024-05-06 00:00:46,729` Consume the old parent shard:（ShardId: shardId-00000001714910797443-fb1d3b22）end.
> There was a gap of 6 seconds. In other words, before the data consumption of the parent shard has finished, the child shard has already started consuming data. This causes the data we read to be sent downstream out of order.
> https://github.com/apache/flink-connector-aws/blob/c688a8545ac1001c8450e8c9c5fe8bbafa13aeba/flink-connector-aws/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/internals/KinesisDataFetcher.java#L689-L740 
> This is because the code immediately submits `ShardConsumer` to `shardConsumersExecutor` when `discoverNewShards` is created, and `shardConsumersExecutor` is created through Executors.newCachedThreadPool(), which does not limit the number of threads, causing new and old shards to be consumed at the same time , so data consumption is out of order?
> `flink-connector-kinesis` relies on `dynamodb-streams-kinesis-adapter` to subscribe to messages from dynamodb stream. But why does `dynamodb-streams-kinesis-adapter` directly consume data without similar problems?



--
This message was sent by Atlassian Jira
(v8.20.10#820010)