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
