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

Search before asking

• I searched in the 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?

• I'm willing to submit a PR!