void record(TopicPartition partition, int bytes, int records) {
        // Aggregate the metrics at the fetch level
        fetchFetchMetrics.increment(bytes, records);

        // Also aggregate the metrics on a per-topic basis.
        perTopicFetchMetrics.computeIfAbsent(partition.topic(), t -> new FetchMetrics())
                        .increment(bytes, records);

        maybeRecordMetrics(partition);
    }
After each partition is parsed, we update the current metric totals with the total bytes
and number of records parsed. After all partitions have reported, we write the metric.
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchMetricsAggregator.java