Hi Gyula, Your issue is possibly related to [1] that slots prematurely released. I’ve raised a PR which is still pending review.
[1] https://issues.apache.org/jira/browse/FLINK-10941 > On Dec 20, 2018, at 9:33 PM, Gyula Fóra <gyula.f...@gmail.com> wrote: > > Hi! > > Since we have moved to the new execution mode with Flink 1.7.0 we have > observed some pretty bad stability issues with the Yarn execution. > > It's pretty hard to understand what's going on so sorry for the vague > description but here is what seems to happen: > > In some cases when a bigger job fails (lets say 30 taskmanagers, 10 slots > each) and the job tries to recover we can observe taskmanagers start to fail. > > The errors usually look like this: > 20181220T141057.132+0100 INFO The heartbeat of TaskManager with id > container_e15_1542798679751_0033_01_000021 timed out. > [org.apache.flink.runtime.resourcemanager.ResourceManager$TaskManagerHeartbeatListener$1.run() > @ 1137] > 20181220T141057.133+0100 INFO Closing TaskExecutor connection > container_e15_1542798679751_0033_01_000021 because: The heartbeat of > TaskManager with id container_e15_1542798679751_0033_01_000021 timed out. > [org.apache.flink.runtime.resourcemanager.ResourceManager.closeTaskManagerConnection() > @ 822] > 20181220T141057.135+0100 INFO Execute processors -> (Filter config stream -> > (Filter Failures, Flat Map), Filter BEA) (168/180) > (3e9c164e4c0594f75c758624815265f1) switched from RUNNING to FAILED. > org.apache.flink.util.FlinkException: The assigned slot > container_e15_1542798679751_0033_01_000021_0 was removed. > at > org.apache.flink.runtime.resourcemanager.slotmanager.SlotManager.removeSlot(SlotManager.java:893) > at > org.apache.flink.runtime.resourcemanager.slotmanager.SlotManager.removeSlots(SlotManager.java:863) > at > org.apache.flink.runtime.resourcemanager.slotmanager.SlotManager.internalUnregisterTaskManager(SlotManager.java:1058) > at > org.apache.flink.runtime.resourcemanager.slotmanager.SlotManager.unregisterTaskManager(SlotManager.java:385) > at > org.apache.flink.runtime.resourcemanager.ResourceManager.closeTaskManagerConnection(ResourceManager.java:825) > at > org.apache.flink.runtime.resourcemanager.ResourceManager$TaskManagerHeartbeatListener$1.run(ResourceManager.java:1139) > at > org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRunAsync(AkkaRpcActor.java:332) > at > org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:158) > at > org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:70) > 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) > [org.apache.flink.runtime.executiongraph.Execution.transitionState() @ 1342] > > The job then goes in a restart loop, where taskmanagers come and go, the UI > sometimes displays more than 30 taskmanagers and some extra slots. I have in > some instances seen "GC overhead limit exceeded" during the recovery which is > very strange. > > I suspect there might be something strange happening, maybe some broken logic > in the slot allocations or some memory leak. > > Has anyone observed anything similar so far? > Seems to only affect some of our larger jobs. This hasn't been a problem in > the previous Flink releases where we always used the "legacy" execution mode. > > Thank you! > Gyula
