[AMORO-2688] Fix flink runtime info NullPointException (apache#2689)

* [AMORO-2688] Fix flink runtime info nullPointException * add subtask_index runtime info --------- Co-authored-by: huyuanfeng <[email protected]>
huyuanfeng2018 · Apr 8, 2024 · 5faffab · 5faffab
1 parent 561cab5
commit 5faffab
Showing 1 changed file with 7 additions and 2 deletions.
diff --git a/...mizer/flink-optimizer/src/main/java/com/netease/arctic/optimizer/flink/FlinkExecutor.java b/...mizer/flink-optimizer/src/main/java/com/netease/arctic/optimizer/flink/FlinkExecutor.java
@@ -51,8 +51,13 @@ public void open() throws Exception {
         getRuntimeContext().getTaskManagerRuntimeInfo().getConfiguration().getString(HA_CLUSTER_ID);
     executor = (FlinkOptimizerExecutor) allExecutors[subTaskIndex];
     // set optimizer flink runtime info, including application_id, tm_id, host
-    executor.addRuntimeContext("application_id", applicationId);
-    executor.addRuntimeContext("tm_id", taskManagerId);
+    if (applicationId != null) {
+      executor.addRuntimeContext("application_id", applicationId);
+    }
+    if (taskManagerId != null) {
+      executor.addRuntimeContext("tm_id", taskManagerId);
+    }
+    executor.addRuntimeContext("subtask_index", String.valueOf(subTaskIndex));
     // add label optimize_group;
     getMetricGroup().getAllVariables().put("<optimizer_group>", optimizeGroupName);
     executor.initOperatorMetric(getMetricGroup());