KAFKA-16143: New JMX metrics for AsyncKafkaConsumer by FrankYang0529 · Pull Request #17199 · apache/kafka

FrankYang0529 · 2024-09-15T03:37:55Z

Add following metrics to AsyncKafkaConsumer:

time-between-network-thread-poll-avg
time-between-network-thread-poll-max
application-event-queue-size
application-event-queue-time-avg
application-event-queue-time-max
application-event-queue-processing-time-avg
application-event-queue-processing-time-max
unsent-requests-queue-size
unsent-requests-queue-time-avg
unsent-requests-queue-time-max
background-event-queue-size
background-event-queue-time-avg
background-event-queue-time-max
background-event-queue-processing-time-avg
background-event-queue-processing-time-max

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

kirktrue

Thanks @FrankYang0529 for working on this!

I am intentionally not going to look at this too in-depth because it's a draft at the moment. However, it appears that the KafkaConsumerMetrics object is being used in both the application thread and background thread. From a quick glance, KafkaConsumerMetrics doesn't look thread-safe 🤔

If that's the case, we'll need to figure out how we can avoid potential synchronization issues.

Thanks!

kirktrue · 2024-10-08T22:59:35Z

@FrankYang0529—I wanted to check in and see if you had any questions or needed anything to progress on this. Thanks!

FrankYang0529 · 2024-10-09T00:06:48Z

Hi @kirktrue, sorry for late. I will make this PR ready today.

FrankYang0529 · 2024-10-09T16:11:04Z

If that's the case, we'll need to figure out how we can avoid potential synchronization issues.

Hi @kirktrue, good catch. There is synchronized in Sensor#recordInternal. In this case, do we still need a lock in KafkaConsumerMetrics?

kafka/clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java

Lines 230 to 244 in bb6ebd8

    
           private void recordInternal(double value, long timeMs, boolean checkQuotas) { 
        
               this.lastRecordTime = timeMs; 
        
               synchronized (this) { 
        
                   synchronized (metricLock()) { 
        
                       // increment all the stats 
        
                       for (StatAndConfig statAndConfig : this.stats) { 
        
                           statAndConfig.stat.record(statAndConfig.config(), value, timeMs); 
        
                       } 
        
                   } 
        
                   if (checkQuotas) 
        
                       checkQuotas(timeMs); 
        
               } 
        
               for (Sensor parent : parents) 
        
                   parent.record(value, timeMs, checkQuotas); 
        
           }

I wanted to check in and see if you had any questions or needed anything to progress on this.

The PR is almost ready. I will wait to see CI result. Thanks.

kirktrue

Thanks for the PR @FrankYang0529!

This looks pretty comprehensive. I haven't had time to dive as deep as I would like, but I left a first pass of comments.

Thanks!

kirktrue · 2024-10-15T01:00:44Z

+            kafkaConsumerMetrics.recordBackgroundEventQueueTime(time.milliseconds() - event.addedToQueueMs());
+            long startMs = time.milliseconds();


Does the metric determine how long it takes to process the background event queue, or how long it takes to process background events? If it's the latter, we want to update the metric inside each loop, but if it's the former we should update once outside the loop.

No, this metric determine how long a background event is taking to be dequeued. In BackgroundEventHandler#add, we run BackgroundEvent#setAddedToQueueMs. When we start to process it, we can use current time - event.addedToQueueMs to know how long an event is in the queue.

kirktrue · 2024-10-15T01:02:39Z

        processApplicationEvents();

        final long currentTimeMs = time.milliseconds();
+        final long timeSinceLastPollMs = lastPollTimeMs != 0L ? currentTimeMs - lastPollTimeMs : currentTimeMs;


In the first invocation of runOnce(), timeSinceLastPollMs will be something like 1728954137284. Is that correct?

kirktrue · 2024-10-15T01:05:02Z

