Github user zentol commented on the issue:
https://github.com/apache/flink/pull/5100
You can't measure input metrics in the ChainingOutput (nicely) since for a
chain A->B does B at no point have access to the output of A.
We would need to have the task setup all watermark metrics and either pass
them on to operators for registration or have the task register them for
operators, which is just backwards. This also clashes heavily with the idea of
operators deciding whether the same metric is used for input&output.
---
