def consume_messages(self, max_next_messages):
        """ Get messages batch from Kafka (list at output) """
        # 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))
Get messages batch from Kafka (list at output)