[FLINK-TBD][Connectors/Kinesis] Removed resubscription for EFO subscriptions when they are completed

[pelaezryan]

• *Flink ID and Jira link is TBD as I wait for my requested Jiara account to be created

Purpose of the change

Address a bug in the Kinesis Flink Connector that would cause Enhanced Fan Out (EFO) customers to continously experience IllegalArgumentException when their stream is resharded. This occurs due to the subscription being restarted when it should be completed and cleaned up.

Example of error:

[aws-java-sdk-NettyEventLoop-31-1] INFO org.apache.flink.connector.kinesis.source.reader.fanout.FanOutKinesisShardSubscription - Subscription complete - shardId-000000002053 (arn:aws:kinesis:us-east-1:123456:stream/test-source-stream/consumer/my-flink-efo-consumer:1774401121)
[aws-java-sdk-NettyEventLoop-31-1] INFO org.apache.flink.connector.kinesis.source.reader.fanout.FanOutKinesisShardSubscription - Activating subscription to shard shardId-000000002053 with starting position StartingPosition{shardIteratorType=AFTER_SEQUENCE_NUMBER, startingMarker=null} for consumer arn:aws:kinesis:us-east-1:123456:stream/test-source-stream/consumer/my-flink-efo-consumer:1774401121.
[aws-java-sdk-NettyEventLoop-31-1] ERROR software.amazon.awssdk.utils.async.FlatteningSubscriber - Unexpected exception encountered that violates the reactive streams specification. Attempting to terminate gracefully.
java.lang.IllegalArgumentException: Invalid StartingPosition. When ShardIteratorType is AT_SEQUENCE_NUMBER or AFTER_SEQUENCE_NUMBER, startingMarker must be a String.
        at org.apache.flink.util.Preconditions.checkArgument(Preconditions.java:138)
        at org.apache.flink.connector.kinesis.source.split.StartingPositionUtil.toSdkStartingPosition(StartingPositionUtil.java:65)
        at org.apache.flink.connector.kinesis.source.proxy.KinesisAsyncStreamProxy.subscribeToShard(KinesisAsyncStreamProxy.java:56)
        at org.apache.flink.connector.kinesis.source.reader.fanout.FanOutKinesisShardSubscription.activateSubscription(FanOutKinesisShardSubscription.java:137)
        at org.apache.flink.connector.kinesis.source.reader.fanout.FanOutKinesisShardSubscription$FanOutShardSubscriber.onComplete(FanOutKinesisShardSubscription.java:334)
        at software.amazon.awssdk.utils.async.DelegatingSubscriber.onComplete(DelegatingSubscriber.java:47)
...

Verifying this change

• Manually verified by running the Kinesis connector on a local Flink cluster.

After making the change, i 1. re-ran my flink application, 2. resharded my stream. Afterwards verified that subscriptions were being shut down without issues:

[aws-java-sdk-NettyEventLoop-31-9] INFO org.apache.flink.connector.kinesis.source.reader.fanout.FanOutKinesisShardSubscription - Subscription complete - shardId-000000002825 (arn:aws:kinesis:us-east-1:123456:stream/test-source-stream/consumer/my-flink-efo-consumer:1774401121)
[aws-java-sdk-NettyEventLoop-31-5] INFO org.apache.flink.connector.kinesis.source.reader.fanout.FanOutKinesisShardSubscription - Subscription complete - shardId-000000002828 (arn:aws:kinesis:us-east-1:123456:stream/test-source-stream/consumer/my-flink-efo-consumer:1774401121)
[aws-java-sdk-NettyEventLoop-31-18] INFO org.apache.flink.connector.kinesis.source.reader.fanout.FanOutKinesisShardSubscription - Subscription complete - shardId-000000002822 (arn:aws:kinesis:us-east-1:123456:stream/test-source-stream/consumer/my-flink-efo-consumer:1774401121)

Significant changes

N/A - Not a significant change

[boring-cyborg]

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

[gguptp]

@pelaezryan can you see if we can combine work with this PR: https://github.com/apache/flink-connector-aws/pull/224/changes

Also, lets add UTs for the change