Added kafka consumer