org.apache.flink.streaming.api.datastream.GroupedDataStream to org.apache.flink.streaming.api.datastream.KeyedStream