You are on the right path with using the --allowNonRestoredState flag;
we'll just have to find the right place to put it w.r.t. your setup.
Which docker images are you using (flink/statefun/something custom), and
how do you submit the job?
On 03/06/2022 01:17, Bhavani Balasubramanyam wrote:
Hi,
I am Bhavani, a Software Engineer at Reddit. I'm trying to upgrade the
Flink version in my application from 3.0.0 to version 3.2.0, and in the
process I see the below error, where the the operator has been removed,
and the checkpoint is unable to recover:
- Jun 2 14:32:03
snooron-worker-perspective-flink-staging-statefun-master-5f8hzd master
ERROR org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - Fatal
error occurred in the cluster entrypoint.
org.apache.flink.util.FlinkException: JobMaster for job
00000000000000000000000000000000 failed. at
org.apache.flink.runtime.dispatcher.Dispatcher.jobMasterFailed(Dispatcher.java:913)
~[flink-dist_2.12-1.14.3.jar:1.14.3] at
org.apache.flink.runtime.dispatcher.Dispatcher.jobManagerRunnerFailed(Dispatcher.java:473)
~[flink-dist_2.12-1.14.3.jar:1.14.3] at
org.apache.flink.runtime.dispatcher.Dispatcher.handleJobManagerRunnerResult(Dispatcher.java:450)
~[flink-dist_2.12-1.14.3.jar:1.14.3] at
org.apache.flink.runtime.dispatcher.Dispatcher.lambda$runJob$3(Dispatcher.java:427)
~[flink-dist_2.12-1.14.3.jar:1.14.3] at
java.util.concurrent.CompletableFuture.uniHandle(Unknown Source) ~[?:?] at
java.util.concurrent.CompletableFuture$UniHandle.tryFire(Unknown Source)
~[?:?] at java.util.concurrent.CompletableFuture$Completion.run(Unknown
Source) ~[?:?] at
org.apache.flink.runtime.rpc.akka.AkkaRpcActor.lambda$handleRunAsync$4(AkkaRpcActor.java:455)
~[flink-rpc-akka_0833457a-e96a-488d-b0fd-a4aafb20352d.jar:1.14.3] at
org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:68)
~[flink-rpc-akka_0833457a-e96a-488d-b0fd-a4aafb20352d.jar:1.14.3] at
org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRunAsync(AkkaRpcActor.java:455)
~[flink-rpc-akka_0833457a-e96a-488d-b0fd-a4aafb20352d.jar:1.14.3] at
org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:213)
~[flink-rpc-akka_0833457a-e96a-488d-b0fd-a4aafb20352d.jar:1.14.3] at
org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:78)
~[flink-rpc-akka_0833457a-e96a-488d-b0fd-a4aafb20352d.jar:1.14.3] at
org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:163)
~[flink-rpc-akka_0833457a-e96a-488d-b0fd-a4aafb20352d.jar:1.14.3] at
akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:24)
[flink-rpc-akka_0833457a-e96a-488d-b0fd-a4aafb20352d.jar:1.14.3] at
akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:20)
[flink-rpc-akka_0833457a-e96a-488d-b0fd-a4aafb20352d.jar:1.14.3] at
scala.PartialFunction.applyOrElse(PartialFunction.scala:123)
[flink-rpc-akka_0833457a-e96a-488d-b0fd-a4aafb20352d.jar:1.14.3] at
scala.PartialFunction.applyOrElse$(PartialFunction.scala:122)
[flink-rpc-akka_0833457a-e96a-488d-b0fd-a4aafb20352d.jar:1.14.3] at
akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:20)
[flink-rpc-akka_0833457a-e96a-488d-b0fd-a4aafb20352d.jar:1.14.3] at
scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)
[flink-rpc-akka_0833457a-e96a-488d-b0fd-a4aafb20352d.jar:1.14.3] at
scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:172)
[flink-rpc-akka_0833457a-e96a-488d-b0fd-a4aafb20352d.jar:1.14.3] at
scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:172)
[flink-rpc-akka_0833457a-e96a-488d-b0fd-a4aafb20352d.jar:1.14.3] at
akka.actor.Actor.aroundReceive(Actor.scala:537)
[flink-rpc-akka_0833457a-e96a-488d-b0fd-a4aafb20352d.jar:1.14.3] at
akka.actor.Actor.aroundReceive$(Actor.scala:535)
[flink-rpc-akka_0833457a-e96a-488d-b0fd-a4aafb20352d.jar:1.14.3] at
akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:220)
[flink-rpc-akka_0833457a-e96a-488d-b0fd-a4aafb20352d.jar:1.14.3] at
akka.actor.ActorCell.receiveMessage(ActorCell.scala:580)
[flink-rpc-akka_0833457a-e96a-488d-b0fd-a4aafb20352d.jar:1.14.3] at
akka.actor.ActorCell.invoke(ActorCell.scala:548)
[flink-rpc-akka_0833457a-e96a-488d-b0fd-a4aafb20352d.jar:1.14.3] at
akka.dispatch.Mailbox.processMailbox(Mailbox.scala:270)
[flink-rpc-akka_0833457a-e96a-488d-b0fd-a4aafb20352d.jar:1.14.3] at
akka.dispatch.Mailbox.run(Mailbox.scala:231)
[flink-rpc-akka_0833457a-e96a-488d-b0fd-a4aafb20352d.jar:1.14.3] at
akka.dispatch.Mailbox.exec(Mailbox.scala:243)
[flink-rpc-akka_0833457a-e96a-488d-b0fd-a4aafb20352d.jar:1.14.3] at
java.util.concurrent.ForkJoinTask.doExec(Unknown Source) [?:?] at
java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(Unknown Source)
[?:?] at java.util.concurrent.ForkJoinPool.scan(Unknown Source) [?:?] at
java.util.concurrent.ForkJoinPool.runWorker(Unknown Source) [?:?] at
java.util.concurrent.ForkJoinWorkerThread.run(Unknown Source) [?:?] Caused
by: org.apache.flink.runtime.client.JobInitializationException: Could not
start the JobMaster. at
org.apache.flink.runtime.jobmaster.DefaultJobMasterServiceProcess.lambda$new$0(DefaultJobMasterServiceProcess.java:97)
~[flink-dist_2.12-1.14.3.jar:1.14.3] at
java.util.concurrent.CompletableFuture.uniWhenComplete(Unknown Source)
~[?:?] at
java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(Unknown
Source) ~[?:?] at
java.util.concurrent.CompletableFuture.postComplete(Unknown Source) ~[?:?]
at java.util.concurrent.CompletableFuture$AsyncSupply.run(Unknown Source)
~[?:?] at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
~[?:?] at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown
Source) ~[?:?] at java.lang.Thread.run(Unknown Source) ~[?:?] Caused by:
java.util.concurrent.CompletionException: java.lang.IllegalStateException:
There is no operator for the state <operator_id> at
java.util.concurrent.CompletableFuture.encodeThrowable(Unknown Source)
~[?:?] at java.util.concurrent.CompletableFuture.completeThrowable(Unknown
Source) ~[?:?] at
java.util.concurrent.CompletableFuture$AsyncSupply.run(Unknown Source)
~[?:?] at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
~[?:?] at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown
Source) ~[?:?] at java.lang.Thread.run(Unknown Source) ~[?:?] Caused by:
java.lang.IllegalStateException: There is no operator for the state
9727c19a8b3d91cecfdb945b014d10c1 at
org.apache.flink.runtime.checkpoint.StateAssignmentOperation.checkStateMappingCompleteness(StateAssignmentOperation.java:726)
~[flink-dist_2.12-1.14.3.jar:1.14.3] at
org.apache.flink.runtime.checkpoint.StateAssignmentOperation.assignStates(StateAssignmentOperation.java:100)
~[flink-dist_2.12-1.14.3.jar:1.14.3] at
org.apache.flink.runtime.checkpoint.CheckpointCoordinator.restoreLatestCheckpointedStateInternal(CheckpointCoordinator.java:1562)
~[flink-dist_2.12-1.14.3.jar:1.14.3] at
org.apache.flink.runtime.checkpoint.CheckpointCoordinator.restoreInitialCheckpointIfPresent(CheckpointCoordinator.java:1474)
~[flink-dist_2.12-1.14.3.jar:1.14.3] at
org.apache.flink.runtime.scheduler.DefaultExecutionGraphFactory.createAndRestoreExecutionGraph(DefaultExecutionGraphFactory.java:134)
~[flink-dist_2.12-1.14.3.jar:1.14.3] at
org.apache.flink.runtime.scheduler.SchedulerBase.createAndRestoreExecutionGraph(SchedulerBase.java:335)
~[flink-dist_2.12-1.14.3.jar:1.14.3] at
org.apache.flink.runtime.scheduler.SchedulerBase.<init>(SchedulerBase.java:191)
~[flink-dist_2.12-1.14.3.jar:1.14.3] at
org.apache.flink.runtime.scheduler.DefaultScheduler.<init>(DefaultScheduler.java:140)
~[flink-dist_2.12-1.14.3.jar:1.14.3] at
org.apache.flink.runtime.scheduler.DefaultSchedulerFactory.createInstance(DefaultSchedulerFactory.java:134)
~[flink-dist_2.12-1.14.3.jar:1.14.3] at
org.apache.flink.runtime.jobmaster.DefaultSlotPoolServiceSchedulerFactory.createScheduler(DefaultSlotPoolServiceSchedulerFactory.java:110)
~[flink-dist_2.12-1.14.3.jar:1.14.3] at
org.apache.flink.runtime.jobmaster.JobMaster.createScheduler(JobMaster.java:346)
~[flink-dist_2.12-1.14.3.jar:1.14.3] at
org.apache.flink.runtime.jobmaster.JobMaster.<init>(JobMaster.java:323)
~[flink-dist_2.12-1.14.3.jar:1.14.3] at
org.apache.flink.runtime.jobmaster.factories.DefaultJobMasterServiceFactory.internalCreateJobMasterService(DefaultJobMasterServiceFactory.java:106)
~[flink-dist_2.12-1.14.3.jar:1.14.3] at
org.apache.flink.runtime.jobmaster.factories.DefaultJobMasterServiceFactory.lambda$createJobMasterService$0(DefaultJobMasterServiceFactory.java:94)
~[flink-dist_2.12-1.14.3.jar:1.14.3] at
org.apache.flink.util.function.FunctionUtils.lambda$uncheckedSupplier$4(FunctionUtils.java:112)
~[flink-dist_2.12-1.14.3.jar:1.14.3] at
java.util.concurrent.CompletableFuture$AsyncSupply.run(Unknown Source)
~[?:?] at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
~[?:?] at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown
Source) ~[?:?] at java.lang.Thread.run(Unknown Source) ~[?:?]
I tried setting execution.savepoint.ignore-unclaimed-state: true in the
Flink config file, but this did not work.
I wanted to know how we could pass arguments to Flink during the creation
of docker image, as I wanted to set the --allowNonRestoredState flag to
skip the savepoint state that cannot be restored with the new job.
Also tried exec /opt/flink/bin/flink run --allowNonRestoredState from the
Flink entrypoint file in my repo but no luck.
Could someone please help me with this?
Thanks,
Bhavani
