docs: def fetch_messages(self):
          """Sends FetchRequests for all topic/partitions set for consumption
  
          Returns:
              Generator that yields KafkaMessage structs
              after deserializing with the configured `deserializer_class`
  
          Note:
              Refreshes metadata on errors, and resets fetch offset on
              OffsetOutOfRange, per the configured `auto_offset_reset` policy
  
          See Also:
              Key KafkaConsumer configuration parameters:
              * `fetch_message_max_bytes`
              * `fetch_max_wait_ms`
              * `fetch_min_bytes`
              * `deserializer_class`
              * `auto_offset_reset`
  
          """
  
          max_bytes = self._config['fetch_message_max_bytes']
          max_wait_time = self._config['fetch_wait_max_ms']
          min_bytes = self._config['fetch_min_bytes']
  
          if not self._topics:
              raise KafkaConfigurationError('No topics or partitions configured')
  
          if not self._offsets.fetch:
              raise KafkaConfigurationError(
                  'No fetch offsets found when calling fetch_messages'
              )
  
          fetches = [FetchRequest(topic, partition,
                                  self._offsets.fetch[(topic, partition)],
                                  max_bytes)
                     for (topic, partition) in self._topics]
  
          # send_fetch_request will batch topic/partition requests by leader
          responses = self._client.send_fetch_request(
              fetches,
              max_wait_time=max_wait_time,
              min_bytes=min_bytes,
              fail_on_error=False
          )
  
          for resp in responses:
  
              if isinstance(resp, FailedPayloadsError):
                  logger.warning('FailedPayloadsError attempting to fetch data')
                  self._refresh_metadata_on_error()
                  continue
  
              topic = kafka_bytestring(resp.topic)
              partition = resp.partition
              try:
                  check_error(resp)
              except OffsetOutOfRangeError:
                  logger.warning('OffsetOutOfRange: topic %s, partition %d, '
                                 'offset %d (Highwatermark: %d)',
                                 topic, partition,
                                 self._offsets.fetch[(topic, partition)],
                                 resp.highwaterMark)
                  # Reset offset
                  self._offsets.fetch[(topic, partition)] = (
                      self._reset_partition_offset((topic, partition))
                  )
                  continue
  
              except NotLeaderForPartitionError:
                  logger.warning("NotLeaderForPartitionError for %s - %d. "
                                 "Metadata may be out of date",
                                 topic, partition)
                  self._refresh_metadata_on_error()
                  continue
  
              except RequestTimedOutError:
                  logger.warning("RequestTimedOutError for %s - %d",
                                 topic, partition)
                  continue
  
              # Track server highwater mark
              self._offsets.highwater[(topic, partition)] = resp.highwaterMark
  
              # Yield each message
              # Kafka-python could raise an exception during iteration
              # we are not catching -- user will need to address
              for (offset, message) in resp.messages:
                  # deserializer_class could raise an exception here
                  val = self._config['deserializer_class'](message.value)
                  msg = KafkaMessage(topic, partition, offset, message.key, val)
  
                  # in some cases the server will return earlier messages
                  # than we requested. skip them per kafka spec
                  if offset < self._offsets.fetch[(topic, partition)]:
                      logger.debug('message offset less than fetched offset '
                                   'skipping: %s', msg)
                      continue
                  # Only increment fetch offset
                  # if we safely got the message and deserialized
                  self._offsets.fetch[(topic, partition)] = offset + 1
  
                  # Then yield to user
                  yield msg