You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "ShivsundarR (via GitHub)" <gi...@apache.org> on 2025/07/31 19:38:44 UTC
[PR] KAFKA-18220: Refactor AsyncConsumerMetrics to not extend KafkaConsumerMetrics [kafka]
ShivsundarR opened a new pull request, #20283:
URL: https://github.com/apache/kafka/pull/20283
*What*
https://issues.apache.org/jira/browse/KAFKA-18220
- Currently, `AsyncConsumerMetrics` extends `KafkaConsumerMetrics`, but is being used both by `AsyncKafkaConsumer` and `ShareConsumerImpl`.
- `ShareConsumerImpl` only needs the async consumer metrics(the metrics associated with the new consumer threading model).
- This needs to be fixed, we are unnecessarily having `KafkaConsumerMetrics` as a parent class for `ShareConsumer` metrics.
Fix :
- In this PR, we have removed the dependancy of `AsyncConsumerMetrics` on `KafkaConsumerMetrics` and made it an independent class which both `AsyncKafkaConsumer` and `ShareConsumerImpl` will use.
- The "`asyncConsumerMetrics`" field represents the metrics associated with the new consumer threading model (like application event queue size, background queue size, etc).
- The "`kafkaConsumerMetrics`" and "`kafkaShareConsumerMetrics`" fields denote the actual consumer metrics for `KafkaConsumer` and `KafkaShareConsumer` respectively.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18220: Refactor AsyncConsumerMetrics to not extend KafkaConsumerMetrics [kafka]
Posted by "ShivsundarR (via GitHub)" <gi...@apache.org>.
ShivsundarR commented on PR #20283:
URL: https://github.com/apache/kafka/pull/20283#issuecomment-3209213770
Hi @AndrewJSchofield , Thanks for the review. I have addressed the comments and updated the PR.
> Could you please check that the share consumer metrics are properly tidied up when the share consumer close?
- Yes I checked this, we are closing all the associated metrics when we close the share consumer.
> I'm not sure why ShareConsumerMetrics and KafkaShareConsumerMetrics both has CONSUMER_SHARE_METRIC_GROUP_PREFIX as constructor arguments
- Yes, ideally we need not have the prefix passed as an argument, but I think to maintain consistency with `KafkaConsumerMetrics` and `ConsumerMetrics`, we might have done this.
- I have removed the parameter from `KafkaShareConsumerMetrics`, `ShareConsumerMetrics` and `KafkaConsumerMetrics`, `ConsumerMetrics`.
Additionally, we were having the same group name (consumer-metrics) for the `AsyncConsumerMetrics` initialised by both `ShareConsumerImpl` and `AsyncKafkaConsumer`.
I have now added prefix as an argument for `AsyncConsumerMetrics` class.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18220: Refactor AsyncConsumerMetrics to not extend KafkaConsumerMetrics [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield merged PR #20283:
URL: https://github.com/apache/kafka/pull/20283
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18220: Refactor AsyncConsumerMetrics to not extend KafkaConsumerMetrics [kafka]
Posted by "ShivsundarR (via GitHub)" <gi...@apache.org>.
ShivsundarR commented on code in PR #20283:
URL: https://github.com/apache/kafka/pull/20283#discussion_r2321249906
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java:
##########
@@ -1116,19 +1116,23 @@ private boolean processBackgroundEvents() {
LinkedList<BackgroundEvent> events = new LinkedList<>();
backgroundEventQueue.drainTo(events);
+ if (!events.isEmpty()) {
Review Comment:
Thanks @chia7712 , that makes sense. I have created another PR - https://github.com/apache/kafka/pull/20474 to fix the same, I have used `BackgroundEventHandler::drainEvents` to keep it consistent with `AsyncKafkaConsumer`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18220: Refactor AsyncConsumerMetrics to not extend KafkaConsumerMetrics [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20283:
URL: https://github.com/apache/kafka/pull/20283#discussion_r2319988221
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java:
##########
@@ -1116,19 +1116,23 @@ private boolean processBackgroundEvents() {
LinkedList<BackgroundEvent> events = new LinkedList<>();
backgroundEventQueue.drainTo(events);
+ if (!events.isEmpty()) {
Review Comment:
`BackgroundEventHandler#drainEvents` is not used, so the metrics `background-event-queue-size` is never reset to zero. One option could be to call `asyncConsumerMetrics.recordBackgroundEventQueueSize(0)` manually. Alternatively, `ShareConsumerImpl` could use `BackgroundEventHandler` instead of `backgroundEventQueue`, similar to `AsyncKafkaConsumer`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18220: Refactor AsyncConsumerMetrics to not extend KafkaConsumerMetrics [kafka]
Posted by "ShivsundarR (via GitHub)" <gi...@apache.org>.
ShivsundarR commented on PR #20283:
URL: https://github.com/apache/kafka/pull/20283#issuecomment-3248851776
Thanks @AndrewJSchofield, I used the `consumer-perf-test.sh` and `share-consumer-perf-test.sh` tools to print the metrics before and after the changes.
- For `ClassicKafkaConsumer` - There are no changes in the names/number of metrics.
- For `AsyncKafkaConsumer`(group.protocol=consumer) - There are no changes in the names/number of metrics.
- For `ShareConsumerImpl`, there is a difference of 6 metrics.
```
These 4 metrics are already recorded as part of KafkaShareConsumerMetrics, previously as AsyncConsumerMetrics inherited KafkaConsumerMetrics, we were getting duplicate metrics.
consumer-metrics:last-poll-seconds-ago
consumer-metrics:poll-idle-ratio-avg
consumer-metrics:time-between-poll-avg
consumer-metrics:time-between-poll-max
These 2 metrics were also part of KafkaConsumerMetrics, but we do not need them for ShareConsumerImpl. These were present before as AsyncConsumerMetrics inherited KafkaConsumerMetrics.
consumer-metrics:commit-sync-time-ns-total
consumer-metrics:committed-time-ns-total
```
Hence the difference is expected.
We have also changes prefix name to include "share" for the `AsyncConsumerMetrics`. This changes prefix for a few metrics but the count remains the same.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-18220: Refactor AsyncConsumerMetrics to not extend KafkaConsumerMetrics [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #20283:
URL: https://github.com/apache/kafka/pull/20283#discussion_r2288227393
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -2270,10 +2276,14 @@ public Metrics metricsRegistry() {
}
@Override
- public AsyncConsumerMetrics kafkaConsumerMetrics() {
+ public KafkaConsumerMetrics kafkaConsumerMetrics() {
return kafkaConsumerMetrics;
}
+ public AsyncConsumerMetrics asyncConsumerMetrics() {
Review Comment:
I don't think this needs to be public. The test is in the same package. For `kafkaConsumerMetrics()`, there is a test which accesses the metrics object from the external object via the delegate.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/metrics/AsyncConsumerMetricsTest.java:
##########
@@ -47,21 +47,6 @@ public void tearDown() {
public void shouldMetricNames() {
// create
consumerMetrics = new AsyncConsumerMetrics(metrics);
- Set<MetricName> expectedMetrics = Set.of(
- metrics.metricName("last-poll-seconds-ago", CONSUMER_METRIC_GROUP),
- metrics.metricName("time-between-poll-avg", CONSUMER_METRIC_GROUP),
- metrics.metricName("time-between-poll-max", CONSUMER_METRIC_GROUP),
- metrics.metricName("poll-idle-ratio-avg", CONSUMER_METRIC_GROUP),
- metrics.metricName("commit-sync-time-ns-total", CONSUMER_METRIC_GROUP),
- metrics.metricName("committed-time-ns-total", CONSUMER_METRIC_GROUP)
- );
- expectedMetrics.forEach(
- metricName -> assertTrue(
- metrics.metrics().containsKey(metricName),
- "Missing metric: " + metricName
- )
- );
-
Set<MetricName> expectedConsumerMetrics = Set.of(
Review Comment:
I suggest renaming this to `expectedMetrics`. It is specifically NOT the expected consumer metrics :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org