https://flink.apache.org/ logo
Join Slack
Powered by
Hello guys, I am trying my flink job with flink ku...
# troubleshooting
y
Hello guys, I am trying my flink job with flink kubernets operator. I have tried to do application upgrade on application mode and session mode. Each time I update FlinkDeployment or FlinkSessionJob to trigger a upgrade, I see the flink job failed to create savepoint and has following exception
Copy code
org.apache.flink.streaming.connectors.kafka.internals.Handover$ClosedException
Do you know what could be the cause of this? πŸ™
βœ… 1
m
This information is not sufficient to effectively answer this question i'm afraid.
y
Other thing is that I haven't enable kubernetes HA or zookeeper HA, I'll try active HA and see if it can fix this issue
k
y
thanks! I have succedd creating HA but still has that kafka Handover$ClosedException I'll try tuning some kafka conf like mentioned here https://stackoverflow.com/questions/68270362/flink-streaming-kafka-internals-handoverclosedexception
πŸ‘ 1
it failed agains even if I increase request time out
Copy code
org.apache.flink.runtime.checkpoint.CheckpointException: Task has failed.
	at org.apache.flink.runtime.messages.checkpoint.SerializedCheckpointException.unwrap(SerializedCheckpointException.java:51)
	at org.apache.flink.runtime.checkpoint.CheckpointCoordinator.receiveDeclineMessage(CheckpointCoordinator.java:1036)
	at org.apache.flink.runtime.scheduler.ExecutionGraphHandler.lambda$declineCheckpoint$2(ExecutionGraphHandler.java:103)
	at org.apache.flink.runtime.scheduler.ExecutionGraphHandler.lambda$processCheckpointCoordinatorMessage$3(ExecutionGraphHandler.java:119)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
	at java.base/java.lang.Thread.run(Unknown Source)
Caused by: org.apache.flink.runtime.checkpoint.CheckpointException: org.apache.flink.runtime.checkpoint.CheckpointException: Task name with subtask : Source: kafka_integration_inputs -> Flat Map (5/16)#1 Failure reason: Task has failed.
	at org.apache.flink.runtime.taskmanager.Task.declineCheckpoint(Task.java:1375)
	at org.apache.flink.runtime.taskmanager.Task.lambda$triggerCheckpointBarrier$3(Task.java:1318)
	at java.base/java.util.concurrent.CompletableFuture.uniHandle(Unknown Source)
	at java.base/java.util.concurrent.CompletableFuture$UniHandle.tryFire(Unknown Source)
	at java.base/java.util.concurrent.CompletableFuture.postComplete(Unknown Source)
	at java.base/java.util.concurrent.CompletableFuture.completeExceptionally(Unknown Source)
	at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:344)
Caused by: java.util.concurrent.CompletionException: java.util.concurrent.CompletionException: org.apache.flink.streaming.connectors.kafka.internals.Handover$ClosedException
	at java.base/java.util.concurrent.CompletableFuture.encodeThrowable(Unknown Source)
	at java.base/java.util.concurrent.CompletableFuture.completeThrowable(Unknown Source)
	at java.base/java.util.concurrent.CompletableFuture$UniCompose.tryFire(Unknown Source)
	... 3 more
Caused by: org.apache.flink.streaming.connectors.kafka.internals.Handover$ClosedException: org.apache.flink.streaming.connectors.kafka.internals.Handover$ClosedException
	at org.apache.flink.streaming.connectors.kafka.internals.Handover.close(Handover.java:177)
	at org.apache.flink.streaming.connectors.kafka.internals.KafkaFetcher.cancel(KafkaFetcher.java:164)
	at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.cancel(FlinkKafkaConsumerBase.java:945)
	at org.apache.flink.streaming.api.operators.StreamSource.stop(StreamSource.java:128)
	at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.stopOperatorForStopWithSavepoint(SourceStreamTask.java:306)
	at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.lambda$triggerStopWithSavepointAsync$1(SourceStreamTask.java:286)
	at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:93)
	at org.apache.flink.streaming.runtime.tasks.mailbox.Mail.run(Mail.java:90)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMail(MailboxProcessor.java:398)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMailsWhenDefaultActionUnavailable(MailboxProcessor.java:367)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMail(MailboxProcessor.java:352)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:229)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:831)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:780)
	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:935)
	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:914)
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:728)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:550)
	at java.base/java.lang.Thread.run(Unknown Source)
😞
m
Have you enabled committing the offset to Kafka? It has seemingly nothing to do with the operator.
y
I didn't enable commiting the offset, but I have activated checkpoint. can you clarify me a bit more? I though it was not necessary though
Just to let you knows, that this issues is fixed when I upgraded my FlinkKafkaConsumer and FlinkKafkaProducer to KafkaSource and KafkaSink, Although it creates some painpoints for me now like for instance it won't be able anymore to create kafka topic if not exist which break my testsπŸ˜‚
m
What version was not working and what fixed it?
y
before I have used FlinkKafkaConsumer and FlinkKafkaProducer in my flink job with flink 13 After using kubernets operator with flink 16, the flink job failed when I tried to do application upgrade (meaning operator do stop job with savepoint and start new job with that savepoint) with that kafka exception So for flink 16 we need to switche to KafkaSource and KafkaSink to avoid having that kafka exception when we do stop job with savepoint
3 Views
Open in Slack
PreviousNext
Powered by