Hello Team, I have question around flink checkpoint. when I downscale my application (e.g. parallelism 4 ->2) and try to restore the application from previous checkpoint with new parallelism, I get following exception:

Caused by: java.util.concurrent.CompletionException: java.lang.IllegalStateException: Failed to rollback to checkpoint/savepoint <hdfs://blackhole-auditlog/data/dcn-log-parser/checkpointing/755c50f2ddfbed68607f7c6ad2428c4c/chk-1493>. Cannot map checkpoint/savepoint state for operator cbc357ccb763df2852fee8c4fc7d55f2 to the new program, because the operator is not available in the new program. If you want to allow to skip this, you can set the --allowNonRestoredState option on the CLI.
	at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:273)
	at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:280)
	at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1592)
	... 3 more
Caused by: java.lang.IllegalStateException: Failed to rollback to checkpoint/savepoint <hdfs://blackhole-auditlog/data/dcn-log-parser/checkpointing/755c50f2ddfbed68607f7c6ad2428c4c/chk-1493>. Cannot map checkpoint/savepoint state for operator cbc357ccb763df2852fee8c4fc7d55f2 to the new program, because the operator is not available in the new program. If you want to allow to skip this, you can set the --allowNonRestoredState option on the CLI.
	at org.apache.flink.runtime.checkpoint.Checkpoints.throwNonRestoredStateException(Checkpoints.java:238)
	at org.apache.flink.runtime.checkpoint.Checkpoints.loadAndValidateCheckpoint(Checkpoints.java:197)
	at org.apache.flink.runtime.checkpoint.CheckpointCoordinator.restoreSavepoint(CheckpointCoordinator.java:1798)
	at org.apache.flink.runtime.scheduler.DefaultExecutionGraphFactory.tryRestoreExecutionGraphFromSavepoint(DefaultExecutionGraphFactory.java:214)
	at org.apache.flink.runtime.scheduler.DefaultExecutionGraphFactory.createAndRestoreExecutionGraph(DefaultExecutionGraphFactory.java:189)
	at org.apache.flink.runtime.scheduler.SchedulerBase.createAndRestoreExecutionGraph(SchedulerBase.java:361)
	at org.apache.flink.runtime.scheduler.SchedulerBase.<init>(SchedulerBase.java:206)
	at org.apache.flink.runtime.scheduler.DefaultScheduler.<init>(DefaultScheduler.java:134)
	at org.apache.flink.runtime.scheduler.DefaultSchedulerFactory.createInstance(DefaultSchedulerFactory.java:152)
	at org.apache.flink.runtime.jobmaster.DefaultSlotPoolServiceSchedulerFactory.createScheduler(DefaultSlotPoolServiceSchedulerFactory.java:119)
	at org.apache.flink.runtime.jobmaster.JobMaster.createScheduler(JobMaster.java:369)
	at org.apache.flink.runtime.jobmaster.JobMaster.<init>(JobMaster.java:346)
	at org.apache.flink.runtime.jobmaster.factories.DefaultJobMasterServiceFactory.internalCreateJobMasterService(DefaultJobMasterServiceFactory.java:123)
	at org.apache.flink.runtime.jobmaster.factories.DefaultJobMasterServiceFactory.lambda$createJobMasterService$0(DefaultJobMasterServiceFactory.java:95)
	at org.apache.flink.util.function.FunctionUtils.lambda$uncheckedSupplier$4(FunctionUtils.java:112)
	at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1590)

I know I can restore using my application using option

--allowNonRestoredState

or

-n

. My question is if I use

--allowNonRestoredState

option, do I loose the state(application state (keyed state)) of my application for the operators that were non-restorable?

AFAIK, if you change the parallelism of the operators/ global parallelism then it affects your checkpoints and savepoints.

"operator cbc357ccb763df2852fee8c4fc7d55f2" sounds as if you didn't set an explicit operator UID. That's very likely to cause restore problems (cf. docs)

thanks Michael, I didn’t set

uid

explicitly to some of the operators, which was causing problem in restoring. After setting explict

uid

to each operator, downscale works smoothly.

great 🙂