void recordPartitionLag(TopicPartition tp, long lag) {
        this.recordsLag.record(lag);

        String name = partitionRecordsLagMetricName(tp);
        Sensor recordsLag = new SensorBuilder(metrics, name, () -> topicPartitionTags(tp))
                .withValue(metricsRegistry.partitionRecordsLag)
                .withMax(metricsRegistry.partitionRecordsLagMax)
                .withAvg(metricsRegistry.partitionRecordsLagAvg)
                .build();

        recordsLag.record(lag);
    }
The {@link FetchMetricsManager} class provides wrapper methods to record lag, lead, latency, and fetch metrics.
It keeps an internal ID of the assigned set of partitions which is updated to ensure the set of metrics it
records matches up with the topic-partitions in use.
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchMetricsManager.java