[FLINK-14740] Create OperatorID for OperatorMetricGroup which in batch job

• Type: Improvement
• Status: Closed
• Priority: Major
• Resolution: Abandoned
• Component/s: Runtime / Metrics
• Assignee: Unassigned
• Reporter: lining
• Created: 13/Nov/19 09:52
• Updated: 13/Apr/21 20:30
• Resolved: 28/Aug/20 09:49

Description

In current design:

The DataSet job uses VertexID as the OperatorID in the OperatorMetricGroup (ps:TaskMetricGroup.getOrAddOperator (string name)).

If two operators in the same vertex have the same name, they will overwrite each other in the TaskMetricGroup.

Proposal:

We could add the OperatorID to the operator of the dataset.

for (TaskInChain tic : this.chainedTasksInSequence) {
   TaskConfig t = new TaskConfig(tic.getContainingVertex().getConfiguration());
   Integer nodeId = tic.getPlanNode().getOptimizerNode().getId();
   OperatorID operatorID = this.nodeId2OperatorId.get(nodeId);
   if(operatorID == null) {
      operatorID = new OperatorID();
      this.nodeId2OperatorId.put(nodeId, operatorID);
   }
   t.addChainedTask(tic.getChainedTask(), tic.getTaskConfig(), tic.getTaskName(), operatorID.toString());
}

Then we could get id from TaskInfo.