batching_kafka_consumer
