Skip to content

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

Description

@bube

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!

Activity

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    enhancementNew feature or request

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions