答复: flink集群性能问题

[戴嘉诚 <a7...@gmail.com>]

好的，我尝试吧18个flink job拆分出来，在yarn中单独运行，观察里面的jm/tm的信息，谢谢.

发件人: Yun Gao
发送时间: 2019年5月10日 17:52
收件人: user-zh
主题: Re: flink集群性能问题

异常一和异常三应该是相关的，因为Heartbeat也是通过akka来发送的；如果确定不是GC的问题的话，那么超时一般是由于JM/TM在做一些耗时的操作导致akka线程阻塞，但是具体在做什么操作应该还需要进一步分析。


------------------------------------------------------------------
From:戴嘉诚 <a7...@gmail.com>
Send Time:2019 May 10 (Fri.) 17:00
To:user-zh@flink.apache.org <us...@flink.apache.org>
Subject:flink集群性能问题

大家好：
 我这里遇到了一个问题，我的运行方式是flink session on yarn上，一共有18个任务在这个session上运行，这个任务运行了几天后，最近开始发现有几个job，不定时报这个错误，（ps：就这几个job报这个异常，其他job没有出现）。一直都提示超时，然后看了gc，发现没有长时间的的fullgc，而且gc也改为了用g1垃圾收集器，但是也是会有这个问题。
 状态后端使用的是文件后端，以前用rocksDB的时候，也是出现过如此异常。

异常一：
java.util.concurrent.TimeoutException: Heartbeat of TaskManager with id container_e30_1556059873942_0007_01_003306 timed out.
 at org.apache.flink.runtime.jobmaster.JobMaster$TaskManagerHeartbeatListener.notifyHeartbeatTimeout(JobMaster.java:1656)
 at org.apache.flink.runtime.heartbeat.HeartbeatManagerImpl$HeartbeatMonitor.run(HeartbeatManagerImpl.java:339)
 at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
 at java.util.concurrent.FutureTask.run(FutureTask.java:266)
 at org.apache.flink.runtime.concurrent.akka.ActorSystemScheduledExecutorAdapter$ScheduledFutureTask.run(ActorSystemScheduledExecutorAdapter.java:154)
 at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:39)
 at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:415)
 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)


异常二：
Caused by: org.apache.hadoop.ipc.RemoteException(org.apache.hadoop.hdfs.server.namenode.LeaseExpiredException): No lease on /flink/checkpoint/public_rule_persona_guanzhu_count/225a05c3ac6192017f38017417ad8f71/chk-6015/1ed16855-3261-4eda-bceb-1519f401eae5 (inode 1967760389): File does not exist. [Lease.  Holder: DFSClient_NONMAPREDUCE_2064130997_92, pendingcreates: 11]
 at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.checkLease(FSNamesystem.java:3697)
 at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.analyzeFileState(FSNamesystem.java:3498)
 at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.getNewBlockTargets(FSNamesystem.java:3336)
 at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.getAdditionalBlock(FSNamesystem.java:3296)
 at org.apache.hadoop.hdfs.server.namenode.NameNodeRpcServer.addBlock(NameNodeRpcServer.java:850)
 at org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolServerSideTranslatorPB.addBlock(ClientNamenodeProtocolServerSideTranslatorPB.java:504)
 at org.apache.hadoop.hdfs.protocol.proto.ClientNamenodeProtocolProtos$ClientNamenodeProtocol$2.callBlockingMethod(ClientNamenodeProtocolProtos.java)
 at org.apache.hadoop.ipc.ProtobufRpcEngine$Server$ProtoBufRpcInvoker.call(ProtobufRpcEngine.java:640)
 at org.apache.hadoop.ipc.RPC$Server.call(RPC.java:982)
 at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:2351)
 at org.apache.hadoop.ipc.Server$Handler$1.run(Server.java:2347)
 at java.security.AccessController.doPrivileged(Native Method)
 at javax.security.auth.Subject.doAs(Subject.java:422)
 at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1869)
 at org.apache.hadoop.ipc.Server$Handler.run(Server.java:2347)

 at org.apache.hadoop.ipc.Client.getRpcResponse(Client.java:1554)
 at org.apache.hadoop.ipc.Client.call(Client.java:1498)
 at org.apache.hadoop.ipc.Client.call(Client.java:1398)
 at org.apache.hadoop.ipc.ProtobufRpcEngine$Invoker.invoke(ProtobufRpcEngine.java:233)
 at com.sun.proxy.$Proxy17.addBlock(Unknown Source)
 at org.apache.hadoop.hdfs.protocolPB.ClientNamenodeProtocolTranslatorPB.addBlock(ClientNamenodeProtocolTranslatorPB.java:459)
 at sun.reflect.GeneratedMethodAccessor51.invoke(Unknown Source)
 at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
 at java.lang.reflect.Method.invoke(Method.java:498)
 at org.apache.hadoop.io.retry.RetryInvocationHandler.invokeMethod(RetryInvocationHandler.java:290)
 at org.apache.hadoop.io.retry.RetryInvocationHandler.invoke(RetryInvocationHandler.java:202)
 at org.apache.hadoop.io.retry.RetryInvocationHandler.invoke(RetryInvocationHandler.java:184)
 at com.sun.proxy.$Proxy18.addBlock(Unknown Source)
 at org.apache.hadoop.hdfs.DFSOutputStream$DataStreamer.locateFollowingBlock(DFSOutputStream.java:1580)
 at org.apache.hadoop.hdfs.DFSOutputStream$DataStreamer.nextBlockOutputStream(DFSOutputStream.java:1375)
 at org.apache.hadoop.hdfs.DFSOutputStream$DataStreamer.run(DFSOutputStream.java:564)

