Hi.
I'm running a Flink application (version 1.8.0) that
uses FlinkKafkaConsumer to fetch topic data and perform transformation on
the data, with state backend as below:
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(5_000, CheckpointingMode.AT_LEAST_ONCE);
env.setStateBackend((StateBackend) new
FsStateBackend("file:///test"));
env.getCheckpointConfig().setCheckpointTimeout(30_000);
env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000);
My problem is with the kafka brokers, where in the cluster there are 3
operating brokers and 2 are down - total 5 brokers.
I was able to consume the data, but when the checkpoint triggered it throws
this exception:
[INFO ] 2019-07-22 12:29:14.634 [flink-akka.actor.default-dispatcher-18]
o.a.f.r.c.CheckpointCoordinator - Decline checkpoint 6 by task
457b1f801fee89d6f9544409877e29d8 of job 1c46aa5719bac1f0bea436d460b79db1.
[INFO ] 2019-07-22 12:29:14.636 [flink-akka.actor.default-dispatcher-28]
o.a.f.r.t.TaskExecutor - Un-registering task and sending final execution
state FAILED to JobManager for task Source: Custom Source -> Sink: Print to
Std. Out 457b1f801fee89d6f9544409877e29d8.
[INFO ] 2019-07-22 12:29:14.634 [flink-akka.actor.default-dispatcher-18]
o.a.f.r.c.CheckpointCoordinator - Discarding checkpoint 6 of job
1c46aa5719bac1f0bea436d460b79db1.
org.apache.flink.runtime.checkpoint.decline.CheckpointDeclineTaskNotReadyException:
Task Source: Custom Source -> Sink: Print to Std. Out (2/4) was not running
at
org.apache.flink.runtime.taskmanager.Task.triggerCheckpointBarrier(Task.java:1198)
~[flink-runtime_2.11-1.8.0.jar:1.8.0]
at
org.apache.flink.runtime.taskexecutor.TaskExecutor.triggerCheckpoint(TaskExecutor.java:700)
~[flink-runtime_2.11-1.8.0.jar:1.8.0]
at sun.reflect.GeneratedMethodAccessor15.invoke(Unknown Source) ~[?:?]
at
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
~[?:1.8.0_201]
at java.lang.reflect.Method.invoke(Method.java:498) ~[?:1.8.0_201]
at
org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcInvocation(AkkaRpcActor.java:274)
~[flink-runtime_2.11-1.8.0.jar:1.8.0]
at
org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:189)
~[flink-runtime_2.11-1.8.0.jar:1.8.0]
at
org.apache.flink.runtime.rpc.akka.AkkaRpcActor.onReceive(AkkaRpcActor.java:147)
~[flink-runtime_2.11-1.8.0.jar:1.8.0]
at
akka.actor.UntypedActor$$anonfun$receive$1.applyOrElse(UntypedActor.scala:165)
~[akka-actor_2.11-2.4.20.jar:?]
at akka.actor.Actor$class.aroundReceive$$$capture(Actor.scala:502)
~[akka-actor_2.11-2.4.20.jar:?]
at akka.actor.Actor$class.aroundReceive(Actor.scala)
~[akka-actor_2.11-2.4.20.jar:?]
at akka.actor.UntypedActor.aroundReceive(UntypedActor.scala:95)
~[akka-actor_2.11-2.4.20.jar:?]
at akka.actor.ActorCell.receiveMessage(ActorCell.scala:526)
~[akka-actor_2.11-2.4.20.jar:?]
at akka.actor.ActorCell.invoke(ActorCell.scala:495)
~[akka-actor_2.11-2.4.20.jar:?]
at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:257)
~[akka-actor_2.11-2.4.20.jar:?]
at akka.dispatch.Mailbox.run(Mailbox.scala:224)
~[akka-actor_2.11-2.4.20.jar:?]
at akka.dispatch.Mailbox.exec(Mailbox.scala:234)
~[akka-actor_2.11-2.4.20.jar:?]
at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
[scala-library-2.11.12.jar:?]
at
scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
[scala-library-2.11.12.jar:?]
at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
[scala-library-2.11.12.jar:?]
at
scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
[scala-library-2.11.12.jar:?]
[INFO ] 2019-07-22 12:29:14.637 [flink-akka.actor.default-dispatcher-28]
o.a.f.r.e.ExecutionGraph - Source: Custom Source -> Sink: Print to Std. Out
(2/4) (457b1f801fee89d6f9544409877e29d8) switched from RUNNING to FAILED.
org.apache.kafka.common.errors.TimeoutException: Timeout expired while
fetching topic metadata
My question is (as I think what does the checkpoint tries to do) why is it
trying to fetch topic metadata from the brokers that are down?
Thanks,
Yitzchak.
