flink-connector-aws/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/source/metrics/KinesisShardMetrics.java [58:79]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
                metricGroup.getMetricIdentifier(MetricConstants.MILLIS_BEHIND_LATEST));
    }

    public MetricGroup getMetricGroup() {
        return metricGroup;
    }

    public long getMillisBehindLatest() {
        return millisBehindLatest;
    }

    public void setMillisBehindLatest(long millisBehindLatest) {
        log.debug(
                "Updating millisBehindLatest metric for shard {} to {}",
                shardInfo.getShardId(),
                millisBehindLatest);

        this.millisBehindLatest = millisBehindLatest;
    }

    public void unregister() {
        ((AbstractMetricGroup) metricGroup).close();
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -



flink-connector-aws/flink-connector-dynamodb/src/main/java/org/apache/flink/connector/dynamodb/source/metrics/DynamoDbStreamsShardMetrics.java [56:77]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
                metricGroup.getMetricIdentifier(MetricConstants.MILLIS_BEHIND_LATEST));
    }

    public MetricGroup getMetricGroup() {
        return metricGroup;
    }

    public long getMillisBehindLatest() {
        return millisBehindLatest;
    }

    public void setMillisBehindLatest(long millisBehindLatest) {
        log.debug(
                "Updating millisBehindLatest metric for shard {} to {}",
                shardInfo.getShardId(),
                millisBehindLatest);

        this.millisBehindLatest = millisBehindLatest;
    }

    public void unregister() {
        ((AbstractMetricGroup) metricGroup).close();
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -