private void maybeRecordMetrics(TopicPartition partition) {
        unrecordedPartitions.remove(partition);

        if (!unrecordedPartitions.isEmpty())
            return;

        // Record the metrics aggregated at the fetch level.
        metricsManager.recordBytesFetched(fetchFetchMetrics.bytes);
        metricsManager.recordRecordsFetched(fetchFetchMetrics.records);

        // Also record the metrics aggregated on a per-topic basis.
        for (Map.Entry<String, FetchMetrics> entry: perTopicFetchMetrics.entrySet()) {
            String topic = entry.getKey();
            FetchMetrics fetchMetrics = entry.getValue();
            metricsManager.recordBytesFetched(topic, fetchMetrics.bytes);
            metricsManager.recordRecordsFetched(topic, fetchMetrics.records);
        }
    }
Once we've detected that all of the {@link TopicPartition partitions} for the fetch have been handled, we
can then record the aggregated metrics values. This is done at the fetch level and on a per-topic basis.

@param partition {@link TopicPartition}
clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchMetricsAggregator.java