@PublicEvolving
public class MetricOptions {
    ......

    /**
     * The scope format string that is applied to all metrics scoped to an operator coordinator on a
     * JobManager.
     */
    public static final ConfigOption<String> SCOPE_NAMING_JM_JOB_OPERATOR_COORDINATOR =
            key("metrics.scope.jm-job.operator-coordinator")
                    .stringType()
                    .defaultValue("<host>.jobmanager.<job_name>.<operator_name>.coordinator")
                    .withDescription(
                            "Defines the scope format string that is applied to all metrics scoped to an operator coordinator on a JobManager.");
    ......
}

/**
 * Special {@link MetricGroup} representing an Operator coordinator.
 *
 * <p>You should only update the metrics in the coordinator thread.
 */
@PublicEvolving
public interface OperatorCoordinatorMetricGroup extends MetricGroup {
    /**
	 * The total number of events received since the operator coordinator started.
	 */
	Counter getNumEventsInCounter();

    /**
	 * The total number of events sent by the operator coordinator.
	 */
	Counter getNumEventsOutCounter(); 
}

public class OperatorCoordinatorScopeFormat extends ScopeFormat {
    public OperatorCoordinatorScopeFormat(String format, JobManagerJobScopeFormat parentFormat) {
        super(
                format,
                parentFormat,
                new String[] {
                    SCOPE_HOST, SCOPE_JOB_ID, SCOPE_JOB_NAME, SCOPE_OPERATOR_ID, SCOPE_OPERATOR_NAME
                });
    }

    public String[] formatScope(
            JobManagerJobMetricGroup parent, OperatorID operatorID, String operatorName) {

        final String[] template = copyTemplate();
        final String[] values = {
            parent.parent().hostname(),
            valueOrNull(parent.jobId()),
            valueOrNull(parent.jobName()),
            valueOrNull(operatorID),
            valueOrNull(operatorName)
        };
        return bindVariables(template, values);
    }
}

public final class ScopeFormats {
    ......
    private final OperatorCoordinatorScopeFormat operatorCoordinatorScopeFormat;

    private ScopeFormats(
            String jobManagerFormat,
            String jobManagerJobFormat,
            String taskManagerFormat,
            String taskManagerJobFormat,
            String taskFormat,
            String operatorFormat,
            String operatorCoordinatorFormat) {
        ......
        this.operatorCoordinatorScopeFormat =
                new OperatorCoordinatorScopeFormat(
                        operatorCoordinatorFormat, this.jobManagerJobFormat);
    }

    public OperatorCoordinatorScopeFormat getOperatorCoordinatorScopeFormat() {
        return operatorCoordinatorScopeFormat;
    }
}

public class InternalOperatorCoordinatorMetricGroup
        extends ComponentMetricGroup<JobManagerJobMetricGroup>
        implements OperatorCoordinatorMetricGroup {

    private final String operatorName;
    private final OperatorID operatorID;

    public InternalOperatorCoordinatorMetricGroup(
            MetricRegistry registry,
            JobManagerJobMetricGroup parent,
            OperatorID operatorID,
            String operatorName) {
        super(
                registry,
                registry.getScopeFormats()
                        .getOperatorCoordinatorScopeFormat()
                        .formatScope(checkNotNull(parent), operatorID, operatorName),
                parent);
        this.operatorID = operatorID;
        this.operatorName = operatorName;
    }

    ......
}

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
        // operators
        final String key = operatorID + truncatedOperatorName;

        synchronized (this) {
            return operatorCoordinators.computeIfAbsent(
                    key,
                    operator ->
                            new InternalOperatorCoordinatorMetricGroup(
                                    this.registry, this, operatorID, truncatedOperatorName));
        }
    }

    private String getTruncatedOperatorName(String operatorName) {
        if (operatorName != null && operatorName.length() > METRICS_OPERATOR_NAME_MAX_LENGTH) {
            LOG.warn(
                    "The operator name {} exceeded the {} characters length limit and was truncated.",
                    operatorName,
                    METRICS_OPERATOR_NAME_MAX_LENGTH);
            return operatorName.substring(0, METRICS_OPERATOR_NAME_MAX_LENGTH);
        } else {
            return operatorName;
        }
    }
    ......
}

@Internal
public interface OperatorCoordinator extends CheckpointListener, AutoCloseable {
    interface Context {
        /** Gets the metric group of the operator coordinator. */
        OperatorCoordinatorMetricGroup metricGroup();
        
        ....
    }
}

@Internal
public class SourceCoordinator<SplitT extends SourceSplit, EnumChkT>
        implements OperatorCoordinator {
    ......
    private final Counter numEventsInCounter;

    public SourceCoordinator(
            String operatorName,
            Source<?, SplitT, EnumChkT> source,
            SourceCoordinatorContext<SplitT> context,
            CoordinatorStore coordinatorStore,
            WatermarkAlignmentParams watermarkAlignmentParams,
            @Nullable String coordinatorListeningID) {
        ......
        this.numEventsInCounter = context.metricGroup().getNumEventsInCounter();
    	......
    }

    @Override
    public void handleEventFromOperator(int subtask, int attemptNumber, OperatorEvent event) {
        numEventsInCounter.inc();
		......
	}
}

@Internal
public class KafkaSourceEnumerator
        implements SplitEnumerator<KafkaPartitionSplit, KafkaSourceEnumState> {
    public KafkaSourceEnumerator(
            KafkaSubscriber subscriber,
            OffsetsInitializer startingOffsetInitializer,
            OffsetsInitializer stoppingOffsetInitializer,
            Properties properties,
            SplitEnumeratorContext<KafkaPartitionSplit> context,
            Boundedness boundedness,
            Set<TopicPartition> assignedPartitions) {
        ......
        context.metricGroup().gauge("assignedPartitionsNum", assignedPartitions::size);
    }
    
    ......
}