def fetch_messages(self):
max_bytes = self._config['<STR_LIT>']<EOL>max_wait_time = self._config['<STR_LIT>']<EOL>min_bytes = self._config['<STR_LIT>']<EOL>if not self._topics:<EOL><INDENT>raise KafkaConfigurationError('<STR_LIT>')<EOL><DEDENT>if not self._offsets.fetch:<EOL><INDENT>raise KafkaConfigurationError(<EOL>'<STR_LIT>'<EOL>)<EOL><DEDENT>fetches = [FetchRequest(topic, partition,<EOL>self._offsets.fetch[(topic, partition)],<EOL>max_bytes)<EOL>for (topic, partition) in self._topics]<EOL>responses = self._client.send_fetch_request(<EOL>fetches,<EOL>max_wait_time=max_wait_time,<EOL>min_bytes=min_bytes,<EOL>fail_on_error=False<EOL>)<EOL>for resp in responses:<EOL><INDENT>if isinstance(resp, FailedPayloadsError):<EOL><INDENT>logger.warning('<STR_LIT>')<EOL>self._refresh_metadata_on_error()<EOL>continue<EOL><DEDENT>topic = kafka_bytestring(resp.topic)<EOL>partition = resp.partition<EOL>try:<EOL><INDENT>check_error(resp)<EOL><DEDENT>except OffsetOutOfRangeError:<EOL><INDENT>logger.warning('<STR_LIT>'<EOL>'<STR_LIT>',<EOL>topic, partition,<EOL>self._offsets.fetch[(topic, partition)],<EOL>resp.highwaterMark)<EOL>self._offsets.fetch[(topic, partition)] = (<EOL>self._reset_partition_offset((topic, partition))<EOL>)<EOL>continue<EOL><DEDENT>except NotLeaderForPartitionError:<EOL><INDENT>logger.warning("<STR_LIT>"<EOL>"<STR_LIT>",<EOL>topic, partition)<EOL>self._refresh_metadata_on_error()<EOL>continue<EOL><DEDENT>except RequestTimedOutError:<EOL><INDENT>logger.warning("<STR_LIT>",<EOL>topic, partition)<EOL>continue<EOL><DEDENT>self._offsets.highwater[(topic, partition)] = resp.highwaterMark<EOL>for (offset, message) in resp.messages:<EOL><INDENT>val = self._config['<STR_LIT>'](message.value)<EOL>msg = KafkaMessage(topic, partition, offset, message.key, val)<EOL>if offset < self._offsets.fetch[(topic, partition)]:<EOL><INDENT>logger.debug('<STR_LIT>'<EOL>'<STR_LIT>', msg)<EOL>continue<EOL><DEDENT>self._offsets.fetch[(topic, partition)] = offset + <NUM_LIT:1><EOL>yield msg<EOL><DEDENT><DEDENT>
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`