roland created FLINK-37648:
------------------------------
Summary: Flink Kinesis connector lost track of 1 shard.
Key: FLINK-37648
URL: https://issues.apache.org/jira/browse/FLINK-37648
Project: Flink
Issue Type: Bug
Components: Connectors / Kinesis
Affects Versions: 1.20.1
Environment: Flink version: 1.20
flink-connector-aws-kinesis-streams' version: 5.0.0-1.20
kinesis stream's shard number: 2
consumer type: EFO
parallelism: 1
Reporter: roland
Hi,
I have a Flink 1.20 job which uses flink-connector-aws-kinesis-streams to
consume from a Kinesis stream with 2 shards using EFO.
The job runs for 40 days without any issue and from one point, it failed to
subscribe to both of the shard and only subscribed to shard-0.
1. Before 2025-04-08 07:20 (UTC) , Flink job subscribed 2 shards:
{code:java}
2025-04-08 07:12:57,788 INFO
org.apache.flink.connector.kinesis.source.reader.fanout.FanOutKinesisShardSubscription
[] - Subscription complete - shardId-000000000001
15:12:57.789
se-content-platform
2025-04-08 07:12:57,788 INFO
org.apache.flink.connector.kinesis.source.reader.fanout.FanOutKinesisShardSubscription
[] - Activating subscription to shard shardId-000000000001 with starting
position StartingPosition{shardIteratorType=AFTER_SEQUENCE_NUMBER,
startingMarker=49661185042803593150728679800204632415527945304203591698} for
consumer arn:xxx
15:12:57.841
se-content-platform
2025-04-08 07:12:57,841 INFO
org.apache.flink.connector.kinesis.source.reader.fanout.FanOutKinesisShardSubscription
[] - Successfully subscribed to shard shardId-000000000001 at
StartingPosition{shardIteratorType=AFTER_SEQUENCE_NUMBER,
startingMarker=49661185042803593150728679800204632415527945304203591698} using
consumer arn:xxx
15:12:57.841
se-content-platform
2025-04-08 07:12:57,841 INFO
org.apache.flink.connector.kinesis.source.reader.fanout.FanOutKinesisShardSubscription
[] - Successfully subscribed to shard shardId-000000000001 with starting
position StartingPosition{shardIteratorType=AFTER_SEQUENCE_NUMBER,
startingMarker=49661185042803593150728679800204632415527945304203591698} for
consumer arn:xxx
15:15:06.360
se-content-platform
2025-04-08 07:15:06,360 INFO
org.apache.flink.connector.kinesis.source.reader.fanout.FanOutKinesisShardSubscription
[] - Activating subscription to shard shardId-000000000000 with starting
position StartingPosition{shardIteratorType=AFTER_SEQUENCE_NUMBER,
startingMarker=49661066447418326619680247193290876689240616667852046338} for
consumer arn:xxx{code}
As shown above, both shard-0 and shard-1 are subscribed and consumed.
2. After that timestamp, Flink job only subscribed 1 shard:
{code:java}
2025-04-08 07:20:06,432 INFO
org.apache.flink.connector.kinesis.source.reader.fanout.FanOutKinesisShardSubscription
[] - Subscription complete - shardId-000000000000 (arn:
15:20:06.433
se-content-platform
2025-04-08 07:20:06,433 INFO
org.apache.flink.connector.kinesis.source.reader.fanout.FanOutKinesisShardSubscription
[] - Activating subscription to shard shardId-000000000000 with starting
position StartingPosition{shardIteratorType=AFTER_SEQUENCE_NUMBER,
startingMarker=49661066447418326619680247330352841488049219966007771138} for
consumer arn:
15:20:06.442
se-content-platform
2025-04-08 07:20:06,442 INFO
org.apache.flink.connector.kinesis.source.reader.fanout.FanOutKinesisShardSubscription
[] - Successfully subscribed to shard shardId-000000000000 at
StartingPosition{shardIteratorType=AFTER_SEQUENCE_NUMBER,
startingMarker=49661066447418326619680247330352841488049219966007771138} using
consumer arn:
15:20:06.442
se-content-platform
2025-04-08 07:20:06,442 INFO
org.apache.flink.connector.kinesis.source.reader.fanout.FanOutKinesisShardSubscription
[] - Successfully subscribed to shard shardId-000000000000 with starting
position StartingPosition{shardIteratorType=AFTER_SEQUENCE_NUMBER,
startingMarker=49661066447418326619680247330352841488049219966007771138} for
consumer arn:. {code}
There're no exception logs in between, however I found this warn log appeared
every 40 min:
{code:java}
2025-04-09 09:35:29,477 WARN io.netty.channel.DefaultChannelPipeline
[] - An exceptionCaught() event was fired, and it reached at the
tail of the pipeline. It usually means the last handler in the pipeline did not
handle the exception.
java.io.IOException: An error occurred on the connection:
java.nio.channels.ClosedChannelException, [channel: 086eb48c]. All streams will
be closed
at
software.amazon.awssdk.http.nio.netty.internal.http2.MultiplexedChannelRecord.decorateConnectionException(MultiplexedChannelRecord.java:213)
~[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
software.amazon.awssdk.http.nio.netty.internal.http2.MultiplexedChannelRecord.lambda$closeChildChannels$10(MultiplexedChannelRecord.java:205)
~[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
software.amazon.awssdk.http.nio.netty.internal.http2.MultiplexedChannelRecord.lambda$closeAndExecuteOnChildChannels$11(MultiplexedChannelRecord.java:229)
~[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
software.amazon.awssdk.http.nio.netty.internal.utils.NettyUtils.doInEventLoop(NettyUtils.java:248)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
software.amazon.awssdk.http.nio.netty.internal.http2.MultiplexedChannelRecord.closeAndExecuteOnChildChannels(MultiplexedChannelRecord.java:220)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
software.amazon.awssdk.http.nio.netty.internal.http2.MultiplexedChannelRecord.closeChildChannels(MultiplexedChannelRecord.java:205)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
software.amazon.awssdk.http.nio.netty.internal.http2.Http2MultiplexedChannelPool.closeAndReleaseParent(Http2MultiplexedChannelPool.java:353)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
software.amazon.awssdk.http.nio.netty.internal.http2.Http2MultiplexedChannelPool.closeAndReleaseParent(Http2MultiplexedChannelPool.java:333)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
software.amazon.awssdk.http.nio.netty.internal.http2.Http2MultiplexedChannelPool.access$200(Http2MultiplexedChannelPool.java:76)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
software.amazon.awssdk.http.nio.netty.internal.http2.Http2MultiplexedChannelPool$ReleaseOnExceptionHandler.closeAndReleaseParent(Http2MultiplexedChannelPool.java:509)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
software.amazon.awssdk.http.nio.netty.internal.http2.Http2MultiplexedChannelPool$ReleaseOnExceptionHandler.channelInactive(Http2MultiplexedChannelPool.java:486)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:303)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:281)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:274)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.handler.logging.LoggingHandler.channelInactive(LoggingHandler.java:206)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:303)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:281)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:274)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.channel.ChannelInboundHandlerAdapter.channelInactive(ChannelInboundHandlerAdapter.java:81)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.handler.timeout.IdleStateHandler.channelInactive(IdleStateHandler.java:280)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:303)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:281)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:274)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
software.amazon.awssdk.http.nio.netty.internal.http2.Http2PingHandler.channelInactive(Http2PingHandler.java:83)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:303)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:281)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:274)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.handler.codec.ByteToMessageDecoder.channelInputClosed(ByteToMessageDecoder.java:412)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.handler.codec.ByteToMessageDecoder.channelInactive(ByteToMessageDecoder.java:377)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.handler.codec.http2.Http2ConnectionHandler.channelInactive(Http2ConnectionHandler.java:428)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:303)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:281)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:274)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.handler.codec.ByteToMessageDecoder.channelInputClosed(ByteToMessageDecoder.java:412)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.handler.codec.ByteToMessageDecoder.channelInactive(ByteToMessageDecoder.java:377)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.handler.ssl.SslHandler.channelInactive(SslHandler.java:1172)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:303)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:281)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:274)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.channel.DefaultChannelPipeline$HeadContext.channelInactive(DefaultChannelPipeline.java:1402)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:301)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:281)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.channel.DefaultChannelPipeline.fireChannelInactive(DefaultChannelPipeline.java:900)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.channel.AbstractChannel$AbstractUnsafe$7.run(AbstractChannel.java:811)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:173)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:166)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:469)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:566)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:994)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at
io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
[blob_p-30497883442822f5c82caef062655823d22e6214-d096e666a38e1d8371c3989c721c6564:?]
at java.lang.Thread.run(Unknown Source) [?:?]
Caused by: java.nio.channels.ClosedChannelException
... 41 more {code}
My question is:
1. Why does the job stop subscribing shard-1 ?
2. Why does the Flink job keep running without exceptions ?
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
