Github user aljoscha commented on the issue:
https://github.com/apache/flink/pull/5100
Could we set the watermark in the stream input processors (for the input
watermark). Operators might override `processWatermark`, and in fact
`AsyncWaitOperator` does that and doesn't call the `super`.---