+    public ConsumerNetworkThread(LogContext logContext,
+                                 Time time,
+                                 BlockingQueue<ApplicationEvent> applicationEventQueue,
+                                 CompletableEventReaper applicationEventReaper,
+                                 Supplier<ApplicationEventProcessor> applicationEventProcessorSupplier,
+                                 Supplier<NetworkClientDelegate> networkClientDelegateSupplier,
+                                 Supplier<RequestManagers> requestManagersSupplier) {
+        this(logContext, time, applicationEventQueue, applicationEventReaper, applicationEventProcessorSupplier,
+            networkClientDelegateSupplier, requestManagersSupplier, Optional.empty());


Is it possible to remove this constructor? Can't we call the existing constructor with Optional.empty()?

kirktrue · 2024-10-15T01:05:22Z

+    public NetworkClientDelegate(
+            final Time time,
+            final ConsumerConfig config,
+            final LogContext logContext,
+            final KafkaClient client,
+            final Metadata metadata,
+            final BackgroundEventHandler backgroundEventHandler) {
+        this(time, config, logContext, client, metadata, backgroundEventHandler, Optional.empty());


Is it possible to remove this constructor? Can't callers invoke the existing constructor with Optional.empty()?

kirktrue · 2024-10-15T01:06:54Z

+    public void setAddedToQueueMs(long addedToQueueMs) {
+        this.addedToQueueMs = addedToQueueMs;
+    }
+
+    public long addedToQueueMs() {
+        return addedToQueueMs;
+    }
+


Suggested change

public void setAddedToQueueMs(long addedToQueueMs) {

this.addedToQueueMs = addedToQueueMs;

}

public long addedToQueueMs() {

return addedToQueueMs;

}

public void setEnqueuedMs(long enqueuedMs) {

enqueuedMs = enqueuedMs;

}

public long enqueuedMs() {

return enqueuedMs;

}

kirktrue · 2024-10-15T01:08:20Z

     */
    public void add(final ApplicationEvent event) {
        Objects.requireNonNull(event, "ApplicationEvent provided to add must be non-null");
+        event.setAddedToQueueMs(System.currentTimeMillis());


This needs to get the current time in milliseconds from the Time object that was passed in to the constructor.

kirktrue · 2024-10-15T01:08:47Z

+        void setAddedToQueueMs(final long addedToQueueMs) {
+            this.addedToQueueMs = addedToQueueMs;
+        }
+
+        long addedToQueueMs() {
+            return addedToQueueMs;
+        }
+


I know it's a bit nit-picky, but can we change it to:

Suggested change

void setAddedToQueueMs(final long addedToQueueMs) {

this.addedToQueueMs = addedToQueueMs;

}

long addedToQueueMs() {

return addedToQueueMs;

}

void setEnqueuedMs(final long enqueuedMs) {

this.enqueuedMs = enqueuedMs;

}

long enqueuedMs() {

return enqueuedMs;

}

kirktrue · 2024-10-15T01:09:23Z

+    public void setAddedToQueueMs(long addedToQueueMs) {
+        this.addedToQueueMs = addedToQueueMs;
+    }
+
+    public long addedToQueueMs() {
+        return addedToQueueMs;
+    }


Suggested change

public void setAddedToQueueMs(long addedToQueueMs) {

this.addedToQueueMs = addedToQueueMs;

}

public long addedToQueueMs() {

return addedToQueueMs;

}

public void setEnqueuedMs(long enqueuedMs) {

this. enqueuedMs = enqueuedMs;

}

public long enqueuedMs() {

return enqueuedMs;

}

kirktrue · 2024-10-15T01:09:59Z

+    public BackgroundEventHandler(final Queue<BackgroundEvent> backgroundEventQueue) {
+        this(backgroundEventQueue, Optional.empty());


Same with the other changes. I'd rather make the callers pass in Optional.empty().

kirktrue · 2024-10-15T01:10:34Z

     */
    public void add(BackgroundEvent event) {
        Objects.requireNonNull(event, "BackgroundEvent provided to add must be non-null");
+        event.setAddedToQueueMs(System.currentTimeMillis());


Same comment here: we need to update the constructor to provide a Time object and then use that here.

FrankYang0529 · 2024-10-16T15:17:16Z

Hi @kirktrue, I address all comments. Could you take a look when you have time? Thank you.

kirktrue

Thanks @FrankYang0529!

Overall, I think it's headed in the right direction. I have mostly minor requests, but the one that's slightly larger is class hierarchy around the KafkaConsumerMetrics. If you want to punt on that, file a Jira and I'll take care of it later.

I didn't get to the unit tests yet, though 😢

Thanks!

kirktrue · 2024-10-30T22:31:37Z

+    public void setEnqueuedMs(long enqueuedMs) {
+        this.enqueuedMs = enqueuedMs;
+    }


I don't like that we're introducing mutability into all the events. I understand that this changes a lot less code than having the constructor take the creation time or something.

Can you add a comment to the enqueued variable that states that because of its mutability that it should not be used in hashCode() or equals() or toStringBase() (or something along those lines)?

I take the point about hashCode() or equals() but we probably should be able to see the enqueued time in the string representation of the events.

Add enqueued time to toStringBase. Thanks.

we probably should be able to see the enqueued time in the string representation of the events

Agreed. Good point. Thanks @AndrewJSchofield.

kirktrue · 2024-10-30T22:32:01Z

     * {@link #equals(Object)} and can be used in log messages when debugging.
     */
    private final Uuid id;
+    private long enqueuedMs;


Same comment as per above with ApplicationEvent.

kirktrue · 2024-10-30T22:33:09Z

+            kafkaConsumerMetrics.ifPresent(metrics -> metrics.recordApplicationEventQueueTime(time.milliseconds() - event.enqueuedMs()));
+            long startMs = time.milliseconds();


Can we swap the ordering of these two lines to avoid the extra call to time.milliseconds()?

Suggested change

kafkaConsumerMetrics.ifPresent(metrics -> metrics.recordApplicationEventQueueTime(time.milliseconds() - event.enqueuedMs()));

long startMs = time.milliseconds();

long startMs = time.milliseconds();

kafkaConsumerMetrics.ifPresent(metrics -> metrics.recordApplicationEventQueueTime(startMs - event.enqueuedMs()));

kirktrue · 2024-10-30T22:34:23Z

    private void processApplicationEvents() {
        LinkedList<ApplicationEvent> events = new LinkedList<>();
        applicationEventQueue.drainTo(events);
+        kafkaConsumerMetrics.ifPresent(metrics -> metrics.recordApplicationEventQueueSize(applicationEventQueue.size()));


Once we drain applicationEventQueue, it'll be empty, right? If, so why not just do:

Suggested change

kafkaConsumerMetrics.ifPresent(metrics -> metrics.recordApplicationEventQueueSize(applicationEventQueue.size()));

kafkaConsumerMetrics.ifPresent(metrics -> metrics.recordApplicationEventQueueSize(0));

kirktrue · 2024-10-30T22:36:00Z

    private final IdempotentCloser closer = new IdempotentCloser();
    private volatile Duration closeTimeout = Duration.ofMillis(DEFAULT_CLOSE_TIMEOUT_MS);
    private volatile long cachedMaximumTimeToWait = MAX_POLL_TIMEOUT_MS;
+    private long lastPollTimeMs = 0L;


This is only ever written to and read from the same thread, right?

Pinging on this. I believe it's only written on the background thread, but just want to be sure.

Hi @kirktrue, this is only used for metrics time-between-network-thread-poll-max and time-between-network-thread-poll-avg. Thanks.

kirktrue · 2024-10-30T22:38:25Z

     */
    public void add(BackgroundEvent event) {
        Objects.requireNonNull(event, "BackgroundEvent provided to add must be non-null");
+        event.setEnqueuedMs(System.currentTimeMillis());


We need to add a Time variable to this class so that we can do this instead:

Suggested change

event.setEnqueuedMs(System.currentTimeMillis());

event.setEnqueuedMs(time.milliseconds());

kirktrue · 2024-10-30T22:41:37Z

 import static org.apache.kafka.clients.consumer.internals.ConsumerUtils.CONSUMER_METRICS_SUFFIX;

 public class KafkaConsumerMetrics implements AutoCloseable {
+    private final GroupProtocol groupProtocol;


Would it be possible to create a subclass of KafkaConsumerMetrics instead of adding group protocol-specific bits on here? I see that the ShareConsumerImpl uses a custom KafkaShareConsumerMetrics that at first glance is entirely a subset of KafkaConsumerMetrics, so there's some refactoring that could be done here.

kirktrue · 2024-10-30T22:42:37Z

+    private Sensor timeBetweenNetworkThreadPollSensor;
+    private Sensor applicationEventQueueSizeSensor;
+    private Sensor applicationEventQueueTimeSensor;
+    private Sensor applicationEventQueueProcessingTimeSensor;
+    private Sensor backgroundEventQueueSizeSensor;
+    private Sensor backgroundEventQueueTimeSensor;
+    private Sensor backgroundEventQueueProcessingTimeSensor;
+    private Sensor unsentRequestsQueueSizeSensor;
+    private Sensor unsentRequestsQueueTimeSensor;


Can we make these final too? It'll be a bit ugly in the else block of the constructor to mark them all as null. But that goes away if we make this a subclass.

kirktrue · 2024-10-30T22:43:18Z

-            metricGroupName, "The number of seconds since the last poll() invocation.");
+                metricGroupName, "The number of seconds since the last poll() invocation.");


Can we remove these whitespace changes? They seem extraneous.

kirktrue · 2024-10-30T22:44:08Z

+
+        if (groupProtocol == GroupProtocol.CONSUMER) {
+            Arrays.asList(
+                    timeBetweenNetworkThreadPollSensor.name(),
+                    applicationEventQueueSizeSensor.name(),
+                    applicationEventQueueTimeSensor.name(),
+                    applicationEventQueueProcessingTimeSensor.name(),
+                    backgroundEventQueueSizeSensor.name(),
+                    backgroundEventQueueTimeSensor.name(),
+                    backgroundEventQueueProcessingTimeSensor.name(),
+                    unsentRequestsQueueSizeSensor.name(),
+                    unsentRequestsQueueTimeSensor.name()
+            ).forEach(metrics::removeSensor);
+        }


This is another place that having a subclass would help.

lianetm · 2024-11-18T21:14:53Z

Hey @FrankYang0529, sorry I haven't had the bandwidth for this. I will be taking a look this week. Thanks!

lianetm

Hey @FrankYang0529, thanks for taking over this one! Some initial comments.

lianetm · 2024-11-18T21:45:38Z

        Objects.requireNonNull(event, "BackgroundEvent provided to add must be non-null");
+        event.setEnqueuedMs(time.milliseconds());
        backgroundEventQueue.add(event);
+        kafkaConsumerMetrics.ifPresent(metrics -> metrics.recordBackgroundEventQueueSize(backgroundEventQueue.size()));


this change here makes sense to me, but makes me wonder if we should push it further. It would be helpful if we could try to keep all the updates for this queue size metric in this component that holds the queue, so we can easily maintain/track how "add" and "remove/drain" update that metric.

We could then use that drain from the processBackgroundEvents, instead of manually draining the queue and recording the metric there . What do you think?

Yes, I agree.

lianetm · 2024-11-18T21:58:36Z

        LinkedList<BackgroundEvent> events = new LinkedList<>();
        backgroundEventQueue.drainTo(events);
+        kafkaAsyncConsumerMetrics.recordBackgroundEventQueueSize(backgroundEventQueue.size());


This is what I was wondering if we could encapsulate a BackgroundEventHandler.drain or similar, that would take care of draining the queue and recording the metric (all metric updates done there consistently)

Suggested change

LinkedList<BackgroundEvent> events = new LinkedList<>();

backgroundEventQueue.drainTo(events);

kafkaAsyncConsumerMetrics.recordBackgroundEventQueueSize(backgroundEventQueue.size());

LinkedList<BackgroundEvent> events = backgroundEventHandler.drainBackgroundEvents();

lianetm · 2024-11-18T22:12:46Z

        LinkedList<ApplicationEvent> events = new LinkedList<>();
        applicationEventQueue.drainTo(events);
+        kafkaAsyncConsumerMetrics.ifPresent(metrics -> metrics.recordApplicationEventQueueSize(0));


keeping the symmetry with the background event, would it make sense to encapsulate these actions in the ApplicationEventHandler so that we keep that component responsible of add and drain the queue (including the metric actions related to those ops)?

It would mean that this ConsumerNetworkThread would keep the ref to the ApplicationEventHandler that has the queue (instead of directly having the queue like it does now), but that is already available, so I guess we just need to pass it in the constructor instead of the queue. What do you think?

We initialize ConsumerNetworkThread in ApplicationEventHandler. If we want to reference ApplicationEventHandler in ConsumerNetworkThread, we have to give this as parameter. Probably, we can do some refactor in next PR.

Oh good point, I forgot about that dependency! (feels kind of unexpected actually). Given that structure we shouldn't reference AppEventHandler in the ConsumerNetworkThread because we would end up with a circular dependency.

Sorry for the extra work, but I would suggest we revert this back to your initial change, without having a drainEvents, and we rethink this class structure in a separate jira, to see if it would make sense to decouple the AppEventHandler from the network thread (and if so then we could properly add a drainEvents, without circular deps). Makes sense?

Create a followup Jira for it: https://issues.apache.org/jira/browse/KAFKA-18048.

lianetm · 2024-11-19T03:10:23Z

            } catch (Throwable t) {
                log.warn("Error processing event {}", t.getMessage(), t);
+            } finally {
+                kafkaAsyncConsumerMetrics.ifPresent(metrics -> metrics.recordApplicationEventQueueProcessingTime(time.milliseconds() - startMs));


Is this the best place to record this? From the description of the metric I get that we want to measure the time "that the consumer network takes to process all available application events". So wouldn't it be simpler to record the metric once per runOnce instead of recording it N times on each run? (startTime right before the loop over events, and ending/recording right after the loop).

I went to the KIP discussion thread to double check this interpretation, and this was the intention behind what was proposed (by me actually I discovered he he).

LM3. Thinking about the actual usage of "time-between-network-thread-poll-xxx" metric, I imagine it would be helpful to know more about what could be impacting it. As I see it, the network thread cadence could be mainly impacted by: 1- app event processing (generate requests), 2- network client poll (actual send/receive). For 2, the new consumer reuses the same component as the legacy one, but 1 is specific to the new consumer, so what about a metric for application-event-processing-time-ms (we could consider avg I would say). It would be the time that the network thread takes to process all available events on each run.

What do you think?

lianetm · 2024-11-19T03:29:04Z

        Iterator<UnsentRequest> iterator = unsentRequests.iterator();
        while (iterator.hasNext()) {
            UnsentRequest unsent = iterator.next();
+            kafkaConsumerMetrics.ifPresent(metrics -> metrics.recordUnsentRequestsQueueTime(currentTimeMs - unsent.enqueuedMs()));


recording this here means we would consider the request removed from the unsent queue even in the case where it cannot be sent and it actually stays in the unsent queue (!doSend), right? If so, I guess we should probably record this only when we do remove it from the queue with iterator.remove() (either because it's expired, or because we did sent it).

Also, shouldn't we record this same metric on checkDisconnects if the request is removed from the unsent queue because the node is disconnected?

lianetm · 2024-11-19T03:41:47Z

+        this.applicationEventQueueProcessingTimeSensor.record(processingTime);
+    }
+
+    public void recordUnsentRequestsQueueSize(int size) {


This metric is about the size of the queue at a given time, so I expect we should have another param here timeMs, for the time where we read the metric, and we should pass it into the .record, that has an overload for it.

lianetm · 2024-11-19T04:03:55Z

                if (!firstError.compareAndSet(null, e))
                    log.warn("An error occurred when processing the background event: {}", e.getMessage(), e);
+            } finally {
+                kafkaAsyncConsumerMetrics.recordBackgroundEventQueueProcessingTime(time.milliseconds() - startMs);


Similar to recordApplicationEventQueueProcessingTime. The metric description states this is about the time "that the consumer took to process all available background events' . Shouldn't we simply take the time from right before the loop to right after it ends, and record the metric once per run of the processBackgroundEvents?

lianetm · 2024-11-19T04:16:28Z

        r.setTimer(this.time, this.requestTimeoutMs);
+        r.setEnqueuedMs(this.time.milliseconds());
        unsentRequests.add(r);
+        kafkaConsumerMetrics.ifPresent(metrics -> metrics.recordUnsentRequestsQueueSize(unsentRequests.size()));


I'm still debating whether this is the best place to record this. We want snapshots in time of the queue size. Recording here has the limitation that we won't be recording when the size decreases (ie. requests sent, failed due to disconnections). So I wonder if recording this on poll, which is called regularly, would given a better view of the queue size?

The way add/poll are used from the ConsumerNetworkThread.runOnce they end up being called sequentially anyways, but I'm thinking about the case where, let's say managers are not returning any requests (so addAll is called with empty, add never called), but there could be unsent requests in the queue, that could be sent out, cancelled, time out, etc). Thoughts?

How about we still record a metric in NetworkClientDelegate#add for increasing path. For decreasing path, we can record in NetworkClientDelegate#poll, because it covers both trySend and checkDisconnects. WDYT?

Makes sense to record the unsent event queue size at the end of a poll iteration, after events were added and removed/sent, that's truly what could help spot issues (too many requests being left "unsent" on each run). Then it actually makes me wonder about the value of also calling it on add, would it be useful to see the number of events added on each run? or just seeing how many where left unsent is all that matters?

I think we just want to see how many unsent requests left. We can just record at the end of NetworkClientDelegate#poll.

lianetm · 2024-11-19T04:18:10Z

+    public static final String UNSENT_REQUESTS_QUEUE_SIZE_SENSOR_NAME = "unsent-requests-queue-size";
+    public static final String UNSENT_REQUESTS_QUEUE_TIME_SENSOR_NAME = "unsent-requests-queue-time";
+    private final Sensor timeBetweenNetworkThreadPollSensor;
+    private final  Sensor applicationEventQueueSizeSensor;


nit: extra space before Sensor

kirktrue

Thanks for the updates @FrankYang0529. I left a few comments.

Out of necessity, there's a good amount of duplication of metric logic between the application and background thread queues. It would be good to include cleanup of the metrics in KAFKA-18048, if possible.

Thanks!

kirktrue · 2024-11-26T23:01:35Z

+                    backgroundEventHandler,
+                asyncConsumerMetrics);


Nit: minor alignment issue:

Suggested change

backgroundEventHandler,

asyncConsumerMetrics);

backgroundEventHandler,

asyncConsumerMetrics);

kirktrue · 2024-11-26T23:02:33Z

+            this.backgroundEventHandler = new BackgroundEventHandler(
+                backgroundEventQueue, time, asyncConsumerMetrics);


Nitpick: for better or worse, we've adopted this style for multi-line parameter lists:

Suggested change

this.backgroundEventHandler = new BackgroundEventHandler(

backgroundEventQueue, time, asyncConsumerMetrics);

this.backgroundEventHandler = new BackgroundEventHandler(

backgroundEventQueue,

time,

asyncConsumerMetrics

);

kirktrue · 2024-11-26T23:03:06Z

+        this.backgroundEventHandler = new BackgroundEventHandler(
+            backgroundEventQueue, time, asyncConsumerMetrics);


Suggested change

this.backgroundEventHandler = new BackgroundEventHandler(

backgroundEventQueue, time, asyncConsumerMetrics);

this.backgroundEventHandler = new BackgroundEventHandler(

backgroundEventQueue,

time,

asyncConsumerMetrics

);

kirktrue · 2024-11-26T23:04:10Z

-    private final KafkaConsumerMetrics kafkaConsumerMetrics;
+    private final AsyncConsumerMetrics asyncConsumerMetrics;


I'd prefer it be left as kafkaConsumerMetrics. Calling out the distinction in the variable name isn't really adding anything (IMO).

kirktrue · 2024-11-26T23:13:44Z

+    public void setEnqueuedMs(long enqueuedMs) {
+        this.enqueuedMs = enqueuedMs;
+    }


we probably should be able to see the enqueued time in the string representation of the events

Agreed. Good point. Thanks @AndrewJSchofield.

kirktrue · 2024-11-26T23:15:08Z


    private final Logger log;
+    private final Time time;
    private final BlockingQueue<ApplicationEvent> applicationEventQueue;


I agree that this is an area that could use some refactoring. Thanks for filing KAFKA-18048, @FrankYang0529.

kirktrue · 2024-11-26T23:16:20Z

+     * The time in milliseconds when this event was enqueued.
+     * This field can be changed after the event is created, so it should not be used in hashCode, equals, or toStringBase.
+     */
+    private long enqueuedMs;


We also need the enqueuedMs in the toStringBase() method as per the ApplicationEvent’s method of the same name.

FrankYang0529 · 2024-12-03T14:48:06Z

Hi @AndrewJSchofield / @lianetm / @kirktrue, could you please review this PR when you have time? Thank you.

kirktrue

Thanks for the updates. I have just a few nitpicks. I'd also like to see a resolution to @AndrewJSchofield’s outstanding question, if possible.

kirktrue · 2024-12-04T00:40:14Z

+                    requestManagersSupplier,
+                kafkaConsumerMetrics


Nit: alignment.

Suggested change

requestManagersSupplier,

kafkaConsumerMetrics

requestManagersSupplier,

kafkaConsumerMetrics

kirktrue · 2024-12-04T00:40:55Z

+                    backgroundEventHandler,
+                kafkaConsumerMetrics


Nit: alignment.

Suggested change

backgroundEventHandler,

kafkaConsumerMetrics

backgroundEventHandler,

kafkaConsumerMetrics

kirktrue · 2024-12-04T00:41:46Z

+                requestManagersSupplier,
+            kafkaConsumerMetrics);


Nit: alignment.

Suggested change

requestManagersSupplier,

kafkaConsumerMetrics);

requestManagersSupplier,

kafkaConsumerMetrics);

kirktrue · 2024-12-04T01:53:14Z

    private final IdempotentCloser closer = new IdempotentCloser();
    private volatile Duration closeTimeout = Duration.ofMillis(DEFAULT_CLOSE_TIMEOUT_MS);
    private volatile long cachedMaximumTimeToWait = MAX_POLL_TIMEOUT_MS;
+    private long lastPollTimeMs = 0L;


Pinging on this. I believe it's only written on the background thread, but just want to be sure.

FrankYang0529 · 2024-12-04T13:27:21Z

Hi @kirktrue, I addressed all comments. Could you take a look when you have time? Thank you.

lianetm

Thanks for the updates @FrankYang0529!

lianetm · 2024-12-05T20:03:25Z

+            kafkaConsumerMetrics.recordBackgroundEventQueueProcessingTime(time.milliseconds() - startMs);
        }

        backgroundEventReaper.reap(time.milliseconds());


Interesting, and if we agree on what we want we could just send an update in the KIP email thread to add it to the KIP and here.

To align internally first, I guess we would be interested in the num/avg of expired events, but we need to consider how that metric would go crazy and be a false alarm in cases like poll(0) right? Should we consider the expiration relevant only if there was a non-zero timeout? Thoughts?

lianetm · 2024-12-05T21:45:51Z

+        /**
+         * Set the time when the request was enqueued to {@link NetworkClientDelegate#unsentRequests}.
+         */
+        void setEnqueueTimeMs(final long enqueueTimeMs) {


this should be private right?

lianetm · 2024-12-05T21:46:19Z

+        /**
+         * Return the time when the request was enqueued to {@link NetworkClientDelegate#unsentRequests}.
+         */
+        long enqueueTimeMs() {


this one is less sensitive, but if it's only used here as it seems we could consider private too

lianetm · 2024-12-05T21:55:23Z

+        this.applicationEventQueueSizeSensor = metrics.sensor(APPLICATION_EVENT_QUEUE_SIZE_SENSOR_NAME);
+        this.applicationEventQueueSizeSensor.add(metrics.metricName("application-event-queue-size",
+                metricGroupName,
+                "The current number of events in the consumer network application event queue."),


nit: I guess that, in a time from now, even us that know this by heart will get tricked with if this is the outgoing or incoming queue. Should we be more explicit with something like

Suggested change

"The current number of events in the consumer network application event queue."),

"The current number of events in the queue to send from the application thread to the background thread."),

(and then we can consistently have the flipped version of the message for the background-event-queue-size metric)

lianetm · 2024-12-05T22:05:38Z

+            PollEvent event = new PollEvent(time.milliseconds());
+
+            // add event
+            applicationEventHandler.add(event);


Suggested change

PollEvent event = new PollEvent(time.milliseconds());

// add event

applicationEventHandler.add(event);

// add event

applicationEventHandler.add(new PollEvent(time.milliseconds()));

lianetm · 2024-12-05T22:07:08Z

+
+            // add event
+            applicationEventHandler.add(event);
+            assertEquals(1, (double) metrics.metric(metrics.metricName("application-event-queue-size", "consumer-metrics")).metricValue());


could we reuse the metric name constants we already have?

lianetm · 2024-12-05T22:13:11Z

+        assertTrue((double) metrics.metric(metrics.metricName("background-event-queue-time-avg", "consumer-metrics")).metricValue() > 0);
+        assertTrue((double) metrics.metric(metrics.metricName("background-event-queue-time-max", "consumer-metrics")).metricValue() > 0);


couldn't we be more precise here and expect >= 10?

lianetm · 2024-12-09T14:38:03Z

Hello @FrankYang0529 , could you please solve the conflicts and address the minor comments left? Given that this is introducing new metrics we should meet the Feature freeze deadline which is this week so let's give it the final push. Thanks!

FrankYang0529 · 2024-12-09T15:31:25Z

Hi @lianetm / @AndrewJSchofield, thanks for the review and suggestion. I will update this PR and make it ready tomorrow. Sorry for late.

FrankYang0529 · 2024-12-10T23:11:44Z

Resolved almost all comments. Remaining two discussion thread:

Add application expired event metric: KAFKA-16143: New JMX metrics for AsyncKafkaConsumer #17199 (comment)
AsyncConsumerMetrics doesn't inherit KafkaConsumerMetrics: KAFKA-16143: New JMX metrics for AsyncKafkaConsumer #17199 (comment)

kirktrue · 2024-12-11T00:53:11Z

+            kafkaConsumerMetrics.recordBackgroundEventQueueProcessingTime(time.milliseconds() - startMs);
        }

        backgroundEventReaper.reap(time.milliseconds());


Yes, this sounds like a useful metric to have. Thanks!

kirktrue · 2024-12-11T01:03:58Z

+    private final Sensor unsentRequestsQueueSizeSensor;
+    private final Sensor unsentRequestsQueueTimeSensor;
+
+    public AsyncConsumerMetrics(Metrics metrics, String metricGrpPrefix) {


Nit: I made a quick pass and I don't see anywhere that we're passing in anything other than CONSUMER_METRIC_GROUP_PREFIX or "consumer". Does it make sense to provide this parameter if the value is always the same?

I guess this param will be "share" or similar when integrated with the ShareConsumer as @AndrewJSchofield suggested #17199 (comment)

Hi @lianetm, sorry, I didn't notice that. I addressed other comments and will leave this for https://issues.apache.org/jira/browse/KAFKA-18220.

sounds good. Please also merge trunk latest changes when you address what't left for this PR. Thanks!

kirktrue · 2024-12-11T01:08:01Z

+        expectedMetrics.forEach(metricName -> assertFalse(metrics.metrics().containsKey(metricName), "Missing metric: " + metricName));
+        expectedConsumerMetrics.forEach(metricName -> assertFalse(metrics.metrics().containsKey(metricName), "Missing metric: " + metricName));


The error message on test failure is incorrect here, right? Shouldn't it be:

Suggested change

expectedMetrics.forEach(metricName -> assertFalse(metrics.metrics().containsKey(metricName), "Missing metric: " + metricName));

expectedConsumerMetrics.forEach(metricName -> assertFalse(metrics.metrics().containsKey(metricName), "Missing metric: " + metricName));

expectedMetrics.forEach(metricName -> assertFalse(metrics.metrics().containsKey(metricName), "Metric present after close: " + metricName));

expectedConsumerMetrics.forEach(metricName -> assertFalse(metrics.metrics().containsKey(metricName), "Metric present after close: " + metricName));

kirktrue · 2024-12-11T01:09:41Z

+    private static final String CONSUMER_GROUP_PREFIX = "consumer";
+    private static final String CONSUMER_METRIC_GROUP = "consumer-metrics";


Aren't these constants already declared elsewhere?

lianetm

Just a few last minor comments, and 2 follow-ups:

https://issues.apache.org/jira/browse/KAFKA-18048
refactor to allow ShareConsumer to leverage the new metrics.

With that it LGTM, but let's also wait to hear what @AndrewJSchofield and @kirktrue think. Thanks @FrankYang0529 !

lianetm · 2024-12-11T19:32:29Z

+    private final Sensor unsentRequestsQueueSizeSensor;
+    private final Sensor unsentRequestsQueueTimeSensor;
+
+    public AsyncConsumerMetrics(Metrics metrics, String metricGrpPrefix) {


I guess this param will be "share" or similar when integrated with the ShareConsumer as @AndrewJSchofield suggested #17199 (comment)

AndrewJSchofield · 2024-12-12T13:21:05Z

@lianetm @FrankYang0529 I have opened https://issues.apache.org/jira/browse/KAFKA-18220 to track any refactoring to this to improve the code structure where we want to use this for both the AsyncKafkaConsumer and ShareConsumerImpl. I'm happy that we take this for AK 4.1.

AndrewJSchofield

Approved, with the follow-up task https://issues.apache.org/jira/browse/KAFKA-18220 to address the code structure when used in the share consumer. @lianetm still has some outstanding comments I think.

Signed-off-by: PoAn Yang <payang@apache.org>

lianetm

Thanks for the updates @FrankYang0529 ! LGTM.

lianetm · 2024-12-14T03:00:26Z

Hey @FrankYang0529 , please remember to send the email sharing the update for the expired event metric discovered with this implementation. I missed adding it to the follow ups I mentioned above :). Thanks!

FrankYang0529 · 2024-12-14T16:47:03Z

Hi @lianetm, thanks for reviewing this PR. I mentioned the metric in the vote thread.

https://lists.apache.org/thread/4sn92okp89pt3pzk6f0g3wp0lbtn0g2c

Reviewers: Andrew Schofield <aschofield@confluent.io>, Kirk True <ktrue@confluent.io>, Lianet Magrans <lmagrans@confluent.io>

lianetm added consumer KIP-848 The Next Generation of the Consumer Rebalance Protocol ctr Consumer Threading Refactor (KIP-848) labels Sep 16, 2024

kirktrue added the clients label Sep 26, 2024

kirktrue reviewed Sep 27, 2024

View reviewed changes

FrankYang0529 force-pushed the KAFKA-16143 branch from b1b848f to f68c568 Compare October 9, 2024 16:03

FrankYang0529 marked this pull request as ready for review October 9, 2024 16:30

FrankYang0529 requested a review from kirktrue October 10, 2024 01:51

kirktrue requested changes Oct 15, 2024

View reviewed changes

FrankYang0529 force-pushed the KAFKA-16143 branch from f68c568 to 33e8f2c Compare October 16, 2024 10:26

FrankYang0529 requested a review from kirktrue October 20, 2024 02:03

FrankYang0529 mentioned this pull request Oct 20, 2024

KAFKA-17648: AsyncKafkaConsumer#unsubscribe swallow TopicAuthorizationException and GroupAuthorizationException #17516

Merged

3 tasks

FrankYang0529 force-pushed the KAFKA-16143 branch from 33e8f2c to 16b5b03 Compare October 21, 2024 00:05

kirktrue requested changes Oct 30, 2024

View reviewed changes

FrankYang0529 force-pushed the KAFKA-16143 branch 2 times, most recently from f121073 to 90af010 Compare November 1, 2024 07:07

FrankYang0529 force-pushed the KAFKA-16143 branch 2 times, most recently from 5f171f3 to 68b27ad Compare November 11, 2024 02:46

FrankYang0529 requested a review from kirktrue November 12, 2024 10:00

FrankYang0529 force-pushed the KAFKA-16143 branch 3 times, most recently from 20951a9 to 4ce1c45 Compare November 18, 2024 11:45

FrankYang0529 requested a review from lianetm November 18, 2024 14:09

lianetm mentioned this pull request Nov 18, 2024

KIP-1068: New metrics for the new KafkaConsumer #16685

Closed

3 tasks

lianetm reviewed Nov 19, 2024

View reviewed changes

FrankYang0529 force-pushed the KAFKA-16143 branch 2 times, most recently from f64243c to 4187497 Compare November 26, 2024 14:00

kirktrue requested changes Nov 26, 2024

View reviewed changes

FrankYang0529 force-pushed the KAFKA-16143 branch from 4187497 to 96d8baf Compare November 27, 2024 12:24

kirktrue requested changes Dec 4, 2024

View reviewed changes

FrankYang0529 force-pushed the KAFKA-16143 branch from 96d8baf to fe4834b Compare December 4, 2024 12:51

lianetm reviewed Dec 5, 2024

View reviewed changes

FrankYang0529 force-pushed the KAFKA-16143 branch from fe4834b to 5fb1293 Compare December 8, 2024 03:56

FrankYang0529 force-pushed the KAFKA-16143 branch from 5fb1293 to 89d372a Compare December 10, 2024 13:23

kirktrue reviewed Dec 11, 2024

View reviewed changes

FrankYang0529 force-pushed the KAFKA-16143 branch 2 times, most recently from ea1a661 to a54f409 Compare December 11, 2024 03:18

lianetm reviewed Dec 11, 2024

View reviewed changes

AndrewJSchofield approved these changes Dec 12, 2024

View reviewed changes

FrankYang0529 force-pushed the KAFKA-16143 branch 2 times, most recently from d5a1688 to c5e6fbb Compare December 12, 2024 15:22

KAFKA-16143: New JMX metrics for AsyncKafkaConsumer

a473f22

Signed-off-by: PoAn Yang <payang@apache.org>

FrankYang0529 force-pushed the KAFKA-16143 branch from c5e6fbb to a473f22 Compare December 13, 2024 01:09

lianetm approved these changes Dec 13, 2024

View reviewed changes

lianetm merged commit 770d64d into apache:trunk Dec 13, 2024

FrankYang0529 deleted the KAFKA-16143 branch December 14, 2024 16:26

tedyu pushed a commit to tedyu/kafka that referenced this pull request Jan 6, 2025

KAFKA-16143: New JMX metrics for AsyncKafkaConsumer (apache#17199)

1e64a69

Reviewers: Andrew Schofield <aschofield@confluent.io>, Kirk True <ktrue@confluent.io>, Lianet Magrans <lmagrans@confluent.io>

		kafkaConsumerMetrics.recordBackgroundEventQueueTime(time.milliseconds() - event.addedToQueueMs());
		long startMs = time.milliseconds();

		public BackgroundEventHandler(final Queue<BackgroundEvent> backgroundEventQueue) {
		this(backgroundEventQueue, Optional.empty());

		kafkaConsumerMetrics.ifPresent(metrics -> metrics.recordApplicationEventQueueTime(time.milliseconds() - event.enqueuedMs()));
		long startMs = time.milliseconds();

	event.setEnqueuedMs(System.currentTimeMillis());
	event.setEnqueuedMs(time.milliseconds());

		metricGroupName, "The number of seconds since the last poll() invocation.");
		metricGroupName, "The number of seconds since the last poll() invocation.");

		this.backgroundEventHandler = new BackgroundEventHandler(
		backgroundEventQueue, time, asyncConsumerMetrics);

		private final KafkaConsumerMetrics kafkaConsumerMetrics;
		private final AsyncConsumerMetrics asyncConsumerMetrics;

	"The current number of events in the consumer network application event queue."),
	"The current number of events in the queue to send from the application thread to the background thread."),

		assertTrue((double) metrics.metric(metrics.metricName("background-event-queue-time-avg", "consumer-metrics")).metricValue() > 0);
		assertTrue((double) metrics.metric(metrics.metricName("background-event-queue-time-max", "consumer-metrics")).metricValue() > 0);

		expectedMetrics.forEach(metricName -> assertFalse(metrics.metrics().containsKey(metricName), "Missing metric: " + metricName));
		expectedConsumerMetrics.forEach(metricName -> assertFalse(metrics.metrics().containsKey(metricName), "Missing metric: " + metricName));

		private static final String CONSUMER_GROUP_PREFIX = "consumer";
		private static final String CONSUMER_METRIC_GROUP = "consumer-metrics";

Uh oh!

Conversation

FrankYang0529 commented Sep 15, 2024

Committer Checklist (excluded from commit message)

Uh oh!

kirktrue left a comment

Choose a reason for hiding this comment

Uh oh!

kirktrue commented Oct 8, 2024

Uh oh!

FrankYang0529 commented Oct 9, 2024

Uh oh!

FrankYang0529 commented Oct 9, 2024

Uh oh!

kirktrue left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

FrankYang0529 commented Oct 16, 2024

Uh oh!

kirktrue left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lianetm commented Nov 18, 2024

Uh oh!

lianetm left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment