[
https://issues.apache.org/jira/browse/FLINK-39540?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Ryan Pelaez updated FLINK-39540:
--------------------------------
Description:
Resharding a Kinesis stream when a Flink application is running leads to a
repeated ``IllegalArgumentException`` being thrown. This is caused by the
subscription attempting to be restarted after it has been completed
Logs 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)
...
```
was:
Resharding a Kinesis stream when a Flink application is running leads to a
repeated ``IllegalArgumentException`` being thrown. This is caused by the
subscription attempting to be restarted after it has been completed
Logs 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)
...}}
{{```}}
> Kinesis Connector - EFO Issue when resharding - IllegalArgumentException
> ------------------------------------------------------------------------
>
> Key: FLINK-39540
> URL: https://issues.apache.org/jira/browse/FLINK-39540
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kinesis
> Affects Versions: aws-connector-5.0.0, aws-connector-6.0.0
> Reporter: Ryan Pelaez
> Priority: Major
>
> Resharding a Kinesis stream when a Flink application is running leads to a
> repeated ``IllegalArgumentException`` being thrown. This is caused by the
> subscription attempting to be restarted after it has been completed
>
> Logs 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)
> ...
> ```
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
