class_docstr: Since we parse the message data for each partition from each fetch
  response lazily, fetch-level metrics need to be aggregated as the messages
  from each partition are parsed. This class is used to facilitate this
  incremental aggregation.
class_docstr_tok: ["Since","we","parse","the","message","data","for","each","partition","from","each","fetch","response","lazily",",","fetch-level","metrics","need","to","be","aggregated","as","the","messages","from","each","partition","are","parsed",".","This","class","is","used","to","facilitate","this","incremental","aggregation","."]
docstr: 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.
code: 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.total_bytes += num_bytes
      self.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.total_bytes)
          self.sensors.records_fetched.record(self.total_records)
  
imports: from __future__ import absolute_import
  import collections
  import copy
  import logging
  import random
  import sys
  import time
  from kafka.vendor import six
  import kafka.errors
  from kafka.future import Future
  from kafka.metrics.stats import Avg, Count, Max, Rate
  from kafka.protocol.fetch import FetchRequest
  from kafka.protocol.offset import OffsetRequest, OffsetResetStrategy, UNKNOWN_OFFSET
  from kafka.record import MemoryRecords
  from kafka.serializer import Deserializer
  from kafka.structs import TopicPartition, OffsetAndTimestamp
label_desc: Use image node_id 2 for calling the FetchResponseMetricAggregator obj's underlying member method code with example usage: obj.record(partition, num_bytes, num_records) without return types