more changes on kafka consumer