[streaming] Reduce operator added to ConnectedDataStream + test