flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Paul Lam <paullin3...@gmail.com>
Subject New received containers silently lost during job auto restarts
Date Thu, 27 Sep 2018 11:24:25 GMT
Hi,

One of my Flink on YARN jobs got into a weird situation after a fail-fast restart. The restart
was triggered by loss of a TaskManager, but when the job was recovering, one of its subtask
(1/24) couldn’t be deployed, and finally failed with NoResourceAvailableException. 

Looking into the logs, I find the JobManager was requesting containers for the subtask again
and again, but indeed it got a container in each round of requests, which seemed to be lost
silently (with no error logs) or never registered. 

At last, I canceled the job and resubmitted it, and the problem was gone. 

I thought it might be something wrong with the TaskManager initialization, but if it is the
case there would be some errors in JobManagar’s log, right? 

Sorry for not being able to reproduce the problem, but does anyone have any idea on this?
Thanks a lot!

This is the cause of the job restart:
```
2018-09-27 12:33:11,639 INFO  org.apache.flink.yarn.YarnResourceManager                  
  - Closing TaskExecutor connection container_1536852167599_809133_02_000016 because: Container
released on a *lost* node
2018-09-27 12:33:11,639 INFO  org.apache.flink.runtime.resourcemanager.slotmanager.SlotManager
 - Unregister TaskManager 554e53380a836229ecbc4ccae58f9b0f from the SlotManager.
2018-09-27 12:33:11,639 INFO  org.apache.flink.runtime.executiongraph.ExecutionGraph     
  - LogAndConfigCoMap -> Filter -> FieldRegexFlatMap -> Log2JsonMap -> Sink: LogKafkaSink
(11/24) (51fedf378844bc7e5d2855f59baafcd0) switched from RUNNING to FAILED.
org.apache.flink.util.FlinkException: The assigned slot container_1536852167599_809133_02_000016_0
was removed.
	at org.apache.flink.runtime.resourcemanager.slotmanager.SlotManager.removeSlot(SlotManager.java:786)
	at org.apache.flink.runtime.resourcemanager.slotmanager.SlotManager.removeSlots(SlotManager.java:756)
	at org.apache.flink.runtime.resourcemanager.slotmanager.SlotManager.internalUnregisterTaskManager(SlotManager.java:948)
	at org.apache.flink.runtime.resourcemanager.slotmanager.SlotManager.unregisterTaskManager(SlotManager.java:372)
	at org.apache.flink.runtime.resourcemanager.ResourceManager.closeTaskManagerConnection(ResourceManager.java:802)
	at org.apache.flink.yarn.YarnResourceManager.lambda$onContainersCompleted$0(YarnResourceManager.java:339)
	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)
```

