[FLINK-35404] Report all metrics of KafkaConsumer in new Kafka source

[hmit]

Flink docs state that all KafkaConsumer related metrics are reported by Flink under MetricGroup KafkaSourceReader.KafkaConsumer per flink docs.
However, only metrics registered with KafkaConsumer at creation are registered under aforementioned Flink MetricGroup.
This causes important metrics like consumer-lag to be missing. Flink does publish consumer-lag metric but loses fidelity like topic and partition info available as tags.

The PR fixes this issue by providing a Flink MetricGroup implementation for Kafka MetricsReporter interface. This ensures that metrics registered at any point in kafka consumer lifecycle will be correctly added to Flink MetricRegistry.

[boring-cyborg]

Thanks for opening this pull request! Please check out our contributing guidelines. (https://flink.apache.org/contributing/how-to-contribute.html)

[tomncooper]

I would be helpful if you included some kind of description with this PR, even if it just a copy of the JIRA issue description. It helps reviewers.

With regard to the metrics reporter, your issue mentions missing consumer lag. However, AFAIK the Kafka connector does not use a Kafka side consumer group so there would be no lag to report?

[hmit]

@tomncooper thanks for taking the time to look at it and providing feedback! I've updated the PR description with some details, is it possible to give this another look?

The kafka connector very much still uses a kafka side consumer group to consume. In fact, the metrics from the said consumer are cherry-picked to be reported using Flink MetricGroup - KafkaSourceReader.KafkaConsumer. This is the relevant section from flink docs on the matter