[FLINK-37604] Generate static UIDs for pipeline operators

[morozov]

This PR adds a new configuration parameter operator.uid.prefix¹. Once it's specified, the pipeline composer will generate static UIDs for all its operators using this prefix. This way, for a given job, all operator UIDs will be preserved if the job graph changes.

By default, the operator UID prefix is not set. In this case, the pipeline composer will not generate operator UIDs in order to preserve backward compatibility of the existing jobs with their state that includes operator UIDs generated by Flink.

It should be recommended to set this parameter as it is recommended by Flink.

Please let me know if this a directionally right change, and I will update the documentation.

Footnotes

1. A similar approach is used in Apache Iceberg's FlinkSink.Builder (reference). ↩

[yuxiqian]

It seems we already have the pipeline option schema.operator.uid to configure UID, but only for Schema Operators. So maybe the description of operator.uid.prefix isn't precise as it doesn't apply to “all pipeline operators”.

Agree that keeping all operators UID fixed (not only for schema operators) is the right thing, so perhaps we can deprecate schema.operator.uid and favor operator.uid.prefix? We may keep state backwards compatibility with extra checking:

	schema.operator.uid set	schema.operator.uid not set
operator.uid.prefix set	Incompatible configurations Throw exceptions (?)	Set fixed UID for all operators (including schema operators)
operator.uid.prefix not set	Only set UID for schema operators (behavior unchanged for state compatibility)	Only set UID for schema operators with the default value of schema.operator.uid (behavior unchanged for state compatibility)

and remove schema.operator.uid as a breaking change later. WDYT?

[morozov]

@yuxiqian that sounds like a plan. I will work on making these changes.

[lvyanquan]

Hi @yuxiqian please take a look at this if you have time.

[yuxiqian]

Thanks for @morozov's quick response. Do you think we need an IT case to verify if operator UIDs are correctly set by examining Flink execution plan JSON (like #3887)?

[morozov]

Do you think we need an IT case to verify if operator UIDs are correctly set by examining Flink execution plan JSON (like #3887)?

@yuxiqian, it doesn't look like the execution plan JSON contains operator UIDs. It contains IDs (the numeric identifiers that start with 1 and increment by 1). For example (from the test you referenced):

{
  "nodes" : [ {
    "id" : 1,
    "type" : "Source: Distributed Source",
    "pact" : "Data Source",
    "contents" : "Source: Distributed Source",
    "parallelism" : 9
  }, {
    "id" : 2,
    "type" : "Partitioning",
    "pact" : "Operator",
    "contents" : "Partitioning",
    "parallelism" : 4,
    "predecessors" : [ {
      "id" : 1,
      "ship_strategy" : "REBALANCE",
      "side" : "second"
    } ]
  }, {
    "id" : 4,
    "type" : "SchemaMapper",
    "pact" : "Operator",
    "contents" : "SchemaMapper",
    "parallelism" : 4,
    "predecessors" : [ {
      "id" : 2,
      "ship_strategy" : "CUSTOM",
      "side" : "second"
    } ]
  }, {
    "id" : 5,
    "type" : "Sink: Sink Writer: Value Sink",
    "pact" : "Data Sink",
    "contents" : "Sink: Sink Writer: Value Sink",
    "parallelism" : 10,
    "predecessors" : [ {
      "id" : 4,
      "ship_strategy" : "REBALANCE",
      "side" : "second"
    } ]
  } ]
}

[yuxiqian]

Thanks for double-checking this; it seems we could only get the hashed VertexId in the JobGraph, and could not verify this in unit tests reliably. It should not be a blocker.

[morozov]

It looks like there's some active development happening in the code modified by this PR. Recently, I resolved conflicts with the changes from #3812, and then with the ones from #3986.

@yuxiqian, is there anything else I can do before the merge?

I added the new parameter to the documentation in English and can update the Chinese if I know what to put there.

[yuxiqian]

Thanks for @morozov's nice work, LGTM. Would @lvyanquan like to take a further look?

---

Suggested Chinese translation:

Index: docs/content.zh/docs/core-concept/data-pipeline.md
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/docs/content.zh/docs/core-concept/data-pipeline.md b/docs/content.zh/docs/core-concept/data-pipeline.md
--- a/docs/content.zh/docs/core-concept/data-pipeline.md	(revision 54768703ec976b712e62f4184b7b24c8319f8e69)
+++ b/docs/content.zh/docs/core-concept/data-pipeline.md	(date 1745460782234)
@@ -111,9 +111,10 @@
 # Pipeline 配置
 下面 是 Data Pipeline 的一些可选配置：
 
-| 参数                      | 含义                                                  | optional/required |
-|-------------------------|-----------------------------------------------------|-------------------|
-| name                    | 这个 pipeline 的名称，会用在 Flink 集群中作为作业的名称。               | optional          |
-| parallelism             | pipeline的全局并发度，默认值是1。                               | optional          |
-| local-time-zone         | 作业级别的本地时区。                                          | optional          |
-| execution.runtime-mode  | pipeline 的运行模式，包含 STREAMING 和 BATCH，默认值是 STREAMING。 | optional          |
\ No newline at end of file
+| 参数                     | 含义                                                  | optional/required |
+|------------------------|-----------------------------------------------------|-------------------|
+| name                   | 这个 Pipeline 的名称，会用在 Flink 集群中作为作业的名称。               | optional          |
+| parallelism            | Pipeline 的全局并发度，默认值是1。                              | optional          |
+| local-time-zone        | 作业级别的本地时区。                                          | optional          |
+| execution.runtime-mode | Pipeline 的运行模式，包含 STREAMING 和 BATCH，默认值是 STREAMING。 | optional          |
+| operator.uid.prefix    | Pipeline 中算子 UID 的前缀。如果不设置，Flink 会为每个算子生成唯一的 UID。   | optional          |
\ No newline at end of file

[lvyanquan]

Please add an annotation of @deprecated.

[lvyanquan]

We should clarify that this is a recommended parameter to set and explain the reason behind it.

[lvyanquan]

......
It is recommended to set this parameter to ensure stable and recognizable operator UIDs, which can help with stateful upgrades, troubleshooting, and Flink UI diagnostics.

[lvyanquan]

......
It is recommended to set this parameter to ensure stable and recognizable operator UIDs, which can help with stateful upgrades, troubleshooting, and Flink UI diagnostics.

[lvyanquan]

......
建议设置这个参数以提供稳定和可识别的算子 ID，这有助于有状态升级、问题排查和在 Flink UI 上的诊断。

[lvyanquan]

+1