[streaming] DataStream fix to return DataStreamSink on closing operators