Flush operator state on end of stream

Stateful operators should flush all buffered state when end of stream is reached. Otherwise the container may shut down with open windows and lead to message loss.

As of 0.14, only window operator needs to do this. When we add outer/left/right joins they'll need to do it as well.