def get_next_batch(self):
        """
        This method is called from the manager. It must return a list or a generator
        of BaseRecord objects.
        When it has nothing else to read, it must set class variable "finished" to True.
        """
        try:
            batch = self.get_from_kafka()
            for message in batch:
                item = BaseRecord(message)
                self.increase_read()
                yield item
        except:
            self.finished = True
        self.logger.debug('Done reading batch')
This method is called from the manager. It must return a list or a generator
        of BaseRecord objects.
        When it has nothing else to read, it must set class variable "finished" to True.