Change to collect/flatten instead of flatMap