def kafka_consumer_refresh_dashboard_handler(topic, value):
    logger.info('Creating Kafka Consumer for %s topic', topic)
    consumer = KafkaConsumer(topic, bootstrap_servers=[(dcm_ip_address + ':9092')], auto_offset_reset='earliest', enable_auto_commit=True, group_id=None, value_deserializer=(lambda x: json.loads(x.decode('utf-8'))))
    message_received = False
    while (not message_received):
        message = consumer.poll(timeout_ms=1000)
        if (message != {}):
            logger.info('Message received in %s topic: %s', topic, message)
            message_received = True
    time.sleep(5)
    logger.info('Creating dashboard for topic: %s', topic)
    r = requests.post(dcs_dashboard_url, json={'records': [{'value': json.loads(value)}]})
    logger.info('Response: Code %s', r)
    logger.info('Closing Kafka Consumer for %s topic', topic)
    consumer.close()
def kafka_consumer_refresh_dashboard_handler(topic, value):
    logger.info('Creating Kafka Consumer for %s topic', topic)
    consumer = KafkaConsumer(topic, bootstrap_servers=[(dcm_ip_address + ':9092')], auto_offset_reset='earliest', enable_auto_commit=True, group_id=None, value_deserializer=(lambda x: json.loads(x.decode('utf-8'))))
    message_received = False
    while (not message_received):
        message = consumer.poll(timeout_ms=1000)
        if (message != {}):
            logger.info('Message received in %s topic: %s', topic, message)
            message_received = True
    time.sleep(5)
    logger.info('Creating dashboard for topic: %s', topic)
    r = requests.post(dcs_dashboard_url, json={'records': [{'value': json.loads(value)}]})
    logger.info('Response: Code %s', r)
    logger.info('Closing Kafka Consumer for %s topic', topic)
    consumer.close()