Hi Leon, The kinesis consumer allows you to specify how many of these exceptions can be thrown without a job failure. This is independent from Flink's restart policy.
So in your case, if you feel like the Kinesis consumer is giving up too quickly, you could simply increase flink.shard.subscribetoshard.maxretries to a larger value. Note that Kinesis distinguishes between RetryableFanOutSubscriberException (which you saw and count towards the max retries) and RecoverableFanOutSubscriberException (which are repeated indefinitely). So the Kinesis connector deems your exceptions as bad enough that they probably require a restart at some point in time. At least the second exception looks like some network issues; the first one probably is also unrecoverable. So if your cluster experiences some network outages from time to time, you could just increase the max retries. Of course, that means Flink takes longer to restart if something is really off. If you have more information on the circumstances under which you see these exceptions, please share. I'm CCing Danny in case he has more ideas. On Mon, Dec 13, 2021 at 7:13 AM Leon Xu <[email protected]> wrote: > Hi Flink users, > > I used flink-1.12.5 kinesis connector to consume data from kinesis. > > From time to time I am getting IOException or > StacklessClosedChannelException, which will fail the Flink operator when it > by default reaches 10 times and trigger the entire job to restart. > I have two questions: > > 1. I am wondering if we can have a better way to handle the error > without restarting the Flink job. Restarting the job is time-consuming and > will often slow down the data consumption. > 2. These two errors are treated as retryable exceptions, instead of > recoverable exceptions. Are they not recoverable if we don't restart the > Flink source operator? > > > Followed are the exceptions: > > org.apache.flink.streaming.connectors.kinesis.internals.publisher.fanout.FanOutShardSubscriber$RetryableFanOutSubscriberException: > org.apache.flink.kinesis.shaded.io.netty.channel.StacklessClosedChannelException > at > org.apache.flink.streaming.connectors.kinesis.internals.publisher.fanout.FanOutShardSubscriber.handleError(FanOutShardSubscriber.java:296) > ~[blob_p-6581ced9ade704dee24c2632e701b413d2e656ba-f7515b4b21d25cb44b07411a9ad294e8:?] > at > org.apache.flink.streaming.connectors.kinesis.internals.publisher.fanout.FanOutShardSubscriber.consumeAllRecordsFromKinesisShard(FanOutShardSubscriber.java:363) > ~[blob_p-6581ced9ade704dee24c2632e701b413d2e656ba-f7515b4b21d25cb44b07411a9ad294e8:?] > at > org.apache.flink.streaming.connectors.kinesis.internals.publisher.fanout.FanOutShardSubscriber.subscribeToShardAndConsumeRecords(FanOutShardSubscriber.java:188) > ~[blob_p-6581ced9ade704dee24c2632e701b413d2e656ba-f7515b4b21d25cb44b07411a9ad294e8:?] > at > org.apache.flink.streaming.connectors.kinesis.internals.publisher.fanout.FanOutRecordPublisher.runWithBackoff(FanOutRecordPublisher.java:154) > [blob_p-6581ced9ade704dee24c2632e701b413d2e656ba-f7515b4b21d25cb44b07411a9ad294e8:?] > > or > > org.apache.flink.streaming.connectors.kinesis.internals.publisher.fanout.FanOutShardSubscriber$RetryableFanOutSubscriberException: > java.io.IOException: An error occurred on the connection: null at > org.apache.flink.streaming.connectors.kinesis.internals.publisher.fanout.FanOutShardSubscriber.handleError(FanOutShardSubscriber.java:296) > ~[blob_p-6581ced9ade704dee24c2632e701b413d2e656ba-f7515b4b21d25cb44b07411a9ad294e8:?] > at > org.apache.flink.streaming.connectors.kinesis.internals.publisher.fanout.FanOutShardSubscriber.consumeAllRecordsFromKinesisShard(FanOutShardSubscriber.java:363) > ~[blob_p-6581ced9ade704dee24c2632e701b413d2e656ba-f7515b4b21d25cb44b07411a9ad294e8:?] > at > org.apache.flink.streaming.connectors.kinesis.internals.publisher.fanout.FanOutShardSubscriber.subscribeToShardAndConsumeRecords(FanOutShardSubscriber.java:188) > ~[blob_p-6581ced9ade704dee24c2632e701b413d2e656ba-f7515b4b21d25cb44b07411a9ad294e8:?] > at > org.apache.flink.streaming.connectors.kinesis.internals.publisher.fanout.FanOutRecordPublisher.runWithBackoff(FanOutRecordPublisher.java:154) > [blob_p-6581ced9ade704dee24c2632e701b413d2e656ba-f7515b4b21d25cb44b07411a9ad294e8:?] > > > > Thanks > > Leon > > >
