docstr: Iterate over fetched_records
code: def _message_generator(self):
      """Iterate over fetched_records"""
      while self._next_partition_records or self._completed_fetches:
          if not self._next_partition_records:
              completion = self._completed_fetches.popleft()
              self._next_partition_records = self._parse_fetched_data(
                  completion
              )
              continue
  
          # Send additional FetchRequests when the internal queue is low
          # this should enable moderate pipelining
          if (
              len(self._completed_fetches)
              <= self.config["iterator_refetch_records"]
          ):
              self.send_fetches()
  
          tp = self._next_partition_records.topic_partition
  
          # We can ignore any prior signal to drop pending message sets
          # because we are starting from a fresh one where fetch_offset == position
          # i.e., the user seek()'d to this position
          self._subscriptions.assignment[
              tp
          ].drop_pending_message_set = False
  
          for msg in self._next_partition_records.take():
              # Because we are in a generator, it is possible for
              # subscription state to change between yield calls
              # so we need to re-check on each loop
              # this should catch assignment changes, pauses
              # and resets via seek_to_beginning / seek_to_end
              if not self._subscriptions.is_fetchable(tp):
                  log.debug(
                      "Not returning fetched records for partition %s"
                      " since it is no longer fetchable",
                      tp,
                  )
                  self._next_partition_records = None
                  break
  
              # If there is a seek during message iteration,
              # we should stop unpacking this message set and
              # wait for a new fetch response that aligns with the
              # new seek position
              elif self._subscriptions.assignment[
                  tp
              ].drop_pending_message_set:
                  log.debug(
                      "Skipping remainder of message set for partition %s",
                      tp,
                  )
                  self._subscriptions.assignment[
                      tp
                  ].drop_pending_message_set = False
                  self._next_partition_records = None
                  break
  
              # Compressed messagesets may include earlier messages
              elif (
                  msg.offset
                  < self._subscriptions.assignment[tp].position
              ):
                  log.debug(
                      "Skipping message offset: %s (expecting %s)",
                      msg.offset,
                      self._subscriptions.assignment[tp].position,
                  )
                  continue
  
              self._subscriptions.assignment[tp].position = (
                  msg.offset + 1
              )
              yield msg
  
          self._next_partition_records = None
  
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 15 for calling the Fetcher obj's underlying member method code with example usage: obj._message_generator() without return types