Github user zentol commented on the issue:
https://github.com/apache/flink/pull/5092
If we separate the watermark metric into in- and output metrics we can
simplify the the whole thing for sources. We can just intercept emitted
watermarks in the `Output` like we do for `numRecordsOut`, and re-use the same
gauge as the input for sources/assigners.---
