[FLINK-36270][Connectors/DynamoDB] Assign only children splits of a split when it is marked as finished

[gguptp]

Purpose of the change

When a split is marked as finished, we are iterating over all splits and assigning available splits to readers. On a large enough stream with enough shards, we'd be spending a lot of time in finishedSplits because we are always calling getSplitsForAssignment function which is O(n). This leads to performance issues. Making this an O(1) algo to solve the issues

Before the fix:

• Generally we see long checkpoint duration
  

• We see that the start delay is very long
  

• App is stuck like this for long
  

After the fix:

• We see checkpoint duration now reduced to 1second

• No more failed checkpoints due to long checkpoints

Verifying this change

Please make sure both new and modified tests in this PR follows the conventions defined in our code quality guide: https://flink.apache.org/contributing/code-style-and-quality-common.html#testing

(Please pick either of the following options)

• This change is already covered by existing UTs

Significant changes

(Please check any boxes [x] if the answer is "yes". You can first publish the PR and check them afterwards, for convenience.)

• Dependencies have been added or upgraded
• Public API has been changed (Public API is any class annotated with @Public(Evolving))
• Serializers have been changed
• New feature has been introduced
  • If yes, how is this documented? (not applicable / docs / JavaDocs / not documented)

[hlteoh37]

Thanks for the contribution @gguptp ! Left some comments

[gguptp]

Thanks for the review @hlteoh37! The review comments have been addressed now

[hlteoh37]

Add some pictures for the performance bottleneck!

We see the checkpoints taking a long time to complete

We see that the time is spent on "Start Delay"

We also see new checkpoints being stuck in an "unacknowledged" state

[gguptp]

Add some pictures for the performance bottleneck!

We see the checkpoints taking a long time to complete

We see that the time is spent on "Start Delay"

We also see new checkpoints being stuck in an "unacknowledged" state

Updated the PR overview

[hlteoh37]

LGTM! Thanks for the contribution @gguptp