code: def _unpack_message_set(self, tp, records):
      try:
          batch = records.next_batch()
          while batch is not None:
              # LegacyRecordBatch cannot access either base_offset or last_offset_delta
              try:
                  self._subscriptions.assignment[
                      tp
                  ].last_offset_from_message_batch = (
                      batch.base_offset + batch.last_offset_delta
                  )
              except AttributeError:
                  pass
  
              for record in batch:
                  key_size = (
                      len(record.key) if record.key is not None else -1
                  )
                  value_size = (
                      len(record.value)
                      if record.value is not None
                      else -1
                  )
                  key = self._deserialize(
                      self.config["key_deserializer"],
                      tp.topic,
                      record.key,
                  )
                  value = self._deserialize(
                      self.config["value_deserializer"],
                      tp.topic,
                      record.value,
                  )
                  headers = record.headers
                  header_size = (
                      sum(
                          len(h_key.encode("utf-8"))
                          + (len(h_val) if h_val is not None else 0)
                          for h_key, h_val in headers
                      )
                      if headers
                      else -1
                  )
                  yield ConsumerRecord(
                      tp.topic,
                      tp.partition,
                      record.offset,
                      record.timestamp,
                      record.timestamp_type,
                      key,
                      value,
                      headers,
                      record.checksum,
                      key_size,
                      value_size,
                      header_size,
                  )
  
              batch = records.next_batch()
  
      # If unpacking raises StopIteration, it is erroneously
      # caught by the generator. We want all exceptions to be raised
      # back to the user. See Issue 545
      except StopIteration as e:
          log.exception("StopIteration raised unpacking messageset")
          raise RuntimeError(
              "StopIteration raised unpacking messageset"
          )
  
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 16 for calling the Fetcher obj's underlying member method code with example usage: obj._unpack_message_set(tp, records) without return types