(DEPRECATED) Apache Flink User Mailing List archive.

Jobmanager Crashes with Kubernetes HA When Restart Kubernetes Master Node

Classic

List

Threaded

4 messages Options

Jerome Li

Jobmanager Crashes with Kubernetes HA When Restart Kubernetes Master Node

Hi,

I am running Flink v1.12.2 in Standalone mode on Kubernetes. I set Kubernetes native as HA.

The HA works well when either jobmanager or taskmanager pod lost or crashes.

But, when I restart master node, jobmanager pod will always crash and restart. This results in the entire Flink cluster restart and most of taskmanager pod will restart as well.

I didn’t see this issue when using zookeeper as HA. Not sure if this is a bug should be handle or there is some work around.

Below is my Flink setting

Job-Manager

flink-conf.yaml:

----

jobmanager.rpc.address: streakerflink-jobmanager

high-availability: org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory

high-availability.cluster-id: /streaker

high-availability.jobmanager.port: 6123

high-availability.storageDir: hdfs://hdfs-namenode-0.hdfs-namenode:8020/flink

kubernetes.cluster-id: streaker

rest.address: streakerflink-jobmanager

rest.bind-port: 8081

rest.port: 8081

state.checkpoints.dir: hdfs://hdfs-namenode-0.hdfs-namenode:8020/flink/streaker

blob.server.port: 6124

metrics.internal.query-service.port: 6125

metrics.reporters: prom

metrics.reporter.prom.class: org.apache.flink.metrics.prometheus.PrometheusReporter

metrics.reporter.prom.port: 9999

restart-strategy: fixed-delay

restart-strategy.fixed-delay.attempts: 2147483647

restart-strategy.fixed-delay.delay: 5 s

jobmanager.memory.process.size: 1768m

parallelism.default: 1

task.cancellation.timeout: 2000

web.log.path: /opt/flink/log/output.log

jobmanager.web.log.path: /opt/flink/log/output.log

web.submit.enable: false

Task-Manager

flink-conf.yaml:

----

jobmanager.rpc.address: streakerflink-jobmanager

high-availability: org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory

high-availability.cluster-id: /streaker

high-availability.storageDir: hdfs://hdfs-namenode-0.hdfs-namenode:8020/flink

kubernetes.cluster-id: streaker

taskmanager.network.bind-policy: ip

taskmanager.data.port: 6121

taskmanager.rpc.port: 6122

restart-strategy: fixed-delay

restart-strategy.fixed-delay.attempts: 2147483647

restart-strategy.fixed-delay.delay: 5 s

taskmanager.memory.task.heap.size: 9728m

taskmanager.memory.framework.off-heap.size: 512m

taskmanager.memory.managed.size: 512m

taskmanager.memory.jvm-metaspace.size: 256m

taskmanager.memory.jvm-overhead.max: 3g

taskmanager.memory.jvm-overhead.fraction: 0.035

taskmanager.memory.network.fraction: 0.03

taskmanager.memory.network.max: 3g

taskmanager.numberOfTaskSlots: 1

taskmanager.jvm-exit-on-oom: true

metrics.internal.query-service.port: 6125

metrics.reporters: prom

metrics.reporter.prom.class: org.apache.flink.metrics.prometheus.PrometheusReporter

metrics.reporter.prom.port: 9999

web.log.path: /opt/flink/log/output.log

taskmanager.log.path: /opt/flink/log/output.log

task.cancellation.timeout: 2000

Any help will be appreciated!

Thanks,

Jerome

Yang Wang

Re: Jobmanager Crashes with Kubernetes HA When Restart Kubernetes Master Node

By "restart master node", do you mean to restart the K8s master component(e.g. APIServer, ETCD, etc.)?

Even though the master components are restarted, the Flink JobManager and TaskManager should eventually get to work.

Could you please share the JobManager logs so that we could debug why it crashed.

Best,

Yang

Jerome Li <[hidden email]> 于2021年5月25日周二上午3:43写道：

Hi,

I am running Flink v1.12.2 in Standalone mode on Kubernetes. I set Kubernetes native as HA.

The HA works well when either jobmanager or taskmanager pod lost or crashes.

