docs: 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