[streaming] ConnectedDataStream API  and operator cleanup + modified windowReduceGroup functionality