But, when I restart master node, jobmanager pod will always crash and restart. This results in the entire Flink cluster restart and most of taskmanager pod will restart as well.

I didn’t see this issue when using zookeeper as HA. Not sure if this is a bug should be handle or there is some work around.

Below is my Flink setting

Job-Manager

flink-conf.yaml:

----

jobmanager.rpc.address: streakerflink-jobmanager

high-availability: org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory

high-availability.cluster-id: /streaker

high-availability.jobmanager.port: 6123

high-availability.storageDir: hdfs://hdfs-namenode-0.hdfs-namenode:8020/flink

kubernetes.cluster-id: streaker

rest.address: streakerflink-jobmanager

rest.bind-port: 8081

rest.port: 8081

state.checkpoints.dir: hdfs://hdfs-namenode-0.hdfs-namenode:8020/flink/streaker

blob.server.port: 6124

metrics.internal.query-service.port: 6125

metrics.reporters: prom

metrics.reporter.prom.class: org.apache.flink.metrics.prometheus.PrometheusReporter

metrics.reporter.prom.port: 9999

restart-strategy: fixed-delay

restart-strategy.fixed-delay.attempts: 2147483647

restart-strategy.fixed-delay.delay: 5 s

jobmanager.memory.process.size: 1768m

parallelism.default: 1

task.cancellation.timeout: 2000

web.log.path: /opt/flink/log/output.log

jobmanager.web.log.path: /opt/flink/log/output.log

web.submit.enable: false

Task-Manager

flink-conf.yaml:

----

jobmanager.rpc.address: streakerflink-jobmanager

high-availability: org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory

high-availability.cluster-id: /streaker

high-availability.storageDir: hdfs://hdfs-namenode-0.hdfs-namenode:8020/flink

kubernetes.cluster-id: streaker

taskmanager.network.bind-policy: ip

taskmanager.data.port: 6121

taskmanager.rpc.port: 6122

restart-strategy: fixed-delay

restart-strategy.fixed-delay.attempts: 2147483647

restart-strategy.fixed-delay.delay: 5 s

taskmanager.memory.task.heap.size: 9728m

taskmanager.memory.framework.off-heap.size: 512m

taskmanager.memory.managed.size: 512m

taskmanager.memory.jvm-metaspace.size: 256m

taskmanager.memory.jvm-overhead.max: 3g

taskmanager.memory.jvm-overhead.fraction: 0.035

taskmanager.memory.network.fraction: 0.03

taskmanager.memory.network.max: 3g

taskmanager.numberOfTaskSlots: 1

taskmanager.jvm-exit-on-oom: true

metrics.internal.query-service.port: 6125

metrics.reporters: prom

metrics.reporter.prom.class: org.apache.flink.metrics.prometheus.PrometheusReporter

metrics.reporter.prom.port: 9999

web.log.path: /opt/flink/log/output.log

taskmanager.log.path: /opt/flink/log/output.log

task.cancellation.timeout: 2000

Any help will be appreciated!

Thanks,

Jerome

Jerome Li

Re: Jobmanager Crashes with Kubernetes HA When Restart Kubernetes Master Node

Hi Yang,

Thanks for getting back to me.

By “restart master node”, I mean do “kubctl get nodes” to find the node’s role as master and “ssh” into one of master nodes as ubuntu user. Then run “sudo /sbin/reboot -f” to restart the master node.

It looks like The JobManager would cancel the running job and log this after that.

2021-05-26 18:28:37,997 [INFO] org.apache.flink.runtime.executiongraph.ExecutionGraph - Discarding the results produced by task execution 34eb9f5009dc7cf07117e720e7d393de.

2021-05-26 18:28:37,999 [INFO] org.apache.flink.runtime.checkpoint.DefaultCompletedCheckpointStore - Suspending

2021-05-26 18:28:37,999 [INFO] org.apache.flink.kubernetes.highavailability.KubernetesCheckpointIDCounter - Shutting down.

2021-05-26 18:28:38,000 [INFO] org.apache.flink.runtime.executiongraph.ExecutionGraph - Job 74fc5c858e50f5efc91db9ee16c17a8c has been suspended.

