FLINK-1272: Add a "reduceWithKey" function