You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2024/08/13 20:33:54 UTC
(kafka) branch trunk updated: KAFKA-17239 add request-latency metrics for node in admin client (#16832)
This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 75bcb9eb42f KAFKA-17239 add request-latency metrics for node in admin client (#16832)
75bcb9eb42f is described below
commit 75bcb9eb42fb305bce0eecd4dda265f935cbd108
Author: xijiu <42...@qq.com>
AuthorDate: Wed Aug 14 04:33:47 2024 +0800
KAFKA-17239 add request-latency metrics for node in admin client (#16832)
Reviewers: TengYao Chi <ki...@gmail.com>, Chia-Ping Tsai <ch...@gmail.com>
---
.../kafka/clients/admin/KafkaAdminClient.java | 4 +
.../admin/internals/AdminFetchMetricsManager.java | 37 +++++
.../internals/AdminFetchMetricsManagerTest.java | 170 +++++++++++++++++++++
3 files changed, 211 insertions(+)
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
index 30ec02684ba..a3adc98b887 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
@@ -41,6 +41,7 @@ import org.apache.kafka.clients.admin.internals.AdminApiFuture;
import org.apache.kafka.clients.admin.internals.AdminApiFuture.SimpleAdminApiFuture;
import org.apache.kafka.clients.admin.internals.AdminApiHandler;
import org.apache.kafka.clients.admin.internals.AdminBootstrapAddresses;
+import org.apache.kafka.clients.admin.internals.AdminFetchMetricsManager;
import org.apache.kafka.clients.admin.internals.AdminMetadataManager;
import org.apache.kafka.clients.admin.internals.AllBrokersStrategy;
import org.apache.kafka.clients.admin.internals.AlterConsumerGroupOffsetsHandler;
@@ -405,6 +406,7 @@ public class KafkaAdminClient extends AdminClient {
private final ExponentialBackoff retryBackoff;
private final boolean clientTelemetryEnabled;
private final MetadataRecoveryStrategy metadataRecoveryStrategy;
+ private final AdminFetchMetricsManager adminFetchMetricsManager;
/**
* The telemetry requests client instance id.
@@ -619,6 +621,7 @@ public class KafkaAdminClient extends AdminClient {
CommonClientConfigs.RETRY_BACKOFF_JITTER);
this.clientTelemetryEnabled = config.getBoolean(AdminClientConfig.ENABLE_METRICS_PUSH_CONFIG);
this.metadataRecoveryStrategy = MetadataRecoveryStrategy.forName(config.getString(AdminClientConfig.METADATA_RECOVERY_STRATEGY_CONFIG));
+ this.adminFetchMetricsManager = new AdminFetchMetricsManager(metrics);
config.logUnused();
AppInfoParser.registerAppInfo(JMX_PREFIX, clientId, metrics, time.milliseconds());
log.debug("Kafka admin client initialized");
@@ -1397,6 +1400,7 @@ public class KafkaAdminClient extends AdminClient {
} else {
try {
call.handleResponse(response.responseBody());
+ adminFetchMetricsManager.recordLatency(response.destination(), response.requestLatencyMs());
if (log.isTraceEnabled())
log.trace("{} got response {}", call, response.responseBody());
} catch (Throwable t) {
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminFetchMetricsManager.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminFetchMetricsManager.java
new file mode 100644
index 00000000000..a38fcce1600
--- /dev/null
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminFetchMetricsManager.java
@@ -0,0 +1,37 @@
+/*
+ * 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.admin.internals;
+
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+
+public class AdminFetchMetricsManager {
+ private final Metrics metrics;
+
+ public AdminFetchMetricsManager(Metrics metrics) {
+ this.metrics = metrics;
+ }
+
+ public void recordLatency(String node, long requestLatencyMs) {
+ if (!node.isEmpty()) {
+ String nodeTimeName = "node-" + node + ".latency";
+ Sensor nodeRequestTime = this.metrics.getSensor(nodeTimeName);
+ if (nodeRequestTime != null)
+ nodeRequestTime.record(requestLatencyMs);
+ }
+ }
+}
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/internals/AdminFetchMetricsManagerTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/internals/AdminFetchMetricsManagerTest.java
new file mode 100644
index 00000000000..a9b0d447345
--- /dev/null
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/internals/AdminFetchMetricsManagerTest.java
@@ -0,0 +1,170 @@
+/*
+ * 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.admin.internals;
+
+import org.apache.kafka.common.MetricName;
+import org.apache.kafka.common.metrics.KafkaMetric;
+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.utils.MockTime;
+import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Utils;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class AdminFetchMetricsManagerTest {
+ private static final double EPSILON = 0.0001;
+
+ private final Time time = new MockTime(1, 0, 0);
+ private Metrics metrics;
+ private AdminFetchMetricsManager adminFetchMetricsManager;
+ private final String group = "group";
+
+
+ @BeforeEach
+ public void setup() {
+ metrics = new Metrics(time);
+ adminFetchMetricsManager = new AdminFetchMetricsManager(metrics);
+ }
+
+ @AfterEach
+ public void tearDown() {
+ Utils.closeQuietly(metrics, "metrics");
+ metrics = null;
+ adminFetchMetricsManager = null;
+ }
+
+ @Test
+ public void testSingleNodeLatency() {
+ String connectionId = "0";
+ MetricName nodeLatencyAvg = metrics.metricName("request-latency-avg", group);
+ MetricName nodeLatencyMax = metrics.metricName("request-latency-max", group);
+ registerNodeLatencyMetric(connectionId, nodeLatencyAvg, nodeLatencyMax);
+
+ adminFetchMetricsManager.recordLatency(connectionId, 333);
+ mockSleepTimeWindow();
+ adminFetchMetricsManager.recordLatency(connectionId, 444);
+ assertEquals(388.5, metricValue(nodeLatencyAvg), EPSILON);
+ assertEquals(444, metricValue(nodeLatencyMax), EPSILON);
+ adminFetchMetricsManager.recordLatency(connectionId, 666);
+ assertEquals(481, metricValue(nodeLatencyAvg), EPSILON);
+ assertEquals(666, metricValue(nodeLatencyMax), EPSILON);
+
+ // first record(333) expired
+ mockSleepTimeWindow();
+ assertEquals(555, metricValue(nodeLatencyAvg), EPSILON);
+ assertEquals(666, metricValue(nodeLatencyMax), EPSILON);
+
+ // all records expired
+ mockSleepTimeWindow();
+ assertTrue(Double.isNaN(metricValue(nodeLatencyAvg)));
+ assertTrue(Double.isNaN(metricValue(nodeLatencyMax)));
+ }
+
+ @Test
+ public void testMultiNodeLatency() {
+ String connectionId0 = "0";
+ MetricName nodeLatencyAvg0 = metrics.metricName("request-latency-avg", group, genericTag(connectionId0));
+ MetricName nodeLatencyMax0 = metrics.metricName("request-latency-max", group, genericTag(connectionId0));
+ registerNodeLatencyMetric(connectionId0, nodeLatencyAvg0, nodeLatencyMax0);
+ adminFetchMetricsManager.recordLatency(connectionId0, 5);
+ adminFetchMetricsManager.recordLatency(connectionId0, 8);
+
+ // Record metric against another node.
+ String connectionId1 = "1";
+ MetricName nodeLatencyAvg1 = metrics.metricName("request-latency-avg", group, genericTag(connectionId1));
+ MetricName nodeLatencyMax1 = metrics.metricName("request-latency-max", group, genericTag(connectionId1));
+ registerNodeLatencyMetric(connectionId1, nodeLatencyAvg1, nodeLatencyMax1);
+ adminFetchMetricsManager.recordLatency(connectionId1, 105);
+ adminFetchMetricsManager.recordLatency(connectionId1, 108);
+
+ assertEquals(6.5, metricValue(nodeLatencyAvg0), EPSILON);
+ assertEquals(8, metricValue(nodeLatencyMax0), EPSILON);
+ assertEquals(106.5, metricValue(nodeLatencyAvg1), EPSILON);
+ assertEquals(108, metricValue(nodeLatencyMax1), EPSILON);
+
+ mockSleepTimeWindow();
+ adminFetchMetricsManager.recordLatency(connectionId0, 11);
+ adminFetchMetricsManager.recordLatency(connectionId1, 111);
+ assertEquals(8, metricValue(nodeLatencyAvg0), EPSILON);
+ assertEquals(11, metricValue(nodeLatencyMax0), EPSILON);
+ assertEquals(108, metricValue(nodeLatencyAvg1), EPSILON);
+ assertEquals(111, metricValue(nodeLatencyMax1), EPSILON);
+
+ mockSleepTimeWindow();
+ assertEquals(11, metricValue(nodeLatencyAvg0), EPSILON);
+ assertEquals(11, metricValue(nodeLatencyMax0), EPSILON);
+ assertEquals(111, metricValue(nodeLatencyAvg1), EPSILON);
+ assertEquals(111, metricValue(nodeLatencyMax1), EPSILON);
+
+ mockSleepTimeWindow();
+ assertTrue(Double.isNaN(metricValue(nodeLatencyAvg0)));
+ assertTrue(Double.isNaN(metricValue(nodeLatencyMax0)));
+ assertTrue(Double.isNaN(metricValue(nodeLatencyAvg1)));
+ assertTrue(Double.isNaN(metricValue(nodeLatencyMax1)));
+
+ adminFetchMetricsManager.recordLatency(connectionId0, 500);
+ adminFetchMetricsManager.recordLatency(connectionId0, 600);
+ mockSleepTimeWindow();
+ adminFetchMetricsManager.recordLatency(connectionId1, 800);
+ adminFetchMetricsManager.recordLatency(connectionId1, 900);
+ assertEquals(550, metricValue(nodeLatencyAvg0), EPSILON);
+ assertEquals(600, metricValue(nodeLatencyMax0), EPSILON);
+ assertEquals(850, metricValue(nodeLatencyAvg1), EPSILON);
+ assertEquals(900, metricValue(nodeLatencyMax1), EPSILON);
+
+ mockSleepTimeWindow();
+ assertTrue(Double.isNaN(metricValue(nodeLatencyAvg0)));
+ assertTrue(Double.isNaN(metricValue(nodeLatencyMax0)));
+ assertEquals(850, metricValue(nodeLatencyAvg1), EPSILON);
+ assertEquals(900, metricValue(nodeLatencyMax1), EPSILON);
+
+ mockSleepTimeWindow();
+ assertTrue(Double.isNaN(metricValue(nodeLatencyAvg1)));
+ assertTrue(Double.isNaN(metricValue(nodeLatencyMax1)));
+ }
+
+ private Map<String, String> genericTag(String connectionId) {
+ return Collections.singletonMap("node-id", "node-" + connectionId);
+ }
+
+ private void mockSleepTimeWindow() {
+ time.sleep(metrics.config().timeWindowMs() + 1);
+ }
+
+ private void registerNodeLatencyMetric(String connectionId, MetricName nodeLatencyAvg, MetricName nodeLatencyMax) {
+ String nodeTimeName = "node-" + connectionId + ".latency";
+ Sensor nodeRequestTime = metrics.sensor(nodeTimeName);
+ nodeRequestTime.add(nodeLatencyAvg, new Avg());
+ nodeRequestTime.add(nodeLatencyMax, new Max());
+ }
+
+ private double metricValue(MetricName metricName) {
+ KafkaMetric metric = metrics.metric(metricName);
+ return (double) metric.metricValue();
+ }
+}