And then JobManager kept requesting and returning containers before timeout:
```
2018-09-27 12:38:11,072 INFO  org.apache.flink.yarn.YarnResourceManager                  
  - Received new container: container_1536852167599_809133_02_000832 - Remaining pending container
requests: 1
2018-09-27 12:38:11,072 INFO  org.apache.flink.yarn.YarnResourceManager                  
  - Adding keytab hdfs://horton2/user/gdc_sa/.flink/application_1536852167599_809133/gdc_sa.keytab
to the AM container local resource bucket
2018-09-27 12:38:11,338 INFO  org.apache.flink.yarn.YarnResourceManager                  
  - Creating container launch context for TaskManagers
2018-09-27 12:38:11,338 INFO  org.apache.flink.yarn.YarnResourceManager                  
  - Starting TaskManagers
2018-09-27 12:38:11,340 INFO  org.apache.hadoop.yarn.client.api.impl.ContainerManagementProtocolProxy
 - Opening proxy : gdc-dn254:8041
2018-09-27 12:38:11,346 INFO  org.apache.flink.yarn.YarnResourceManager                  
  - Received new container: container_1536852167599_809133_02_000833 - Remaining pending container
requests: 0
2018-09-27 12:38:11,346 INFO  org.apache.flink.yarn.YarnResourceManager                  
  - Returning excess container container_1536852167599_809133_02_000833.
2018-09-27 12:38:11,346 INFO  org.apache.flink.yarn.YarnResourceManager                  
  - Received new container: container_1536852167599_809133_02_000834 - Remaining pending container
requests: 0
2018-09-27 12:38:11,346 INFO  org.apache.flink.yarn.YarnResourceManager                  
  - Returning excess container container_1536852167599_809133_02_000834.
2018-09-27 12:38:11,346 INFO  org.apache.flink.yarn.YarnResourceManager                  
  - Received new container: container_1536852167599_809133_02_000835 - Remaining pending container
requests: 0
2018-09-27 12:38:11,346 INFO  org.apache.flink.yarn.YarnResourceManager                  
  - Returning excess container container_1536852167599_809133_02_000835.
———————— skip some similar logs ——————— 
2018-09-27 12:43:15,449 INFO  org.apache.flink.yarn.YarnResourceManager                  
  - Received new container: container_1536852167599_809133_02_002284 - Remaining pending container
requests: 1
2018-09-27 12:43:15,449 INFO  org.apache.flink.yarn.YarnResourceManager                  
  - Adding keytab hdfs://horton2/user/gdc_sa/.flink/application_1536852167599_809133/gdc_sa.keytab
to the AM container local resource bucket
2018-09-27 12:43:15,725 INFO  org.apache.flink.yarn.YarnResourceManager                  
  - Creating container launch context for TaskManagers
2018-09-27 12:43:15,725 INFO  org.apache.flink.yarn.YarnResourceManager                  
  - Starting TaskManagers
2018-09-27 12:43:15,727 INFO  org.apache.hadoop.yarn.client.api.impl.ContainerManagementProtocolProxy
 - Opening proxy : gdc-dn521:8041
2018-09-27 12:43:15,733 INFO  org.apache.flink.yarn.YarnResourceManager                  
  - Received new container: container_1536852167599_809133_02_002285 - Remaining pending container
requests: 0
2018-09-27 12:43:15,733 INFO  org.apache.flink.yarn.YarnResourceManager                  
  - Returning excess container container_1536852167599_809133_02_002285.
2018-09-27 12:43:15,733 INFO  org.apache.flink.yarn.YarnResourceManager                  
  - Received new container: container_1536852167599_809133_02_002286 - Remaining pending container
requests: 0
2018-09-27 12:43:15,734 INFO  org.apache.flink.yarn.YarnResourceManager                  
  - Received new container: container_1536852167599_809133_02_002351 - Remaining pending container
requests: 0
2018-09-27 12:43:15,734 INFO  org.apache.flink.yarn.YarnResourceManager                  
  - Received new container: container_1536852167599_809133_02_002352 - Remaining pending container
requests: 0
2018-09-27 12:43:15,734 INFO  org.apache.flink.yarn.YarnResourceManager                  
  - Returning excess container container_1536852167599_809133_02_002352.
2018-09-27 12:43:15,734 INFO  org.apache.flink.yarn.YarnResourceManager                  
  - Received new container: container_1536852167599_809133_02_002353 - Remaining pending container
requests: 0
2018-09-27 12:43:15,734 INFO  org.apache.flink.yarn.YarnResourceManager                  
  - Returning excess container container_1536852167599_809133_02_002353.
2018-09-27 12:43:16,831 INFO  org.apache.flink.runtime.checkpoint.CheckpointCoordinator  
  - Checkpoint triggering task Source: Custom Source (1/24) of job 4f6c0c7a0c09254ff6c1de9bc92dc9e7
is not in state RUNNING but SCHEDULED instead. Aborting checkpoint.
2018-09-27 12:43:21,831 INFO  org.apache.flink.runtime.checkpoint.CheckpointCoordinator  
  - Checkpoint triggering task Source: Custom Source (1/24) of job 4f6c0c7a0c09254ff6c1de9bc92dc9e7
is not in state RUNNING but SCHEDULED instead. Aborting checkpoint.
2018-09-27 12:43:21,833 INFO  org.apache.flink.runtime.jobmaster.slotpool.SlotPool       
  - Pending slot request [SlotRequestId{7c611bb7a1cb28c061ada9c485b116ac}] timed out.
2018-09-27 12:43:21,837 INFO  org.apache.flink.runtime.executiongraph.ExecutionGraph     
  - Job LogStream (4f6c0c7a0c09254ff6c1de9bc92dc9e7) switched from state RUNNING to FAILING.
org.apache.flink.runtime.jobmanager.scheduler.NoResourceAvailableException: Could not allocate
all requires slots within timeout of 300000 ms. Slots required: 72, slots allocated: 69
	at org.apache.flink.runtime.executiongraph.ExecutionGraph.lambda$scheduleEager$3(ExecutionGraph.java:984)
	at java.util.concurrent.CompletableFuture.uniExceptionally(CompletableFuture.java:870)
	at java.util.concurrent.CompletableFuture$UniExceptionally.tryFire(CompletableFuture.java:852)
	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$ResultConjunctFuture.handleCompletedFuture(FutureUtils.java:534)
	at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:760)
	at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:736)
	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:770)
	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:83)
	at scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:44)
	at scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:252)
	at akka.pattern.PromiseActorRef.$bang(AskSupport.scala:534)
	at akka.pattern.PipeToSupport$PipeableFuture$$anonfun$pipeTo$1.applyOrElse(PipeToSupport.scala:20)
	at akka.pattern.PipeToSupport$PipeableFuture$$anonfun$pipeTo$1.applyOrElse(PipeToSupport.scala:18)
	at scala.concurrent.Future$$anonfun$andThen$1.apply(Future.scala:436)
	at scala.concurrent.Future$$anonfun$andThen$1.apply(Future.scala:435)
	at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:36)
	at akka.dispatch.BatchingExecutor$AbstractBatch.processBatch(BatchingExecutor.scala:55)
	at akka.dispatch.BatchingExecutor$BlockableBatch$$anonfun$run$1.apply$mcV$sp(BatchingExecutor.scala:91)
	at akka.dispatch.BatchingExecutor$BlockableBatch$$anonfun$run$1.apply(BatchingExecutor.scala:91)
	at akka.dispatch.BatchingExecutor$BlockableBatch$$anonfun$run$1.apply(BatchingExecutor.scala:91)
	at scala.concurrent.BlockContext$.withBlockContext(BlockContext.scala:72)
	at akka.dispatch.BatchingExecutor$BlockableBatch.run(BatchingExecutor.scala:90)
	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)
```

Best,
Paul Lam
Mime
View raw message