KAFKA-16143: New JMX metrics for AsyncKafkaConsumer #17199

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

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java

+            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

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java

@@ -139,6 +155,9 @@ void runOnce() {
        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

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java

+    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

clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java

+    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

clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEvent.java

+    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

...rc/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventHandler.java

@@ -73,7 +90,9 @@ public ApplicationEventHandler(final LogContext logContext,
     */
    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

clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java

+        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

clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/BackgroundEvent.java

+    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

...src/main/java/org/apache/kafka/clients/consumer/internals/events/BackgroundEventHandler.java

+    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

...src/main/java/org/apache/kafka/clients/consumer/internals/events/BackgroundEventHandler.java

@@ -42,6 +51,8 @@ public BackgroundEventHandler(final Queue<BackgroundEvent> backgroundEventQueue)
     */
    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

clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEvent.java

+    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)?

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

clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/BackgroundEvent.java

@@ -37,6 +37,7 @@ public enum Type {
     * {@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

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java

+            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

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java

@@ -162,15 +171,20 @@ void runOnce() {
    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

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java

    private ApplicationEventProcessor applicationEventProcessor;
    private NetworkClientDelegate networkClientDelegate;
    private RequestManagers requestManagers;
    private volatile boolean running;
    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?

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

...src/main/java/org/apache/kafka/clients/consumer/internals/events/BackgroundEventHandler.java

@@ -42,6 +47,8 @@ public BackgroundEventHandler(final Queue<BackgroundEvent> backgroundEventQueue)
     */
    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

.../src/main/java/org/apache/kafka/clients/consumer/internals/metrics/KafkaConsumerMetrics.java

 import java.util.concurrent.TimeUnit;

 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

.../src/main/java/org/apache/kafka/clients/consumer/internals/metrics/KafkaConsumerMetrics.java

+    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

.../src/main/java/org/apache/kafka/clients/consumer/internals/metrics/KafkaConsumerMetrics.java

-            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

.../src/main/java/org/apache/kafka/clients/consumer/internals/metrics/KafkaConsumerMetrics.java

+
+        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.

Signed-off-by: PoAn Yang <[email protected]>

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 suggested 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 suggested 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 from 68b27ad to 99387e0 Compare November 15, 2024 01:25

KAFKA-16143: New JMX metrics for AsyncKafkaConsumer

20951a9

Signed-off-by: PoAn Yang <[email protected]>

FrankYang0529 force-pushed the KAFKA-16143 branch from 99387e0 to 20951a9 Compare November 15, 2024 14:50

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-16143: New JMX metrics for AsyncKafkaConsumer #17199

KAFKA-16143: New JMX metrics for AsyncKafkaConsumer #17199

FrankYang0529 commented Sep 15, 2024

kirktrue left a comment

kirktrue commented Oct 8, 2024

FrankYang0529 commented Oct 9, 2024

FrankYang0529 commented Oct 9, 2024

kirktrue left a comment

kirktrue Oct 15, 2024

FrankYang0529 Oct 16, 2024

kirktrue Oct 15, 2024

kirktrue Oct 15, 2024

kirktrue Oct 15, 2024

kirktrue Oct 15, 2024

kirktrue Oct 15, 2024

kirktrue Oct 15, 2024

kirktrue Oct 15, 2024

kirktrue Oct 15, 2024

kirktrue Oct 15, 2024

FrankYang0529 commented Oct 16, 2024

kirktrue left a comment

kirktrue Oct 30, 2024

kirktrue Oct 30, 2024

kirktrue Oct 30, 2024

kirktrue Oct 30, 2024

kirktrue Oct 30, 2024

kirktrue Oct 30, 2024

kirktrue Oct 30, 2024

kirktrue Oct 30, 2024

kirktrue Oct 30, 2024

kirktrue Oct 30, 2024

		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.");

KAFKA-16143: New JMX metrics for AsyncKafkaConsumer #17199

Are you sure you want to change the base?

KAFKA-16143: New JMX metrics for AsyncKafkaConsumer #17199

Conversation

FrankYang0529 commented Sep 15, 2024

Committer Checklist (excluded from commit message)

kirktrue left a comment

Choose a reason for hiding this comment

kirktrue commented Oct 8, 2024

FrankYang0529 commented Oct 9, 2024

FrankYang0529 commented Oct 9, 2024

kirktrue left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

FrankYang0529 commented Oct 16, 2024

kirktrue left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment