@Public
public interface RuntimeContext {
	...
    /**
     * Returns the metric group for this parallel subtask.
     *
     * @return The metric group for this parallel subtask.
     */
    @PublicEvolving
    OperatorMetricGroup getMetricGroup(); // more specific return type
    ...
}

/**
 * Special {@link org.apache.flink.metrics.MetricGroup} representing an Operator.
 *
 * <p>All metrics can only be accessed in the main operator thread.
 */
@NotThreadSafe
public interface OperatorMetricGroup extends MetricGroup {
    OperatorIOMetricGroup getIOMetricGroup();
}

/**
 * Metric group that contains shareable pre-defined IO-related metrics for operators.
 *
 * <p>All metrics can only be accessed in the main operator thread.
 */
@NotThreadSafe
public interface OperatorIOMetricGroup extends MetricGroup {
    /**
     * The total number of input records since the operator started. Will also populate
     * numRecordsInPerSecond meter.
     */
    Counter getNumRecordsInCounter();

    /**
     * The total number of output records since the operator started. Will also populate
     * numRecordsOutPerSecond meter.
     */
    Counter getNumRecordsOutCounter();

    /**
     * The total number of input bytes since the task started. Will also populate
     * numBytesInPerSecond meter.
     */
    Counter getNumBytesInCounter();

    /**
     * The total number of output bytes since the task started. Will also populate
     * numBytesOutPerSecond meter.
     */
    Counter getNumBytesOutCounter();
}

/**
 * Special {@link MetricGroup} representing an Operator coordinator.
 *
 * <p>All metrics can only be accessed in the main coordinator thread.
 */
@NotThreadSafe
public interface OperatorCoordinatorMetricGroup extends MetricGroup {}

/**
 * Pre-defined metrics for sources.
 *
 * <p>All metrics can only be accessed in the main operator thread.
 */
@NotThreadSafe
public interface SourceReaderMetricGroup extends OperatorMetricGroup {
    /** The total number of record that failed to consume, process, or emit. */
    Counter getNumRecordsInErrorsCounter();

    /**
     * Sets an optional gauge for the number of bytes that have not been fetched by the source. e.g.
     * the remaining bytes in a file after the file descriptor reading position.
     *
     * <p>Note that not every source can report this metric in an plausible and efficient way.
     *
     * @see SettableGauge SettableGauge to continuously update the value.
     */
    <G extends Gauge<Long>> G setPendingBytesGauge(G pendingBytesGauge);

    /**
     * Sets an optional gauge for the number of records that have not been fetched by the source.
     * e.g. the available records after the consumer offset in a Kafka partition.
     *
     * <p>Note that not every source can report this metric in an plausible and efficient way.
     *
     * @see SettableGauge SettableGauge to continuously update the value.
     */
    <G extends Gauge<Long>> G setPendingRecordsGauge(G pendingRecordsGauge);
}

@PublicEvolving
public interface SourceReaderContext {

    /** @return The metric group of this source reader. */
    SourceReaderMetricGroup metricGroup(); // more specific return type
    ...
}

/**
 * Pre-defined metrics for {@code SplitEnumerator}.
 *
 * <p>All metrics can only be accessed in the main operator thread.
 */
@NotThreadSafe
public interface SplitEnumeratorMetricGroup extends OperatorCoordinatorMetricGroup {
    /**
     * Sets an optional gauge for the number of splits that have been enumerated but not yet
     * assigned. For example, this would be the number of files that are in the backlog.
     *
     * <p>Note that not every source can report this metric in an plausible and efficient way.
     *
     * @return the supplied gauge
     * @see SettableGauge SettableGauge to continuously update the value.
     */
    <G extends Gauge<Long>> G setUnassignedSplitsGauge(G unassignedSplitsGauge);
}

/**
 * A context class for the {@link SplitEnumerator}. This class serves the following purposes: 1.
 * Host information necessary for the SplitEnumerator to make split assignment decisions. 2. Accept
 * and track the split assignment from the enumerator. 3. Provide a managed threading model so the
 * split enumerators do not need to create their own internal threads.
 *
 * @param <SplitT> the type of the splits.
 */
@PublicEvolving
public interface SplitEnumeratorContext<SplitT extends SourceSplit> {

    /** @return The metric group of the split enumerator. */
    SplitEnumeratorMetricGroup metricGroup(); // more specific return type
    ...}

/**
 * Pre-defined metrics for sinks.
 *
 * <p>All metrics can only be accessed in the main operator thread.
 */
@NotThreadSafe
public interface SinkMetricGroup extends OperatorMetricGroup {
    /** The total number of records failed to send. */
    Counter getNumRecordsOutErrorsCounter();

    /**
     * Sets an optional gauge for the time it takes to send the last record.
     *
     * <p>This metric is an instantaneous value recorded for the last processed record.
     *
     * <p>If this metric is eagerly calculated, this metric should NOT be updated for each record.
     * Instead, update this metric for each batch of record or sample every X records.
     *
     * <p>Note for asynchronous sinks, the time must be accessible from the main operator thread.
     * For example, a `volatile` field could be set in the async thread and lazily read through the
     * gauge.
     *
     * @see SettableGauge SettableGauge to continuously update the value.
     */
    <G extends Gauge<Long>> G setCurrentSendTimeGauge(G currentSendTime);
}

@Experimental
public interface Sink<InputT, CommT, WriterStateT, GlobalCommT> extends Serializable {
    ...
    interface InitContext {
        ...

        /** @return The metric group this writer belongs to. */
        SinkMetricGroup metricGroup(); // more specific return type
    }
}