You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "FrankYang0529 (via GitHub)" <gi...@apache.org> on 2024/09/15 03:37:57 UTC
[PR] KAFKA-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
FrankYang0529 opened a new pull request, #17199:
URL: https://github.com/apache/kafka/pull/17199
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)
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on PR #17199:
URL: https://github.com/apache/kafka/pull/17199#issuecomment-2402750772
> 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`?
https://github.com/apache/kafka/blob/bb6ebd83f9a0f9b2a37704f1f37b78a48cf480e1/clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java#L230-L244
> 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.
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1876152617
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1893,25 +1901,30 @@ private void subscribeInternal(Collection<String> topics, Optional<ConsumerRebal
* It is possible that {@link ErrorEvent an error}
* could occur when processing the events. In such cases, the processor will take a reference to the first
* error, continue to process the remaining events, and then throw the first error that occurred.
+ *
+ * Visible for testing.
*/
- private boolean processBackgroundEvents() {
+ boolean processBackgroundEvents() {
AtomicReference<KafkaException> firstError = new AtomicReference<>();
- LinkedList<BackgroundEvent> events = new LinkedList<>();
- backgroundEventQueue.drainTo(events);
-
- for (BackgroundEvent event : events) {
- try {
- if (event instanceof CompletableEvent)
- backgroundEventReaper.add((CompletableEvent<?>) event);
-
- backgroundEventProcessor.process(event);
- } catch (Throwable t) {
- KafkaException e = ConsumerUtils.maybeWrapAsKafkaException(t);
-
- if (!firstError.compareAndSet(null, e))
- log.warn("An error occurred when processing the background event: {}", e.getMessage(), e);
+ List<BackgroundEvent> events = backgroundEventHandler.drainEvents();
+ if (!events.isEmpty()) {
+ long startMs = time.milliseconds();
+ for (BackgroundEvent event : events) {
+ kafkaConsumerMetrics.recordBackgroundEventQueueTime(time.milliseconds() - event.enqueuedMs());
+ try {
+ if (event instanceof CompletableEvent)
+ backgroundEventReaper.add((CompletableEvent<?>) event);
+
+ backgroundEventProcessor.process(event);
+ } catch (Throwable t) {
+ KafkaException e = ConsumerUtils.maybeWrapAsKafkaException(t);
+
+ if (!firstError.compareAndSet(null, e))
+ log.warn("An error occurred when processing the background event: {}", e.getMessage(), e);
+ }
}
+ kafkaConsumerMetrics.recordBackgroundEventQueueProcessingTime(time.milliseconds() - startMs);
}
backgroundEventReaper.reap(time.milliseconds());
Review Comment:
Yes, the background event reaper triggered my thought, but @lianetm is correct. The interesting part is the events with deadlines that are timing out.
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on PR #17199:
URL: https://github.com/apache/kafka/pull/17199#issuecomment-2417140265
Hi @kirktrue, I address all comments. Could you take a look when you have time? Thank you.
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on PR #17199:
URL: https://github.com/apache/kafka/pull/17199#issuecomment-2401018863
Hi @kirktrue, sorry for late. I will make this PR ready today.
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on PR #17199:
URL: https://github.com/apache/kafka/pull/17199#issuecomment-2533183424
Resolved almost all comments. Remaining two discussion thread:
1. Add application expired event metric: https://github.com/apache/kafka/pull/17199#discussion_r1878098712
2. AsyncConsumerMetrics doesn't inherit KafkaConsumerMetrics: https://github.com/apache/kafka/pull/17199#discussion_r1878111713
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #17199:
URL: https://github.com/apache/kafka/pull/17199#issuecomment-2542699373
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!
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on PR #17199:
URL: https://github.com/apache/kafka/pull/17199#issuecomment-2543180990
Hi @lianetm, thanks for reviewing this PR. I mentioned the metric in the vote thread.
https://lists.apache.org/thread/4sn92okp89pt3pzk6f0g3wp0lbtn0g2c
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1874604929
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1893,25 +1901,30 @@ private void subscribeInternal(Collection<String> topics, Optional<ConsumerRebal
* It is possible that {@link ErrorEvent an error}
* could occur when processing the events. In such cases, the processor will take a reference to the first
* error, continue to process the remaining events, and then throw the first error that occurred.
+ *
+ * Visible for testing.
*/
- private boolean processBackgroundEvents() {
+ boolean processBackgroundEvents() {
AtomicReference<KafkaException> firstError = new AtomicReference<>();
- LinkedList<BackgroundEvent> events = new LinkedList<>();
- backgroundEventQueue.drainTo(events);
-
- for (BackgroundEvent event : events) {
- try {
- if (event instanceof CompletableEvent)
- backgroundEventReaper.add((CompletableEvent<?>) event);
-
- backgroundEventProcessor.process(event);
- } catch (Throwable t) {
- KafkaException e = ConsumerUtils.maybeWrapAsKafkaException(t);
-
- if (!firstError.compareAndSet(null, e))
- log.warn("An error occurred when processing the background event: {}", e.getMessage(), e);
+ List<BackgroundEvent> events = backgroundEventHandler.drainEvents();
+ if (!events.isEmpty()) {
+ long startMs = time.milliseconds();
+ for (BackgroundEvent event : events) {
+ kafkaConsumerMetrics.recordBackgroundEventQueueTime(time.milliseconds() - event.enqueuedMs());
+ try {
+ if (event instanceof CompletableEvent)
+ backgroundEventReaper.add((CompletableEvent<?>) event);
+
+ backgroundEventProcessor.process(event);
+ } catch (Throwable t) {
+ KafkaException e = ConsumerUtils.maybeWrapAsKafkaException(t);
+
+ if (!firstError.compareAndSet(null, e))
+ log.warn("An error occurred when processing the background event: {}", e.getMessage(), e);
+ }
}
+ kafkaConsumerMetrics.recordBackgroundEventQueueProcessingTime(time.milliseconds() - startMs);
}
backgroundEventReaper.reap(time.milliseconds());
Review Comment:
It looks like the only background event which has deadline is `ConsumerRebalanceListenerCallbackNeededEvent`, but its deadline is `Long.MAX_VALUE`, so it will never be expired. Do we want to update it? If yes, I can create a Jira for it. If not, we may not need to add this metric. WDYT?
https://github.com/apache/kafka/blob/104fa57933d6831ed3364a26e88fbee2911d27b8/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ConsumerRebalanceListenerCallbackNeededEvent.java#L42
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1880783173
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/metrics/KafkaAsyncConsumerMetrics.java:
##########
@@ -0,0 +1,186 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer.internals.metrics;
+
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.metrics.stats.Value;
+
+import java.util.Arrays;
+
+import static org.apache.kafka.clients.consumer.internals.ConsumerUtils.CONSUMER_METRICS_SUFFIX;
+
+public class KafkaAsyncConsumerMetrics extends KafkaConsumerMetrics implements AutoCloseable {
Review Comment:
Hey folks, do you want to address this in this PR or do the refactor in a follow-up? Thoughts @AndrewJSchofield?
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on PR #17199:
URL: https://github.com/apache/kafka/pull/17199#issuecomment-2528383298
Hi @lianetm / @AndrewJSchofield, thanks for the review and suggestion. I will update this PR and make it ready tomorrow. Sorry for late.
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1848358392
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java:
##########
@@ -267,7 +272,9 @@ public void addAll(final List<UnsentRequest> requests) {
public void add(final UnsentRequest r) {
Objects.requireNonNull(r);
r.setTimer(this.time, this.requestTimeoutMs);
+ r.setEnqueuedMs(this.time.milliseconds());
unsentRequests.add(r);
+ kafkaConsumerMetrics.ifPresent(metrics -> metrics.recordUnsentRequestsQueueSize(unsentRequests.size()));
Review Comment:
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?
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1858311982
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEvent.java:
##########
@@ -61,6 +62,14 @@ public Uuid id() {
return id;
}
+ public void setEnqueuedMs(long enqueuedMs) {
+ this.enqueuedMs = enqueuedMs;
+ }
Review Comment:
Add enqueued time to `toStringBase`. Thanks.
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1882367375
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/metrics/AsyncConsumerMetrics.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer.internals.metrics;
+
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.metrics.stats.Value;
+
+import java.util.Arrays;
+
+import static org.apache.kafka.clients.consumer.internals.ConsumerUtils.CONSUMER_METRICS_SUFFIX;
+
+public class AsyncConsumerMetrics extends KafkaConsumerMetrics implements AutoCloseable {
+ private final Metrics metrics;
+
+ public static final String TIME_BETWEEN_NETWORK_THREAD_POLL_SENSOR_NAME = "time-between-network-thread-poll";
+ public static final String APPLICATION_EVENT_QUEUE_SIZE_SENSOR_NAME = "application-event-queue-size";
+ public static final String APPLICATION_EVENT_QUEUE_TIME_SENSOR_NAME = "application-event-queue-time";
+ public static final String APPLICATION_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME = "application-event-queue-processing-time";
+ public static final String APPLICATION_EVENT_EXPIRED_SIZE_SENSOR_NAME = "application-event-expired-size";
+ public static final String BACKGROUND_EVENT_QUEUE_SIZE_SENSOR_NAME = "background-event-queue-size";
+ public static final String BACKGROUND_EVENT_QUEUE_TIME_SENSOR_NAME = "background-event-queue-time";
+ public static final String BACKGROUND_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME = "background-event-queue-processing-time";
+ 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;
+ private final Sensor applicationEventQueueTimeSensor;
+ private final Sensor applicationEventQueueProcessingTimeSensor;
+ private final Sensor applicationEventExpiredSizeSensor;
+ private final Sensor backgroundEventQueueSizeSensor;
+ private final Sensor backgroundEventQueueTimeSensor;
+ private final Sensor backgroundEventQueueProcessingTimeSensor;
+ private final Sensor unsentRequestsQueueSizeSensor;
+ private final Sensor unsentRequestsQueueTimeSensor;
+
+ public AsyncConsumerMetrics(Metrics metrics, String metricGrpPrefix) {
Review 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.
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm merged PR #17199:
URL: https://github.com/apache/kafka/pull/17199
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1868541417
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -380,7 +387,9 @@ public void onGroupAssignmentUpdated(Set<TopicPartition> partitions) {
new CompletableEventReaper(logContext),
applicationEventProcessorSupplier,
networkClientDelegateSupplier,
- requestManagersSupplier);
+ requestManagersSupplier,
+ kafkaConsumerMetrics
Review Comment:
Nit: alignment.
```suggestion
requestManagersSupplier,
kafkaConsumerMetrics
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -553,7 +570,8 @@ public void onGroupAssignmentUpdated(Set<TopicPartition> partitions) {
new CompletableEventReaper(logContext),
applicationEventProcessorSupplier,
networkClientDelegateSupplier,
- requestManagersSupplier);
+ requestManagersSupplier,
+ kafkaConsumerMetrics);
Review Comment:
Nit: alignment.
```suggestion
requestManagersSupplier,
kafkaConsumerMetrics);
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java:
##########
@@ -60,21 +61,24 @@ public class ConsumerNetworkThread extends KafkaThread implements Closeable {
private final Supplier<ApplicationEventProcessor> applicationEventProcessorSupplier;
private final Supplier<NetworkClientDelegate> networkClientDelegateSupplier;
private final Supplier<RequestManagers> requestManagersSupplier;
+ private final Optional<KafkaConsumerMetrics> kafkaConsumerMetrics;
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;
Review Comment:
Pinging on this. I believe it's only written on the background thread, but just want to be sure.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -350,7 +355,9 @@ public void onGroupAssignmentUpdated(Set<TopicPartition> partitions) {
metrics,
fetchMetricsManager.throttleTimeSensor(),
clientTelemetryReporter.map(ClientTelemetryReporter::telemetrySender).orElse(null),
- backgroundEventHandler);
+ backgroundEventHandler,
+ kafkaConsumerMetrics
Review Comment:
Nit: alignment.
```suggestion
backgroundEventHandler,
kafkaConsumerMetrics
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -458,10 +465,15 @@ public void onGroupAssignmentUpdated(Set<TopicPartition> partitions) {
this.defaultApiTimeoutMs = defaultApiTimeoutMs;
this.deserializers = deserializers;
this.applicationEventHandler = applicationEventHandler;
- this.kafkaConsumerMetrics = new KafkaConsumerMetrics(metrics, "consumer");
+ this.kafkaConsumerMetrics = new AsyncConsumerMetrics(metrics, CONSUMER_METRIC_GROUP_PREFIX);
Review Comment:
👍
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1850073759
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventHandler.java:
##########
@@ -39,29 +43,36 @@
* An event handler that receives {@link ApplicationEvent application events} from the application thread which
* are then readable from the {@link ApplicationEventProcessor} in the {@link ConsumerNetworkThread network thread}.
*/
+@SuppressWarnings("this-escape")
public class ApplicationEventHandler implements Closeable {
private final Logger log;
+ private final Time time;
private final BlockingQueue<ApplicationEvent> applicationEventQueue;
private final ConsumerNetworkThread networkThread;
private final IdempotentCloser closer = new IdempotentCloser();
+ private final Optional<KafkaAsyncConsumerMetrics> kafkaAsyncConsumerMetrics;
public ApplicationEventHandler(final LogContext logContext,
final Time time,
final BlockingQueue<ApplicationEvent> applicationEventQueue,
final CompletableEventReaper applicationEventReaper,
final Supplier<ApplicationEventProcessor> applicationEventProcessorSupplier,
final Supplier<NetworkClientDelegate> networkClientDelegateSupplier,
- final Supplier<RequestManagers> requestManagersSupplier) {
+ final Supplier<RequestManagers> requestManagersSupplier,
+ final Optional<KafkaAsyncConsumerMetrics> kafkaAsyncConsumerMetrics) {
this.log = logContext.logger(ApplicationEventHandler.class);
+ this.time = time;
this.applicationEventQueue = applicationEventQueue;
+ this.kafkaAsyncConsumerMetrics = kafkaAsyncConsumerMetrics;
this.networkThread = new ConsumerNetworkThread(logContext,
time,
- applicationEventQueue,
+ this,
Review Comment:
Revert the change. Create a followup Jira: https://issues.apache.org/jira/browse/KAFKA-18048.
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1858275760
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/metrics/KafkaAsyncConsumerMetrics.java:
##########
@@ -0,0 +1,186 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer.internals.metrics;
+
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.metrics.stats.Value;
+
+import java.util.Arrays;
+
+import static org.apache.kafka.clients.consumer.internals.ConsumerUtils.CONSUMER_METRICS_SUFFIX;
+
+public class KafkaAsyncConsumerMetrics extends KafkaConsumerMetrics implements AutoCloseable {
Review Comment:
Hi @AndrewJSchofield, thanks for the suggestion. I rename it to `AsyncConsumerMetrics` and remove `Optional.empty()`. However, it still needs to inherit from `KafkaConsumerMetrics`, because `AsyncKafkaConsumer` implements `ConsumerDelegate` which returns `KafkaConsumerMetrics` in `kafkaConsumerMetrics` function.
https://github.com/apache/kafka/blob/cebec91470c440a1735e50210ae3fc0f62ee40a7/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java#L2079-L2082
https://github.com/apache/kafka/blob/cebec91470c440a1735e50210ae3fc0f62ee40a7/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerDelegate.java#L40
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1876149436
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/metrics/KafkaAsyncConsumerMetrics.java:
##########
@@ -0,0 +1,186 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer.internals.metrics;
+
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.metrics.stats.Value;
+
+import java.util.Arrays;
+
+import static org.apache.kafka.clients.consumer.internals.ConsumerUtils.CONSUMER_METRICS_SUFFIX;
+
+public class KafkaAsyncConsumerMetrics extends KafkaConsumerMetrics implements AutoCloseable {
Review Comment:
Alternatively, you could do a bit more refactoring. There are essentially two metrics objects now - the KafkaConsumerMetrics and the AsyncConsumerMetrics. There could be two methods in `ConsumerDelegate` to give access to them.
`ShareConsumerDelegate` could follow the same pattern.
What do you think? I feel there's a bit more tidying up to do in this area.
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on PR #17199:
URL: https://github.com/apache/kafka/pull/17199#issuecomment-2538901859
@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.
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1872198214
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java:
##########
@@ -317,6 +327,20 @@ Timer timer() {
return timer;
}
+ /**
+ * Set the time when the request was enqueued to {@link NetworkClientDelegate#unsentRequests}.
+ */
+ void setEnqueueTimeMs(final long enqueueTimeMs) {
+ this.enqueueTimeMs = enqueueTimeMs;
+ }
+
+ /**
+ * Return the time when the request was enqueued to {@link NetworkClientDelegate#unsentRequests}.
+ */
+ long enqueueTimeMs() {
Review Comment:
this one is less sensitive, but if it's only used here as it seems we could consider private too
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ApplicationEventHandlerTest.java:
##########
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer.internals;
+
+import org.apache.kafka.clients.consumer.internals.events.ApplicationEvent;
+import org.apache.kafka.clients.consumer.internals.events.ApplicationEventHandler;
+import org.apache.kafka.clients.consumer.internals.events.ApplicationEventProcessor;
+import org.apache.kafka.clients.consumer.internals.events.CompletableEventReaper;
+import org.apache.kafka.clients.consumer.internals.events.PollEvent;
+import org.apache.kafka.clients.consumer.internals.metrics.AsyncConsumerMetrics;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.common.utils.Time;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+
+public class ApplicationEventHandlerTest {
+ private final Time time = new MockTime();
+ private final BlockingQueue<ApplicationEvent> applicationEventsQueue = new LinkedBlockingQueue<>();
+ private final ApplicationEventProcessor applicationEventProcessor = mock(ApplicationEventProcessor.class);
+ private final NetworkClientDelegate networkClientDelegate = mock(NetworkClientDelegate.class);
+ private final RequestManagers requestManagers = mock(RequestManagers.class);
+ private final CompletableEventReaper applicationEventReaper = mock(CompletableEventReaper.class);
+
+ @Test
+ public void testRecordApplicationEventQueueSize() {
+ try (Metrics metrics = new Metrics();
+ AsyncConsumerMetrics asyncConsumerMetrics = new AsyncConsumerMetrics(metrics, "consumer");
+ ApplicationEventHandler applicationEventHandler = new ApplicationEventHandler(
+ new LogContext(),
+ time,
+ applicationEventsQueue,
+ applicationEventReaper,
+ () -> applicationEventProcessor,
+ () -> networkClientDelegate,
+ () -> requestManagers,
+ asyncConsumerMetrics
+ )) {
+ PollEvent event = new PollEvent(time.milliseconds());
+
+ // add event
+ applicationEventHandler.add(event);
+ assertEquals(1, (double) metrics.metric(metrics.metricName("application-event-queue-size", "consumer-metrics")).metricValue());
Review Comment:
could we reuse the metric name constants we already have?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1893,25 +1901,30 @@ private void subscribeInternal(Collection<String> topics, Optional<ConsumerRebal
* It is possible that {@link ErrorEvent an error}
* could occur when processing the events. In such cases, the processor will take a reference to the first
* error, continue to process the remaining events, and then throw the first error that occurred.
+ *
+ * Visible for testing.
*/
- private boolean processBackgroundEvents() {
+ boolean processBackgroundEvents() {
AtomicReference<KafkaException> firstError = new AtomicReference<>();
- LinkedList<BackgroundEvent> events = new LinkedList<>();
- backgroundEventQueue.drainTo(events);
-
- for (BackgroundEvent event : events) {
- try {
- if (event instanceof CompletableEvent)
- backgroundEventReaper.add((CompletableEvent<?>) event);
-
- backgroundEventProcessor.process(event);
- } catch (Throwable t) {
- KafkaException e = ConsumerUtils.maybeWrapAsKafkaException(t);
-
- if (!firstError.compareAndSet(null, e))
- log.warn("An error occurred when processing the background event: {}", e.getMessage(), e);
+ List<BackgroundEvent> events = backgroundEventHandler.drainEvents();
+ if (!events.isEmpty()) {
+ long startMs = time.milliseconds();
+ for (BackgroundEvent event : events) {
+ kafkaConsumerMetrics.recordBackgroundEventQueueTime(time.milliseconds() - event.enqueuedMs());
+ try {
+ if (event instanceof CompletableEvent)
+ backgroundEventReaper.add((CompletableEvent<?>) event);
+
+ backgroundEventProcessor.process(event);
+ } catch (Throwable t) {
+ KafkaException e = ConsumerUtils.maybeWrapAsKafkaException(t);
+
+ if (!firstError.compareAndSet(null, e))
+ log.warn("An error occurred when processing the background event: {}", e.getMessage(), e);
+ }
}
+ kafkaConsumerMetrics.recordBackgroundEventQueueProcessingTime(time.milliseconds() - startMs);
}
backgroundEventReaper.reap(time.milliseconds());
Review Comment:
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 relevant the expiration only if there was a non-zero timeout? Thoughts?
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ApplicationEventHandlerTest.java:
##########
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer.internals;
+
+import org.apache.kafka.clients.consumer.internals.events.ApplicationEvent;
+import org.apache.kafka.clients.consumer.internals.events.ApplicationEventHandler;
+import org.apache.kafka.clients.consumer.internals.events.ApplicationEventProcessor;
+import org.apache.kafka.clients.consumer.internals.events.CompletableEventReaper;
+import org.apache.kafka.clients.consumer.internals.events.PollEvent;
+import org.apache.kafka.clients.consumer.internals.metrics.AsyncConsumerMetrics;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.common.utils.Time;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+
+public class ApplicationEventHandlerTest {
+ private final Time time = new MockTime();
+ private final BlockingQueue<ApplicationEvent> applicationEventsQueue = new LinkedBlockingQueue<>();
+ private final ApplicationEventProcessor applicationEventProcessor = mock(ApplicationEventProcessor.class);
+ private final NetworkClientDelegate networkClientDelegate = mock(NetworkClientDelegate.class);
+ private final RequestManagers requestManagers = mock(RequestManagers.class);
+ private final CompletableEventReaper applicationEventReaper = mock(CompletableEventReaper.class);
+
+ @Test
+ public void testRecordApplicationEventQueueSize() {
+ try (Metrics metrics = new Metrics();
+ AsyncConsumerMetrics asyncConsumerMetrics = new AsyncConsumerMetrics(metrics, "consumer");
+ ApplicationEventHandler applicationEventHandler = new ApplicationEventHandler(
+ new LogContext(),
+ time,
+ applicationEventsQueue,
+ applicationEventReaper,
+ () -> applicationEventProcessor,
+ () -> networkClientDelegate,
+ () -> requestManagers,
+ asyncConsumerMetrics
+ )) {
+ PollEvent event = new PollEvent(time.milliseconds());
+
+ // add event
+ applicationEventHandler.add(event);
Review Comment:
```suggestion
// add event
applicationEventHandler.add(new PollEvent(time.milliseconds()));
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java:
##########
@@ -317,6 +327,20 @@ Timer timer() {
return timer;
}
+ /**
+ * Set the time when the request was enqueued to {@link NetworkClientDelegate#unsentRequests}.
+ */
+ void setEnqueueTimeMs(final long enqueueTimeMs) {
Review Comment:
this should be private right?
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/BackgroundEventHandlerTest.java:
##########
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer.internals;
+
+import org.apache.kafka.clients.consumer.internals.events.BackgroundEvent;
+import org.apache.kafka.clients.consumer.internals.events.BackgroundEventHandler;
+import org.apache.kafka.clients.consumer.internals.events.ErrorEvent;
+import org.apache.kafka.clients.consumer.internals.metrics.AsyncConsumerMetrics;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.utils.MockTime;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class BackgroundEventHandlerTest {
+ private final BlockingQueue<BackgroundEvent> backgroundEventsQueue = new LinkedBlockingQueue<>();
+
+ @Test
+ public void testRecordBackgroundEventQueueSize() {
+ try (Metrics metrics = new Metrics();
+ AsyncConsumerMetrics asyncConsumerMetrics = new AsyncConsumerMetrics(metrics, "consumer")) {
+ BackgroundEventHandler backgroundEventHandler = new BackgroundEventHandler(
+ backgroundEventsQueue,
+ new MockTime(0),
+ asyncConsumerMetrics);
+ BackgroundEvent event = new ErrorEvent(new Throwable());
+
+ // add event
+ backgroundEventHandler.add(event);
Review Comment:
```suggestion
// add event
backgroundEventHandler.add(new ErrorEvent(new Throwable()));
```
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThreadTest.java:
##########
@@ -199,4 +203,57 @@ public void testSendUnsentRequests() {
consumerNetworkThread.cleanup();
verify(networkClientDelegate, times(2)).poll(anyLong(), anyLong());
}
+
+ @Test
+ public void testRunOnceRecordTimeBetweenNetworkThreadPoll() {
+ try (Metrics metrics = new Metrics();
+ AsyncConsumerMetrics asyncConsumerMetrics = new AsyncConsumerMetrics(metrics, "consumer");
+ ConsumerNetworkThread consumerNetworkThread = new ConsumerNetworkThread(
+ new LogContext(),
+ time,
+ applicationEventQueue,
+ applicationEventReaper,
+ () -> applicationEventProcessor,
+ () -> networkClientDelegate,
+ () -> requestManagers,
+ asyncConsumerMetrics
+ )) {
+ consumerNetworkThread.initializeResources();
+
+ consumerNetworkThread.runOnce();
+ time.sleep(10);
+ consumerNetworkThread.runOnce();
+ assertTrue((double) metrics.metric(metrics.metricName("time-between-network-thread-poll-avg", "consumer-metrics")).metricValue() > 0);
+ assertTrue((double) metrics.metric(metrics.metricName("time-between-network-thread-poll-max", "consumer-metrics")).metricValue() > 0);
Review Comment:
couldn't we be more precise here (I guess it should be 10 exactly given how this is calculated right? not sure if we would need >= in this case)
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumerTest.java:
##########
@@ -1951,6 +1952,31 @@ public void testSubscribePatternAgainstBrokerNotSupportingRegex() throws Interru
}, "Consumer did not throw the expected UnsupportedVersionException on poll");
}
+ @Test
+ public void testRecordBackgroundEventQueueSizeAndBackgroundEventQueueTime() {
+ consumer = newConsumer(
+ mock(FetchBuffer.class),
+ mock(ConsumerInterceptors.class),
+ mock(ConsumerRebalanceListenerInvoker.class),
+ mock(SubscriptionState.class),
+ "group-id",
+ "client-id",
+ false);
+ Metrics metrics = consumer.metricsRegistry();
+ AsyncConsumerMetrics kafkaConsumerMetrics = consumer.kafkaConsumerMetrics();
+
+ ConsumerRebalanceListenerCallbackNeededEvent event = new ConsumerRebalanceListenerCallbackNeededEvent(ON_PARTITIONS_REVOKED, Collections.emptySortedSet());
+ event.setEnqueuedMs(time.milliseconds());
+ backgroundEventQueue.add(event);
+ kafkaConsumerMetrics.recordBackgroundEventQueueSize(1);
+
+ time.sleep(10);
+ consumer.processBackgroundEvents();
+ assertEquals(0, (double) metrics.metric(metrics.metricName("background-event-queue-size", "consumer-metrics")).metricValue());
+ 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);
Review Comment:
couldn't we be more precise here and expect >= 10?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/metrics/AsyncConsumerMetrics.java:
##########
@@ -0,0 +1,186 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer.internals.metrics;
+
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.metrics.stats.Value;
+
+import java.util.Arrays;
+
+import static org.apache.kafka.clients.consumer.internals.ConsumerUtils.CONSUMER_METRICS_SUFFIX;
+
+public class AsyncConsumerMetrics extends KafkaConsumerMetrics implements AutoCloseable {
+ private final Metrics metrics;
+
+ public static final String TIME_BETWEEN_NETWORK_THREAD_POLL_SENSOR_NAME = "time-between-network-thread-poll";
+ public static final String APPLICATION_EVENT_QUEUE_SIZE_SENSOR_NAME = "application-event-queue-size";
+ public static final String APPLICATION_EVENT_QUEUE_TIME_SENSOR_NAME = "application-event-queue-time";
+ public static final String APPLICATION_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME = "application-event-queue-processing-time";
+ public static final String BACKGROUND_EVENT_QUEUE_SIZE_SENSOR_NAME = "background-event-queue-size";
+ public static final String BACKGROUND_EVENT_QUEUE_TIME_SENSOR_NAME = "background-event-queue-time";
+ public static final String BACKGROUND_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME = "background-event-queue-processing-time";
+ 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;
+ private final Sensor applicationEventQueueTimeSensor;
+ private final Sensor applicationEventQueueProcessingTimeSensor;
+ private final Sensor backgroundEventQueueSizeSensor;
+ private final Sensor backgroundEventQueueTimeSensor;
+ private final Sensor backgroundEventQueueProcessingTimeSensor;
+ private final Sensor unsentRequestsQueueSizeSensor;
+ private final Sensor unsentRequestsQueueTimeSensor;
+
+ public AsyncConsumerMetrics(Metrics metrics, String metricGrpPrefix) {
+ super(metrics, metricGrpPrefix);
+
+ this.metrics = metrics;
+ final String metricGroupName = metricGrpPrefix + CONSUMER_METRICS_SUFFIX;
+ this.timeBetweenNetworkThreadPollSensor = metrics.sensor(TIME_BETWEEN_NETWORK_THREAD_POLL_SENSOR_NAME);
+ this.timeBetweenNetworkThreadPollSensor.add(metrics.metricName("time-between-network-thread-poll-avg",
+ metricGroupName,
+ "The average time taken, in milliseconds, between each poll in the network thread."),
+ new Avg());
+ this.timeBetweenNetworkThreadPollSensor.add(metrics.metricName("time-between-network-thread-poll-max",
+ metricGroupName,
+ "The maximum time taken, in milliseconds, between each poll in the network thread."),
+ new Max());
+
+ 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."),
Review Comment:
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
```suggestion
"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)
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on PR #17199:
URL: https://github.com/apache/kafka/pull/17199#issuecomment-2400960219
@FrankYang0529—I wanted to check in and see if you had any questions or needed anything to progress on this. Thanks!
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1850054088
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java:
##########
@@ -267,7 +272,9 @@ public void addAll(final List<UnsentRequest> requests) {
public void add(final UnsentRequest r) {
Objects.requireNonNull(r);
r.setTimer(this.time, this.requestTimeoutMs);
+ r.setEnqueuedMs(this.time.milliseconds());
unsentRequests.add(r);
+ kafkaConsumerMetrics.ifPresent(metrics -> metrics.recordUnsentRequestsQueueSize(unsentRequests.size()));
Review Comment:
I think we just want to see how many unsent requests left. We can just record at the end of `NetworkClientDelegate#poll`.
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1858310870
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventHandler.java:
##########
@@ -42,26 +44,32 @@
public class ApplicationEventHandler implements Closeable {
private final Logger log;
+ private final Time time;
private final BlockingQueue<ApplicationEvent> applicationEventQueue;
Review Comment:
I think `ApplicationEventHandler` is a class to encapsulate application event queue. However, it couples with `ConsumerNetworkThread`, so we can't just add the draining event function to `ApplicationEventHandler`. How about we solve it in followup Jira https://issues.apache.org/jira/browse/KAFKA-18048? We had related discussion in previous comment https://github.com/apache/kafka/pull/17199#discussion_r1849058419.
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1854114243
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventHandler.java:
##########
@@ -42,26 +44,32 @@
public class ApplicationEventHandler implements Closeable {
private final Logger log;
+ private final Time time;
private final BlockingQueue<ApplicationEvent> applicationEventQueue;
Review Comment:
I wonder if we could actually make a class for the application event queue, rather than using `BlockingQueue` directly. In this way, we could do things like incorporate the queue size metric into the queue, rather than having update the metrics in the caller.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/metrics/KafkaAsyncConsumerMetrics.java:
##########
@@ -0,0 +1,186 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer.internals.metrics;
+
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.metrics.stats.Value;
+
+import java.util.Arrays;
+
+import static org.apache.kafka.clients.consumer.internals.ConsumerUtils.CONSUMER_METRICS_SUFFIX;
+
+public class KafkaAsyncConsumerMetrics extends KafkaConsumerMetrics implements AutoCloseable {
Review Comment:
These metrics are really to do with measuring the processing of events sent between the application and background threads. AsyncKafkaConsumer and ShareConsumerImpl both use the same architecture and the same metrics would apply equally well to both. So, I suggest:
* Renaming this to `AsyncConsumerMetrics`
* Do not inherit from `KafkaConsumerMetrics`
* Supply an instance of this class when constructing the `NetworkClientDelegate` in `ShareConsumerImpl`. This should not be `Optional.empty()`. In fact, I don't think it should be optional at all.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEvent.java:
##########
@@ -61,6 +62,14 @@ public Uuid id() {
return id;
}
+ public void setEnqueuedMs(long enqueuedMs) {
+ this.enqueuedMs = enqueuedMs;
+ }
Review Comment:
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.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java:
##########
@@ -60,21 +61,24 @@ public class ConsumerNetworkThread extends KafkaThread implements Closeable {
private final Supplier<ApplicationEventProcessor> applicationEventProcessorSupplier;
private final Supplier<NetworkClientDelegate> networkClientDelegateSupplier;
private final Supplier<RequestManagers> requestManagersSupplier;
+ private final Optional<KafkaAsyncConsumerMetrics> kafkaAsyncConsumerMetrics;
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;
public ConsumerNetworkThread(LogContext logContext,
Time time,
BlockingQueue<ApplicationEvent> applicationEventQueue,
CompletableEventReaper applicationEventReaper,
Supplier<ApplicationEventProcessor> applicationEventProcessorSupplier,
Supplier<NetworkClientDelegate> networkClientDelegateSupplier,
- Supplier<RequestManagers> requestManagersSupplier) {
+ Supplier<RequestManagers> requestManagersSupplier,
+ Optional<KafkaAsyncConsumerMetrics> kafkaAsyncConsumerMetrics) {
Review Comment:
As mentioned in another comment, this should not be optional. It's almost general-purpose enough that it can be used for all users of ConsumerNetworkThread.
##########
clients/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.setEnqueuedMs(time.milliseconds());
backgroundEventQueue.add(event);
+ kafkaConsumerMetrics.ifPresent(metrics -> metrics.recordBackgroundEventQueueSize(backgroundEventQueue.size()));
Review Comment:
Yes, I agree.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1893,25 +1901,30 @@ private void subscribeInternal(Collection<String> topics, Optional<ConsumerRebal
* It is possible that {@link ErrorEvent an error}
* could occur when processing the events. In such cases, the processor will take a reference to the first
* error, continue to process the remaining events, and then throw the first error that occurred.
+ *
+ * Visible for testing.
*/
- private boolean processBackgroundEvents() {
+ boolean processBackgroundEvents() {
AtomicReference<KafkaException> firstError = new AtomicReference<>();
- LinkedList<BackgroundEvent> events = new LinkedList<>();
- backgroundEventQueue.drainTo(events);
-
- for (BackgroundEvent event : events) {
- try {
- if (event instanceof CompletableEvent)
- backgroundEventReaper.add((CompletableEvent<?>) event);
-
- backgroundEventProcessor.process(event);
- } catch (Throwable t) {
- KafkaException e = ConsumerUtils.maybeWrapAsKafkaException(t);
-
- if (!firstError.compareAndSet(null, e))
- log.warn("An error occurred when processing the background event: {}", e.getMessage(), e);
+ List<BackgroundEvent> events = backgroundEventHandler.drainEvents();
+ if (!events.isEmpty()) {
+ long startMs = time.milliseconds();
+ for (BackgroundEvent event : events) {
+ kafkaConsumerMetrics.recordBackgroundEventQueueTime(time.milliseconds() - event.enqueuedMs());
+ try {
+ if (event instanceof CompletableEvent)
+ backgroundEventReaper.add((CompletableEvent<?>) event);
+
+ backgroundEventProcessor.process(event);
+ } catch (Throwable t) {
+ KafkaException e = ConsumerUtils.maybeWrapAsKafkaException(t);
+
+ if (!firstError.compareAndSet(null, e))
+ log.warn("An error occurred when processing the background event: {}", e.getMessage(), e);
+ }
}
+ kafkaConsumerMetrics.recordBackgroundEventQueueProcessingTime(time.milliseconds() - startMs);
}
backgroundEventReaper.reap(time.milliseconds());
Review Comment:
How are we going to account for events which expired and are removed from the queue by the event reaper? They probably ought to be included in the 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
Re: [PR] KAFKA-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1823530021
##########
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;
Review Comment:
Same comment as per above with `ApplicationEvent`.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/metrics/KafkaConsumerMetrics.java:
##########
@@ -16,19 +16,23 @@
*/
package org.apache.kafka.clients.consumer.internals.metrics;
+import org.apache.kafka.clients.consumer.GroupProtocol;
import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.metrics.Measurable;
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.metrics.stats.Avg;
import org.apache.kafka.common.metrics.stats.CumulativeSum;
import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.metrics.stats.Value;
+import java.util.Arrays;
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;
Review Comment:
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.
##########
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()));
for (ApplicationEvent event : events) {
+ kafkaConsumerMetrics.ifPresent(metrics -> metrics.recordApplicationEventQueueTime(time.milliseconds() - event.enqueuedMs()));
+ long startMs = time.milliseconds();
Review Comment:
Can we swap the ordering of these two lines to avoid the extra call to `time.milliseconds()`?
```suggestion
long startMs = time.milliseconds();
kafkaConsumerMetrics.ifPresent(metrics -> metrics.recordApplicationEventQueueTime(startMs - event.enqueuedMs()));
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java:
##########
@@ -60,21 +61,24 @@ public class ConsumerNetworkThread extends KafkaThread implements Closeable {
private final Supplier<ApplicationEventProcessor> applicationEventProcessorSupplier;
private final Supplier<NetworkClientDelegate> networkClientDelegateSupplier;
private final Supplier<RequestManagers> requestManagersSupplier;
+ private final Optional<KafkaConsumerMetrics> kafkaConsumerMetrics;
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;
Review Comment:
This is only ever written to and read from the same thread, right?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEvent.java:
##########
@@ -61,6 +62,14 @@ public Uuid id() {
return id;
}
+ public void setEnqueuedMs(long enqueuedMs) {
+ this.enqueuedMs = enqueuedMs;
+ }
Review Comment:
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)?
##########
clients/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());
Review Comment:
We need to add a `Time` variable to this class so that we can do this instead:
```suggestion
event.setEnqueuedMs(time.milliseconds());
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/metrics/KafkaConsumerMetrics.java:
##########
@@ -50,44 +75,124 @@ public KafkaConsumerMetrics(Metrics metrics, String metricGrpPrefix) {
return TimeUnit.SECONDS.convert(now - lastPollMs, TimeUnit.MILLISECONDS);
};
this.lastPollMetricName = metrics.metricName("last-poll-seconds-ago",
- metricGroupName, "The number of seconds since the last poll() invocation.");
+ metricGroupName, "The number of seconds since the last poll() invocation.");
Review Comment:
Can we remove these whitespace changes? They seem extraneous.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/metrics/KafkaConsumerMetrics.java:
##########
@@ -39,8 +43,29 @@ public class KafkaConsumerMetrics implements AutoCloseable {
private long pollStartMs;
private long timeSinceLastPollMs;
- public KafkaConsumerMetrics(Metrics metrics, String metricGrpPrefix) {
+ // CONSUMER group protocol metrics
+ public static final String TIME_BETWEEN_NETWORK_THREAD_POLL_SENSOR_NAME = "time-between-network-thread-poll";
+ public static final String APPLICATION_EVENT_QUEUE_SIZE_SENSOR_NAME = "application-event-queue-size";
+ public static final String APPLICATION_EVENT_QUEUE_TIME_SENSOR_NAME = "application-event-queue-time";
+ public static final String APPLICATION_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME = "application-event-queue-processing-time";
+ public static final String BACKGROUND_EVENT_QUEUE_SIZE_SENSOR_NAME = "background-event-queue-size";
+ public static final String BACKGROUND_EVENT_QUEUE_TIME_SENSOR_NAME = "background-event-queue-time";
+ public static final String BACKGROUND_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME = "background-event-queue-processing-time";
+ 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 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;
Review Comment:
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.
##########
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()));
Review Comment:
Once we drain `applicationEventQueue`, it'll be empty, right? If, so why not just do:
```suggestion
kafkaConsumerMetrics.ifPresent(metrics -> metrics.recordApplicationEventQueueSize(0));
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/metrics/KafkaConsumerMetrics.java:
##########
@@ -111,12 +216,62 @@ public void recordCommitted(long duration) {
this.committedSensor.record(duration);
}
+ public void recordTimeBetweenNetworkThreadPoll(long timeBetweenNetworkThreadPoll) {
+ this.timeBetweenNetworkThreadPollSensor.record(timeBetweenNetworkThreadPoll);
+ }
+
+ public void recordApplicationEventQueueSize(int size) {
+ this.applicationEventQueueSizeSensor.record(size);
+ }
+
+ public void recordApplicationEventQueueTime(long time) {
+ this.applicationEventQueueTimeSensor.record(time);
+ }
+
+ public void recordApplicationEventQueueProcessingTime(long processingTime) {
+ this.applicationEventQueueProcessingTimeSensor.record(processingTime);
+ }
+
+ public void recordUnsentRequestsQueueSize(int size) {
+ this.unsentRequestsQueueSizeSensor.record(size);
+ }
+
+ public void recordUnsentRequestsQueueTime(long time) {
+ this.unsentRequestsQueueTimeSensor.record(time);
+ }
+
+ public void recordBackgroundEventQueueSize(int size) {
+ this.backgroundEventQueueSizeSensor.record(size);
+ }
+
+ public void recordBackgroundEventQueueTime(long time) {
+ this.backgroundEventQueueTimeSensor.record(time);
+ }
+
+ public void recordBackgroundEventQueueProcessingTime(long processingTime) {
+ this.backgroundEventQueueProcessingTimeSensor.record(processingTime);
+ }
+
@Override
public void close() {
metrics.removeMetric(lastPollMetricName);
metrics.removeSensor(timeBetweenPollSensor.name());
metrics.removeSensor(pollIdleSensor.name());
metrics.removeSensor(commitSyncSensor.name());
metrics.removeSensor(committedSensor.name());
+
+ 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);
+ }
Review Comment:
This is another place that having a subclass would help.
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1872002514
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1893,25 +1901,30 @@ private void subscribeInternal(Collection<String> topics, Optional<ConsumerRebal
* It is possible that {@link ErrorEvent an error}
* could occur when processing the events. In such cases, the processor will take a reference to the first
* error, continue to process the remaining events, and then throw the first error that occurred.
+ *
+ * Visible for testing.
*/
- private boolean processBackgroundEvents() {
+ boolean processBackgroundEvents() {
AtomicReference<KafkaException> firstError = new AtomicReference<>();
- LinkedList<BackgroundEvent> events = new LinkedList<>();
- backgroundEventQueue.drainTo(events);
-
- for (BackgroundEvent event : events) {
- try {
- if (event instanceof CompletableEvent)
- backgroundEventReaper.add((CompletableEvent<?>) event);
-
- backgroundEventProcessor.process(event);
- } catch (Throwable t) {
- KafkaException e = ConsumerUtils.maybeWrapAsKafkaException(t);
-
- if (!firstError.compareAndSet(null, e))
- log.warn("An error occurred when processing the background event: {}", e.getMessage(), e);
+ List<BackgroundEvent> events = backgroundEventHandler.drainEvents();
+ if (!events.isEmpty()) {
+ long startMs = time.milliseconds();
+ for (BackgroundEvent event : events) {
+ kafkaConsumerMetrics.recordBackgroundEventQueueTime(time.milliseconds() - event.enqueuedMs());
+ try {
+ if (event instanceof CompletableEvent)
+ backgroundEventReaper.add((CompletableEvent<?>) event);
+
+ backgroundEventProcessor.process(event);
+ } catch (Throwable t) {
+ KafkaException e = ConsumerUtils.maybeWrapAsKafkaException(t);
+
+ if (!firstError.compareAndSet(null, e))
+ log.warn("An error occurred when processing the background event: {}", e.getMessage(), e);
+ }
}
+ kafkaConsumerMetrics.recordBackgroundEventQueueProcessingTime(time.milliseconds() - startMs);
}
backgroundEventReaper.reap(time.milliseconds());
Review Comment:
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?
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1859369437
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -507,7 +511,8 @@ public void onGroupAssignmentUpdated(Set<TopicPartition> partitions) {
BlockingQueue<ApplicationEvent> applicationEventQueue = new LinkedBlockingQueue<>();
this.backgroundEventQueue = new LinkedBlockingQueue<>();
- BackgroundEventHandler backgroundEventHandler = new BackgroundEventHandler(backgroundEventQueue);
+ this.backgroundEventHandler = new BackgroundEventHandler(
+ backgroundEventQueue, time, asyncConsumerMetrics);
Review Comment:
```suggestion
this.backgroundEventHandler = new BackgroundEventHandler(
backgroundEventQueue,
time,
asyncConsumerMetrics
);
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -214,10 +213,11 @@ private void process(final ConsumerRebalanceListenerCallbackNeededEvent event) {
private final ApplicationEventHandler applicationEventHandler;
private final Time time;
private final AtomicReference<Optional<ConsumerGroupMetadata>> groupMetadata = new AtomicReference<>(Optional.empty());
- private final KafkaConsumerMetrics kafkaConsumerMetrics;
+ private final AsyncConsumerMetrics asyncConsumerMetrics;
Review Comment:
I'd prefer it be left as `kafkaConsumerMetrics`. Calling out the distinction in the variable name isn't really adding anything (IMO).
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -350,7 +352,8 @@ public void onGroupAssignmentUpdated(Set<TopicPartition> partitions) {
metrics,
fetchMetricsManager.throttleTimeSensor(),
clientTelemetryReporter.map(ClientTelemetryReporter::telemetrySender).orElse(null),
- backgroundEventHandler);
+ backgroundEventHandler,
+ asyncConsumerMetrics);
Review Comment:
Nit: minor alignment issue:
```suggestion
backgroundEventHandler,
asyncConsumerMetrics);
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEvent.java:
##########
@@ -61,6 +62,14 @@ public Uuid id() {
return id;
}
+ public void setEnqueuedMs(long enqueuedMs) {
+ this.enqueuedMs = enqueuedMs;
+ }
Review Comment:
> we probably should be able to see the enqueued time in the string representation of the events
Agreed. Good point. Thanks @AndrewJSchofield.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -338,7 +339,8 @@ public void onGroupAssignmentUpdated(Set<TopicPartition> partitions) {
ApiVersions apiVersions = new ApiVersions();
final BlockingQueue<ApplicationEvent> applicationEventQueue = new LinkedBlockingQueue<>();
- final BackgroundEventHandler backgroundEventHandler = new BackgroundEventHandler(backgroundEventQueue);
+ this.backgroundEventHandler = new BackgroundEventHandler(
+ backgroundEventQueue, time, asyncConsumerMetrics);
Review Comment:
Nitpick: for better or worse, we've adopted this style for multi-line parameter lists:
```suggestion
this.backgroundEventHandler = new BackgroundEventHandler(
backgroundEventQueue,
time,
asyncConsumerMetrics
);
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/BackgroundEvent.java:
##########
@@ -38,6 +38,12 @@ public enum Type {
*/
private final Uuid id;
+ /**
+ * 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;
Review Comment:
We also need the `enqueuedMs` in the `toStringBase()` method as per the `ApplicationEvent`’s method of the same name.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventHandler.java:
##########
@@ -42,26 +44,32 @@
public class ApplicationEventHandler implements Closeable {
private final Logger log;
+ private final Time time;
private final BlockingQueue<ApplicationEvent> applicationEventQueue;
Review Comment:
I agree that this is an area that could use some refactoring. Thanks for filing KAFKA-18048, @FrankYang0529.
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1878098712
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1893,25 +1901,30 @@ private void subscribeInternal(Collection<String> topics, Optional<ConsumerRebal
* It is possible that {@link ErrorEvent an error}
* could occur when processing the events. In such cases, the processor will take a reference to the first
* error, continue to process the remaining events, and then throw the first error that occurred.
+ *
+ * Visible for testing.
*/
- private boolean processBackgroundEvents() {
+ boolean processBackgroundEvents() {
AtomicReference<KafkaException> firstError = new AtomicReference<>();
- LinkedList<BackgroundEvent> events = new LinkedList<>();
- backgroundEventQueue.drainTo(events);
-
- for (BackgroundEvent event : events) {
- try {
- if (event instanceof CompletableEvent)
- backgroundEventReaper.add((CompletableEvent<?>) event);
-
- backgroundEventProcessor.process(event);
- } catch (Throwable t) {
- KafkaException e = ConsumerUtils.maybeWrapAsKafkaException(t);
-
- if (!firstError.compareAndSet(null, e))
- log.warn("An error occurred when processing the background event: {}", e.getMessage(), e);
+ List<BackgroundEvent> events = backgroundEventHandler.drainEvents();
+ if (!events.isEmpty()) {
+ long startMs = time.milliseconds();
+ for (BackgroundEvent event : events) {
+ kafkaConsumerMetrics.recordBackgroundEventQueueTime(time.milliseconds() - event.enqueuedMs());
+ try {
+ if (event instanceof CompletableEvent)
+ backgroundEventReaper.add((CompletableEvent<?>) event);
+
+ backgroundEventProcessor.process(event);
+ } catch (Throwable t) {
+ KafkaException e = ConsumerUtils.maybeWrapAsKafkaException(t);
+
+ if (!firstError.compareAndSet(null, e))
+ log.warn("An error occurred when processing the background event: {}", e.getMessage(), e);
+ }
}
+ kafkaConsumerMetrics.recordBackgroundEventQueueProcessingTime(time.milliseconds() - startMs);
}
backgroundEventReaper.reap(time.milliseconds());
Review Comment:
Add a new metric `application-event-expired-size`. After we all agree this metric, I will send a mail to original vote thread to mention the new metric.
I also change `CompletableEventReaper#reap` to return expired event count, so we can record the value. Thanks.
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1850073308
##########
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);
+ kafkaAsyncConsumerMetrics.ifPresent(metrics -> metrics.recordApplicationEventQueueSize(0));
Review Comment:
Create a followup Jira for it: https://issues.apache.org/jira/browse/KAFKA-18048.
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1847318144
##########
clients/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.setEnqueuedMs(time.milliseconds());
backgroundEventQueue.add(event);
+ kafkaConsumerMetrics.ifPresent(metrics -> metrics.recordBackgroundEventQueueSize(backgroundEventQueue.size()));
Review Comment:
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?
##########
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);
+ kafkaAsyncConsumerMetrics.ifPresent(metrics -> metrics.recordApplicationEventQueueSize(0));
for (ApplicationEvent event : events) {
+ long startMs = time.milliseconds();
+ kafkaAsyncConsumerMetrics.ifPresent(metrics -> metrics.recordApplicationEventQueueTime(startMs - event.enqueuedMs()));
try {
if (event instanceof CompletableEvent)
applicationEventReaper.add((CompletableEvent<?>) event);
applicationEventProcessor.process(event);
} catch (Throwable t) {
log.warn("Error processing event {}", t.getMessage(), t);
+ } finally {
+ kafkaAsyncConsumerMetrics.ifPresent(metrics -> metrics.recordApplicationEventQueueProcessingTime(time.milliseconds() - startMs));
Review Comment:
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?
##########
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);
+ kafkaAsyncConsumerMetrics.ifPresent(metrics -> metrics.recordApplicationEventQueueSize(0));
Review Comment:
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?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/metrics/KafkaAsyncConsumerMetrics.java:
##########
@@ -0,0 +1,186 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer.internals.metrics;
+
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.metrics.stats.Value;
+
+import java.util.Arrays;
+
+import static org.apache.kafka.clients.consumer.internals.ConsumerUtils.CONSUMER_METRICS_SUFFIX;
+
+public class KafkaAsyncConsumerMetrics extends KafkaConsumerMetrics implements AutoCloseable {
+ private final Metrics metrics;
+
+ public static final String TIME_BETWEEN_NETWORK_THREAD_POLL_SENSOR_NAME = "time-between-network-thread-poll";
+ public static final String APPLICATION_EVENT_QUEUE_SIZE_SENSOR_NAME = "application-event-queue-size";
+ public static final String APPLICATION_EVENT_QUEUE_TIME_SENSOR_NAME = "application-event-queue-time";
+ public static final String APPLICATION_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME = "application-event-queue-processing-time";
+ public static final String BACKGROUND_EVENT_QUEUE_SIZE_SENSOR_NAME = "background-event-queue-size";
+ public static final String BACKGROUND_EVENT_QUEUE_TIME_SENSOR_NAME = "background-event-queue-time";
+ public static final String BACKGROUND_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME = "background-event-queue-processing-time";
+ 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;
+ private final Sensor applicationEventQueueTimeSensor;
+ private final Sensor applicationEventQueueProcessingTimeSensor;
+ private final Sensor backgroundEventQueueSizeSensor;
+ private final Sensor backgroundEventQueueTimeSensor;
+ private final Sensor backgroundEventQueueProcessingTimeSensor;
+ private final Sensor unsentRequestsQueueSizeSensor;
+ private final Sensor unsentRequestsQueueTimeSensor;
+
+ public KafkaAsyncConsumerMetrics(Metrics metrics, String metricGrpPrefix) {
+ super(metrics, metricGrpPrefix);
+
+ this.metrics = metrics;
+ final String metricGroupName = metricGrpPrefix + CONSUMER_METRICS_SUFFIX;
+ this.timeBetweenNetworkThreadPollSensor = metrics.sensor(TIME_BETWEEN_NETWORK_THREAD_POLL_SENSOR_NAME);
+ this.timeBetweenNetworkThreadPollSensor.add(metrics.metricName("time-between-network-thread-poll-avg",
+ metricGroupName,
+ "The average time taken, in milliseconds, between each poll in the network thread."),
+ new Avg());
+ this.timeBetweenNetworkThreadPollSensor.add(metrics.metricName("time-between-network-thread-poll-max",
+ metricGroupName,
+ "The maximum time taken, in milliseconds, between each poll in the network thread."),
+ new Max());
+
+ 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."),
+ new Value());
+
+ this.applicationEventQueueTimeSensor = metrics.sensor(APPLICATION_EVENT_QUEUE_TIME_SENSOR_NAME);
+ this.applicationEventQueueTimeSensor.add(metrics.metricName("application-event-queue-time-avg",
+ metricGroupName,
+ "The average time, in milliseconds, that application events are taking to be dequeued."),
+ new Avg());
+ this.applicationEventQueueTimeSensor.add(metrics.metricName("application-event-queue-time-max",
+ metricGroupName,
+ "The maximum time, in milliseconds, that an application event took to be dequeued."),
+ new Max());
+
+ this.applicationEventQueueProcessingTimeSensor = metrics.sensor(APPLICATION_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME);
+ this.applicationEventQueueProcessingTimeSensor.add(metrics.metricName("application-event-queue-processing-time-avg",
+ metricGroupName,
+ "The average time, in milliseconds, that the consumer network takes to process all available application events."),
+ new Avg());
+ this.applicationEventQueueProcessingTimeSensor.add(metrics.metricName("application-event-queue-processing-time-max",
+ metricGroupName,
+ "The maximum time, in milliseconds, that the consumer network took to process all available application events."),
+ new Max());
+
+ this.unsentRequestsQueueSizeSensor = metrics.sensor(UNSENT_REQUESTS_QUEUE_SIZE_SENSOR_NAME);
+ this.unsentRequestsQueueSizeSensor.add(metrics.metricName("unsent-requests-queue-size",
+ metricGroupName,
+ "The current number of unsent requests in the consumer network."),
+ new Value());
+
+ this.unsentRequestsQueueTimeSensor = metrics.sensor(UNSENT_REQUESTS_QUEUE_TIME_SENSOR_NAME);
+ this.unsentRequestsQueueTimeSensor.add(metrics.metricName("unsent-requests-queue-time-avg",
+ metricGroupName,
+ "The average time, in milliseconds, that requests are taking to be sent in the consumer network."),
+ new Avg());
+ this.unsentRequestsQueueTimeSensor.add(metrics.metricName("unsent-requests-queue-time-max",
+ metricGroupName,
+ "The maximum time, in milliseconds, that a request remained unsent in the consumer network."),
+ new Max());
+
+ this.backgroundEventQueueSizeSensor = metrics.sensor(BACKGROUND_EVENT_QUEUE_SIZE_SENSOR_NAME);
+ this.backgroundEventQueueSizeSensor.add(metrics.metricName("background-event-queue-size",
+ metricGroupName,
+ "The current number of events in the consumer background event queue."),
+ new Value());
+
+ this.backgroundEventQueueTimeSensor = metrics.sensor(BACKGROUND_EVENT_QUEUE_TIME_SENSOR_NAME);
+ this.backgroundEventQueueTimeSensor.add(metrics.metricName("background-event-queue-time-avg",
+ metricGroupName,
+ "The average time, in milliseconds, that background events are taking to be dequeued."),
+ new Avg());
+ this.backgroundEventQueueTimeSensor.add(metrics.metricName("background-event-queue-time-max",
+ metricGroupName,
+ "The maximum time, in milliseconds, that background events are taking to be dequeued."),
+ new Max());
+
+ this.backgroundEventQueueProcessingTimeSensor = metrics.sensor(BACKGROUND_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME);
+ this.backgroundEventQueueProcessingTimeSensor.add(metrics.metricName("background-event-queue-processing-time-avg",
+ metricGroupName,
+ "The average time, in milliseconds, that the consumer took to process all available background events."),
+ new Avg());
+ this.backgroundEventQueueProcessingTimeSensor.add(metrics.metricName("background-event-queue-processing-time-max",
+ metricGroupName,
+ "The maximum time, in milliseconds, that the consumer took to process all available background events."),
+ new Max());
+ }
+
+ public void recordTimeBetweenNetworkThreadPoll(long timeBetweenNetworkThreadPoll) {
+ this.timeBetweenNetworkThreadPollSensor.record(timeBetweenNetworkThreadPoll);
+ }
+
+ public void recordApplicationEventQueueSize(int size) {
+ this.applicationEventQueueSizeSensor.record(size);
+ }
+
+ public void recordApplicationEventQueueTime(long time) {
+ this.applicationEventQueueTimeSensor.record(time);
+ }
+
+ public void recordApplicationEventQueueProcessingTime(long processingTime) {
+ this.applicationEventQueueProcessingTimeSensor.record(processingTime);
+ }
+
+ public void recordUnsentRequestsQueueSize(int size) {
Review Comment:
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.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java:
##########
@@ -170,6 +174,7 @@ private void trySend(final long currentTimeMs) {
Iterator<UnsentRequest> iterator = unsentRequests.iterator();
while (iterator.hasNext()) {
UnsentRequest unsent = iterator.next();
+ kafkaConsumerMetrics.ifPresent(metrics -> metrics.recordUnsentRequestsQueueTime(currentTimeMs - unsent.enqueuedMs()));
Review Comment:
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?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1911,6 +1922,8 @@ private boolean processBackgroundEvents() {
if (!firstError.compareAndSet(null, e))
log.warn("An error occurred when processing the background event: {}", e.getMessage(), e);
+ } finally {
+ kafkaAsyncConsumerMetrics.recordBackgroundEventQueueProcessingTime(time.milliseconds() - startMs);
Review Comment:
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`?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1893,14 +1899,19 @@ private void subscribeInternal(Collection<String> topics, Optional<ConsumerRebal
* It is possible that {@link ErrorEvent an error}
* could occur when processing the events. In such cases, the processor will take a reference to the first
* error, continue to process the remaining events, and then throw the first error that occurred.
+ *
+ * Visible for testing.
*/
- private boolean processBackgroundEvents() {
+ boolean processBackgroundEvents() {
AtomicReference<KafkaException> firstError = new AtomicReference<>();
LinkedList<BackgroundEvent> events = new LinkedList<>();
backgroundEventQueue.drainTo(events);
+ kafkaAsyncConsumerMetrics.recordBackgroundEventQueueSize(backgroundEventQueue.size());
Review Comment:
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)
```suggestion
LinkedList<BackgroundEvent> events = backgroundEventHandler.drainBackgroundEvents();
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/metrics/KafkaAsyncConsumerMetrics.java:
##########
@@ -0,0 +1,186 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer.internals.metrics;
+
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.metrics.stats.Value;
+
+import java.util.Arrays;
+
+import static org.apache.kafka.clients.consumer.internals.ConsumerUtils.CONSUMER_METRICS_SUFFIX;
+
+public class KafkaAsyncConsumerMetrics extends KafkaConsumerMetrics implements AutoCloseable {
+ private final Metrics metrics;
+
+ public static final String TIME_BETWEEN_NETWORK_THREAD_POLL_SENSOR_NAME = "time-between-network-thread-poll";
+ public static final String APPLICATION_EVENT_QUEUE_SIZE_SENSOR_NAME = "application-event-queue-size";
+ public static final String APPLICATION_EVENT_QUEUE_TIME_SENSOR_NAME = "application-event-queue-time";
+ public static final String APPLICATION_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME = "application-event-queue-processing-time";
+ public static final String BACKGROUND_EVENT_QUEUE_SIZE_SENSOR_NAME = "background-event-queue-size";
+ public static final String BACKGROUND_EVENT_QUEUE_TIME_SENSOR_NAME = "background-event-queue-time";
+ public static final String BACKGROUND_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME = "background-event-queue-processing-time";
+ 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;
Review Comment:
nit: extra space before Sensor
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java:
##########
@@ -267,7 +272,9 @@ public void addAll(final List<UnsentRequest> requests) {
public void add(final UnsentRequest r) {
Objects.requireNonNull(r);
r.setTimer(this.time, this.requestTimeoutMs);
+ r.setEnqueuedMs(this.time.milliseconds());
unsentRequests.add(r);
+ kafkaConsumerMetrics.ifPresent(metrics -> metrics.recordUnsentRequestsQueueSize(unsentRequests.size()));
Review Comment:
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?
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1882373644
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/metrics/AsyncConsumerMetrics.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer.internals.metrics;
+
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.metrics.stats.Value;
+
+import java.util.Arrays;
+
+import static org.apache.kafka.clients.consumer.internals.ConsumerUtils.CONSUMER_METRICS_SUFFIX;
+
+public class AsyncConsumerMetrics extends KafkaConsumerMetrics implements AutoCloseable {
+ private final Metrics metrics;
+
+ public static final String TIME_BETWEEN_NETWORK_THREAD_POLL_SENSOR_NAME = "time-between-network-thread-poll";
+ public static final String APPLICATION_EVENT_QUEUE_SIZE_SENSOR_NAME = "application-event-queue-size";
+ public static final String APPLICATION_EVENT_QUEUE_TIME_SENSOR_NAME = "application-event-queue-time";
+ public static final String APPLICATION_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME = "application-event-queue-processing-time";
+ public static final String APPLICATION_EVENT_EXPIRED_SIZE_SENSOR_NAME = "application-event-expired-size";
+ public static final String BACKGROUND_EVENT_QUEUE_SIZE_SENSOR_NAME = "background-event-queue-size";
+ public static final String BACKGROUND_EVENT_QUEUE_TIME_SENSOR_NAME = "background-event-queue-time";
+ public static final String BACKGROUND_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME = "background-event-queue-processing-time";
+ 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;
+ private final Sensor applicationEventQueueTimeSensor;
+ private final Sensor applicationEventQueueProcessingTimeSensor;
+ private final Sensor applicationEventExpiredSizeSensor;
+ private final Sensor backgroundEventQueueSizeSensor;
+ private final Sensor backgroundEventQueueTimeSensor;
+ private final Sensor backgroundEventQueueProcessingTimeSensor;
+ private final Sensor unsentRequestsQueueSizeSensor;
+ private final Sensor unsentRequestsQueueTimeSensor;
+
+ public AsyncConsumerMetrics(Metrics metrics, String metricGrpPrefix) {
Review Comment:
sounds good. Please also merge trunk latest changes when you address what't left for this PR. Thanks!
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1802621948
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1740,14 +1742,19 @@ private void subscribeInternal(Collection<String> topics, Optional<ConsumerRebal
* It is possible that {@link ErrorEvent an error}
* could occur when processing the events. In such cases, the processor will take a reference to the first
* error, continue to process the remaining events, and then throw the first error that occurred.
+ *
+ * Visible for testing.
*/
- private boolean processBackgroundEvents() {
+ boolean processBackgroundEvents() {
AtomicReference<KafkaException> firstError = new AtomicReference<>();
LinkedList<BackgroundEvent> events = new LinkedList<>();
backgroundEventQueue.drainTo(events);
+ kafkaConsumerMetrics.recordBackgroundEventQueueSize(backgroundEventQueue.size());
for (BackgroundEvent event : events) {
+ kafkaConsumerMetrics.recordBackgroundEventQueueTime(time.milliseconds() - event.addedToQueueMs());
+ long startMs = time.milliseconds();
Review Comment:
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.
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1858276205
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java:
##########
@@ -60,21 +61,24 @@ public class ConsumerNetworkThread extends KafkaThread implements Closeable {
private final Supplier<ApplicationEventProcessor> applicationEventProcessorSupplier;
private final Supplier<NetworkClientDelegate> networkClientDelegateSupplier;
private final Supplier<RequestManagers> requestManagersSupplier;
+ private final Optional<KafkaAsyncConsumerMetrics> kafkaAsyncConsumerMetrics;
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;
public ConsumerNetworkThread(LogContext logContext,
Time time,
BlockingQueue<ApplicationEvent> applicationEventQueue,
CompletableEventReaper applicationEventReaper,
Supplier<ApplicationEventProcessor> applicationEventProcessorSupplier,
Supplier<NetworkClientDelegate> networkClientDelegateSupplier,
- Supplier<RequestManagers> requestManagersSupplier) {
+ Supplier<RequestManagers> requestManagersSupplier,
+ Optional<KafkaAsyncConsumerMetrics> kafkaAsyncConsumerMetrics) {
Review Comment:
Updated it. Thanks.
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1878111713
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/metrics/KafkaAsyncConsumerMetrics.java:
##########
@@ -0,0 +1,186 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer.internals.metrics;
+
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.metrics.stats.Value;
+
+import java.util.Arrays;
+
+import static org.apache.kafka.clients.consumer.internals.ConsumerUtils.CONSUMER_METRICS_SUFFIX;
+
+public class KafkaAsyncConsumerMetrics extends KafkaConsumerMetrics implements AutoCloseable {
Review Comment:
Hi @AndrewJSchofield, thanks for the suggestion. If we want to do this change, we also need to add a new function `asyncConsumerMetrics` to `KafkaConsumer` like `KafkaConsumer#kafkaConsumerMetrics`. Before I update the PR, I would like to confirm we are on the same page. Thanks.
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1850051456
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java:
##########
@@ -173,6 +178,7 @@ private void trySend(final long currentTimeMs) {
unsent.timer.update(currentTimeMs);
if (unsent.timer.isExpired()) {
iterator.remove();
+ kafkaConsumerMetrics.ifPresent(metrics -> metrics.recordUnsentRequestsQueueTime(this.time.milliseconds() - unsent.enqueuedMs()));
Review Comment:
Hi @lianetm, we may need to keep `time.milliseconds()` here. The `currentTimeMs` is set when `runOnce` starts. However, the `enqueueTimeMs` (original `enqueuedMs`) is set when the request is added to the `unsentRequests`. It's after `currentTimeMs`. We will get negative result if using `currentTimeMs` here.
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1848346990
##########
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);
+ kafkaAsyncConsumerMetrics.ifPresent(metrics -> metrics.recordApplicationEventQueueSize(0));
Review Comment:
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.
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on PR #17199:
URL: https://github.com/apache/kafka/pull/17199#issuecomment-2485708134
Hi @lianetm, thanks for prompt review. I address all comments and leave message under some thread. Feel free to review it when you have time. Thank you.
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #17199:
URL: https://github.com/apache/kafka/pull/17199#issuecomment-2484145198
Hey @FrankYang0529, sorry I haven't had the bandwidth for this. I will be taking a look this week. Thanks!
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1858281950
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1893,25 +1901,30 @@ private void subscribeInternal(Collection<String> topics, Optional<ConsumerRebal
* It is possible that {@link ErrorEvent an error}
* could occur when processing the events. In such cases, the processor will take a reference to the first
* error, continue to process the remaining events, and then throw the first error that occurred.
+ *
+ * Visible for testing.
*/
- private boolean processBackgroundEvents() {
+ boolean processBackgroundEvents() {
AtomicReference<KafkaException> firstError = new AtomicReference<>();
- LinkedList<BackgroundEvent> events = new LinkedList<>();
- backgroundEventQueue.drainTo(events);
-
- for (BackgroundEvent event : events) {
- try {
- if (event instanceof CompletableEvent)
- backgroundEventReaper.add((CompletableEvent<?>) event);
-
- backgroundEventProcessor.process(event);
- } catch (Throwable t) {
- KafkaException e = ConsumerUtils.maybeWrapAsKafkaException(t);
-
- if (!firstError.compareAndSet(null, e))
- log.warn("An error occurred when processing the background event: {}", e.getMessage(), e);
+ List<BackgroundEvent> events = backgroundEventHandler.drainEvents();
+ if (!events.isEmpty()) {
+ long startMs = time.milliseconds();
+ for (BackgroundEvent event : events) {
+ kafkaConsumerMetrics.recordBackgroundEventQueueTime(time.milliseconds() - event.enqueuedMs());
+ try {
+ if (event instanceof CompletableEvent)
+ backgroundEventReaper.add((CompletableEvent<?>) event);
+
+ backgroundEventProcessor.process(event);
+ } catch (Throwable t) {
+ KafkaException e = ConsumerUtils.maybeWrapAsKafkaException(t);
+
+ if (!firstError.compareAndSet(null, e))
+ log.warn("An error occurred when processing the background event: {}", e.getMessage(), e);
+ }
}
+ kafkaConsumerMetrics.recordBackgroundEventQueueProcessingTime(time.milliseconds() - startMs);
}
backgroundEventReaper.reap(time.milliseconds());
Review Comment:
I think we don't have this metric in original KIP. Do we want to add one?
https://cwiki.apache.org/confluence/display/KAFKA/KIP-1068%3A+New+metrics+for+the+new+KafkaConsumer
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1800283720
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1740,14 +1742,19 @@ private void subscribeInternal(Collection<String> topics, Optional<ConsumerRebal
* It is possible that {@link ErrorEvent an error}
* could occur when processing the events. In such cases, the processor will take a reference to the first
* error, continue to process the remaining events, and then throw the first error that occurred.
+ *
+ * Visible for testing.
*/
- private boolean processBackgroundEvents() {
+ boolean processBackgroundEvents() {
AtomicReference<KafkaException> firstError = new AtomicReference<>();
LinkedList<BackgroundEvent> events = new LinkedList<>();
backgroundEventQueue.drainTo(events);
+ kafkaConsumerMetrics.recordBackgroundEventQueueSize(backgroundEventQueue.size());
for (BackgroundEvent event : events) {
+ kafkaConsumerMetrics.recordBackgroundEventQueueTime(time.milliseconds() - event.addedToQueueMs());
+ long startMs = time.milliseconds();
Review Comment:
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.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java:
##########
@@ -85,6 +88,17 @@ public NetworkClientDelegate(
this.unsentRequests = new ArrayDeque<>();
this.requestTimeoutMs = config.getInt(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG);
this.retryBackoffMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MS_CONFIG);
+ this.kafkaConsumerMetrics = kafkaConsumerMetrics;
+ }
+
+ 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());
Review Comment:
Is it possible to remove this constructor? Can't callers invoke the existing constructor with `Optional.empty()`?
##########
clients/src/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());
Review Comment:
This needs to get the current time in milliseconds from the `Time` object that was passed in to the constructor.
##########
clients/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());
Review Comment:
Same comment here: we need to update the constructor to provide a `Time` object and then use that here.
##########
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;
Review Comment:
In the first invocation of `runOnce()`, `timeSinceLastPollMs` will be something like `1728954137284`. Is that correct?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java:
##########
@@ -317,6 +335,14 @@ Timer timer() {
return timer;
}
+ void setAddedToQueueMs(final long addedToQueueMs) {
+ this.addedToQueueMs = addedToQueueMs;
+ }
+
+ long addedToQueueMs() {
+ return addedToQueueMs;
+ }
+
Review Comment:
I know it's a bit nit-picky, but can we change it to:
```suggestion
void setEnqueuedMs(final long enqueuedMs) {
this.enqueuedMs = enqueuedMs;
}
long enqueuedMs() {
return enqueuedMs;
}
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java:
##########
@@ -84,6 +88,18 @@ public ConsumerNetworkThread(LogContext logContext,
this.networkClientDelegateSupplier = networkClientDelegateSupplier;
this.requestManagersSupplier = requestManagersSupplier;
this.running = true;
+ this.kafkaConsumerMetrics = kafkaConsumerMetrics;
+ }
+
+ 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());
Review Comment:
Is it possible to remove this constructor? Can't we call the existing constructor with `Optional.empty()`?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/BackgroundEventHandler.java:
##########
@@ -30,9 +32,16 @@
public class BackgroundEventHandler {
private final Queue<BackgroundEvent> backgroundEventQueue;
+ private final Optional<KafkaConsumerMetrics> kafkaConsumerMetrics;
- public BackgroundEventHandler(final Queue<BackgroundEvent> backgroundEventQueue) {
+ public BackgroundEventHandler(final Queue<BackgroundEvent> backgroundEventQueue,
+ final Optional<KafkaConsumerMetrics> kafkaConsumerMetrics) {
this.backgroundEventQueue = backgroundEventQueue;
+ this.kafkaConsumerMetrics = kafkaConsumerMetrics;
+ }
+
+ public BackgroundEventHandler(final Queue<BackgroundEvent> backgroundEventQueue) {
+ this(backgroundEventQueue, Optional.empty());
Review Comment:
Same with the other changes. I'd rather make the callers pass in `Optional.empty()`.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/BackgroundEvent.java:
##########
@@ -51,6 +52,14 @@ public Uuid id() {
return id;
}
+ public void setAddedToQueueMs(long addedToQueueMs) {
+ this.addedToQueueMs = addedToQueueMs;
+ }
+
+ public long addedToQueueMs() {
+ return addedToQueueMs;
+ }
Review Comment:
```suggestion
public void setEnqueuedMs(long enqueuedMs) {
this. enqueuedMs = enqueuedMs;
}
public long enqueuedMs() {
return enqueuedMs;
}
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEvent.java:
##########
@@ -61,6 +62,14 @@ public Uuid id() {
return id;
}
+ public void setAddedToQueueMs(long addedToQueueMs) {
+ this.addedToQueueMs = addedToQueueMs;
+ }
+
+ public long addedToQueueMs() {
+ return addedToQueueMs;
+ }
+
Review Comment:
```suggestion
public void setEnqueuedMs(long enqueuedMs) {
enqueuedMs = enqueuedMs;
}
public long enqueuedMs() {
return enqueuedMs;
}
```
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1849317079
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -213,10 +212,11 @@ private void process(final ConsumerRebalanceListenerCallbackNeededEvent event) {
private final ApplicationEventHandler applicationEventHandler;
private final Time time;
private final AtomicReference<Optional<ConsumerGroupMetadata>> groupMetadata = new AtomicReference<>(Optional.empty());
- private final KafkaConsumerMetrics kafkaConsumerMetrics;
+ private final KafkaAsyncConsumerMetrics kafkaAsyncConsumerMetrics;
Review Comment:
`<bike-shedding>`I'm not staunchly against changing it, but I think keeping the variable as `kafkaConsumerMetrics` is fine.`</bike-shedding>`
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventHandler.java:
##########
@@ -39,29 +43,36 @@
* An event handler that receives {@link ApplicationEvent application events} from the application thread which
* are then readable from the {@link ApplicationEventProcessor} in the {@link ConsumerNetworkThread network thread}.
*/
+@SuppressWarnings("this-escape")
public class ApplicationEventHandler implements Closeable {
private final Logger log;
+ private final Time time;
private final BlockingQueue<ApplicationEvent> applicationEventQueue;
private final ConsumerNetworkThread networkThread;
private final IdempotentCloser closer = new IdempotentCloser();
+ private final Optional<KafkaAsyncConsumerMetrics> kafkaAsyncConsumerMetrics;
public ApplicationEventHandler(final LogContext logContext,
final Time time,
final BlockingQueue<ApplicationEvent> applicationEventQueue,
final CompletableEventReaper applicationEventReaper,
final Supplier<ApplicationEventProcessor> applicationEventProcessorSupplier,
final Supplier<NetworkClientDelegate> networkClientDelegateSupplier,
- final Supplier<RequestManagers> requestManagersSupplier) {
+ final Supplier<RequestManagers> requestManagersSupplier,
+ final Optional<KafkaAsyncConsumerMetrics> kafkaAsyncConsumerMetrics) {
this.log = logContext.logger(ApplicationEventHandler.class);
+ this.time = time;
this.applicationEventQueue = applicationEventQueue;
+ this.kafkaAsyncConsumerMetrics = kafkaAsyncConsumerMetrics;
this.networkThread = new ConsumerNetworkThread(logContext,
time,
- applicationEventQueue,
+ this,
Review Comment:
I'm not crazy about this. (Haha, pun intended.)
With this change there's now a bidirectional relationship between the event handler and the background thread. I'd hope we could find a way to maintain the decoupled nature of the event queue from the event handler 🤔
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java:
##########
@@ -267,7 +275,9 @@ public void addAll(final List<UnsentRequest> requests) {
public void add(final UnsentRequest r) {
Objects.requireNonNull(r);
r.setTimer(this.time, this.requestTimeoutMs);
+ r.setEnqueuedMs(this.time.milliseconds());
Review Comment:
Is it OK to use `this.time.milliseconds()` here? Either way, I think we can call `time.milliseconds()` once and reuse it in the metrics update too, right?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1913,6 +1923,7 @@ private boolean processBackgroundEvents() {
log.warn("An error occurred when processing the background event: {}", e.getMessage(), e);
}
}
+ kafkaAsyncConsumerMetrics.recordBackgroundEventQueueProcessingTime(time.milliseconds() - startMs);
Review Comment:
Good point. It'll artificially skew the averages lower.
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1849018100
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java:
##########
@@ -173,6 +181,7 @@ private void processApplicationEvents() {
log.warn("Error processing event {}", t.getMessage(), t);
}
}
+ kafkaAsyncConsumerMetrics.ifPresent(metrics -> metrics.recordApplicationEventQueueProcessingTime(time.milliseconds() - startMs));
Review Comment:
we probably don't want to record this if there were no events right? Maybe just an early return after drainEvents if there are none?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java:
##########
@@ -173,6 +178,7 @@ private void trySend(final long currentTimeMs) {
unsent.timer.update(currentTimeMs);
if (unsent.timer.isExpired()) {
iterator.remove();
+ kafkaConsumerMetrics.ifPresent(metrics -> metrics.recordUnsentRequestsQueueTime(this.time.milliseconds() - unsent.enqueuedMs()));
Review Comment:
shouldn't we use `currentTimeMs` instead of `this.time.milliseconds()`?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventHandler.java:
##########
@@ -73,10 +84,19 @@ public ApplicationEventHandler(final LogContext logContext,
*/
public void add(final ApplicationEvent event) {
Objects.requireNonNull(event, "ApplicationEvent provided to add must be non-null");
+ event.setEnqueuedMs(time.milliseconds());
applicationEventQueue.add(event);
+ kafkaAsyncConsumerMetrics.ifPresent(metrics -> metrics.recordApplicationEventQueueSize(applicationEventQueue.size()));
wakeupNetworkThread();
}
+ public List<ApplicationEvent> drainEvents() {
Review Comment:
could you add a javadoc for this pls
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java:
##########
@@ -183,6 +189,7 @@ private void trySend(final long currentTimeMs) {
continue;
}
iterator.remove();
+ kafkaConsumerMetrics.ifPresent(metrics -> metrics.recordUnsentRequestsQueueTime(this.time.milliseconds() - unsent.enqueuedMs()));
Review Comment:
`currentTimeMs`?
##########
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);
+ kafkaAsyncConsumerMetrics.ifPresent(metrics -> metrics.recordApplicationEventQueueSize(0));
Review Comment:
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?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1913,6 +1923,7 @@ private boolean processBackgroundEvents() {
log.warn("An error occurred when processing the background event: {}", e.getMessage(), e);
}
}
+ kafkaAsyncConsumerMetrics.recordBackgroundEventQueueProcessingTime(time.milliseconds() - startMs);
Review Comment:
we probably don't want to record this if there were no events right?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java:
##########
@@ -144,6 +148,7 @@ public void poll(final long timeoutMs, final long currentTimeMs) {
this.client.poll(pollTimeoutMs, currentTimeMs);
maybePropagateMetadataError();
checkDisconnects(currentTimeMs);
+ kafkaConsumerMetrics.ifPresent(metrics -> metrics.recordUnsentRequestsQueueSize(unsentRequests.size(), this.time.milliseconds()));
Review Comment:
should we use `currentTimeMs` instead of `this.time.milliseconds`?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java:
##########
@@ -317,6 +328,14 @@ Timer timer() {
return timer;
}
+ void setEnqueuedMs(final long enqueuedMs) {
Review Comment:
could add a java doc for this, the getter and var. Mainly to clarify its the time when it was added to the queue (not a duration/time it's been on the queue, which is another concept we are working with, that's why I think it's worth clarifying)....we could also consider `enqueueTimeMs` as name.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java:
##########
@@ -267,7 +272,9 @@ public void addAll(final List<UnsentRequest> requests) {
public void add(final UnsentRequest r) {
Objects.requireNonNull(r);
r.setTimer(this.time, this.requestTimeoutMs);
+ r.setEnqueuedMs(this.time.milliseconds());
unsentRequests.add(r);
+ kafkaConsumerMetrics.ifPresent(metrics -> metrics.recordUnsentRequestsQueueSize(unsentRequests.size()));
Review Comment:
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?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java:
##########
@@ -210,6 +217,7 @@ protected void checkDisconnects(final long currentTimeMs) {
UnsentRequest u = iter.next();
if (u.node.isPresent() && client.connectionFailed(u.node.get())) {
iter.remove();
+ kafkaConsumerMetrics.ifPresent(metrics -> metrics.recordUnsentRequestsQueueTime(this.time.milliseconds() - u.enqueuedMs()));
Review Comment:
`currentTimeMs`?
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1850055837
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java:
##########
@@ -267,7 +275,9 @@ public void addAll(final List<UnsentRequest> requests) {
public void add(final UnsentRequest r) {
Objects.requireNonNull(r);
r.setTimer(this.time, this.requestTimeoutMs);
+ r.setEnqueuedMs(this.time.milliseconds());
Review Comment:
After discussing with @lianetm, we may just want to record unsent request size at the end of `poll`, so I remove the metrics update here.
https://github.com/apache/kafka/pull/17199#discussion_r1849097694
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on PR #17199:
URL: https://github.com/apache/kafka/pull/17199#issuecomment-2514780896
Hi @AndrewJSchofield / @lianetm / @kirktrue, could you please review this PR when you have time? Thank you.
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1869492850
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java:
##########
@@ -60,21 +61,24 @@ public class ConsumerNetworkThread extends KafkaThread implements Closeable {
private final Supplier<ApplicationEventProcessor> applicationEventProcessorSupplier;
private final Supplier<NetworkClientDelegate> networkClientDelegateSupplier;
private final Supplier<RequestManagers> requestManagersSupplier;
+ private final Optional<KafkaConsumerMetrics> kafkaConsumerMetrics;
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;
Review Comment:
Hi @kirktrue, this is only used for metrics `time-between-network-thread-poll-max` and `time-between-network-thread-poll-avg`. Thanks.
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on PR #17199:
URL: https://github.com/apache/kafka/pull/17199#issuecomment-2517356209
Hi @kirktrue, I addressed all comments. Could you take a look when you have time? Thank you.
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "kirktrue (via GitHub)" <gi...@apache.org>.
kirktrue commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1879153121
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1893,25 +1901,30 @@ private void subscribeInternal(Collection<String> topics, Optional<ConsumerRebal
* It is possible that {@link ErrorEvent an error}
* could occur when processing the events. In such cases, the processor will take a reference to the first
* error, continue to process the remaining events, and then throw the first error that occurred.
+ *
+ * Visible for testing.
*/
- private boolean processBackgroundEvents() {
+ boolean processBackgroundEvents() {
AtomicReference<KafkaException> firstError = new AtomicReference<>();
- LinkedList<BackgroundEvent> events = new LinkedList<>();
- backgroundEventQueue.drainTo(events);
-
- for (BackgroundEvent event : events) {
- try {
- if (event instanceof CompletableEvent)
- backgroundEventReaper.add((CompletableEvent<?>) event);
-
- backgroundEventProcessor.process(event);
- } catch (Throwable t) {
- KafkaException e = ConsumerUtils.maybeWrapAsKafkaException(t);
-
- if (!firstError.compareAndSet(null, e))
- log.warn("An error occurred when processing the background event: {}", e.getMessage(), e);
+ List<BackgroundEvent> events = backgroundEventHandler.drainEvents();
+ if (!events.isEmpty()) {
+ long startMs = time.milliseconds();
+ for (BackgroundEvent event : events) {
+ kafkaConsumerMetrics.recordBackgroundEventQueueTime(time.milliseconds() - event.enqueuedMs());
+ try {
+ if (event instanceof CompletableEvent)
+ backgroundEventReaper.add((CompletableEvent<?>) event);
+
+ backgroundEventProcessor.process(event);
+ } catch (Throwable t) {
+ KafkaException e = ConsumerUtils.maybeWrapAsKafkaException(t);
+
+ if (!firstError.compareAndSet(null, e))
+ log.warn("An error occurred when processing the background event: {}", e.getMessage(), e);
+ }
}
+ kafkaConsumerMetrics.recordBackgroundEventQueueProcessingTime(time.milliseconds() - startMs);
}
backgroundEventReaper.reap(time.milliseconds());
Review Comment:
Yes, this sounds like a useful metric to have. Thanks!
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/metrics/AsyncConsumerMetrics.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer.internals.metrics;
+
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.metrics.stats.Value;
+
+import java.util.Arrays;
+
+import static org.apache.kafka.clients.consumer.internals.ConsumerUtils.CONSUMER_METRICS_SUFFIX;
+
+public class AsyncConsumerMetrics extends KafkaConsumerMetrics implements AutoCloseable {
+ private final Metrics metrics;
+
+ public static final String TIME_BETWEEN_NETWORK_THREAD_POLL_SENSOR_NAME = "time-between-network-thread-poll";
+ public static final String APPLICATION_EVENT_QUEUE_SIZE_SENSOR_NAME = "application-event-queue-size";
+ public static final String APPLICATION_EVENT_QUEUE_TIME_SENSOR_NAME = "application-event-queue-time";
+ public static final String APPLICATION_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME = "application-event-queue-processing-time";
+ public static final String APPLICATION_EVENT_EXPIRED_SIZE_SENSOR_NAME = "application-event-expired-size";
+ public static final String BACKGROUND_EVENT_QUEUE_SIZE_SENSOR_NAME = "background-event-queue-size";
+ public static final String BACKGROUND_EVENT_QUEUE_TIME_SENSOR_NAME = "background-event-queue-time";
+ public static final String BACKGROUND_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME = "background-event-queue-processing-time";
+ 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;
+ private final Sensor applicationEventQueueTimeSensor;
+ private final Sensor applicationEventQueueProcessingTimeSensor;
+ private final Sensor applicationEventExpiredSizeSensor;
+ private final Sensor backgroundEventQueueSizeSensor;
+ private final Sensor backgroundEventQueueTimeSensor;
+ private final Sensor backgroundEventQueueProcessingTimeSensor;
+ private final Sensor unsentRequestsQueueSizeSensor;
+ private final Sensor unsentRequestsQueueTimeSensor;
+
+ public AsyncConsumerMetrics(Metrics metrics, String metricGrpPrefix) {
Review Comment:
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?
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/metrics/AsyncConsumerMetricsTest.java:
##########
@@ -0,0 +1,198 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer.internals.metrics;
+
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.Metrics;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.HashSet;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class AsyncConsumerMetricsTest {
+ private static final long METRIC_VALUE = 123L;
+ private static final String CONSUMER_GROUP_PREFIX = "consumer";
+ private static final String CONSUMER_METRIC_GROUP = "consumer-metrics";
Review Comment:
Aren't these constants already declared elsewhere?
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/metrics/AsyncConsumerMetricsTest.java:
##########
@@ -0,0 +1,198 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer.internals.metrics;
+
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.Metrics;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.HashSet;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class AsyncConsumerMetricsTest {
+ private static final long METRIC_VALUE = 123L;
+ private static final String CONSUMER_GROUP_PREFIX = "consumer";
+ private static final String CONSUMER_METRIC_GROUP = "consumer-metrics";
+
+ private final Metrics metrics = new Metrics();
+ private AsyncConsumerMetrics consumerMetrics;
+
+ @AfterEach
+ public void tearDown() {
+ if (consumerMetrics != null) {
+ consumerMetrics.close();
+ }
+ metrics.close();
+ }
+
+ @Test
+ public void shouldMetricNames() {
+ // create
+ consumerMetrics = new AsyncConsumerMetrics(metrics, CONSUMER_GROUP_PREFIX);
+ HashSet<MetricName> expectedMetrics = new HashSet<>(Arrays.asList(
+ 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));
+
+ HashSet<MetricName> expectedConsumerMetrics = new HashSet<>(Arrays.asList(
+ metrics.metricName("time-between-network-thread-poll-avg", CONSUMER_METRIC_GROUP),
+ metrics.metricName("time-between-network-thread-poll-max", CONSUMER_METRIC_GROUP),
+ metrics.metricName("application-event-queue-size", CONSUMER_METRIC_GROUP),
+ metrics.metricName("application-event-queue-time-avg", CONSUMER_METRIC_GROUP),
+ metrics.metricName("application-event-queue-time-max", CONSUMER_METRIC_GROUP),
+ metrics.metricName("application-event-queue-processing-time-avg", CONSUMER_METRIC_GROUP),
+ metrics.metricName("application-event-queue-processing-time-max", CONSUMER_METRIC_GROUP),
+ metrics.metricName("unsent-requests-queue-size", CONSUMER_METRIC_GROUP),
+ metrics.metricName("unsent-requests-queue-time-avg", CONSUMER_METRIC_GROUP),
+ metrics.metricName("unsent-requests-queue-time-max", CONSUMER_METRIC_GROUP),
+ metrics.metricName("background-event-queue-size", CONSUMER_METRIC_GROUP),
+ metrics.metricName("background-event-queue-time-avg", CONSUMER_METRIC_GROUP),
+ metrics.metricName("background-event-queue-time-max", CONSUMER_METRIC_GROUP),
+ metrics.metricName("background-event-queue-processing-time-avg", CONSUMER_METRIC_GROUP),
+ metrics.metricName("background-event-queue-processing-time-max", CONSUMER_METRIC_GROUP)
+ ));
+ expectedConsumerMetrics.forEach(metricName -> assertTrue(metrics.metrics().containsKey(metricName), "Missing metric: " + metricName));
+
+ // close
+ consumerMetrics.close();
+ expectedMetrics.forEach(metricName -> assertFalse(metrics.metrics().containsKey(metricName), "Missing metric: " + metricName));
+ expectedConsumerMetrics.forEach(metricName -> assertFalse(metrics.metrics().containsKey(metricName), "Missing metric: " + metricName));
Review Comment:
The error message on test failure is incorrect here, right? Shouldn't it be:
```suggestion
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));
```
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1879255266
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1893,25 +1901,30 @@ private void subscribeInternal(Collection<String> topics, Optional<ConsumerRebal
* It is possible that {@link ErrorEvent an error}
* could occur when processing the events. In such cases, the processor will take a reference to the first
* error, continue to process the remaining events, and then throw the first error that occurred.
+ *
+ * Visible for testing.
*/
- private boolean processBackgroundEvents() {
+ boolean processBackgroundEvents() {
AtomicReference<KafkaException> firstError = new AtomicReference<>();
- LinkedList<BackgroundEvent> events = new LinkedList<>();
- backgroundEventQueue.drainTo(events);
-
- for (BackgroundEvent event : events) {
- try {
- if (event instanceof CompletableEvent)
- backgroundEventReaper.add((CompletableEvent<?>) event);
-
- backgroundEventProcessor.process(event);
- } catch (Throwable t) {
- KafkaException e = ConsumerUtils.maybeWrapAsKafkaException(t);
-
- if (!firstError.compareAndSet(null, e))
- log.warn("An error occurred when processing the background event: {}", e.getMessage(), e);
+ List<BackgroundEvent> events = backgroundEventHandler.drainEvents();
+ if (!events.isEmpty()) {
+ long startMs = time.milliseconds();
+ for (BackgroundEvent event : events) {
+ kafkaConsumerMetrics.recordBackgroundEventQueueTime(time.milliseconds() - event.enqueuedMs());
+ try {
+ if (event instanceof CompletableEvent)
+ backgroundEventReaper.add((CompletableEvent<?>) event);
+
+ backgroundEventProcessor.process(event);
+ } catch (Throwable t) {
+ KafkaException e = ConsumerUtils.maybeWrapAsKafkaException(t);
+
+ if (!firstError.compareAndSet(null, e))
+ log.warn("An error occurred when processing the background event: {}", e.getMessage(), e);
+ }
}
+ kafkaConsumerMetrics.recordBackgroundEventQueueProcessingTime(time.milliseconds() - startMs);
}
backgroundEventReaper.reap(time.milliseconds());
Review Comment:
makes sense to me too. I would just suggest we name it `application-events-expired-count`.
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1875064838
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -1893,25 +1901,30 @@ private void subscribeInternal(Collection<String> topics, Optional<ConsumerRebal
* It is possible that {@link ErrorEvent an error}
* could occur when processing the events. In such cases, the processor will take a reference to the first
* error, continue to process the remaining events, and then throw the first error that occurred.
+ *
+ * Visible for testing.
*/
- private boolean processBackgroundEvents() {
+ boolean processBackgroundEvents() {
AtomicReference<KafkaException> firstError = new AtomicReference<>();
- LinkedList<BackgroundEvent> events = new LinkedList<>();
- backgroundEventQueue.drainTo(events);
-
- for (BackgroundEvent event : events) {
- try {
- if (event instanceof CompletableEvent)
- backgroundEventReaper.add((CompletableEvent<?>) event);
-
- backgroundEventProcessor.process(event);
- } catch (Throwable t) {
- KafkaException e = ConsumerUtils.maybeWrapAsKafkaException(t);
-
- if (!firstError.compareAndSet(null, e))
- log.warn("An error occurred when processing the background event: {}", e.getMessage(), e);
+ List<BackgroundEvent> events = backgroundEventHandler.drainEvents();
+ if (!events.isEmpty()) {
+ long startMs = time.milliseconds();
+ for (BackgroundEvent event : events) {
+ kafkaConsumerMetrics.recordBackgroundEventQueueTime(time.milliseconds() - event.enqueuedMs());
+ try {
+ if (event instanceof CompletableEvent)
+ backgroundEventReaper.add((CompletableEvent<?>) event);
+
+ backgroundEventProcessor.process(event);
+ } catch (Throwable t) {
+ KafkaException e = ConsumerUtils.maybeWrapAsKafkaException(t);
+
+ if (!firstError.compareAndSet(null, e))
+ log.warn("An error occurred when processing the background event: {}", e.getMessage(), e);
+ }
}
+ kafkaConsumerMetrics.recordBackgroundEventQueueProcessingTime(time.milliseconds() - startMs);
}
backgroundEventReaper.reap(time.milliseconds());
Review Comment:
Agree that it's not applicable for background events because this callback needed is the only `CompletableBackgroundEvent`, and it's intentionally not expired (I don't think we need to change that)
I would say that what might be interesting to know is expiration of Application events.
https://github.com/apache/kafka/blob/ee4264439ddda7bdebcaa845752b824abba14161/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java#L184
There we do have lots of events with deadline, I guess that's what @AndrewJSchofield had in mind maybe? (I notice now that the initial comment was here on the reap of background events).
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on PR #17199:
URL: https://github.com/apache/kafka/pull/17199#issuecomment-2528138109
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!
--
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-16143: New JMX metrics for AsyncKafkaConsumer [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #17199:
URL: https://github.com/apache/kafka/pull/17199#discussion_r1880793583
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/metrics/AsyncConsumerMetrics.java:
##########
@@ -0,0 +1,262 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer.internals.metrics;
+
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.metrics.stats.Value;
+
+import java.util.Arrays;
+
+import static org.apache.kafka.clients.consumer.internals.ConsumerUtils.CONSUMER_METRIC_GROUP;
+import static org.apache.kafka.clients.consumer.internals.ConsumerUtils.CONSUMER_METRIC_GROUP_PREFIX;
+
+public class AsyncConsumerMetrics extends KafkaConsumerMetrics implements AutoCloseable {
+ private final Metrics metrics;
+
+ public static final String TIME_BETWEEN_NETWORK_THREAD_POLL_SENSOR_NAME = "time-between-network-thread-poll";
+ public static final String APPLICATION_EVENT_QUEUE_SIZE_SENSOR_NAME = "application-event-queue-size";
+ public static final String APPLICATION_EVENT_QUEUE_TIME_SENSOR_NAME = "application-event-queue-time";
+ public static final String APPLICATION_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME = "application-event-queue-processing-time";
+ public static final String APPLICATION_EVENT_EXPIRED_SIZE_SENSOR_NAME = "application-events-expired-count";
+ public static final String BACKGROUND_EVENT_QUEUE_SIZE_SENSOR_NAME = "background-event-queue-size";
+ public static final String BACKGROUND_EVENT_QUEUE_TIME_SENSOR_NAME = "background-event-queue-time";
+ public static final String BACKGROUND_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME = "background-event-queue-processing-time";
+ 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;
+ private final Sensor applicationEventQueueTimeSensor;
+ private final Sensor applicationEventQueueProcessingTimeSensor;
+ private final Sensor applicationEventExpiredSizeSensor;
+ private final Sensor backgroundEventQueueSizeSensor;
+ private final Sensor backgroundEventQueueTimeSensor;
+ private final Sensor backgroundEventQueueProcessingTimeSensor;
+ private final Sensor unsentRequestsQueueSizeSensor;
+ private final Sensor unsentRequestsQueueTimeSensor;
+
+ public AsyncConsumerMetrics(Metrics metrics) {
+ super(metrics, CONSUMER_METRIC_GROUP_PREFIX);
+
+ this.metrics = metrics;
+ this.timeBetweenNetworkThreadPollSensor = metrics.sensor(TIME_BETWEEN_NETWORK_THREAD_POLL_SENSOR_NAME);
+ this.timeBetweenNetworkThreadPollSensor.add(
+ metrics.metricName(
+ "time-between-network-thread-poll-avg",
+ CONSUMER_METRIC_GROUP,
+ "The average time taken, in milliseconds, between each poll in the network thread."
+ ),
+ new Avg()
+ );
+ this.timeBetweenNetworkThreadPollSensor.add(
+ metrics.metricName(
+ "time-between-network-thread-poll-max",
+ CONSUMER_METRIC_GROUP,
+ "The maximum time taken, in milliseconds, between each poll in the network thread."
+ ),
+ new Max()
+ );
+
+ this.applicationEventQueueSizeSensor = metrics.sensor(APPLICATION_EVENT_QUEUE_SIZE_SENSOR_NAME);
+ this.applicationEventQueueSizeSensor.add(
+ metrics.metricName(
+ APPLICATION_EVENT_QUEUE_SIZE_SENSOR_NAME,
+ CONSUMER_METRIC_GROUP,
+ "The current number of events in the queue to send from the application thread to the background thread."
+ ),
+ new Value()
+ );
+
+ this.applicationEventQueueTimeSensor = metrics.sensor(APPLICATION_EVENT_QUEUE_TIME_SENSOR_NAME);
+ this.applicationEventQueueTimeSensor.add(
+ metrics.metricName(
+ "application-event-queue-time-avg",
+ CONSUMER_METRIC_GROUP,
+ "The average time, in milliseconds, that application events are taking to be dequeued."
+ ),
+ new Avg()
+ );
+ this.applicationEventQueueTimeSensor.add(
+ metrics.metricName(
+ "application-event-queue-time-max",
+ CONSUMER_METRIC_GROUP,
+ "The maximum time, in milliseconds, that an application event took to be dequeued."
+ ),
+ new Max()
+ );
+
+ this.applicationEventQueueProcessingTimeSensor = metrics.sensor(APPLICATION_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME);
+ this.applicationEventQueueProcessingTimeSensor.add(
+ metrics.metricName(
+ "application-event-queue-processing-time-avg",
+ CONSUMER_METRIC_GROUP,
+ "The average time, in milliseconds, that the consumer network takes to process all available application events."
+ ),
+ new Avg()
+ );
+ this.applicationEventQueueProcessingTimeSensor.add(
+ metrics.metricName("application-event-queue-processing-time-max",
+ CONSUMER_METRIC_GROUP,
+ "The maximum time, in milliseconds, that the consumer network took to process all available application events."
+ ),
+ new Max()
+ );
+
+ this.applicationEventExpiredSizeSensor = metrics.sensor(APPLICATION_EVENT_EXPIRED_SIZE_SENSOR_NAME);
+ this.applicationEventExpiredSizeSensor.add(
+ metrics.metricName(
+ APPLICATION_EVENT_EXPIRED_SIZE_SENSOR_NAME,
+ CONSUMER_METRIC_GROUP,
+ "The current number of expired application events."
+ ),
+ new Value()
+ );
+
+ this.unsentRequestsQueueSizeSensor = metrics.sensor(UNSENT_REQUESTS_QUEUE_SIZE_SENSOR_NAME);
+ this.unsentRequestsQueueSizeSensor.add(
+ metrics.metricName(
+ UNSENT_REQUESTS_QUEUE_SIZE_SENSOR_NAME,
+ CONSUMER_METRIC_GROUP,
+ "The current number of unsent requests in the consumer network."
+ ),
+ new Value()
+ );
+
+ this.unsentRequestsQueueTimeSensor = metrics.sensor(UNSENT_REQUESTS_QUEUE_TIME_SENSOR_NAME);
+ this.unsentRequestsQueueTimeSensor.add(
+ metrics.metricName(
+ "unsent-requests-queue-time-avg",
+ CONSUMER_METRIC_GROUP,
+ "The average time, in milliseconds, that requests are taking to be sent in the consumer network."
+ ),
+ new Avg()
+ );
+ this.unsentRequestsQueueTimeSensor.add(
+ metrics.metricName(
+ "unsent-requests-queue-time-max",
+ CONSUMER_METRIC_GROUP,
+ "The maximum time, in milliseconds, that a request remained unsent in the consumer network."
+ ),
+ new Max()
+ );
+
+ this.backgroundEventQueueSizeSensor = metrics.sensor(BACKGROUND_EVENT_QUEUE_SIZE_SENSOR_NAME);
+ this.backgroundEventQueueSizeSensor.add(
+ metrics.metricName(
+ BACKGROUND_EVENT_QUEUE_SIZE_SENSOR_NAME,
+ CONSUMER_METRIC_GROUP,
+ "The current number of events in the consumer background event queue."
Review Comment:
nit: for clarity and consistency with the app thread metric, "The current number of events in the queue to send from the background thread to the application thread."
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/metrics/AsyncConsumerMetrics.java:
##########
@@ -0,0 +1,262 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer.internals.metrics;
+
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.metrics.stats.Value;
+
+import java.util.Arrays;
+
+import static org.apache.kafka.clients.consumer.internals.ConsumerUtils.CONSUMER_METRIC_GROUP;
+import static org.apache.kafka.clients.consumer.internals.ConsumerUtils.CONSUMER_METRIC_GROUP_PREFIX;
+
+public class AsyncConsumerMetrics extends KafkaConsumerMetrics implements AutoCloseable {
+ private final Metrics metrics;
+
+ public static final String TIME_BETWEEN_NETWORK_THREAD_POLL_SENSOR_NAME = "time-between-network-thread-poll";
+ public static final String APPLICATION_EVENT_QUEUE_SIZE_SENSOR_NAME = "application-event-queue-size";
+ public static final String APPLICATION_EVENT_QUEUE_TIME_SENSOR_NAME = "application-event-queue-time";
+ public static final String APPLICATION_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME = "application-event-queue-processing-time";
+ public static final String APPLICATION_EVENT_EXPIRED_SIZE_SENSOR_NAME = "application-events-expired-count";
+ public static final String BACKGROUND_EVENT_QUEUE_SIZE_SENSOR_NAME = "background-event-queue-size";
+ public static final String BACKGROUND_EVENT_QUEUE_TIME_SENSOR_NAME = "background-event-queue-time";
+ public static final String BACKGROUND_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME = "background-event-queue-processing-time";
+ 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;
+ private final Sensor applicationEventQueueTimeSensor;
+ private final Sensor applicationEventQueueProcessingTimeSensor;
+ private final Sensor applicationEventExpiredSizeSensor;
+ private final Sensor backgroundEventQueueSizeSensor;
+ private final Sensor backgroundEventQueueTimeSensor;
+ private final Sensor backgroundEventQueueProcessingTimeSensor;
+ private final Sensor unsentRequestsQueueSizeSensor;
+ private final Sensor unsentRequestsQueueTimeSensor;
+
+ public AsyncConsumerMetrics(Metrics metrics) {
+ super(metrics, CONSUMER_METRIC_GROUP_PREFIX);
+
+ this.metrics = metrics;
+ this.timeBetweenNetworkThreadPollSensor = metrics.sensor(TIME_BETWEEN_NETWORK_THREAD_POLL_SENSOR_NAME);
+ this.timeBetweenNetworkThreadPollSensor.add(
+ metrics.metricName(
+ "time-between-network-thread-poll-avg",
+ CONSUMER_METRIC_GROUP,
+ "The average time taken, in milliseconds, between each poll in the network thread."
+ ),
+ new Avg()
+ );
+ this.timeBetweenNetworkThreadPollSensor.add(
+ metrics.metricName(
+ "time-between-network-thread-poll-max",
+ CONSUMER_METRIC_GROUP,
+ "The maximum time taken, in milliseconds, between each poll in the network thread."
+ ),
+ new Max()
+ );
+
+ this.applicationEventQueueSizeSensor = metrics.sensor(APPLICATION_EVENT_QUEUE_SIZE_SENSOR_NAME);
+ this.applicationEventQueueSizeSensor.add(
+ metrics.metricName(
+ APPLICATION_EVENT_QUEUE_SIZE_SENSOR_NAME,
+ CONSUMER_METRIC_GROUP,
+ "The current number of events in the queue to send from the application thread to the background thread."
+ ),
+ new Value()
+ );
+
+ this.applicationEventQueueTimeSensor = metrics.sensor(APPLICATION_EVENT_QUEUE_TIME_SENSOR_NAME);
+ this.applicationEventQueueTimeSensor.add(
+ metrics.metricName(
+ "application-event-queue-time-avg",
+ CONSUMER_METRIC_GROUP,
+ "The average time, in milliseconds, that application events are taking to be dequeued."
+ ),
+ new Avg()
+ );
+ this.applicationEventQueueTimeSensor.add(
+ metrics.metricName(
+ "application-event-queue-time-max",
+ CONSUMER_METRIC_GROUP,
+ "The maximum time, in milliseconds, that an application event took to be dequeued."
+ ),
+ new Max()
+ );
+
+ this.applicationEventQueueProcessingTimeSensor = metrics.sensor(APPLICATION_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME);
+ this.applicationEventQueueProcessingTimeSensor.add(
+ metrics.metricName(
+ "application-event-queue-processing-time-avg",
+ CONSUMER_METRIC_GROUP,
+ "The average time, in milliseconds, that the consumer network takes to process all available application events."
+ ),
+ new Avg()
+ );
+ this.applicationEventQueueProcessingTimeSensor.add(
+ metrics.metricName("application-event-queue-processing-time-max",
+ CONSUMER_METRIC_GROUP,
+ "The maximum time, in milliseconds, that the consumer network took to process all available application events."
Review Comment:
ditto
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/metrics/AsyncConsumerMetrics.java:
##########
@@ -0,0 +1,262 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer.internals.metrics;
+
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.metrics.stats.Value;
+
+import java.util.Arrays;
+
+import static org.apache.kafka.clients.consumer.internals.ConsumerUtils.CONSUMER_METRIC_GROUP;
+import static org.apache.kafka.clients.consumer.internals.ConsumerUtils.CONSUMER_METRIC_GROUP_PREFIX;
+
+public class AsyncConsumerMetrics extends KafkaConsumerMetrics implements AutoCloseable {
+ private final Metrics metrics;
+
+ public static final String TIME_BETWEEN_NETWORK_THREAD_POLL_SENSOR_NAME = "time-between-network-thread-poll";
+ public static final String APPLICATION_EVENT_QUEUE_SIZE_SENSOR_NAME = "application-event-queue-size";
+ public static final String APPLICATION_EVENT_QUEUE_TIME_SENSOR_NAME = "application-event-queue-time";
+ public static final String APPLICATION_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME = "application-event-queue-processing-time";
+ public static final String APPLICATION_EVENT_EXPIRED_SIZE_SENSOR_NAME = "application-events-expired-count";
+ public static final String BACKGROUND_EVENT_QUEUE_SIZE_SENSOR_NAME = "background-event-queue-size";
+ public static final String BACKGROUND_EVENT_QUEUE_TIME_SENSOR_NAME = "background-event-queue-time";
+ public static final String BACKGROUND_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME = "background-event-queue-processing-time";
+ 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;
+ private final Sensor applicationEventQueueTimeSensor;
+ private final Sensor applicationEventQueueProcessingTimeSensor;
+ private final Sensor applicationEventExpiredSizeSensor;
+ private final Sensor backgroundEventQueueSizeSensor;
+ private final Sensor backgroundEventQueueTimeSensor;
+ private final Sensor backgroundEventQueueProcessingTimeSensor;
+ private final Sensor unsentRequestsQueueSizeSensor;
+ private final Sensor unsentRequestsQueueTimeSensor;
+
+ public AsyncConsumerMetrics(Metrics metrics) {
+ super(metrics, CONSUMER_METRIC_GROUP_PREFIX);
+
+ this.metrics = metrics;
+ this.timeBetweenNetworkThreadPollSensor = metrics.sensor(TIME_BETWEEN_NETWORK_THREAD_POLL_SENSOR_NAME);
+ this.timeBetweenNetworkThreadPollSensor.add(
+ metrics.metricName(
+ "time-between-network-thread-poll-avg",
+ CONSUMER_METRIC_GROUP,
+ "The average time taken, in milliseconds, between each poll in the network thread."
+ ),
+ new Avg()
+ );
+ this.timeBetweenNetworkThreadPollSensor.add(
+ metrics.metricName(
+ "time-between-network-thread-poll-max",
+ CONSUMER_METRIC_GROUP,
+ "The maximum time taken, in milliseconds, between each poll in the network thread."
+ ),
+ new Max()
+ );
+
+ this.applicationEventQueueSizeSensor = metrics.sensor(APPLICATION_EVENT_QUEUE_SIZE_SENSOR_NAME);
+ this.applicationEventQueueSizeSensor.add(
+ metrics.metricName(
+ APPLICATION_EVENT_QUEUE_SIZE_SENSOR_NAME,
+ CONSUMER_METRIC_GROUP,
+ "The current number of events in the queue to send from the application thread to the background thread."
+ ),
+ new Value()
+ );
+
+ this.applicationEventQueueTimeSensor = metrics.sensor(APPLICATION_EVENT_QUEUE_TIME_SENSOR_NAME);
+ this.applicationEventQueueTimeSensor.add(
+ metrics.metricName(
+ "application-event-queue-time-avg",
+ CONSUMER_METRIC_GROUP,
+ "The average time, in milliseconds, that application events are taking to be dequeued."
+ ),
+ new Avg()
+ );
+ this.applicationEventQueueTimeSensor.add(
+ metrics.metricName(
+ "application-event-queue-time-max",
+ CONSUMER_METRIC_GROUP,
+ "The maximum time, in milliseconds, that an application event took to be dequeued."
+ ),
+ new Max()
+ );
+
+ this.applicationEventQueueProcessingTimeSensor = metrics.sensor(APPLICATION_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME);
+ this.applicationEventQueueProcessingTimeSensor.add(
+ metrics.metricName(
+ "application-event-queue-processing-time-avg",
+ CONSUMER_METRIC_GROUP,
+ "The average time, in milliseconds, that the consumer network takes to process all available application events."
+ ),
+ new Avg()
+ );
+ this.applicationEventQueueProcessingTimeSensor.add(
+ metrics.metricName("application-event-queue-processing-time-max",
+ CONSUMER_METRIC_GROUP,
+ "The maximum time, in milliseconds, that the consumer network took to process all available application events."
+ ),
+ new Max()
+ );
+
+ this.applicationEventExpiredSizeSensor = metrics.sensor(APPLICATION_EVENT_EXPIRED_SIZE_SENSOR_NAME);
+ this.applicationEventExpiredSizeSensor.add(
+ metrics.metricName(
+ APPLICATION_EVENT_EXPIRED_SIZE_SENSOR_NAME,
+ CONSUMER_METRIC_GROUP,
+ "The current number of expired application events."
+ ),
+ new Value()
+ );
+
+ this.unsentRequestsQueueSizeSensor = metrics.sensor(UNSENT_REQUESTS_QUEUE_SIZE_SENSOR_NAME);
+ this.unsentRequestsQueueSizeSensor.add(
+ metrics.metricName(
+ UNSENT_REQUESTS_QUEUE_SIZE_SENSOR_NAME,
+ CONSUMER_METRIC_GROUP,
+ "The current number of unsent requests in the consumer network."
+ ),
+ new Value()
+ );
+
+ this.unsentRequestsQueueTimeSensor = metrics.sensor(UNSENT_REQUESTS_QUEUE_TIME_SENSOR_NAME);
+ this.unsentRequestsQueueTimeSensor.add(
+ metrics.metricName(
+ "unsent-requests-queue-time-avg",
+ CONSUMER_METRIC_GROUP,
+ "The average time, in milliseconds, that requests are taking to be sent in the consumer network."
Review Comment:
ditto
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/metrics/AsyncConsumerMetrics.java:
##########
@@ -0,0 +1,262 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer.internals.metrics;
+
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.metrics.stats.Value;
+
+import java.util.Arrays;
+
+import static org.apache.kafka.clients.consumer.internals.ConsumerUtils.CONSUMER_METRIC_GROUP;
+import static org.apache.kafka.clients.consumer.internals.ConsumerUtils.CONSUMER_METRIC_GROUP_PREFIX;
+
+public class AsyncConsumerMetrics extends KafkaConsumerMetrics implements AutoCloseable {
+ private final Metrics metrics;
+
+ public static final String TIME_BETWEEN_NETWORK_THREAD_POLL_SENSOR_NAME = "time-between-network-thread-poll";
+ public static final String APPLICATION_EVENT_QUEUE_SIZE_SENSOR_NAME = "application-event-queue-size";
+ public static final String APPLICATION_EVENT_QUEUE_TIME_SENSOR_NAME = "application-event-queue-time";
+ public static final String APPLICATION_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME = "application-event-queue-processing-time";
+ public static final String APPLICATION_EVENT_EXPIRED_SIZE_SENSOR_NAME = "application-events-expired-count";
+ public static final String BACKGROUND_EVENT_QUEUE_SIZE_SENSOR_NAME = "background-event-queue-size";
+ public static final String BACKGROUND_EVENT_QUEUE_TIME_SENSOR_NAME = "background-event-queue-time";
+ public static final String BACKGROUND_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME = "background-event-queue-processing-time";
+ 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;
+ private final Sensor applicationEventQueueTimeSensor;
+ private final Sensor applicationEventQueueProcessingTimeSensor;
+ private final Sensor applicationEventExpiredSizeSensor;
+ private final Sensor backgroundEventQueueSizeSensor;
+ private final Sensor backgroundEventQueueTimeSensor;
+ private final Sensor backgroundEventQueueProcessingTimeSensor;
+ private final Sensor unsentRequestsQueueSizeSensor;
+ private final Sensor unsentRequestsQueueTimeSensor;
+
+ public AsyncConsumerMetrics(Metrics metrics) {
+ super(metrics, CONSUMER_METRIC_GROUP_PREFIX);
+
+ this.metrics = metrics;
+ this.timeBetweenNetworkThreadPollSensor = metrics.sensor(TIME_BETWEEN_NETWORK_THREAD_POLL_SENSOR_NAME);
+ this.timeBetweenNetworkThreadPollSensor.add(
+ metrics.metricName(
+ "time-between-network-thread-poll-avg",
+ CONSUMER_METRIC_GROUP,
+ "The average time taken, in milliseconds, between each poll in the network thread."
+ ),
+ new Avg()
+ );
+ this.timeBetweenNetworkThreadPollSensor.add(
+ metrics.metricName(
+ "time-between-network-thread-poll-max",
+ CONSUMER_METRIC_GROUP,
+ "The maximum time taken, in milliseconds, between each poll in the network thread."
+ ),
+ new Max()
+ );
+
+ this.applicationEventQueueSizeSensor = metrics.sensor(APPLICATION_EVENT_QUEUE_SIZE_SENSOR_NAME);
+ this.applicationEventQueueSizeSensor.add(
+ metrics.metricName(
+ APPLICATION_EVENT_QUEUE_SIZE_SENSOR_NAME,
+ CONSUMER_METRIC_GROUP,
+ "The current number of events in the queue to send from the application thread to the background thread."
+ ),
+ new Value()
+ );
+
+ this.applicationEventQueueTimeSensor = metrics.sensor(APPLICATION_EVENT_QUEUE_TIME_SENSOR_NAME);
+ this.applicationEventQueueTimeSensor.add(
+ metrics.metricName(
+ "application-event-queue-time-avg",
+ CONSUMER_METRIC_GROUP,
+ "The average time, in milliseconds, that application events are taking to be dequeued."
+ ),
+ new Avg()
+ );
+ this.applicationEventQueueTimeSensor.add(
+ metrics.metricName(
+ "application-event-queue-time-max",
+ CONSUMER_METRIC_GROUP,
+ "The maximum time, in milliseconds, that an application event took to be dequeued."
+ ),
+ new Max()
+ );
+
+ this.applicationEventQueueProcessingTimeSensor = metrics.sensor(APPLICATION_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME);
+ this.applicationEventQueueProcessingTimeSensor.add(
+ metrics.metricName(
+ "application-event-queue-processing-time-avg",
+ CONSUMER_METRIC_GROUP,
+ "The average time, in milliseconds, that the consumer network takes to process all available application events."
+ ),
+ new Avg()
+ );
+ this.applicationEventQueueProcessingTimeSensor.add(
+ metrics.metricName("application-event-queue-processing-time-max",
+ CONSUMER_METRIC_GROUP,
+ "The maximum time, in milliseconds, that the consumer network took to process all available application events."
+ ),
+ new Max()
+ );
+
+ this.applicationEventExpiredSizeSensor = metrics.sensor(APPLICATION_EVENT_EXPIRED_SIZE_SENSOR_NAME);
+ this.applicationEventExpiredSizeSensor.add(
+ metrics.metricName(
+ APPLICATION_EVENT_EXPIRED_SIZE_SENSOR_NAME,
+ CONSUMER_METRIC_GROUP,
+ "The current number of expired application events."
+ ),
+ new Value()
+ );
+
+ this.unsentRequestsQueueSizeSensor = metrics.sensor(UNSENT_REQUESTS_QUEUE_SIZE_SENSOR_NAME);
+ this.unsentRequestsQueueSizeSensor.add(
+ metrics.metricName(
+ UNSENT_REQUESTS_QUEUE_SIZE_SENSOR_NAME,
+ CONSUMER_METRIC_GROUP,
+ "The current number of unsent requests in the consumer network."
+ ),
+ new Value()
+ );
+
+ this.unsentRequestsQueueTimeSensor = metrics.sensor(UNSENT_REQUESTS_QUEUE_TIME_SENSOR_NAME);
+ this.unsentRequestsQueueTimeSensor.add(
+ metrics.metricName(
+ "unsent-requests-queue-time-avg",
+ CONSUMER_METRIC_GROUP,
+ "The average time, in milliseconds, that requests are taking to be sent in the consumer network."
+ ),
+ new Avg()
+ );
+ this.unsentRequestsQueueTimeSensor.add(
+ metrics.metricName(
+ "unsent-requests-queue-time-max",
+ CONSUMER_METRIC_GROUP,
+ "The maximum time, in milliseconds, that a request remained unsent in the consumer network."
Review Comment:
ditto
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/metrics/AsyncConsumerMetrics.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer.internals.metrics;
+
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.metrics.stats.Value;
+
+import java.util.Arrays;
+
+import static org.apache.kafka.clients.consumer.internals.ConsumerUtils.CONSUMER_METRICS_SUFFIX;
+
+public class AsyncConsumerMetrics extends KafkaConsumerMetrics implements AutoCloseable {
+ private final Metrics metrics;
+
+ public static final String TIME_BETWEEN_NETWORK_THREAD_POLL_SENSOR_NAME = "time-between-network-thread-poll";
+ public static final String APPLICATION_EVENT_QUEUE_SIZE_SENSOR_NAME = "application-event-queue-size";
+ public static final String APPLICATION_EVENT_QUEUE_TIME_SENSOR_NAME = "application-event-queue-time";
+ public static final String APPLICATION_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME = "application-event-queue-processing-time";
+ public static final String APPLICATION_EVENT_EXPIRED_SIZE_SENSOR_NAME = "application-event-expired-size";
+ public static final String BACKGROUND_EVENT_QUEUE_SIZE_SENSOR_NAME = "background-event-queue-size";
+ public static final String BACKGROUND_EVENT_QUEUE_TIME_SENSOR_NAME = "background-event-queue-time";
+ public static final String BACKGROUND_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME = "background-event-queue-processing-time";
+ 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;
+ private final Sensor applicationEventQueueTimeSensor;
+ private final Sensor applicationEventQueueProcessingTimeSensor;
+ private final Sensor applicationEventExpiredSizeSensor;
+ private final Sensor backgroundEventQueueSizeSensor;
+ private final Sensor backgroundEventQueueTimeSensor;
+ private final Sensor backgroundEventQueueProcessingTimeSensor;
+ private final Sensor unsentRequestsQueueSizeSensor;
+ private final Sensor unsentRequestsQueueTimeSensor;
+
+ public AsyncConsumerMetrics(Metrics metrics, String metricGrpPrefix) {
Review Comment:
I guess this param will be "share" or similar when integrated with the ShareConsumer as @AndrewJSchofield suggested https://github.com/apache/kafka/pull/17199#discussion_r1854077158
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/metrics/AsyncConsumerMetrics.java:
##########
@@ -0,0 +1,262 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer.internals.metrics;
+
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.metrics.stats.Value;
+
+import java.util.Arrays;
+
+import static org.apache.kafka.clients.consumer.internals.ConsumerUtils.CONSUMER_METRIC_GROUP;
+import static org.apache.kafka.clients.consumer.internals.ConsumerUtils.CONSUMER_METRIC_GROUP_PREFIX;
+
+public class AsyncConsumerMetrics extends KafkaConsumerMetrics implements AutoCloseable {
+ private final Metrics metrics;
+
+ public static final String TIME_BETWEEN_NETWORK_THREAD_POLL_SENSOR_NAME = "time-between-network-thread-poll";
+ public static final String APPLICATION_EVENT_QUEUE_SIZE_SENSOR_NAME = "application-event-queue-size";
+ public static final String APPLICATION_EVENT_QUEUE_TIME_SENSOR_NAME = "application-event-queue-time";
+ public static final String APPLICATION_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME = "application-event-queue-processing-time";
+ public static final String APPLICATION_EVENT_EXPIRED_SIZE_SENSOR_NAME = "application-events-expired-count";
+ public static final String BACKGROUND_EVENT_QUEUE_SIZE_SENSOR_NAME = "background-event-queue-size";
+ public static final String BACKGROUND_EVENT_QUEUE_TIME_SENSOR_NAME = "background-event-queue-time";
+ public static final String BACKGROUND_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME = "background-event-queue-processing-time";
+ 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;
+ private final Sensor applicationEventQueueTimeSensor;
+ private final Sensor applicationEventQueueProcessingTimeSensor;
+ private final Sensor applicationEventExpiredSizeSensor;
+ private final Sensor backgroundEventQueueSizeSensor;
+ private final Sensor backgroundEventQueueTimeSensor;
+ private final Sensor backgroundEventQueueProcessingTimeSensor;
+ private final Sensor unsentRequestsQueueSizeSensor;
+ private final Sensor unsentRequestsQueueTimeSensor;
+
+ public AsyncConsumerMetrics(Metrics metrics) {
+ super(metrics, CONSUMER_METRIC_GROUP_PREFIX);
+
+ this.metrics = metrics;
+ this.timeBetweenNetworkThreadPollSensor = metrics.sensor(TIME_BETWEEN_NETWORK_THREAD_POLL_SENSOR_NAME);
+ this.timeBetweenNetworkThreadPollSensor.add(
+ metrics.metricName(
+ "time-between-network-thread-poll-avg",
+ CONSUMER_METRIC_GROUP,
+ "The average time taken, in milliseconds, between each poll in the network thread."
+ ),
+ new Avg()
+ );
+ this.timeBetweenNetworkThreadPollSensor.add(
+ metrics.metricName(
+ "time-between-network-thread-poll-max",
+ CONSUMER_METRIC_GROUP,
+ "The maximum time taken, in milliseconds, between each poll in the network thread."
+ ),
+ new Max()
+ );
+
+ this.applicationEventQueueSizeSensor = metrics.sensor(APPLICATION_EVENT_QUEUE_SIZE_SENSOR_NAME);
+ this.applicationEventQueueSizeSensor.add(
+ metrics.metricName(
+ APPLICATION_EVENT_QUEUE_SIZE_SENSOR_NAME,
+ CONSUMER_METRIC_GROUP,
+ "The current number of events in the queue to send from the application thread to the background thread."
+ ),
+ new Value()
+ );
+
+ this.applicationEventQueueTimeSensor = metrics.sensor(APPLICATION_EVENT_QUEUE_TIME_SENSOR_NAME);
+ this.applicationEventQueueTimeSensor.add(
+ metrics.metricName(
+ "application-event-queue-time-avg",
+ CONSUMER_METRIC_GROUP,
+ "The average time, in milliseconds, that application events are taking to be dequeued."
+ ),
+ new Avg()
+ );
+ this.applicationEventQueueTimeSensor.add(
+ metrics.metricName(
+ "application-event-queue-time-max",
+ CONSUMER_METRIC_GROUP,
+ "The maximum time, in milliseconds, that an application event took to be dequeued."
+ ),
+ new Max()
+ );
+
+ this.applicationEventQueueProcessingTimeSensor = metrics.sensor(APPLICATION_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME);
+ this.applicationEventQueueProcessingTimeSensor.add(
+ metrics.metricName(
+ "application-event-queue-processing-time-avg",
+ CONSUMER_METRIC_GROUP,
+ "The average time, in milliseconds, that the consumer network takes to process all available application events."
+ ),
+ new Avg()
+ );
+ this.applicationEventQueueProcessingTimeSensor.add(
+ metrics.metricName("application-event-queue-processing-time-max",
+ CONSUMER_METRIC_GROUP,
+ "The maximum time, in milliseconds, that the consumer network took to process all available application events."
+ ),
+ new Max()
+ );
+
+ this.applicationEventExpiredSizeSensor = metrics.sensor(APPLICATION_EVENT_EXPIRED_SIZE_SENSOR_NAME);
+ this.applicationEventExpiredSizeSensor.add(
+ metrics.metricName(
+ APPLICATION_EVENT_EXPIRED_SIZE_SENSOR_NAME,
+ CONSUMER_METRIC_GROUP,
+ "The current number of expired application events."
+ ),
+ new Value()
+ );
+
+ this.unsentRequestsQueueSizeSensor = metrics.sensor(UNSENT_REQUESTS_QUEUE_SIZE_SENSOR_NAME);
+ this.unsentRequestsQueueSizeSensor.add(
+ metrics.metricName(
+ UNSENT_REQUESTS_QUEUE_SIZE_SENSOR_NAME,
+ CONSUMER_METRIC_GROUP,
+ "The current number of unsent requests in the consumer network."
Review Comment:
ditto
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/metrics/AsyncConsumerMetrics.java:
##########
@@ -0,0 +1,262 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer.internals.metrics;
+
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.metrics.stats.Avg;
+import org.apache.kafka.common.metrics.stats.Max;
+import org.apache.kafka.common.metrics.stats.Value;
+
+import java.util.Arrays;
+
+import static org.apache.kafka.clients.consumer.internals.ConsumerUtils.CONSUMER_METRIC_GROUP;
+import static org.apache.kafka.clients.consumer.internals.ConsumerUtils.CONSUMER_METRIC_GROUP_PREFIX;
+
+public class AsyncConsumerMetrics extends KafkaConsumerMetrics implements AutoCloseable {
+ private final Metrics metrics;
+
+ public static final String TIME_BETWEEN_NETWORK_THREAD_POLL_SENSOR_NAME = "time-between-network-thread-poll";
+ public static final String APPLICATION_EVENT_QUEUE_SIZE_SENSOR_NAME = "application-event-queue-size";
+ public static final String APPLICATION_EVENT_QUEUE_TIME_SENSOR_NAME = "application-event-queue-time";
+ public static final String APPLICATION_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME = "application-event-queue-processing-time";
+ public static final String APPLICATION_EVENT_EXPIRED_SIZE_SENSOR_NAME = "application-events-expired-count";
+ public static final String BACKGROUND_EVENT_QUEUE_SIZE_SENSOR_NAME = "background-event-queue-size";
+ public static final String BACKGROUND_EVENT_QUEUE_TIME_SENSOR_NAME = "background-event-queue-time";
+ public static final String BACKGROUND_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME = "background-event-queue-processing-time";
+ 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;
+ private final Sensor applicationEventQueueTimeSensor;
+ private final Sensor applicationEventQueueProcessingTimeSensor;
+ private final Sensor applicationEventExpiredSizeSensor;
+ private final Sensor backgroundEventQueueSizeSensor;
+ private final Sensor backgroundEventQueueTimeSensor;
+ private final Sensor backgroundEventQueueProcessingTimeSensor;
+ private final Sensor unsentRequestsQueueSizeSensor;
+ private final Sensor unsentRequestsQueueTimeSensor;
+
+ public AsyncConsumerMetrics(Metrics metrics) {
+ super(metrics, CONSUMER_METRIC_GROUP_PREFIX);
+
+ this.metrics = metrics;
+ this.timeBetweenNetworkThreadPollSensor = metrics.sensor(TIME_BETWEEN_NETWORK_THREAD_POLL_SENSOR_NAME);
+ this.timeBetweenNetworkThreadPollSensor.add(
+ metrics.metricName(
+ "time-between-network-thread-poll-avg",
+ CONSUMER_METRIC_GROUP,
+ "The average time taken, in milliseconds, between each poll in the network thread."
+ ),
+ new Avg()
+ );
+ this.timeBetweenNetworkThreadPollSensor.add(
+ metrics.metricName(
+ "time-between-network-thread-poll-max",
+ CONSUMER_METRIC_GROUP,
+ "The maximum time taken, in milliseconds, between each poll in the network thread."
+ ),
+ new Max()
+ );
+
+ this.applicationEventQueueSizeSensor = metrics.sensor(APPLICATION_EVENT_QUEUE_SIZE_SENSOR_NAME);
+ this.applicationEventQueueSizeSensor.add(
+ metrics.metricName(
+ APPLICATION_EVENT_QUEUE_SIZE_SENSOR_NAME,
+ CONSUMER_METRIC_GROUP,
+ "The current number of events in the queue to send from the application thread to the background thread."
+ ),
+ new Value()
+ );
+
+ this.applicationEventQueueTimeSensor = metrics.sensor(APPLICATION_EVENT_QUEUE_TIME_SENSOR_NAME);
+ this.applicationEventQueueTimeSensor.add(
+ metrics.metricName(
+ "application-event-queue-time-avg",
+ CONSUMER_METRIC_GROUP,
+ "The average time, in milliseconds, that application events are taking to be dequeued."
+ ),
+ new Avg()
+ );
+ this.applicationEventQueueTimeSensor.add(
+ metrics.metricName(
+ "application-event-queue-time-max",
+ CONSUMER_METRIC_GROUP,
+ "The maximum time, in milliseconds, that an application event took to be dequeued."
+ ),
+ new Max()
+ );
+
+ this.applicationEventQueueProcessingTimeSensor = metrics.sensor(APPLICATION_EVENT_QUEUE_PROCESSING_TIME_SENSOR_NAME);
+ this.applicationEventQueueProcessingTimeSensor.add(
+ metrics.metricName(
+ "application-event-queue-processing-time-avg",
+ CONSUMER_METRIC_GROUP,
+ "The average time, in milliseconds, that the consumer network takes to process all available application events."
Review Comment:
```suggestion
"The average time, in milliseconds, that the background thread takes to process all available application events."
```
background thread or "network thread" is fine (but not "consumer network"). Let's just be consistent and use the same term when possible in all 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