kafka_log_handler
