def record(self, partition, num_bytes, num_records):
        """
        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.
        """
        self.unrecorded_partitions.remove(partition)
        self.fetch_metrics.total_bytes += num_bytes
        self.fetch_metrics.total_records += num_records
        self.topic_fetch_metrics[partition.topic].total_bytes += num_bytes
        self.topic_fetch_metrics[partition.topic].total_records += num_records

        # once all expected partitions from the fetch have reported in, record the metrics
        if not self.unrecorded_partitions:
            self.sensors.bytes_fetched.record(self.fetch_metrics.total_bytes)
            self.sensors.records_fetched.record(self.fetch_metrics.total_records)
            for topic, metrics in six.iteritems(self.topic_fetch_metrics):
                self.sensors.record_topic_fetch_metrics(topic, metrics.total_bytes, metrics.total_records)

        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.
        
kafka/consumer/fetcher.py