code: def consume_messages(self, max_next_messages):
          
          # get messages list from kafka
          if self.__next_messages == 0:
              self.set_next_messages(min(1000, max_next_messages))
          self.set_next_messages(min(self.__next_messages, max_next_messages))
          mark = time.time()
          for record in self._get_messages_from_consumer():
              yield record.partition, record.offset, record.key, record.value
          newmark = time.time()
          if newmark - mark > 30:
              self.set_next_messages(self.__next_messages / 2 or 1)
          elif newmark - mark < 5:
              self.set_next_messages(min(self.__next_messages + 100, max_next_messages))
docs: Get messages batch from Kafka (list at output)