Flink HA setup on Kubernetes

[Steven Nelson]

Hello! I am trying to setup Flink in an HA mode on Kubernetes. I have an existing Zookeeper cluster that is coming from my HDFS setup. I set the nodes up in a StatefulSet so I can get exactly 3 masters that have consistent names. The first node launches okay and I can see it when I view the port. However the secondary nodes seem to fail to launch.

Here is my flink-conf.yaml

rest.port: 9081

jobmanager.rpc.port: 6123

fs.default-scheme: hdfs://flink-state-hdfs-namenode-0.flink-state-hdfs-namenode.default.svc.cluster.local:8020

parallelism.default: 1

high-availability: zookeeper

high-availability.cluster-id: /cluster1

high-availability.storageDir: /flink/ha/

high-availability.zookeeper.quorum: flink-state-hdfs-zookeeper-1.flink-state-hdfs-zookeeper-headless.default.svc.cluster.local:2181,flink-state-hdfs-zookeeper-2.flink-state-hdfs-zookeeper-headless.default.svc.cluster.local:2181,flink-state-hdfs-zookeeper-0.flink-stat$

high-availability.zookeeper.path.root: /flink

high-availability.jobmanager.port: 50000-50025

state.backend: rocksdb

state.savepoints.dir: /flink/savepoints

state.checkpoints.dir: /flink/checkpoints

state.backend.incremental: true

The secondary fails the readiness check to localhost:9081/overview with:

2018-12-31 16:49:02,661 ERROR org.apache.flink.runtime.rest.handler.cluster.ClusterOverviewHandler  - Could not retrieve the redirect address.

java.util.concurrent.CompletionException: org.apache.flink.runtime.rpc.exceptions.FencingTokenException: Fencing token not set: Ignoring message LocalFencedMessage(8d4331128f160bfd934264b7e9374998, LocalRpcInvocation(requestRestAddress(Time))) sent to akka.tcp://flink@127.0.0.1:50000/user/dispatcher because the fencing token is null.

        at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:292)

        at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:308)

        at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:593)

        at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:577)

        at java.util.concurrent.CompletableFuture.uniApplyStage(CompletableFuture.java:617)

        at java.util.concurrent.CompletableFuture.thenApply(CompletableFuture.java:1983)

        at org.apache.flink.runtime.rest.handler.util.HandlerRedirectUtils.getRedirectAddress(HandlerRedirectUtils.java:60)

        at org.apache.flink.runtime.rest.handler.RedirectHandler.lambda$channelRead0$1(RedirectHandler.java:113)

        at java.util.Optional.ifPresent(Optional.java:159)

        at org.apache.flink.util.OptionalConsumer.ifPresent(OptionalConsumer.java:44)

        at org.apache.flink.runtime.rest.handler.RedirectHandler.channelRead0(RedirectHandler.java:111)

        at org.apache.flink.runtime.rest.handler.RedirectHandler.channelRead0(RedirectHandler.java:56)

        at org.apache.flink.shaded.netty4.io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)

        at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)

        at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)

        at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)

        at org.apache.flink.runtime.rest.handler.router.RouterHandler.routed(RouterHandler.java:110)

        at org.apache.flink.runtime.rest.handler.router.RouterHandler.channelRead0(RouterHandler.java:89)

        at org.apache.flink.runtime.rest.handler.router.RouterHandler.channelRead0(RouterHandler.java:54)

        at org.apache.flink.shaded.netty4.io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)

        at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)

        at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)

        at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)

        at org.apache.flink.shaded.netty4.io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:86)

        at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)

        at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)

        at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)

        at org.apache.flink.shaded.netty4.io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:102)

        at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)

        at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)

        at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)

        at org.apache.flink.runtime.rest.FileUploadHandler.channelRead0(FileUploadHandler.java:159)

        at org.apache.flink.runtime.rest.FileUploadHandler.channelRead0(FileUploadHandler.java:67)

        at org.apache.flink.shaded.netty4.io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105)

        at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)

        at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)

        at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)

        at org.apache.flink.shaded.netty4.io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:438)

        at org.apache.flink.shaded.netty4.io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:310)

        at org.apache.flink.shaded.netty4.io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:284)

        at org.apache.flink.shaded.netty4.io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:253)

        at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)

        at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)

        at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)

        at org.apache.flink.shaded.netty4.io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1434)

        at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)

        at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)

        at org.apache.flink.shaded.netty4.io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:965)

        at org.apache.flink.shaded.netty4.io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:163)

        at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:645)

        at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:580)

        at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:497)

        at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:459)

        at org.apache.flink.shaded.netty4.io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:884)

        at java.lang.Thread.run(Thread.java:748)

Caused by: org.apache.flink.runtime.rpc.exceptions.FencingTokenException: Fencing token not set: Ignoring message LocalFencedMessage(8d4331128f160bfd934264b7e9374998, LocalRpcInvocation(requestRestAddress(Time))) sent to akka.tcp://flink@127.0.0.1:50000/user/dispatcher because the fencing token is null.

        at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:59)

        at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.onReceive(AkkaRpcActor.java:142)

        at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.onReceive(FencedAkkaRpcActor.java:40)

        at akka.actor.UntypedActor$$anonfun$receive$1.applyOrElse(UntypedActor.scala:165)

        at akka.actor.Actor$class.aroundReceive(Actor.scala:502)

        at akka.actor.UntypedActor.aroundReceive(UntypedActor.scala:95)

        at akka.actor.ActorCell.receiveMessage(ActorCell.scala:526)

        at akka.actor.ActorCell.invoke(ActorCell.scala:495)

        at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:257)

        at akka.dispatch.Mailbox.run(Mailbox.scala:224)

        at akka.dispatch.Mailbox.exec(Mailbox.scala:234)

        at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)

        at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)

        at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)

        at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)

I see folders created in both my hdfs backend and the zookeeper structure. It just doesn't seem to properly find the other master.