Don't run DataSourceTask with WindowRunner unless necessary

Currently, a windowed wordcount pipeline like source.flatMap.window.groupBy.count will actually be executed as source.globalWindow.flatMap.groupBy.window.count and globalWindow.flatMap run by a WindowRunner.

The globalWindow is unnecessary unless source is followed by an aggregation. Further, the globalWindow requires Watermark.MAX to trigger emits which is not obvious for users.