Hi Andreas, Thanks for the investigation and the reproducer. I would prefer an opt-in option, with the current behavior preserved by default.
I checked the current master and confirmed the missing UIDs. I also ran minimal checks against Flink 1.20.4 and 2.2.0: restoring through an explicit snapshot path skips unmatched empty operator entries, while HA checkpoint recovery rejects them. An empty byte[] coordinator snapshot still produces a non-null state handle, so Collect Statistics is rejected by the former path as well. For the patch, I suggest: - Keep all existing explicit UIDs unchanged. - When the new option and the corresponding suffix are set, assign stable UIDs to the remaining Paimon operators, including the DataStream row conversions. - Add coverage for UID stability after upstream topology changes, recovery from a retained checkpoint store, and migration from the old UID layout, including PARTITION_DYNAMIC. The option lets existing jobs choose when to migrate, but enabling it still needs a documented migration procedure. Release notes alone would leave users exposed to an unexpected recovery failure on upgrade. For cases requiring allowNonRestoredState, we should first verify exactly which entries are being dropped and that the source, writer and committer state remains mapped; we should not recommend ignoring unmatched state unconditionally. The current FlinkJobRecoveryITCase restores through execution.state-recovery.path, which explains why its coverage does not catch the last-state failure you reported. A patch along these lines would be welcome. Best, Jingsong
