## What is the purpose of the change This PR fixes a severe issue in the metric system where chained batch operators would always operate on the same `OperatorMetricGroup`. As a result most Flink-provided metrics were not exposed for chained operators at all, while other metrics, like task-level IO metrics, were render incorrect.
The problem is that we used the tasks `VertexID` to identify operators; which is obviously identical for all operators in a chain. We now use the vertexID and operator name to identify them. ## Brief change log * fix identification in `TaskMetricGroup` by using both the ID and operator name * extend `MockEnvironment[Builder]` to allow the `TaskMetricGroup` to be set ## Verifying this change This change added tests: * ChainedOperatorsMetricTest * run a basic wordcount as described in the JIRA and verify the results via the UI/reporter of your choice [ Full content available at: https://github.com/apache/flink/pull/6599 ] This message was relayed via gitbox.apache.org for [email protected]
