(incubator-amoro) branch master updated: [AMORO-2688] Fix flink runtime info NullPointException (#2689)

[yuanfeng]

This is an automated email from the ASF dual-hosted git repository.

yuanfeng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-amoro.git


The following commit(s) were added to refs/heads/master by this push:
     new 5faffab93 [AMORO-2688] Fix flink runtime info NullPointException 
(#2689)
5faffab93 is described below

commit 5faffab9325dedd91ed0ebd4bca62c644dca8542
Author: big face cat <731030...@qq.com>
AuthorDate: Mon Apr 8 19:19:28 2024 +0800

    [AMORO-2688] Fix flink runtime info NullPointException (#2689)
    
    * [AMORO-2688] Fix flink runtime info nullPointException
    
    * add subtask_index runtime info
    
    ---------
    
    Co-authored-by: huyuanfeng <huyuanf...@huya.com>
---
 .../java/com/netease/arctic/optimizer/flink/FlinkExecutor.java   | 9 +++++++--
 1 file changed, 7 insertions(+), 2 deletions(-)

diff --git 
a/ams/optimizer/flink-optimizer/src/main/java/com/netease/arctic/optimizer/flink/FlinkExecutor.java
 
b/ams/optimizer/flink-optimizer/src/main/java/com/netease/arctic/optimizer/flink/FlinkExecutor.java
index 046cb9624..84181321b 100644
--- 
a/ams/optimizer/flink-optimizer/src/main/java/com/netease/arctic/optimizer/flink/FlinkExecutor.java
+++ 
b/ams/optimizer/flink-optimizer/src/main/java/com/netease/arctic/optimizer/flink/FlinkExecutor.java
@@ -51,8 +51,13 @@ public class FlinkExecutor extends 
AbstractStreamOperator<Void>
         
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());