2021-05-26 18:28:38,007 [INFO] org.apache.flink.runtime.jobmaster.slotpool.SlotPoolImpl - Suspending SlotPool.

2021-05-26 18:28:38,007 [INFO] org.apache.flink.runtime.jobmaster.JobMaster - Close ResourceManager connection 5bac86fb0b5c984ef429225b8de82cc0: JobManager is no longer the leader..

2021-05-26 18:28:38,019 [INFO] org.apache.flink.runtime.jobmaster.JobManagerRunnerImpl - JobManager runner for job hogger (74fc5c858e50f5efc91db9ee16c17a8c) was granted leadership with session id 14b9004a-3807-42e8-ac03-c0d77efe5611 at akka.tcp://flink@hoggerflink-jobmanager:6123/user/rpc/jobmanager_2.

2021-05-26 18:28:38,292 [INFO] org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor - The rpc endpoint org.apache.flink.runtime.jobmaster.JobMaster has not been started yet. Discarding message org.apache.flink.runtime.rpc.messages.RemoteFencedMessage until processing is started.

2021-05-26 18:28:38,293 [INFO] org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor - The rpc endpoint org.apache.flink.runtime.jobmaster.JobMaster has not been started yet. Discarding message org.apache.flink.runtime.rpc.messages.RemoteFencedMessage until processing is started.

2021-05-26 18:28:38,295 [INFO] org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor - The rpc endpoint org.apache.flink.runtime.jobmaster.JobMaster has not been started yet. Discarding message org.apache.flink.runtime.rpc.messages.LocalFencedMessage until processing is started.

2021-05-26 18:28:38,296 [INFO] org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor - The rpc endpoint org.apache.flink.runtime.jobmaster.JobMaster has not been started yet. Discarding message org.apache.flink.runtime.rpc.messages.RemoteFencedMessage until processing is started.

2021-05-26 18:28:38,299 [ERROR] org.apache.flink.runtime.entrypoint.ClusterEntrypoint - Fatal error occurred in the cluster entrypoint.

org.apache.flink.util.FlinkException: JobMaster for job 74fc5c858e50f5efc91db9ee16c17a8c failed.

at org.apache.flink.runtime.dispatcher.Dispatcher.jobMasterFailed(Dispatcher.java:887) ~[flink-dist_2.12-1.12.2.jar:1.12.2]

at org.apache.flink.runtime.dispatcher.Dispatcher.dispatcherJobFailed(Dispatcher.java:465) ~[flink-dist_2.12-1.12.2.jar:1.12.2]

at org.apache.flink.runtime.dispatcher.Dispatcher.lambda$runJob$3(Dispatcher.java:426) ~[flink-dist_2.12-1.12.2.jar:1.12.2]

at java.util.concurrent.CompletableFuture.uniHandle(CompletableFuture.java:930) ~[?:?]

at java.util.concurrent.CompletableFuture$UniHandle.tryFire(CompletableFuture.java:907) ~[?:?]

at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:478) ~[?:?]

at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRunAsync(AkkaRpcActor.java:440) ~[flink-dist_2.12-1.12.2.jar:1.12.2]

at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:208) ~[flink-dist_2.12-1.12.2.jar:1.12.2]

at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:77) ~[flink-dist_2.12-1.12.2.jar:1.12.2]

at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:158) ~[flink-dist_2.12-1.12.2.jar:1.12.2]

at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26) [flink-dist_2.12-1.12.2.jar:1.12.2]

at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21) [flink-dist_2.12-1.12.2.jar:1.12.2]

at scala.PartialFunction.applyOrElse(PartialFunction.scala:123) [flink-dist_2.12-1.12.2.jar:1.12.2]

at scala.PartialFunction.applyOrElse$(PartialFunction.scala:122) [flink-dist_2.12-1.12.2.jar:1.12.2]

at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21) [flink-dist_2.12-1.12.2.jar:1.12.2]

at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171) [flink-dist_2.12-1.12.2.jar:1.12.2]

at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:172) [flink-dist_2.12-1.12.2.jar:1.12.2]

at akka.actor.Actor.aroundReceive(Actor.scala:517) [flink-dist_2.12-1.12.2.jar:1.12.2]

