Github user tzulitai commented on a diff in the pull request:
https://github.com/apache/flink/pull/5336#discussion_r165022738
--- Diff:
flink-connectors/flink-connector-kafka-0.9/src/main/java/org/apache/flink/streaming/connectors/kafka/internal/Kafka09Fetcher.java
---
@@ -92,21 +93,19 @@ public Kafka09Fetcher(
watermarksPunctuated,
processingTimeProvider,
autoWatermarkInterval,
- userCodeClassLoader,
+ userCodeClassLoader.getParent(),
+ consumerMetricGroup,
useMetrics);
this.deserializer = deserializer;
this.handover = new Handover();
- final MetricGroup kafkaMetricGroup =
metricGroup.addGroup("KafkaConsumer");
- addOffsetStateGauge(kafkaMetricGroup);
-
this.consumerThread = new KafkaConsumerThread(
LOG,
handover,
kafkaProperties,
unassignedPartitionsQueue,
- kafkaMetricGroup,
+ subtaskMetricGroup, // TODO: the thread should
expose Kafka-shipped metrics through the consumer metric group, not subtask
metric group
--- End diff --
Because that would break compatibility of previous metrics.
On the other hand, it might make sense that we also additionally register
the Kafka-shipped metrics under `consumerMetricGroup` now, if we want to be
able to eventually resolve this TODO in the future. What do you think?
---
