void recordBytesFetched(String topic, int bytes) {
        String name = topicBytesFetchedMetricName(topic);
        Sensor bytesFetched = new SensorBuilder(metrics, name, () -> topicTags(topic))
                .withAvg(metricsRegistry.topicFetchSizeAvg)
                .withMax(metricsRegistry.topicFetchSizeMax)
                .withMeter(metricsRegistry.topicBytesConsumedRate, metricsRegistry.topicBytesConsumedTotal)
                .build();
        bytesFetched.record(bytes);
    }
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