Unable to recover from continuous query Hudi MOR using Flink checkpoint

Firstly, I startup the flink sql client, with command, I have specified the option `--allowNonRestoredState`.

```
sql-client.sh embedded shell --allowNonRestoredState
```

I am using the following code snippet to continuous read a Hudi table

```
set execution.checkpointing.interval=10000;
set 'state.checkpoints.dir'='hdfs:///flink/checkpoints/test_flink_incremental_second_20251221_03';
set 'execution.checkpointing.externalized-checkpoint-retention'='RETAIN_ON_CANCELLATION';
set 'state.checkpoints.num-retained'='60';
set execution.checkpointing.num-retained-checkpoint=80;
```
The Flink SQL is as follows
- test_flink_incremental_second_20251221_03 points to a Hudi table
- test_sink_second_20251221_03 points to an mysql table to save the data that query out.

```

insert into test_sink_second_20251221_03 select * from test_flink_incremental_second_20251221_03
/*+
options(
'read.streaming.enabled'='true',
'read.streaming.check-interval' = '4',
'read.start-commit' = '20251215134527179'
)
*/
```

After a while, I cancel the insert select job and try to start by setting the savepoint directory

```
set execution.savepoint.path='hdfs:///flink/checkpoints/test_flink_incremental_second_20251221_03/d7cc6b5d643f3a8976df93f6e9aa0c48/chk-9';

```

When I rerun the query, an error occurred complaining, I have no idea how to fix this problem.

```
[ERROR] Could not execute SQL statement. Reason:
java.lang.IllegalStateException: Failed to rollback to checkpoint/savepoint hdfs://hadoop.master:9000/flink/checkpoints/test_flink_incremental_second_20251221_03/d7cc6b5d643f3a8976df93f6e9aa0c48/chk-9.
Cannot map checkpoint/savepoint state for operator c43c1940ad62bf3bc996ebfabca09279 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.

```

No source file or test is named. Start by reproducing the Flink SQL Client recovery flow with the supplied Hudi MOR table, checkpoint settings, and execution.savepoint.path, then inspect how the restored operator state is mapped when the query is rerun with --allowNonRestoredState. Done means identifying why the operator is unavailable and documenting or verifying a reliable recovery path.

Written by the indexing model from the issue text.