[
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)
