docs: 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.
          """
          messages = self.get_from_kafka()
          if messages:
              for message in messages:
                  item = BaseRecord(message)
                  self.increase_read()
                  yield item
  
          self.logger.debug('Done reading batch')
          self.last_position = self.consumer.offsets