code: def _consumer_loop(self):
          
          self.logger.debug("running main consumer thread")
          while not self.closed:
              if self.kafka_connected:
                  self._process_messages()
              time.sleep(self.settings['KAFKA_CONSUMER_SLEEP_TIME'])
docs: The main consumer loop