异常三：
java.util.concurrent.CompletionException: akka.pattern.AskTimeoutException: Ask timed out on [Actor[akka.tcp://flink@ambari.host08.yy:37375/user/taskmanager_0#-433497337]] after [10000 ms]. Sender[null] sent message of type "org.apache.flink.runtime.rpc.messages.RemoteRpcInvocation".
 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.postComplete(CompletableFuture.java:474)
 at java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:1977)
 at org.apache.flink.runtime.concurrent.FutureUtils$1.onComplete(FutureUtils.java:815)
 at akka.dispatch.OnComplete.internal(Future.scala:258)
 at akka.dispatch.OnComplete.internal(Future.scala:256)
 at akka.dispatch.japi$CallbackBridge.apply(Future.scala:186)
 at akka.dispatch.japi$CallbackBridge.apply(Future.scala:183)
 at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:36)
 at org.apache.flink.runtime.concurrent.Executors$DirectExecutionContext.execute(Executors.java:74)
 at scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:44)
 at scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:252)
 at akka.pattern.PromiseActorRef$$anonfun$1.apply$mcV$sp(AskSupport.scala:603)
 at akka.actor.Scheduler$$anon$4.run(Scheduler.scala:126)
 at scala.concurrent.Future$InternalCallbackExecutor$.unbatchedExecute(Future.scala:601)
 at scala.concurrent.BatchingExecutor$class.execute(BatchingExecutor.scala:109)
 at scala.concurrent.Future$InternalCallbackExecutor$.execute(Future.scala:599)
 at akka.actor.LightArrayRevolverScheduler$TaskHolder.executeTask(LightArrayRevolverScheduler.scala:329)
 at akka.actor.LightArrayRevolverScheduler$$anon$4.executeBucket$1(LightArrayRevolverScheduler.scala:280)
 at akka.actor.LightArrayRevolverScheduler$$anon$4.nextTick(LightArrayRevolverScheduler.scala:284)
 at akka.actor.LightArrayRevolverScheduler$$anon$4.run(LightArrayRevolverScheduler.scala:236)
 at java.lang.Thread.run(Thread.java:748)
Caused by: akka.pattern.AskTimeoutException: Ask timed out on [Actor[akka.tcp://flink@ambari.host08.yy:37375/user/taskmanager_0#-433497337]] after [10000 ms]. Sender[null] sent message of type "org.apache.flink.runtime.rpc.messages.RemoteRpcInvocation".
 at akka.pattern.PromiseActorRef$$anonfun$1.apply$mcV$sp(AskSupport.scala:604)
 ... 9 more


其中，这里异常二中，的hdfs文件我本地或者程序都没有操作过，就只有这个job来操作checkpoint，但是，也有时候会报这个异常，我感觉这里三个异常，都是跟一个原因相关，但是具体是什么原因，我这边还是排查不出来，请各位指点一下。