def run(self):
        """ Run consumer
        """
        if KSER_METRICS_ENABLED == "yes":
            from prometheus_client import start_http_server
            logger.info("Metric.Starting...")
            start_http_server(
                os.getenv("KSER_METRICS_PORT", 8888),
                os.getenv("KSER_METRICS_ADDRESS", "0.0.0.0")
            )

        logger.info("{}.Starting...".format(self.__class__.__name__))
        running = True
        while running:
            msg = self.client.poll()
            if msg:
                # noinspection PyProtectedMember
                if not msg.error():
                    self.REGISTRY.run(msg.value().decode('utf-8'))
                elif msg.error().code() != KafkaError._PARTITION_EOF:
                    logger.error(msg.error())
                    running = False
        self.client.close()
Run consumer