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()
def run ( self ) : 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 : 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