You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Robert Metzger (Jira)" <ji...@apache.org> on 2020/10/13 15:23:00 UTC
[jira] [Commented] (FLINK-18293) TaskExecutor offering non empty
slots can lead to resource violation
[ https://issues.apache.org/jira/browse/FLINK-18293?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17213165#comment-17213165 ]
Robert Metzger commented on FLINK-18293:
----------------------------------------
Is below exception the type of error you would expect from a situation where a slot is re-offered while the cancellation is still ongoing? I'm not 100% sure because the first two log messages imply that the slot has a task in FAILED state, thus it should be available already?
{code}
07:31:50,288 [flink-akka.actor.default-dispatcher-3] INFO org.apache.flink.runtime.taskmanager.Task [] - Attempting to cancel task blocking operator (1/4) (10fc95eb21e801e1de186569b50f073b_b992e5f3e3d3c178f8fff9fba052540c_0_0).
07:31:50,288 [flink-akka.actor.default-dispatcher-3] INFO org.apache.flink.runtime.taskmanager.Task [] - Task blocking operator (1/4) is already in state FAILED
07:31:50,303 [flink-akka.actor.default-dispatcher-3] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - blocking operator (1/4) (10fc95eb21e801e1de186569b50f073b_b992e5f3e3d3c178f8fff9fba052540c_0_0) switched from SCHEDULED to DEPLOYING.
07:31:50,303 [flink-akka.actor.default-dispatcher-3] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Deploying blocking operator (1/4) (attempt #0) with attempt id 10fc95eb21e801e1de186569b50f073b_b992e5f3e3d3c178f8fff9fba052540c_0_0 to c5f965a5-5520-4d68-9bb5-b20edd072dea @ localhost (dataPort=32853) with allocation id 573d83f4231cac7bd79e5e904bbe0bbe
07:31:50,304 [flink-akka.actor.default-dispatcher-3] INFO org.apache.flink.runtime.taskexecutor.TaskExecutor [] - Received task blocking operator (1/4) (10fc95eb21e801e1de186569b50f073b_b992e5f3e3d3c178f8fff9fba052540c_0_0), deploy into slot with allocation id 573d83f4231cac7bd79e5e904bbe0bbe.
07:31:50,304 [flink-akka.actor.default-dispatcher-3] DEBUG org.apache.flink.runtime.taskexecutor.TaskExecutor [] - TaskManager already contains a task for id 10fc95eb21e801e1de186569b50f073b_b992e5f3e3d3c178f8fff9fba052540c_0_0.
07:31:50,310 [flink-akka.actor.default-dispatcher-3] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - blocking operator (1/4) (10fc95eb21e801e1de186569b50f073b_b992e5f3e3d3c178f8fff9fba052540c_0_0) switched from DEPLOYING to FAILED on org.apache.flink.runtime.jobmaster.slotpool.SingleLogicalSlot@5c87cca9.
java.util.concurrent.CompletionException: org.apache.flink.runtime.taskexecutor.exceptions.TaskSubmissionException: TaskManager already contains a task for id 10fc95eb21e801e1de186569b50f073b_b992e5f3e3d3c178f8fff9fba052540c_0_0.
at java.util.concurrent.CompletableFuture.encodeRelay(CompletableFuture.java:326) ~[?:1.8.0_242]
at java.util.concurrent.CompletableFuture.completeRelay(CompletableFuture.java:338) ~[?:1.8.0_242]
at java.util.concurrent.CompletableFuture.uniRelay(CompletableFuture.java:925) ~[?:1.8.0_242]
at java.util.concurrent.CompletableFuture$UniRelay.tryFire(CompletableFuture.java:913) ~[?:1.8.0_242]
at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488) ~[?:1.8.0_242]
at java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:1990) ~[?:1.8.0_242]
at org.apache.flink.runtime.rpc.akka.AkkaInvocationHandler.lambda$invokeRpc$0(AkkaInvocationHandler.java:227) ~[classes/:?]
at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:774) ~[?:1.8.0_242]
at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:750) ~[?:1.8.0_242]
at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488) ~[?:1.8.0_242]
at java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:1990) ~[?:1.8.0_242]
at org.apache.flink.runtime.concurrent.FutureUtils$1.onComplete(FutureUtils.java:924) ~[classes/:?]
at akka.dispatch.OnComplete.internal(Future.scala:263) ~[akka-actor_2.11-2.5.21.jar:2.5.21]
at akka.dispatch.OnComplete.internal(Future.scala:261) ~[akka-actor_2.11-2.5.21.jar:2.5.21]
at akka.dispatch.japi$CallbackBridge.apply(Future.scala:191) ~[akka-actor_2.11-2.5.21.jar:2.5.21]
at akka.dispatch.japi$CallbackBridge.apply(Future.scala:188) ~[akka-actor_2.11-2.5.21.jar:2.5.21]
at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:36) ~[scala-library-2.11.12.jar:?]
at org.apache.flink.runtime.concurrent.Executors$DirectExecutionContext.execute(Executors.java:74) ~[classes/:?]
at scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:44) ~[scala-library-2.11.12.jar:?]
at scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:252) ~[scala-library-2.11.12.jar:?]
at akka.pattern.PromiseActorRef.$bang(AskSupport.scala:572) ~[akka-actor_2.11-2.5.21.jar:2.5.21]
at akka.pattern.PipeToSupport$PipeableFuture$$anonfun$pipeTo$1.applyOrElse(PipeToSupport.scala:23) ~[akka-actor_2.11-2.5.21.jar:2.5.21]
at akka.pattern.PipeToSupport$PipeableFuture$$anonfun$pipeTo$1.applyOrElse(PipeToSupport.scala:21) ~[akka-actor_2.11-2.5.21.jar:2.5.21]
at scala.concurrent.Future$$anonfun$andThen$1.apply(Future.scala:436) ~[scala-library-2.11.12.jar:?]
at scala.concurrent.Future$$anonfun$andThen$1.apply(Future.scala:435) ~[scala-library-2.11.12.jar:?]
at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:36) ~[scala-library-2.11.12.jar:?]
at akka.dispatch.BatchingExecutor$AbstractBatch.processBatch(BatchingExecutor.scala:55) ~[akka-actor_2.11-2.5.21.jar:2.5.21]
at akka.dispatch.BatchingExecutor$BlockableBatch$$anonfun$run$1.apply$mcV$sp(BatchingExecutor.scala:91) ~[akka-actor_2.11-2.5.21.jar:2.5.21]
at akka.dispatch.BatchingExecutor$BlockableBatch$$anonfun$run$1.apply(BatchingExecutor.scala:91) ~[akka-actor_2.11-2.5.21.jar:2.5.21]
at akka.dispatch.BatchingExecutor$BlockableBatch$$anonfun$run$1.apply(BatchingExecutor.scala:91) ~[akka-actor_2.11-2.5.21.jar:2.5.21]
at scala.concurrent.BlockContext$.withBlockContext(BlockContext.scala:72) ~[scala-library-2.11.12.jar:?]
at akka.dispatch.BatchingExecutor$BlockableBatch.run(BatchingExecutor.scala:90) ~[akka-actor_2.11-2.5.21.jar:2.5.21]
at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:40) ~[akka-actor_2.11-2.5.21.jar:2.5.21]
at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(ForkJoinExecutorConfigurator.scala:44) ~[akka-actor_2.11-2.5.21.jar:2.5.21]
at akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260) [akka-actor_2.11-2.5.21.jar:2.5.21]
at akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339) [akka-actor_2.11-2.5.21.jar:2.5.21]
at akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979) [akka-actor_2.11-2.5.21.jar:2.5.21]
at akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107) [akka-actor_2.11-2.5.21.jar:2.5.21]
Caused by: org.apache.flink.runtime.taskexecutor.exceptions.TaskSubmissionException: TaskManager already contains a task for id 10fc95eb21e801e1de186569b50f073b_b992e5f3e3d3c178f8fff9fba052540c_0_0.
at org.apache.flink.runtime.taskexecutor.TaskExecutor.submitTask(TaskExecutor.java:674) ~[classes/:?]
at sun.reflect.GeneratedMethodAccessor19.invoke(Unknown Source) ~[?:?]
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:1.8.0_242]
at java.lang.reflect.Method.invoke(Method.java:498) ~[?:1.8.0_242]
at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcInvocation(AkkaRpcActor.java:284) ~[classes/:?]
at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:199) ~[classes/:?]
at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:152) ~[classes/:?]
at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26) [akka-actor_2.11-2.5.21.jar:2.5.21]
at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21) [akka-actor_2.11-2.5.21.jar:2.5.21]
at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:123) [scala-library-2.11.12.jar:?]
at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21) [akka-actor_2.11-2.5.21.jar:2.5.21]
at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:170) [scala-library-2.11.12.jar:?]
at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171) [scala-library-2.11.12.jar:?]
at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171) [scala-library-2.11.12.jar:?]
at akka.actor.Actor$class.aroundReceive(Actor.scala:517) [akka-actor_2.11-2.5.21.jar:2.5.21]
at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:225) [akka-actor_2.11-2.5.21.jar:2.5.21]
at akka.actor.ActorCell.receiveMessage(ActorCell.scala:592) [akka-actor_2.11-2.5.21.jar:2.5.21]
at akka.actor.ActorCell.invoke(ActorCell.scala:561) [akka-actor_2.11-2.5.21.jar:2.5.21]
at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:258) [akka-actor_2.11-2.5.21.jar:2.5.21]
at akka.dispatch.Mailbox.run(Mailbox.scala:225) [akka-actor_2.11-2.5.21.jar:2.5.21]
at akka.dispatch.Mailbox.exec(Mailbox.scala:235) [akka-actor_2.11-2.5.21.jar:2.5.21]
... 4 more
{code}
> TaskExecutor offering non empty slots can lead to resource violation
> --------------------------------------------------------------------
>
> Key: FLINK-18293
> URL: https://issues.apache.org/jira/browse/FLINK-18293
> Project: Flink
> Issue Type: Bug
> Components: Runtime / Coordination
> Affects Versions: 1.10.1, 1.11.0
> Reporter: Till Rohrmann
> Priority: Major
> Fix For: 1.12.0
>
>
> When a {{JobMaster}} loses leadership, then the {{TaskExecutor}} will fail all running tasks belonging to this job and transition all slots belonging to this job from {{ACTIVE}} into {{ALLOCATED}}. The idea is that these slots can be re-offered to the new leader of the very same job.
> A problem arises when the {{Task}} cancellation takes longer than the election of the new leader. In this case, the slot containing a {{CANCELLING}} task, will be offered to the new {{JobMaster}} as empty. The {{JobMaster}} not knowing that the slot still contains a resource consumer might deploy new tasks into it believing that these tasks can use all of the available resources. In the best case, the newly deployed {{Tasks}} will simply get fewer resources than thought. In the worst case this will lead to a resource violation.
> W/o the {{JobMaster}} being able to reconcile the state of already deployed {{Tasks}} into {{Slots}}, I believe that we should only re-offer the slot when it is free. One might model this scenario with introducing a new {{TaskSlotState.CLEANING}}. {{CLEANING}} means that the slot is still allocated for a given job but that there are still some resources which need to be cleaned up before it can be re-offered (transition to state {{ALLOCATED}}).
--
This message was sent by Atlassian Jira
(v8.3.4#803005)