public class JobManagerOperatorMetricGroup extends ComponentMetricGroup<JobManagerJobMetricGroup> {
	private final AbstractID vertexId;
	private final String taskName;
	private final String operatorName;
	private final     .formatScope(checkNotNull(parent), operatorID, operatorName),
                parent);
        OperatorID operatorID;

	public JobManagerOperatorMetricGroup(
			MetricRegistry registry,
			JobManagerJobMetricGroup parent,
			AbstractID vertexId,
			String taskName,
			OperatorID operatorID,
			String operatorName) {
		super(
				registry,
				registry.getScopeFormats()
						.getJmOperatorFormat()
						.formatScope(checkNotNull(parent), vertexId, taskName, operatorID, operatorName),
				parent);
		this.vertexId = vertexId;
		this.taskName = taskName;
		this.operatorID = operatorID;
        		this.operatorName = operatorName;
	}

	@Override
	protected String getGroupName(CharacterFilter filter) {
		return "operator";
	}

	@Override
	protected QueryScopeInfo createQueryServiceMetricInfo(CharacterFilter filter) {
		return new QueryScopeInfo.JobQueryScopeInfo(this.parent....
}

@Internal
public class JobManagerJobMetricGroup extends JobMetricGroup<JobManagerMetricGroup> {
    private final Map<String, JobManagerOperatorMetricGroup> operators = new HashMap<>();

    public JobManagerOperatorMetricGroup getOrAddOperator (
    

@Internal
public class JobManagerJobMetricGroup extends JobMetricGroup<JobManagerMetricGroup> {
    private final Map<String, InternalOperatorCoordinatorMetricGroup> operatorCoordinators =
            new HashMap<>();

    public InternalOperatorCoordinatorMetricGroup getOrAddOperatorCoordinator(
            OperatorID operatorID, String operatorName) {
        final String truncatedOperatorName = getTruncatedOperatorName(operatorName);

        // unique OperatorIDs only exist in streaming, so we have to rely on the name for batch
        //AbstractID operatorsvertexId,
        final String key = operatorID + truncatedOperatorName;
String taskName,
        synchronized (this) {
  OperatorID operatorID,
         return operatorCoordinators.computeIfAbsent(
  String operatorName) {
        final String truncatedOperatorName = getTruncatedOperatorName(operatorName);

    key,
    // unique OperatorIDs only exist in streaming, so we  have to rely on the name operatorfor ->batch
        // operators
        final String key = operatorID + truncatedOperatorName;

     new InternalOperatorCoordinatorMetricGroup(
  synchronized (this) {
            return operators.computeIfAbsent(
                   this.registry, thiskey, operatorID, truncatedOperatorName));

         }
    }

    private String getTruncatedOperatorName(String operatorName)operator {->
        if  (operatorName != null && operatorName.length() > METRICS_OPERATOR_NAME_MAX_LENGTH) {
           new LOG.warnJobManagerOperatorMetricGroup(
                    "The operator name {} exceeded the {} characters length limit and was truncated.",
                     this.registry, this, vertexId, taskName, operatorID, truncatedOperatorName));
        }
    }

    operatorName, @Override
    protected Iterable<? extends ComponentMetricGroup> subComponents() {
           METRICS_OPERATOR_NAME_MAX_LENGTHreturn operators.values();
     }     

    return operatorName.substring(0, METRICS_OPERATOR_NAME_MAX_LENGTH);
  ......
}

@Internal
public class InternalOperatorCoordinatorMetricGroup
      } else {
extends ProxyMetricGroup<MetricGroup> implements OperatorCoordinatorMetricGroup {

    public InternalOperatorCoordinatorMetricGroup(JobManagerOperatorMetricGroup   return operatorName;parent) {
        }super(parent.addGroup("coordinator"));
    }
    ......
}