code: def run(self, loopinfo=None, batch_size=1):
          
          logger.info("{}.Starting...".format(self.__class__.__name__))
          if loopinfo:
              while True:
                  for topic in self.topics:
                      self.call_kafka(topic, batch_size)
                  time.sleep(loopinfo.sleep)
          else:
              for topic in self.topics:
                  self.call_kafka(topic, batch_size)
docs: Run consumer