Repository: skiptests/astraea
Pull request #714 (2022-09-14 15:07:04)

astraea-704 Add consumer metrics
as title


Patch:
diff --git a/app/src/main/java/org/astraea/app/scenario/Scenario.java b/app/src/main/java/org/astraea/app/scenario/Scenario.java
index dc8fffad..bb349cfe 100644
--- a/app/src/main/java/org/astraea/app/scenario/Scenario.java
+++ b/app/src/main/java/org/astraea/app/scenario/Scenario.java
@@ -16,23 +16,95 @@
  */
 package org.astraea.app.scenario;
 
+import java.util.Map;
 import org.astraea.common.Utils;
 import org.astraea.common.admin.Admin;
-import org.astraea.common.cost.Configuration;
 
-/**
- * The subclass of this class should contain the logic to fulfill a scenario.
- *
- * @param <T> the return result after applying this scenario to the specific Kafka cluster. This
- *     result might be print on the program output. Or being serialized to json and transmit by web
- *     API.
- */
-public interface Scenario<T> {
+/** The subclass of this class should contain the logic to fulfill a scenario. */
+public interface Scenario {
+
+  static Builder build(double binomialProbability) {
+    return new Builder(binomialProbability);
+  }
+
+  class Builder {
+    private String topicName = Utils.randomString();
+    private int numberOfPartitions = 10;
+    private short numberOfReplicas = 1;
+    private double binomialProbability = 0.5;
+
+    private Builder(double binomialProbability) {
+      this.binomialProbability = binomialProbability;
+    }
+
+    public Builder topicName(String topicName) {
+      this.topicName = topicName;
+      return this;
+    }
+
+    public Builder numberOfPartitions(int numberOfPartitions) {
+      this.numberOfPartitions = numberOfPartitions;
+      return this;
+    }
+
+    public Builder numberOfReplicas(short numberOfReplicas) {
+      this.numberOfReplicas = numberOfReplicas;
+      return this;
+    }
+
+    public Builder binomialProbability(double binomialProbability) {
+      this.binomialProbability = binomialProbability;
+      return this;
+    }
 
-  static Scenario<?> of(Class<Scenario<?>> theClass, Configuration configuration) {
-    return Utils.construct(theClass, configuration);
+    public Scenario build() {
+      return new SkewedPartitionScenario(
+          topicName, numberOfPartitions, numberOfReplicas, binomialProbability);
+    }
   }
 
   /** Apply this scenario to the Kafka cluster */
-  T apply(Admin admin);
+  Result apply(Admin admin);
+
+  class Result {
+
+    private final String topicName;
+    private final int numberOfPartitions;
+    private final short numberOfReplicas;
+    private final Map<Integer, Long> leaderSum;
+    private final Map<Integer, Long> logSum;
+
+    public Result(
+        String topicName,
+        int numberOfPartitions,
+        short numberOfReplicas,
+        Map<Integer, Long> leaderSum,
+        Map<Integer, Long> logSum) {
+      this.topicName = topicName;
+      this.numberOfPartitions = numberOfPartitions;
+      this.numberOfReplicas = numberOfReplicas;
+      this.leaderSum = leaderSum;
+      this.logSum = logSum;
+    }
+
+    public String topicName() {
+      return topicName;
+    }
+
+    public int numberOfPartitions() {
+      return numberOfPartitions;
+    }
+
+    public short numberOfReplicas() {
+      return numberOfReplicas;
+    }
+
+    public Map<Integer, Long> leaderSum() {
+      return leaderSum;
+    }
+
+    public Map<Integer, Long> logSum() {
+      return logSum;
+    }
+  }
 }
diff --git a/app/src/main/java/org/astraea/app/scenario/ScenarioMain.java b/app/src/main/java/org/astraea/app/scenario/ScenarioMain.java
index 8821b81d..3190c829 100644
--- a/app/src/main/java/org/astraea/app/scenario/ScenarioMain.java
+++ b/app/src/main/java/org/astraea/app/scenario/ScenarioMain.java
@@ -37,14 +37,14 @@ public class ScenarioMain extends Argument {
     var theClass = Utils.packException(() -> Class.forName(scenarioClass));
     if (Scenario.class.isAssignableFrom(theClass)) {
       //noinspection unchecked
-      var scenarioClass = (Class<Scenario<?>>) theClass;
-      execute(Scenario.of(scenarioClass, Configuration.of(configs())));
+      var scenarioClass = (Class<Scenario>) theClass;
+      execute(Utils.construct(scenarioClass, Configuration.of(configs())));
     } else {
       throw new RuntimeException("Target class is not a scenario: " + theClass.getName());
     }
   }
 
-  public void execute(Scenario<?> scenario) {
+  public void execute(Scenario scenario) {
     System.out.println("Accept scenario: " + scenario.getClass().getName());
     try (Admin admin = Admin.of(bootstrapServers())) {
       System.out.println(gson.toJson(scenario.apply(admin)));
diff --git a/app/src/main/java/org/astraea/app/scenario/impl/SkewedPartitionScenario.java b/app/src/main/java/org/astraea/app/scenario/SkewedPartitionScenario.java
similarity index 75%
rename from app/src/main/java/org/astraea/app/scenario/impl/SkewedPartitionScenario.java
rename to app/src/main/java/org/astraea/app/scenario/SkewedPartitionScenario.java
index 645bc36f..c5701f90 100644
--- a/app/src/main/java/org/astraea/app/scenario/impl/SkewedPartitionScenario.java
+++ b/app/src/main/java/org/astraea/app/scenario/SkewedPartitionScenario.java
@@ -14,13 +14,12 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.astraea.app.scenario.impl;
+package org.astraea.app.scenario;
 
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
-import java.util.Map;
 import java.util.Objects;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
@@ -28,28 +27,18 @@ import org.apache.commons.math3.distribution.BinomialDistribution;
 import org.apache.commons.math3.distribution.EnumeratedDistribution;
 import org.apache.commons.math3.distribution.IntegerDistribution;
 import org.apache.commons.math3.util.Pair;
-import org.astraea.app.scenario.Scenario;
 import org.astraea.common.Utils;
 import org.astraea.common.admin.Admin;
 import org.astraea.common.admin.TopicPartition;
-import org.astraea.common.cost.Configuration;
 
-public class SkewedPartitionScenario implements Scenario<SkewedPartitionScenario.Result> {
+public class SkewedPartitionScenario implements Scenario {
 
   final String topicName;
   final int partitions;
   final short replicas;
   final double binomialProbability;
 
-  public SkewedPartitionScenario(Configuration configuration) {
-    this(
-        configuration.string("topicName").orElse(Utils.randomString()),
-        configuration.string("partitions").map(Integer::parseInt).orElse(10),
-        configuration.string("replicas").map(Short::parseShort).orElse((short) 1),
-        configuration.string("binomialProbability").map(Double::parseDouble).orElse(0.5));
-  }
-
-  private SkewedPartitionScenario(
+  SkewedPartitionScenario(
       String topicName, int partitions, short replicas, double binomialProbability) {
     this.topicName = topicName;
     this.partitions = partitions;
@@ -93,7 +82,6 @@ public class SkewedPartitionScenario implements Scenario<SkewedPartitionScenario
         topicName,
         partitions,
         replicas,
-        binomialProbability,
         replicaLists.values().stream()
             .map(list -> list.get(0))
             .collect(Collectors.groupingBy(x -> x, Collectors.counting())),
@@ -128,29 +116,4 @@ public class SkewedPartitionScenario implements Scenario<SkewedPartitionScenario
     }
     return result;
   }
-
-  public static class Result {
-
-    public final String topicName;
-    public final int partitions;
-    public final short replicas;
-    public final double binomialProbability;
-    public final Map<Integer, Long> leaderSum;
-    public final Map<Integer, Long> logSum;
-
-    public Result(
-        String topicName,
-        int partitions,
-        short replicas,
-        double binomialProbability,
-        Map<Integer, Long> leaderSum,
-        Map<Integer, Long> logSum) {
-      this.topicName = topicName;
-      this.partitions = partitions;
-      this.replicas = replicas;
-      this.binomialProbability = binomialProbability;
-      this.leaderSum = leaderSum;
-      this.logSum = logSum;
-    }
-  }
 }
diff --git a/app/src/main/java/org/astraea/app/web/TopicHandler.java b/app/src/main/java/org/astraea/app/web/TopicHandler.java
index 22a0677c..16e63870 100644
--- a/app/src/main/java/org/astraea/app/web/TopicHandler.java
+++ b/app/src/main/java/org/astraea/app/web/TopicHandler.java
@@ -25,6 +25,7 @@ import java.util.function.Predicate;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
 import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
+import org.astraea.app.scenario.Scenario;
 import org.astraea.common.ExecutionRuntimeException;
 import org.astraea.common.admin.Admin;
 import org.astraea.common.admin.Config;
@@ -38,6 +39,7 @@ class TopicHandler implements Handler {
   static final String NUMBER_OF_REPLICAS_KEY = "replicas";
   static final String PARTITION_KEY = "partition";
   static final String LIST_INTERNAL = "listInternal";
+  static final String PROBABILITY_INTERNAL = "probability";
 
   private final Admin admin;
 
@@ -113,14 +115,27 @@ class TopicHandler implements Handler {
     if (topicNames.size() != requests.size())
       throw new IllegalArgumentException("duplicate topic name: " + topicNames);
     requests.forEach(
-        request ->
+        request -> {
+          var topicName = request.value(TOPIC_NAME_KEY);
+          var numberOfPartitions = request.getInt(NUMBER_OF_PARTITIONS_KEY).orElse(1);
+          var numberOfReplicas = request.getShort(NUMBER_OF_REPLICAS_KEY).orElse((short) 1);
+          if (request.has(PROBABILITY_INTERNAL)) {
+            Scenario.build(request.doubleValue(PROBABILITY_INTERNAL))
+                .topicName(topicName)
+                .numberOfPartitions(numberOfPartitions)
+                .numberOfReplicas(numberOfReplicas)
+                .build()
+                .apply(admin);
+          } else {
             admin
                 .creator()
                 .topic(request.value(TOPIC_NAME_KEY))
                 .numberOfPartitions(request.getInt(NUMBER_OF_PARTITIONS_KEY).orElse(1))
                 .numberOfReplicas(request.getShort(NUMBER_OF_REPLICAS_KEY).orElse((short) 1))
                 .configs(remainingConfigs(request))
-                .create());
+                .create();
+          }
+        });
 
     try {
       // if the topic creation is synced, we return the details.
diff --git a/common/src/main/java/org/astraea/common/metrics/client/consumer/ConsumerMetrics.java b/common/src/main/java/org/astraea/common/metrics/client/consumer/ConsumerMetrics.java
index 2692ae46..872b8b58 100644
--- a/common/src/main/java/org/astraea/common/metrics/client/consumer/ConsumerMetrics.java
+++ b/common/src/main/java/org/astraea/common/metrics/client/consumer/ConsumerMetrics.java
@@ -71,4 +71,17 @@ public class ConsumerMetrics {
         .map(b -> (HasConsumerFetchMetrics) () -> b)
         .collect(Collectors.toUnmodifiableList());
   }
+
+  public static Collection<HasConsumerMetrics> consumer(MBeanClient mBeanClient) {
+    return mBeanClient
+        .queryBeans(
+            BeanQuery.builder()
+                .domainName("kafka.consumer")
+                .property("type", "consumer-metrics")
+                .property("client-id", "*")
+                .build())
+        .stream()
+        .map(b -> (HasConsumerMetrics) () -> b)
+        .collect(Collectors.toUnmodifiableList());
+  }
 }
diff --git a/common/src/main/java/org/astraea/common/metrics/client/consumer/HasConsumerMetrics.java b/common/src/main/java/org/astraea/common/metrics/client/consumer/HasConsumerMetrics.java
new file mode 100644
index 00000000..30243a19
--- /dev/null
+++ b/common/src/main/java/org/astraea/common/metrics/client/consumer/HasConsumerMetrics.java
@@ -0,0 +1,202 @@
+/*
+ * 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.astraea.common.metrics.client.consumer;
+
+import org.astraea.common.metrics.HasBeanObject;
+
+public interface HasConsumerMetrics extends HasBeanObject {
+
+  default String clientId() {
+    return beanObject().properties().get("client-id");
+  }
+
+  default double reauthenticationLatencyAvg() {
+    return (double) beanObject().attributes().get("reauthentication-latency-avg");
+  }
+
+  default double ioTimeNsTotal() {
+    return (double) beanObject().attributes().get("io-time-ns-total");
+  }
+
+  default double successfulAuthenticationTotal() {
+    return (double) beanObject().attributes().get("successful-authentication-total");
+  }
+
+  default double ioWaittimeTotal() {
+    return (double) beanObject().attributes().get("io-waittime-total");
+  }
+
+  default double committedTimeNsTotal() {
+    return (double) beanObject().attributes().get("committed-time-ns-total");
+  }
+
+  default double reauthenticationLatencyMax() {
+    return (double) beanObject().attributes().get("reauthentication-latency-max");
+  }
+
+  default double successfulAuthenticationRate() {
+    return (double) beanObject().attributes().get("successful-authentication-rate");
+  }
+
+  default double commitSyncTimeNsTotal() {
+    return (double) beanObject().attributes().get("commit-sync-time-ns-total");
+  }
+
+  default double failedAuthenticationTotal() {
+    return (double) beanObject().attributes().get("failed-authentication-total");
+  }
+
+  default double timeBetweenPollAvg() {
+    return (double) beanObject().attributes().get("time-between-poll-avg");
+  }
+
+  default double connectionCount() {
+    return (double) beanObject().attributes().get("connection-count");
+  }
+
+  default double responseTotal() {
+    return (double) beanObject().attributes().get("response-total");
+  }
+
+  default double requestRate() {
+    return (double) beanObject().attributes().get("request-rate");
+  }
+
+  default double incomingByteRate() {
+    return (double) beanObject().attributes().get("incoming-byte-rate");
+  }
+
+  default double lastPollSecondsAgo() {
+    return (double) beanObject().attributes().get("last-poll-seconds-ago");
+  }
+
+  default double successfulAuthenticationNoReauthTotal() {
+    return (double) beanObject().attributes().get("successful-authentication-no-reauth-total");
+  }
+
+  default double timeBetweenPollMax() {
+    return (double) beanObject().attributes().get("time-between-poll-max");
+  }
+
+  default double failedReauthenticationTotal() {
+    return (double) beanObject().attributes().get("failed-reauthentication-total");
+  }
+
+  default double selectRate() {
+    return (double) beanObject().attributes().get("select-rate");
+  }
+
+  default double successfulReauthenticationTotal() {
+    return (double) beanObject().attributes().get("successful-reauthentication-total");
+  }
+
+  default double requestTotal() {
+    return (double) beanObject().attributes().get("request-total");
+  }
+
+  default double ioTimeNsAvg() {
+    return (double) beanObject().attributes().get("io-time-ns-avg");
+  }
+
+  default double iotimeTotal() {
+    return (double) beanObject().attributes().get("iotime-total");
+  }
+
+  default double ioWaitTimeNsTotal() {
+    return (double) beanObject().attributes().get("io-wait-time-ns-total");
+  }
+
+  default double ioWaitRatio() {
+    return (double) beanObject().attributes().get("io-wait-ratio");
+  }
+
+  default double networkIoRate() {
+    return (double) beanObject().attributes().get("network-io-rate");
+  }
+
+  default double connectionCreationRate() {
+    return (double) beanObject().attributes().get("connection-creation-rate");
+  }
+
+  default double requestSizeAvg() {
+    return (double) beanObject().attributes().get("request-size-avg");
+  }
+
+  default double ioRatio() {
+    return (double) beanObject().attributes().get("io-ratio");
+  }
+
+  default double responseRate() {
+    return (double) beanObject().attributes().get("response-rate");
+  }
+
+  default double successfulReauthenticationRate() {
+    return (double) beanObject().attributes().get("successful-reauthentication-rate");
+  }
+
+  default double selectTotal() {
+    return (double) beanObject().attributes().get("select-total");
+  }
+
+  default double outgoingByteTotal() {
+    return (double) beanObject().attributes().get("outgoing-byte-total");
+  }
+
+  default double pollIdleRatioAvg() {
+    return (double) beanObject().attributes().get("poll-idle-ratio-avg");
+  }
+
+  default double connectionCloseRate() {
+    return (double) beanObject().attributes().get("connection-close-rate");
+  }
+
+  default double incomingByteTotal() {
+    return (double) beanObject().attributes().get("incoming-byte-total");
+  }
+
+  default double requestSizeMax() {
+    return (double) beanObject().attributes().get("request-size-max");
+  }
+
+  default double failedAuthenticationRate() {
+    return (double) beanObject().attributes().get("failed-authentication-rate");
+  }
+
+  default double outgoingByteRate() {
+    return (double) beanObject().attributes().get("outgoing-byte-rate");
+  }
+
+  default double failedReauthenticationRate() {
+    return (double) beanObject().attributes().get("failed-reauthentication-rate");
+  }
+
+  default double ioWaitTimeNsAvg() {
+    return (double) beanObject().attributes().get("io-wait-time-ns-avg");
+  }
+
+  default double connectionCloseTotal() {
+    return (double) beanObject().attributes().get("connection-close-total");
+  }
+
+  default double connectionCreationTotal() {
+    return (double) beanObject().attributes().get("connection-creation-total");
+  }
+
+  default double networkIoTotal() {
+    return (double) beanObject().attributes().get("network-io-total");
+  }
+}
diff --git a/docs/web_server/web_api_topics_chinese.md b/docs/web_server/web_api_topics_chinese.md
index 3510bf24..b7c395b3 100644
--- a/docs/web_server/web_api_topics_chinese.md
+++ b/docs/web_server/web_api_topics_chinese.md
@@ -12,11 +12,12 @@ POST /topics
 
 參數
 
-| 名稱         | 說明                  | 預設值 |
-|------------|---------------------|-----|
-| name       | (必填) topic 名稱       | 無   |
-| partitions | (選填) partition 數量   | 1   |
-| replicas   | (選填) replication 數量 | 1   |
+| 名稱          | 說明                  | 預設值                |
+|-------------|---------------------|--------------------|
+| name        | (必填) topic 名稱       | 無                  |
+| partitions  | (選填) partition 數量   | 1                  |
+| replicas    | (選填) replication 數量 | 1                  |
+| probability | (選填) 分佈機率           | 無，預設依照 kafka 的分佈方式 |
 - replicas 數量須 <= brokers 數量
 
 cURL 範例


Test patch:
diff --git a/app/src/test/java/org/astraea/app/scenario/impl/SkewedPartitionScenarioTest.java b/app/src/test/java/org/astraea/app/scenario/SkewedPartitionScenarioTest.java
similarity index 75%
rename from app/src/test/java/org/astraea/app/scenario/impl/SkewedPartitionScenarioTest.java
rename to app/src/test/java/org/astraea/app/scenario/SkewedPartitionScenarioTest.java
index baf72609..16916f03 100644
--- a/app/src/test/java/org/astraea/app/scenario/impl/SkewedPartitionScenarioTest.java
+++ b/app/src/test/java/org/astraea/app/scenario/SkewedPartitionScenarioTest.java
@@ -14,15 +14,13 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.astraea.app.scenario.impl;
+package org.astraea.app.scenario;
 
 import java.util.List;
-import java.util.Map;
 import java.util.Set;
 import org.apache.commons.math3.distribution.BinomialDistribution;
 import org.astraea.common.Utils;
 import org.astraea.common.admin.Admin;
-import org.astraea.common.cost.Configuration;
 import org.astraea.it.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.params.ParameterizedTest;
@@ -50,24 +48,16 @@ class SkewedPartitionScenarioTest extends RequireBrokerCluster {
   @ParameterizedTest
   void test(int partitions, short replicas) {
     var topicName = Utils.randomString();
-    var scenario =
-        new SkewedPartitionScenario(
-            Configuration.of(
-                Map.of(
-                    "topicName", topicName,
-                    "partitions", String.valueOf(partitions),
-                    "replicas", String.valueOf(replicas),
-                    "binomialProbability", String.valueOf(0.5))));
+    var scenario = new SkewedPartitionScenario(topicName, partitions, replicas, 0.5);
     try (Admin admin = Admin.of(bootstrapServers())) {
       var result = scenario.apply(admin);
-      Assertions.assertEquals(topicName, result.topicName);
-      Assertions.assertEquals(partitions, result.partitions);
-      Assertions.assertEquals(replicas, result.replicas);
-      Assertions.assertEquals(0.5, result.binomialProbability);
+      Assertions.assertEquals(topicName, result.topicName());
+      Assertions.assertEquals(partitions, result.numberOfPartitions());
+      Assertions.assertEquals(replicas, result.numberOfReplicas());
       Assertions.assertEquals(
-          partitions, result.leaderSum.values().stream().mapToLong(x -> x).sum());
+          partitions, result.leaderSum().values().stream().mapToLong(x -> x).sum());
       Assertions.assertEquals(
-          (long) partitions * replicas, result.logSum.values().stream().mapToLong(x -> x).sum());
+          (long) partitions * replicas, result.logSum().values().stream().mapToLong(x -> x).sum());
     }
   }
 
diff --git a/app/src/test/java/org/astraea/app/web/TopicHandlerTest.java b/app/src/test/java/org/astraea/app/web/TopicHandlerTest.java
index 87d7517d..e778456e 100644
--- a/app/src/test/java/org/astraea/app/web/TopicHandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/TopicHandlerTest.java
@@ -19,6 +19,7 @@ package org.astraea.app.web;
 import static java.nio.charset.StandardCharsets.UTF_8;
 
 import java.time.Duration;
+import java.util.List;
 import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.Optional;
@@ -303,4 +304,32 @@ public class TopicHandlerTest extends RequireBrokerCluster {
       Assertions.assertTrue(latestTopicNames.contains(topicNames.get(1)));
     }
   }
+
+  @Test
+  void testCreateTopicByProbability() {
+    var topicName = Utils.randomString(10);
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var handler = new TopicHandler(admin);
+      var request =
+          Channel.ofRequest(
+              PostRequest.of(
+                  String.format(
+                      "{\"topics\":[{\"name\":\"%s\", \"partitions\":30, \"probability\": 0.5}]}",
+                      topicName)));
+      var topics = handler.post(request);
+      Assertions.assertEquals(1, topics.topics.size());
+      Utils.waitFor(
+          () ->
+              ((TopicHandler.TopicInfo) handler.get(Channel.ofTarget(topicName))).partitions.size()
+                  == 30);
+      var groupByBroker =
+          ((TopicHandler.TopicInfo) handler.get(Channel.ofTarget(topicName)))
+              .partitions.stream()
+                  .flatMap(p -> p.replicas.stream())
+                  .collect(Collectors.groupingBy(r -> r.broker));
+      // those brokers should host different number of partitions
+      Assertions.assertEquals(
+          3, groupByBroker.values().stream().map(List::size).collect(Collectors.toSet()).size());
+    }
+  }
 }
diff --git a/common/src/test/java/org/astraea/common/metrics/client/consumer/HasConsumerMetricsTest.java b/common/src/test/java/org/astraea/common/metrics/client/consumer/HasConsumerMetricsTest.java
new file mode 100644
index 00000000..864172c7
--- /dev/null
+++ b/common/src/test/java/org/astraea/common/metrics/client/consumer/HasConsumerMetricsTest.java
@@ -0,0 +1,104 @@
+/*
+ * 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.astraea.common.metrics.client.consumer;
+
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.Set;
+import java.util.stream.IntStream;
+import org.astraea.common.Utils;
+import org.astraea.common.consumer.Consumer;
+import org.astraea.common.metrics.MBeanClient;
+import org.astraea.common.producer.Producer;
+import org.astraea.it.RequireSingleBrokerCluster;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class HasConsumerMetricsTest extends RequireSingleBrokerCluster {
+
+  @Test
+  void testClientId() {
+    var topic = Utils.randomString(10);
+
+    try (var producer = Producer.of(bootstrapServers())) {
+      IntStream.range(0, 10)
+          .forEach(
+              index ->
+                  producer
+                      .sender()
+                      .topic(topic)
+                      .key(String.valueOf(index).getBytes(StandardCharsets.UTF_8))
+                      .run());
+    }
+    try (var consumer =
+        Consumer.forTopics(Set.of(topic))
+            .bootstrapServers(bootstrapServers())
+            .fromBeginning()
+            .build()) {
+      Assertions.assertEquals(10, consumer.poll(10, Duration.ofSeconds(5)).size());
+      consumer.commitOffsets(Duration.ofSeconds(2));
+      var metrics = ConsumerMetrics.consumer(MBeanClient.local());
+      Assertions.assertEquals(1, metrics.size());
+      var m = metrics.iterator().next();
+      Assertions.assertNotNull(m.clientId());
+      Assertions.assertDoesNotThrow(m::reauthenticationLatencyAvg);
+      Assertions.assertDoesNotThrow(m::ioTimeNsTotal);
+      Assertions.assertDoesNotThrow(m::successfulAuthenticationTotal);
+      Assertions.assertDoesNotThrow(m::ioWaittimeTotal);
+      Assertions.assertDoesNotThrow(m::committedTimeNsTotal);
+      Assertions.assertDoesNotThrow(m::reauthenticationLatencyMax);
+      Assertions.assertDoesNotThrow(m::successfulAuthenticationRate);
+      Assertions.assertDoesNotThrow(m::commitSyncTimeNsTotal);
+      Assertions.assertDoesNotThrow(m::failedAuthenticationTotal);
+      Assertions.assertDoesNotThrow(m::timeBetweenPollAvg);
+      Assertions.assertDoesNotThrow(m::connectionCount);
+      Assertions.assertDoesNotThrow(m::responseTotal);
+      Assertions.assertDoesNotThrow(m::requestRate);
+      Assertions.assertDoesNotThrow(m::incomingByteRate);
+      Assertions.assertDoesNotThrow(m::lastPollSecondsAgo);
+      Assertions.assertDoesNotThrow(m::successfulAuthenticationNoReauthTotal);
+      Assertions.assertDoesNotThrow(m::timeBetweenPollMax);
+      Assertions.assertDoesNotThrow(m::failedReauthenticationTotal);
+      Assertions.assertDoesNotThrow(m::selectRate);
+      Assertions.assertDoesNotThrow(m::successfulReauthenticationTotal);
+      Assertions.assertDoesNotThrow(m::requestTotal);
+      Assertions.assertDoesNotThrow(m::ioTimeNsAvg);
+      Assertions.assertDoesNotThrow(m::iotimeTotal);
+      Assertions.assertDoesNotThrow(m::ioWaitTimeNsTotal);
+      Assertions.assertDoesNotThrow(m::ioWaitRatio);
+      Assertions.assertDoesNotThrow(m::networkIoRate);
+      Assertions.assertDoesNotThrow(m::connectionCreationRate);
+      Assertions.assertDoesNotThrow(m::requestSizeAvg);
+      Assertions.assertDoesNotThrow(m::ioRatio);
+      Assertions.assertDoesNotThrow(m::responseRate);
+      Assertions.assertDoesNotThrow(m::successfulReauthenticationRate);
+      Assertions.assertDoesNotThrow(m::selectTotal);
+      Assertions.assertDoesNotThrow(m::outgoingByteTotal);
+      Assertions.assertDoesNotThrow(m::pollIdleRatioAvg);
+      Assertions.assertDoesNotThrow(m::connectionCloseRate);
+      Assertions.assertDoesNotThrow(m::incomingByteTotal);
+      Assertions.assertDoesNotThrow(m::requestSizeMax);
+      Assertions.assertDoesNotThrow(m::failedAuthenticationRate);
+      Assertions.assertDoesNotThrow(m::outgoingByteRate);
+      Assertions.assertDoesNotThrow(m::failedReauthenticationRate);
+      Assertions.assertDoesNotThrow(m::ioWaitTimeNsAvg);
+      Assertions.assertDoesNotThrow(m::connectionCloseTotal);
+      Assertions.assertDoesNotThrow(m::connectionCreationTotal);
+      Assertions.assertDoesNotThrow(m::networkIoTotal);
+    }
+  }
+}