[hotfix][Connectors/Kinesis] Pass source configuration to SplitFetcherManager

[lukaszkrawiec]

Purpose of the change

SplitFetcherManager relies on correct Configuration be passed to its constructor such that it can honor flink options such as

• ALLOW_UNALIGNED_SOURCE_SPLITS
• ELEMENT_QUEUE_CAPACITY

This commit forwards the configuration object provided to KinesisStreamsSource to SplitFetcherManager

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 a trivial rework / code cleanup without any test coverage.

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)

[boring-cyborg]

Thanks for opening this pull request! Please check out our contributing guidelines. (https://flink.apache.org/contributing/how-to-contribute.html)

[vahmed-hamdy]

Nice Catch! Thanks for the contribution

[leekeiabstraction]

Thank you Lukasz for raising the PR!

I have a question regarding ALLOW_UNALIGNED_SOURCE_SPLITS.

I see that the configuration will affect the correctness of watermarks. Would enabling this cause records which belong to the same key/keygroup to arrive in different order? If so, have we considered excluding the passing of ALLOW_UNALIGNED_SOURCE_SPLITS to prevent users from shooting themselves in the foot?

If my assumption is correct on ordering, I am curious about the use-case where ALLOW_UNALIGNED_SOURCE_SPLITS would make sense for users especially since FIFO is a key characteristic of Kinesis stream. If there is a strong reason to support this, is there a way to warn users?