Hello all, I am using Flink with kinesis and facin...
# troubleshootingl
Lucas Alcântara Freire
02/28/2023, 11:50 AMHello all, I am using Flink with kinesis and facing this, have anyone faced something similar or has any clue what can be the reason?
We use flink 1.16 with java 8
This cause our job to restart
Copy code
2023-02-28 11:48:55,553 INFO org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher [] {} - Starting shutdown of shard consumer threads and AWS SDK resources of subtask 0 ...
java.lang.InterruptedException: null
at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.reportInterruptAfterWait(AbstractQueuedSynchronizer.java:2014) ~[?:1.8.0_362]
at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2088) ~[?:1.8.0_362]
at java.util.concurrent.ThreadPoolExecutor.awaitTermination(ThreadPoolExecutor.java:1475) ~[?:1.8.0_362]
at org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher.awaitTermination(KinesisDataFetcher.java:875) ~[cdp-flink-jobs.jar:?]
at org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher.runFetcher(KinesisDataFetcher.java:760) ~[cdp-flink-jobs.jar:?]
at org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer.run(FlinkKinesisConsumer.java:392) ~[cdp-flink-jobs.jar:?]
at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:110) ~[flink-dist-1.16.1.jar:1.16.1]
at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:67) ~[flink-dist-1.16.1.jar:1.16.1]
at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:333) ~[flink-dist-1.16.1.jar:1.16.1]
2023-02-28 11:48:55,553 INFO org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher [] {} - Starting shutdown of shard consumer threads and AWS SDK resources of subtask 3 ...j
Jeremy Ber
02/28/2023, 2:26 PMhow many shards in your KDS vs how many parallel instances of your KinesisConsumer?
l
Lucas Alcântara Freire
02/28/2023, 2:34 PMwe have 5 shards and and 4 consumers
Lucas Alcântara Freire
02/28/2023, 2:35 PM2 jobs consume the kinesis stream and we have 2 pods for each job
j
Jeremy Ber
02/28/2023, 2:36 PMDo you see any throughput exceptions on the source kinesis data stream?
l
Lucas Alcântara Freire
02/28/2023, 2:37 PMwe had that but after increaseing the interval for fetching the recods we stop having provisionexception but that error pesisted
j
Jeremy Ber
02/28/2023, 2:38 PMDo you have any idle shards (with no data in them)
l
Lucas Alcântara Freire
02/28/2023, 2:42 PMthey all have data
j
Jeremy Ber
02/28/2023, 2:43 PMIs this the full stack trace? Can you share anything more
l
Lucas Alcântara Freire
02/28/2023, 2:46 PMLucas Alcântara Freire
02/28/2023, 2:47 PMwe have more, so basically the job process some event and then when it tries to save the checkpoint it fails with that error
Lucas Alcântara Freire
02/28/2023, 2:47 PMand then the job recover and starts to process the vents again.
j
Jeremy Ber
02/28/2023, 2:49 PMah so this happens on checkpointing, right?
l
Lucas Alcântara Freire
02/28/2023, 2:51 PMyes
Lucas Alcântara Freire
02/28/2023, 2:51 PMseems like
d
Danny Cranmer
02/28/2023, 2:53 PMSeems like the job is already cancelling:
•
2023-02-28 14:36:39,177 INFO org.apache.flink.runtime.taskmanager.Task [] {} - Attempting to cancel task Source:l
Lucas Alcântara Freire
02/28/2023, 2:54 PMi dont see any kinesis error before the error happens
d
Danny Cranmer
02/28/2023, 2:56 PMWhy is the job transitioning to CANCELLING though? What does the Flink dashboard show for job failure exception?
Danny Cranmer
02/28/2023, 2:57 PMIs this failing for general checkpoint or when you try to stop-with-savepoint?
l
Lucas Alcântara Freire
02/28/2023, 3:10 PMgeneral checkpoint
Lucas Alcântara Freire
02/28/2023, 3:12 PMI see this
Copy code
org.apache.flink.util.FlinkRuntimeException: Exceeded checkpoint tolerable failure threshold.
at org.apache.flink.runtime.checkpoint.CheckpointFailureManager.checkFailureAgainstCounter(CheckpointFailureManager.java:206)
at org.apache.flink.runtime.checkpoint.CheckpointFailureManager.handleJobLevelCheckpointException(CheckpointFailureManager.java:169)
at org.apache.flink.runtime.checkpoint.CheckpointFailureManager.handleCheckpointException(CheckpointFailureManager.java:122)
at org.apache.flink.runtime.checkpoint.CheckpointCoordinator.abortPendingCheckpoint(CheckpointCoordinator.java:2104)
at org.apache.flink.runtime.checkpoint.CheckpointCoordinator.abortPendingCheckpoint(CheckpointCoordinator.java:2083)
at org.apache.flink.runtime.checkpoint.CheckpointCoordinator.access$700(CheckpointCoordinator.java:98)
at org.apache.flink.runtime.checkpoint.CheckpointCoordinator$CheckpointCanceller.run(CheckpointCoordinator.java:2165)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:750)d
Danny Cranmer
02/28/2023, 4:04 PMok, so the Kinesis error does not look like the root cause here, it looks like a symptom. Can you find in the logs the root cause for the checkpoint exception?
The Kinesis error is thrown during shutdown "Starting shutdown of shard consumer threads and AWS SDK resources of subtask 1 ...", thus you need to identify what caused the shutdown.
Granted there might still be a bug here though