[I] [Feature] [flink] Give every streaming operator a stable uid via an opt-in operator-uid option [paimon]

["bube (via GitHub)" <gi...@apache.org>]

bube opened a new issue, #10107:
URL: https://github.com/apache/paimon/issues/10107

   ### Search before asking
   
   - [x] I searched in the [issues](https://github.com/apache/paimon/issues) and found nothing similar.
   
   
   ### Motivation
   
   `sink.operator-uid.suffix` and `source.operator-uid.suffix` (#4424) reach only Writer, Global
   Committer, dynamic-bucket-assigner and Source. Every other operator Paimon adds to a streaming
   job (row conversions, append compaction coordinator/worker, changelog compaction, local merge,
   Collect/Strip Statistics, INDEX_BOOTSTRAP, cross-partition-bucket-assigner, Monitor/Reader,
   Timestamps/Watermarks, the final `end` sink) takes its id from the stream graph shape.
   
   Any topology change upstream of the sink then orphans their checkpoint entries. A restore from
   the HA checkpoint store (JobManager failover, Kubernetes Operator `last-state`) rejects every
   unclaimed entry:
   
       IllegalStateException: There is no operator for the state 1312a02de57ecb55f9f6d4dfc28aad32
   
   We hit this in production on an append table with the suffix set, after adding one operator in
   front of the sink. #4543 is the same failure. Discussed on dev@ in
   "[DISCUSS] The operator-uid.suffix options do not cover every operator"
   (https://lists.apache.org/thread/szjmms72lsrktqps37g5j62mcjtklc9o), where Jingsong asked for an
   opt-in option with a documented migration procedure.
   
   
   
   
   ### Solution
   
   - New `sink.operator-uid.cover-all-operators` / `source.operator-uid.cover-all-operators`,
     boolean, default false. With the option and the matching suffix set, every streaming
     operator gets `<prefix>_<table>_<suffix>`. Existing uids and the default graph are unchanged.
   - Batch-only operators are out of scope: they never restore from a checkpoint.
   - Tests: per-shape graph build under `pipeline.auto-generate-uids=false`, golden pins for all
     uids, upstream-change id stability, restore from a retained checkpoint store (HA route),
     and migration from the old layout on the explicit-path route incl. PARTITION_DYNAMIC.
   - Docs: "Operator UIDs" section in flink/savepoint.md with a migration procedure. Enabling the
     option orphans the newly named operators' entries once; all are empty except Collect
     Statistics and the Monitor source, which need `execution.state-recovery.ignore-unclaimed-state`
     on the explicit-path route. The HA route cannot skip entries, so migrate via a savepoint.
   
   ### Anything else?
   
   _No response_
   
   ### Are you willing to submit a PR?
   
   - [x] I'm willing to submit a PR!


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org