Hi team I am working on a problem where we need to dynamically decrease/increase the sources that a flink application is listening to based on different modes. Currently doing this leads to an error like

Cannot map checkpoint/savepoint state for operator f17fc775f6e82bc8d571beb78ac6c693 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.

I want to avoid the usage of

allowNonRestoredState

flag since it could hide real issues What is the way to achieve starting from snapshots safely after a change in operator chain/sources?

When decreasing the number of sources I guess you would always be removing an operator and thus have state that cannot be mapped to it. I'm not aware of any out-of-the-box solutions for solving this issue.

knowing the use case could help a bit too, but flink does have a config that ensures that every operator has a uid which would make allowNonRestoredState safer option to apply since you control every operator's uid now. i'd imagine if your dynamic sources have some sort of id, you could ensure that sources that persist across deployments will maintain the same uid, and new ones or ones that are dropped won't. however, if dropping a source and adding it back in, i'm not sure how you'd properly maintain source/iterator state on that source

Good point, @RJ Garcia. Yes, it's

pipeline.auto-generate-uids

. It will prevent some use-cases. E.g. Flink SQL always auto generates UIDs.

@RJ Garcia @Andreas Bube Thanks for responding, the use-case is related to performing replications for dynamodb tables where we are using flink as the compute. The addition/removal of sources (update stream or kinesis stream) would be needed when we switch between modes in our algorithm. Prevention of data loss/ false checkpointing is important for us We have added UIDs to all our operators and sources and further observed that addition of a source seems to be working fine when updating with snapshot without needing

allowNonRestoredState

at all Are you guys suggesting that using

disabledAutoGeneratedUids

or

pipeline.auto-generate-uids

along with the nonRestoredState flag is a safer option when removing sources/operators?

https://nightlies.apache.org/flink/flink-docs-stable/docs/ops/state/savepoints/#allowing-non-restored-state the docs state allow non restored state would only come into play when dropping operators (since you'd have state in the save point that isn't mapped to an operator) so yes, if you need to drop parts of your graph (i.e sources), and it's expected to do that, then you need to either leverage allowNonRestoredState (but ensure disable auto generated uids is set) OR, you need to use the state migration utility to modify the checkpoint/savepoint to remove the source / save a new version and then restore from that one.