Inline End-of-stream and Watermark logic inside OperatorImpl

Currently both EOS and Watermark aggregation/population logic is inside TaskInstance. Since watermark handling logic is in OperatorImpl, it makes sense to consolidate them together inside OperatorImpl, and make it feasible to provide user API to emit watermarks.