一、背景

   Flink cdc 的sql做进行调整，添加where条件过滤、调整sql，现在进行升级重启，想从check poit恢复数据，出现一下问题

二：错误

Caused by: java.lang.IllegalStateException: Failed to rollback to checkpoint/savepoint hdfs://127.0.0.1/flink-checkpoints/78f7cb6b577fe6db19648ca63607e640/chk-6976. Cannot map checkpoint/savepoint state for operator e75d4004e6c5f0908bd4077fcf200fcd 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:230) ~[flink-runtime_2.12-1.13.2.jar:1.13.2]
        at org.apache.flink.runtime.checkpoint.Checkpoints.loadAndValidateCheckpoint(Checkpoints.java:194) ~[flink-runtime_2.12-1.13.2.jar:1.13.2]
        at org.apache.flink.runtime.checkpoint.CheckpointCoordinator.restoreSavepoint(CheckpointCoordinator.java:1629) ~[flink-runtime_2.12-1.13.2.jar:1.13.2]
        at org.apache.flink.runtime.scheduler.DefaultExecutionGraphFactory.tryRestoreExecutionGraphFromSavepoint(DefaultExecutionGraphFactory.java:163) ~[flink-runtime_2.12-1.13.2.jar:1.13.2]
        at org.apache.flink.runtime.scheduler.DefaultExecutionGraphFactory.createAndRestoreExecutionGraph(DefaultExecutionGraphFactory.java:138) ~[flink-runtime_2.12-1.13.2.jar:1.13.2]
        at org.apache.flink.runtime.scheduler.SchedulerBase.createAndRestoreExecutionGraph(SchedulerBase.java:342) ~[flink-runtime_2.12-1.13.2.jar:1.13.2]
        at org.apache.flink.runtime.scheduler.SchedulerBase.<init>(SchedulerBase.java:190) ~[flink-runtime_2.12-1.13.2.jar:1.13.2]
        at org.apache.flink.runtime.scheduler.DefaultScheduler.<init>(DefaultScheduler.java:122) ~[flink-runtime_2.12-1.13.2.jar:1.13.2]
        at org.apache.flink.runtime.scheduler.DefaultSchedulerFactory.createInstance(DefaultSchedulerFactory.java:132) ~[flink-runtime_2.12-1.13.2.jar:1.13.2]
        at org.apache.flink.runtime.jobmaster.DefaultSlotPoolServiceSchedulerFactory.createScheduler(DefaultSlotPoolServiceSchedulerFactory.java:110) ~[flink-runtime_2.12-1.13.2.jar:1.13.2]
        at org.apache.flink.runtime.jobmaster.JobMaster.createScheduler(JobMaster.java:340) ~[flink-runtime_2.12-1.13.2.jar:1.13.2]
        at org.apache.flink.runtime.jobmaster.JobMaster.<init>(JobMaster.java:317) ~[flink-runtime_2.12-1.13.2.jar:1.13.2]
        at org.apache.flink.runtime.jobmaster.factories.DefaultJobMasterServiceFactory.internalCreateJobMasterService(DefaultJobMasterServiceFactory.java:107) ~[flink-runtime_2.12-1.13.2.jar:1.13.2]
        at org.apache.flink.runtime.jobmaster.factories.DefaultJobMasterServiceFactory.lambda$createJobMasterService$0(DefaultJobMasterServiceFactory.java:95) ~[flink-runtime_2.12-1.13.2.jar:1.13.2]
        at org.apache.flink.util.function.FunctionUtils.lambda$uncheckedSupplier$4(FunctionUtils.java:112) ~[flink-core-1.13.2.jar:1.13.2]
        at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1590) ~[?:1.8.0_151]
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) ~[?:1.8.0_151]
        at java.util.concurrent.FutureTask.run(FutureTask.java:266) ~[?:1.8.0_151]
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) ~[?:1.8.0_151]
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) ~[?:1.8.0_151]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[?:1.8.0_151]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_151]
        at java.lang.Thread.run(Thread.java:748) ~[?:1.8.0_151]

三、问题分析

       调整sql，相当变更算子，这时执行重新checkpoit会报错，需忽略这些新算子

    $ bin/flink run -s :savepointPath -n [:runArgs]

    新增一个-n 参数就正常