pelaezryan commented on PR #243:
URL:
https://github.com/apache/flink-connector-aws/pull/243#issuecomment-4330233196
Verified scenarios
### 1 - Subscription expires - NOT at shard_end and resubscribes
```
[aws-java-sdk-NettyEventLoop-2-3] INFO
org.apache.flink.connector.kinesis.source.reader.fanout.FanOutKinesisShardSubscription
- Subscription complete - shardId-000000003076
(arn:aws:kinesis:us-east-1:1234567890:stream/test-source-stream/consumer/my-flink-efo-consumer:1774401121)
[Source Data Fetcher for Source: Kinesis source -> (Sink: Print to Std. Out,
Kinesis sink: Writer) (1/1)#0] INFO
org.apache.flink.connector.kinesis.source.reader.fanout.FanOutKinesisShardSubscription
- Subscription expired to shard shardId-000000003076 for consumer
arn:aws:kinesis:us-east-1:1234567890:stream/test-source-stream/consumer/my-flink-efo-consumer:1774401121
but we have not reached SHARD_END. Restarting subscription.
[Source Data Fetcher for Source: Kinesis source -> (Sink: Print to Std. Out,
Kinesis sink: Writer) (1/1)#0] INFO
org.apache.flink.connector.kinesis.source.reader.fanout.FanOutKinesisShardSubscription
- Activating subscription to shard shardId-000000003076 with starting position
StartingPosition{shardIteratorType=AFTER_SEQUENCE_NUMBER,
startingMarker=49674026002850188218576229259473476449429298564223254594} for
consumer
arn:aws:kinesis:us-east-1:1234567890:stream/test-source-stream/consumer/my-flink-efo-consumer:1774401121.
[aws-java-sdk-NettyEventLoop-2-6] INFO
org.apache.flink.connector.kinesis.source.reader.fanout.FanOutKinesisShardSubscription
- Successfully subscribed to shard shardId-000000003076 at
StartingPosition{shardIteratorType=AFTER_SEQUENCE_NUMBER,
startingMarker=49674026002850188218576229259473476449429298564223254594} using
consumer
arn:aws:kinesis:us-east-1:1234567890:stream/test-source-stream/consumer/my-flink-efo-consumer:1774401121.
[ForkJoinPool.commonPool-worker-5] INFO
org.apache.flink.connector.kinesis.source.reader.fanout.FanOutKinesisShardSubscription
- Successfully subscribed to shard shardId-000000003076 with starting position
StartingPosition{shardIteratorType=AFTER_SEQUENCE_NUMBER,
startingMarker=49674026002850188218576229259473476449429298564223254594} for
consumer
arn:aws:kinesis:us-east-1:1234567890:stream/test-source-stream/consumer/my-flink-efo-consumer:1774401121.
[aws-java-sdk-NettyEventLoop-2-6] INFO
org.apache.flink.connector.kinesis.source.reader.fanout.FanOutKinesisShardSubscription
- Subscription complete - shardId-000000003077
(arn:aws:kinesis:us-east-1:1234567890:stream/test-source-stream/consumer/my-flink-efo-consumer:1774401121)
[Source Data Fetcher for Source: Kinesis source -> (Sink: Print to Std. Out,
Kinesis sink: Writer) (1/1)#0] INFO
org.apache.flink.connector.kinesis.source.reader.fanout.FanOutKinesisShardSubscription
- Subscription expired to shard shardId-000000003077 for consumer
arn:aws:kinesis:us-east-1:1234567890:stream/test-source-stream/consumer/my-flink-efo-consumer:1774401121
but we have not reached SHARD_END. Restarting subscription.
[Source Data Fetcher for Source: Kinesis source -> (Sink: Print to Std. Out,
Kinesis sink: Writer) (1/1)#0] INFO
org.apache.flink.connector.kinesis.source.reader.fanout.FanOutKinesisShardSubscription
- Activating subscription to shard shardId-000000003077 with starting position
StartingPosition{shardIteratorType=AFTER_SEQUENCE_NUMBER,
startingMarker=49674026002872488963774759882602922909505800633982173266} for
consumer
arn:aws:kinesis:us-east-1:1234567890:stream/test-source-stream/consumer/my-flink-efo-consumer:1774401121.
[aws-java-sdk-NettyEventLoop-2-6] INFO
org.apache.flink.connector.kinesis.source.reader.fanout.FanOutKinesisShardSubscription
- Successfully subscribed to shard shardId-000000003077 at
StartingPosition{shardIteratorType=AFTER_SEQUENCE_NUMBER,
startingMarker=49674026002872488963774759882602922909505800633982173266} using
consumer
arn:aws:kinesis:us-east-1:1234567890:stream/test-source-stream/consumer/my-flink-efo-consumer:1774401121.
[ForkJoinPool.commonPool-worker-5] INFO
org.apache.flink.connector.kinesis.source.reader.fanout.FanOutKinesisShardSubscription
- Successfully subscribed to shard shardId-000000003077 with starting position
StartingPosition{shardIteratorType=AFTER_SEQUENCE_NUMBER,
startingMarker=49674026002872488963774759882602922909505800633982173266} for
consumer
arn:aws:kinesis:us-east-1:1234567890:stream/test-source-stream/consumer/my-flink-efo-consumer:1774401121.
```
### 2 - Subscruption expires, At shard_end and a new subscription for new
shard is started
```
2026-04-27T12:55:17.103384: Records coming in from source> {"message":
"Record 26954", "timestamp": "2026-04-27T13:06:11.490934"}
[Source: Kinesis source -> (Sink: Print to Std. Out, Kinesis sink: Writer)
(1/1)#0] INFO org.apache.flink.connector.base.source.reader.SourceReaderBase -
Finished reading split(s) [shardId-000000003077]
[Source Data Fetcher for Source: Kinesis source -> (Sink: Print to Std. Out,
Kinesis sink: Writer) (1/1)#0] INFO
org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher - Finished
reading from splits [shardId-000000003077]
Shard is completed for: shardId-000000003076
[Source Data Fetcher for Source: Kinesis source -> (Sink: Print to Std. Out,
Kinesis sink: Writer) (1/1)#0] INFO
org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher - Finished
reading from splits [shardId-000000003076]
[Source: Kinesis source -> (Sink: Print to Std. Out, Kinesis sink: Writer)
(1/1)#0] INFO org.apache.flink.connector.base.source.reader.SourceReaderBase -
Finished reading split(s) [shardId-000000003076]
[Source: Kinesis source -> (Sink: Print to Std. Out, Kinesis sink: Writer)
(1/1)#0] INFO
org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherManager -
Closing splitFetcher 0 because it is idle.
[Source: Kinesis source -> (Sink: Print to Std. Out, Kinesis sink: Writer)
(1/1)#0] INFO
org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher - Shutting
down split fetcher 0
[SourceCoordinator-Source: Kinesis source] INFO
org.apache.flink.connector.kinesis.source.enumerator.KinesisStreamsSourceEnumerator
- Assigning shard shardId-000000003078 from stream
arn:aws:kinesis:us-east-1:1234567890:stream/test-source-stream to subtask 0.
[Source: Kinesis source -> (Sink: Print to Std. Out, Kinesis sink: Writer)
(1/1)#0] INFO org.apache.flink.connector.base.source.reader.SourceReaderBase -
Adding split(s) to reader:
[KinesisShardSplit{streamArn='arn:aws:kinesis:us-east-1:1234567890:stream/test-source-stream',
shardId='shardId-000000003078',
startingPosition=StartingPosition{shardIteratorType=TRIM_HORIZON,
startingMarker=null},
parentShardIds=[shardId-000000003076,shardId-000000003077],
startingHashKey='0', endingHashKey='340282366920938463463374607431768211455',
finished=false}]
[Source Data Fetcher for Source: Kinesis source -> (Sink: Print to Std. Out,
Kinesis sink: Writer) (1/1)#0] INFO
org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher - Starting
split fetcher 1
[Source Data Fetcher for Source: Kinesis source -> (Sink: Print to Std. Out,
Kinesis sink: Writer) (1/1)#0] INFO
org.apache.flink.connector.kinesis.source.reader.fanout.FanOutKinesisShardSubscription
- Activating subscription to shard shardId-000000003078 with starting position
StartingPosition{shardIteratorType=TRIM_HORIZON, startingMarker=null} for
consumer
arn:aws:kinesis:us-east-1:1234567890:stream/test-source-stream/consumer/my-flink-efo-consumer:1774401121.
[Source Data Fetcher for Source: Kinesis source -> (Sink: Print to Std. Out,
Kinesis sink: Writer) (1/1)#0] INFO
org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher - Split
fetcher 0 exited.
Subscription is not active:shardId-000000003078
Subscription is not active:shardId-000000003078
[aws-java-sdk-NettyEventLoop-2-4] INFO
org.apache.flink.connector.kinesis.source.reader.fanout.FanOutKinesisShardSubscription
- Successfully subscribed to shard shardId-000000003078 at
StartingPosition{shardIteratorType=TRIM_HORIZON, startingMarker=null} using
consumer
arn:aws:kinesis:us-east-1:1234567890:stream/test-source-stream/consumer/my-flink-efo-consumer:1774401121.
[ForkJoinPool.commonPool-worker-6] INFO
org.apache.flink.connector.kinesis.source.reader.fanout.FanOutKinesisShardSubscription
- Successfully subscribed to shard shardId-000000003078 with starting position
StartingPosition{shardIteratorType=TRIM_HORIZON, startingMarker=null} for
consumer
arn:aws:kinesis:us-east-1:1234567890:stream/test-source-stream/consumer/my-flink-efo-consumer:1774401121.
[aws-java-sdk-NettyEventLoop-2-4] INFO
org.apache.flink.connector.kinesis.source.reader.fanout.FanOutKinesisShardSubscription
- Next Starting position:
StartingPosition{shardIteratorType=AFTER_SEQUENCE_NUMBER,
startingMarker=49674026407519510591125021697014088120894882864332587106}
2026-04-27T12:55:17.103384: Records coming in from source> {"message":
"Record 26955", "timestamp": "2026-04-27T13:06:12.580617"}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
