kafka_helper
