log.info("Sending create_alarm_request %s", msg)
future = self.producer.send(topic='alarm_request', key='create_alarm_request', value=json.dumps(msg))
future.get(timeout=60)
consumer = KafkaConsumer(bootstrap_servers=self.kafka_server,
key_deserializer=bytes.decode,
log.info("Sending create_alarm_request %s", msg)
future = self.producer.send(topic='alarm_request', key='create_alarm_request', value=json.dumps(msg))
future.get(timeout=60)
consumer = KafkaConsumer(bootstrap_servers=self.kafka_server,
key_deserializer=bytes.decode,