at akka.actor.Actor.aroundReceive$(Actor.scala:515) [flink-dist_2.12-1.12.2.jar:1.12.2]

at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:225) [flink-dist_2.12-1.12.2.jar:1.12.2]

at akka.actor.ActorCell.receiveMessage(ActorCell.scala:592) [flink-dist_2.12-1.12.2.jar:1.12.2]

at akka.actor.ActorCell.invoke(ActorCell.scala:561) [flink-dist_2.12-1.12.2.jar:1.12.2]

at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:258) [flink-dist_2.12-1.12.2.jar:1.12.2]

at akka.dispatch.Mailbox.run(Mailbox.scala:225) [flink-dist_2.12-1.12.2.jar:1.12.2]

at akka.dispatch.Mailbox.exec(Mailbox.scala:235) [flink-dist_2.12-1.12.2.jar:1.12.2]

at akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260) [flink-dist_2.12-1.12.2.jar:1.12.2]

at akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339) [flink-dist_2.12-1.12.2.jar:1.12.2]

at akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979) [flink-dist_2.12-1.12.2.jar:1.12.2]

at akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107) [flink-dist_2.12-1.12.2.jar:1.12.2]

Caused by: org.apache.flink.util.FlinkException: Could not start the job manager.

at org.apache.flink.runtime.jobmaster.JobManagerRunnerImpl.lambda$handleException$7(JobManagerRunnerImpl.java:456) ~[flink-dist_2.12-1.12.2.jar:1.12.2]

at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:859) ~[?:?]

at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:837) ~[?:?]

at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506) ~[?:?]

at java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:2088) ~[?:?]

at org.apache.flink.runtime.concurrent.FutureUtils$1.onComplete(FutureUtils.java:1044) ~[flink-dist_2.12-1.12.2.jar:1.12.2]

at akka.dispatch.OnComplete.internal(Future.scala:263) ~[flink-dist_2.12-1.12.2.jar:1.12.2]

at akka.dispatch.OnComplete.internal(Future.scala:261) ~[flink-dist_2.12-1.12.2.jar:1.12.2]

at akka.dispatch.japi$CallbackBridge.apply(Future.scala:191) ~[flink-dist_2.12-1.12.2.jar:1.12.2]

at akka.dispatch.japi$CallbackBridge.apply(Future.scala:188) ~[flink-dist_2.12-1.12.2.jar:1.12.2]

at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:60) ~[flink-dist_2.12-1.12.2.jar:1.12.2]

at org.apache.flink.runtime.concurrent.Executors$DirectExecutionContext.execute(Executors.java:73) ~[flink-dist_2.12-1.12.2.jar:1.12.2]

at scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:68) ~[flink-dist_2.12-1.12.2.jar:1.12.2]

at scala.concurrent.impl.Promise$DefaultPromise.$anonfun$tryComplete$1(Promise.scala:284) ~[flink-dist_2.12-1.12.2.jar:1.12.2]

at scala.concurrent.impl.Promise$DefaultPromise.$anonfun$tryComplete$1$adapted(Promise.scala:284) ~[flink-dist_2.12-1.12.2.jar:1.12.2]

at scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:284) ~[flink-dist_2.12-1.12.2.jar:1.12.2]

at akka.pattern.PromiseActorRef.$bang(AskSupport.scala:573) ~[flink-dist_2.12-1.12.2.jar:1.12.2]

at akka.actor.ActorRef.tell(ActorRef.scala:126) ~[flink-dist_2.12-1.12.2.jar:1.12.2]

at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleCallAsync(AkkaRpcActor.java:423) ~[flink-dist_2.12-1.12.2.jar:1.12.2]

at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:210) ~[flink-dist_2.12-1.12.2.jar:1.12.2]

at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:100) ~[flink-dist_2.12-1.12.2.jar:1.12.2]

... 21 more

Caused by: java.util.concurrent.CompletionException: java.lang.IllegalStateException: DefaultLeaderRetrievalService can only be started once.

at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:331) ~[?:?]

at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:346) ~[?:?]

at java.util.concurrent.CompletableFuture$UniAccept.tryFire(CompletableFuture.java:704) ~[?:?]