Hi all, #4424 [1] added sink.operator-uid.suffix and source.operator-uid.suffix to improve state compatibility when the job graph changes. #4543 [2] is still open on a related failure, and the last question there was whether #4424 already solves it, or whether we should just recommend the suffix options to users. I think the answer is neither, because neither option reaches every operator it needs to.
We hit this in production. A streaming job writing to an append-only table, with the suffix set. We deployed our job with one extra operator in front of the sink. The Kubernetes operator tried a savepoint upgrade, the savepoint failed, it fell back to last-state recovery, and the new JobManager restored from the checkpoint store: JobInitializationException: Could not start the JobMaster. Caused by: IllegalStateException: There is no operator for the state 1312a02de57ecb55f9f6d4dfc28aad32 The id belonged to "Compact Worker: <table>". The job then stayed globally terminal and refused the same job id, so we recovered by hand. FlinkSinkBuilder adds its two row conversion operators without a uid, and AppendTableSink adds the compaction coordinator and worker without one. Writer, Global Committer and dynamic-bucket-assigner are covered. Other shapes leave more uncovered (local merge, Collect Statistics, the changelog compaction operators, INDEX_BOOTSTRAP, cross-partition-bucket-assigner). The source side has a smaller version of the same gap. toDataStream applies the uid to the source itself, then buildForRow adds a Row conversion operator after it without one. The table path calls build() and adds nothing, so this only affects DataStream readers. Two tests, both expected to fail, and the ITCase fails the same way on Flink 1.20 and 2.x: - https://github.com/bube/paimon/actions/runs/33540653767 - https://github.com/bube/paimon/blob/flink-operator-uid-suffix-restore-itcase/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/OperatorUidSuffixRestoreITCase.java - https://github.com/bube/paimon/blob/flink-operator-uid-suffix-coverage/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/OperatorUidSuffixTest.java Where I would like input is compatibility. Assigning uids changes those ids, so the fix orphans the entries in existing checkpoints and throws the same error once for anyone who already sets the suffix. I first assumed a savepoint upgrade would absorb that, since loadAndValidateCheckpoint skips unmatched operators whose state is empty. I no longer think that holds in general. Collect Statistics has a coordinator that completes its checkpoint with an empty byte[], and OperatorCoordinatorCheckpoints wraps that in a handle unconditionally, so coordinatorState is non-null and the check rejects it before it looks at emptiness. I have not audited the rest, so I do not know how many operators are in that position. So I do not have a migration path I trust. An opt-in option would let users choose when to take the one time break, but I would rather hear what you would prefer. Happy to write the patch once there is a direction. Best, Andreas [1] https://github.com/apache/paimon/issues/4424 [2] https://github.com/apache/paimon/issues/4543
