code: def record(self, partition, num_bytes, num_records):
          
          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)
docs: 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.