Repository: skiptests/astraea
Pull request #35 (2021-09-30 16:19:45)

Implementation of KafkaMetricClient (early version)
Context: #10 

這個 PR 實作專門針對 Kafka 的 MetricClient

和 @wycccccc 討論後先初步實作 ``MessagesInPerSec``, ``MesasgeOutPerSec``, ``BytesInPerSec``, ``BytesOutPerSec`` 這些屬於 ``kafka.server:type=BrokerTopicMetrics,name=*`` 的 MBeans。

不過因為我手太癢所以把整個 ``BrokerTopicMetrics`` 下的東西都塞進去了，我想說整個過程我抽象重用了非常多程式碼，新增一個 Metrics 只會多一行程式碼，所以就弄下去了。

範例程式碼:

```java
package org.astraea;

import org.astraea.metrics.kafka.KafkaMetricClient;
import org.astraea.metrics.kafka.metrics.BrokerTopicMetrics;

import javax.management.remote.JMXServiceURL;
import java.net.MalformedURLException;

public class Ignore {

    public static String JMX_URI_FORMAT =
            "service:jmx:rmi:///jndi/rmi://" + "%s" + ":" + "%d" + "/jmxrmi";

    public static String createJmxUrl(String hostname, int port) {
        return String.format(JMX_URI_FORMAT, hostname, port);
    }

    public static void main(String[] args) throws MalformedURLException {
        JMXServiceURL serviceURL = new JMXServiceURL(createJmxUrl("localhost", 15460));
        try (KafkaMetricClient kafkaMetricClient = new KafkaMetricClient(serviceURL)) {
            while (!Thread.interrupted()) {
                System.out.println(kafkaMetricClient.requestMetric(BrokerTopicMetrics.BytesInPerSec));
                System.out.println(kafkaMetricClient.requestMetric(BrokerTopicMetrics.BytesOutPerSec));
                System.out.println();
                Thread.sleep(1000);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
```

```
BytesInPerSec{
  RateUnit=SECONDS
  OneMinuteRate=2999747.855125004
  EventType=bytes
  Count=205421310
  FifteenMinuteRate=226200.60903020878
  FiveMinuteRate=666548.5879528691
  MeanRate=38938.808808609356}
BytesOutPerSec{
  RateUnit=SECONDS
  OneMinuteRate=805314.3588730913
  EventType=bytes
  Count=56052049
  FifteenMinuteRate=61653.309494084264
  FiveMinuteRate=181272.92186247595
  MeanRate=10624.989980656785}

BytesInPerSec{
  RateUnit=SECONDS
  OneMinuteRate=2999747.855125004
  EventType=bytes
  Count=205421310
  FifteenMinuteRate=226200.60903020878
  FiveMinuteRate=666548.5879528691
  MeanRate=38931.39195484974}
BytesOutPerSec{
  RateUnit=SECONDS
  OneMinuteRate=805314.3588730913
  EventType=bytes
  Count=56052049
  FifteenMinuteRate=61653.309494084264
  FiveMinuteRate=181272.92186247595
  MeanRate=10622.963633234727}

BytesInPerSec{
  RateUnit=SECONDS
  OneMinuteRate=2999747.855125004
  EventType=bytes
  Count=205421310
  FifteenMinuteRate=226200.60903020878
  FiveMinuteRate=666548.5879528691
  MeanRate=38923.96572713439}
BytesOutPerSec{
  RateUnit=SECONDS
  OneMinuteRate=805314.3588730913
  EventType=bytes
  Count=56052049
  FifteenMinuteRate=61653.309494084264
  FiveMinuteRate=181272.92186247595
  MeanRate=10620.937277793315}

...
```

```
 

garyparrot: > 可否用你寫的這個client code先弄一隻簡單的main function，可以就顯示BrokerTopicMetricsResult的內容就好

下面是 Main function 的內容，寫的過程中嘗試去從真的 kafka broker 撈所有資料，意外發現 ``HasRate#rateUnit()`` 回傳的形態有問題，單元測試沒發現這個問題是因爲不是和真的 Kafka Instance 互動，而是 mock 的假資料，能夠做到這種等級的驗證的恐怕是 integration test。這個在 1ea8037 修好了。

```java
package org.astraea.metrics.kafka.metrics;

import java.net.MalformedURLException;
import javax.management.remote.JMXServiceURL;
import org.astraea.metrics.kafka.KafkaMetricClient;

public class Ignore {

  public static String JMX_URI_FORMAT =
      "service:jmx:rmi:///jndi/rmi://" + "%s" + ":" + "%d" + "/jmxrmi";

  public static String createJmxUrl(String hostname, int port) {
    return String.format(JMX_URI_FORMAT, hostname, port);
  }

  public static void main(String[] args) throws MalformedURLException {
    JMXServiceURL jmxServiceURL = new JMXServiceURL(createJmxUrl("localhost", 17440));
    try (KafkaMetricClient kafkaMetricClient = new KafkaMetricClient(jmxServiceURL)) {
      BrokerTopicMetricsResult result =
          kafkaMetricClient.requestMetric(BrokerTopicMetrics.BytesInPerSec);
      System.out.printf("count             : %d\n", result.count());
      System.out.printf("meanRate          : %f\n", result.meanRate());
      System.out.printf("oneMinuteRate     : %f\n", result.oneMinuteRate());
      System.out.printf("fiveMinuteRate    : %f\n", result.fiveMinuteRate());
      System.out.printf("fifteenMinuteRate : %f\n", result.fifteenMinuteRate());
      System.out.printf("rateUnit          : %s\n", result.rateUnit());
      System.out.printf("eventType         : %s\n", result.eventType());
    } catch (Exception e) {
      e.printStackTrace();
    }
  }
}
```
chia7712: > 下面是 Main function 的內容

不好意思我沒說清楚，可否把那隻main function放到此ＰＲ理面？
garyparrot: > 不好意思我沒說清楚，可否把那隻main function放到此ＰＲ理面？

是指向像下面這個連結一樣，直接將範例塞到 codebase 裡面，當做 example 參考用嗎

https://github.com/jacoco/jacoco/blob/master/org.jacoco.examples/src/org/jacoco/examples/MBeanClient.java
chia7712: > 是指向像下面這個連結一樣，直接將範例塞到 codebase 裡面，當做 example 參考用嗎

對，你可以看專案內其他main class的作法。等到後面正式的metrics tool出來後再把你這個臨時的版本刪掉就好

Patch:
diff --git a/app/src/main/java/org/astraea/App.java b/app/src/main/java/org/astraea/App.java
index a6ad139e..c0db58cf 100644
--- a/app/src/main/java/org/astraea/App.java
+++ b/app/src/main/java/org/astraea/App.java
@@ -4,12 +4,13 @@ import java.util.Arrays;
 import java.util.List;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
+import org.astraea.metrics.kafka.KafkaMetricClientApp;
 import org.astraea.offset.OffsetExplorer;
 import org.astraea.performance.latency.End2EndLatency;
 
 public class App {
   private static final List<Class<?>> MAIN_CLASSES =
-      Arrays.asList(End2EndLatency.class, OffsetExplorer.class);
+      Arrays.asList(End2EndLatency.class, OffsetExplorer.class, KafkaMetricClientApp.class);
 
   private static String toString(List<Class<?>> mains) {
     return mains.stream().map(Class::getName).collect(Collectors.joining(","));
diff --git a/app/src/main/java/org/astraea/metrics/kafka/KafkaMetricClient.java b/app/src/main/java/org/astraea/metrics/kafka/KafkaMetricClient.java
new file mode 100644
index 00000000..48a9655b
--- /dev/null
+++ b/app/src/main/java/org/astraea/metrics/kafka/KafkaMetricClient.java
@@ -0,0 +1,49 @@
+package org.astraea.metrics.kafka;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import javax.management.remote.JMXServiceURL;
+import org.astraea.metrics.jmx.BeanObject;
+import org.astraea.metrics.jmx.BeanQuery;
+import org.astraea.metrics.jmx.MBeanClient;
+import org.astraea.metrics.kafka.metrics.BrokerTopicMetrics;
+import org.astraea.metrics.kafka.metrics.BrokerTopicMetricsResult;
+
+public class KafkaMetricClient implements AutoCloseable {
+
+  private final MBeanClient mBeanClient;
+
+  public KafkaMetricClient(JMXServiceURL serviceURL) {
+    try {
+      this.mBeanClient = new MBeanClient(serviceURL);
+    } catch (IOException e) {
+      throw new UncheckedIOException(e);
+    }
+  }
+
+  /**
+   * Request BrokerTopicMetrics.
+   *
+   * <p>For the available metrics, checkout {@link BrokerTopicMetrics}.
+   *
+   * @param metric the metric to request
+   * @return a {@link BrokerTopicMetricsResult} indicate the result
+   */
+  public BrokerTopicMetricsResult requestMetric(BrokerTopicMetrics metric) {
+    BeanObject beanObject =
+        mBeanClient
+            .tryQueryBean(
+                BeanQuery.builder("kafka.server")
+                    .property("type", "BrokerTopicMetrics")
+                    .property("name", metric.metricName())
+                    .build())
+            .orElseThrow();
+
+    return new BrokerTopicMetricsResult(metric, beanObject);
+  }
+
+  @Override
+  public void close() throws Exception {
+    this.mBeanClient.close();
+  }
+}
diff --git a/app/src/main/java/org/astraea/metrics/kafka/KafkaMetricClientApp.java b/app/src/main/java/org/astraea/metrics/kafka/KafkaMetricClientApp.java
new file mode 100644
index 00000000..33cfeb48
--- /dev/null
+++ b/app/src/main/java/org/astraea/metrics/kafka/KafkaMetricClientApp.java
@@ -0,0 +1,87 @@
+package org.astraea.metrics.kafka;
+
+import java.net.MalformedURLException;
+import java.time.LocalTime;
+import java.time.format.DateTimeFormatter;
+import java.time.format.FormatStyle;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import javax.management.remote.JMXServiceURL;
+import org.astraea.metrics.kafka.metrics.BrokerTopicMetrics;
+import org.astraea.metrics.kafka.metrics.BrokerTopicMetricsResult;
+
+public final class KafkaMetricClientApp {
+
+  public static String JMX_URI_FORMAT = "service:jmx:rmi:///jndi/rmi://" + "%s" + "/jmxrmi";
+
+  public static String createJmxUrl(String address) {
+    return String.format(JMX_URI_FORMAT, address);
+  }
+
+  public static void main(String[] args) throws MalformedURLException {
+
+    // ensure argument safe
+    if (args.length < 1) {
+      help();
+      throw new IllegalArgumentException();
+    }
+
+    String argumentJmxServerNetworkAddress = args[0];
+    List<String> argumentTargetMetrics = List.of(args).subList(1, args.length);
+
+    JMXServiceURL serviceURL = new JMXServiceURL(createJmxUrl(argumentJmxServerNetworkAddress));
+    try (KafkaMetricClient kafkaMetricClient = new KafkaMetricClient(serviceURL)) {
+
+      // find the actual metrics to fetch.
+      List<BrokerTopicMetrics> metrics =
+          argumentTargetMetrics.stream()
+              .map(BrokerTopicMetrics::valueOf)
+              .collect(Collectors.toUnmodifiableList());
+
+      // if no metric name specified, all metrics are selected
+      if (argumentTargetMetrics.size() == 0) metrics = List.of(BrokerTopicMetrics.values());
+
+      while (!Thread.interrupted()) {
+        // fetch
+        List<BrokerTopicMetricsResult> collect =
+            metrics.stream()
+                .map(kafkaMetricClient::requestMetric)
+                .collect(Collectors.toUnmodifiableList());
+
+        // output
+        System.out.println(
+            "["
+                + LocalTime.now().format(DateTimeFormatter.ofLocalizedTime(FormatStyle.MEDIUM))
+                + "]");
+        for (BrokerTopicMetricsResult result : collect) System.out.println(result);
+        System.out.println();
+
+        try {
+          TimeUnit.SECONDS.sleep(1);
+        } catch (InterruptedException e) {
+          break;
+        }
+      }
+
+    } catch (Exception e) {
+      e.printStackTrace();
+      help();
+    }
+  }
+
+  private static void help() {
+    String simpleName = KafkaMetricClientApp.class.getSimpleName();
+    System.err.printf("Usage: %s <jmx server address> [metric names ...]\n", simpleName);
+    System.err.println();
+    System.err.println("If no metric name specified in argument, all metrics will be selected.");
+    System.err.println();
+    System.err.printf("Example 1: %s localhost:9875\n", simpleName);
+    System.err.printf("Example 2: %s localhost:9875 BytesInPerSec BytesOutPerSec\n", simpleName);
+    System.err.println();
+    System.err.println("Available Metrics:");
+    for (BrokerTopicMetrics value : BrokerTopicMetrics.values()) {
+      System.err.printf("    %s\n", value.metricName());
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/metrics/kafka/metrics/BrokerTopicMetrics.java b/app/src/main/java/org/astraea/metrics/kafka/metrics/BrokerTopicMetrics.java
new file mode 100644
index 00000000..a772609e
--- /dev/null
+++ b/app/src/main/java/org/astraea/metrics/kafka/metrics/BrokerTopicMetrics.java
@@ -0,0 +1,60 @@
+package org.astraea.metrics.kafka.metrics;
+
+public enum BrokerTopicMetrics {
+  /** Message validation failure rate due to non-continuous offset or sequence number in batch */
+  InvalidOffsetOrSequenceRecordsPerSec("InvalidOffsetOrSequenceRecordsPerSec"),
+
+  /** Message validation failure rate due to incorrect crc checksum */
+  InvalidMessageCrcRecordsPerSec("InvalidMessageCrcRecordsPerSec"),
+
+  FetchMessageConversionsPerSec("FetchMessageConversionsPerSec"),
+
+  BytesRejectedPerSec("BytesRejectedPerSec"),
+
+  /** Message in rate */
+  MessagesInPerSec("MessagesInPerSec"),
+
+  /** Incoming byte rate of reassignment traffic */
+  ReassignmentBytesInPerSec("ReassignmentBytesInPerSec"),
+
+  FailedFetchRequestsPerSec("FailedFetchRequestsPerSec"),
+
+  /** Byte in rate from other brokers */
+  ReplicationBytesInPerSec("ReplicationBytesInPerSec"),
+
+  /** Message validation failure rate due to no key specified for compacted topic */
+  NoKeyCompactedTopicRecordsPerSec("NoKeyCompactedTopicRecordsPerSec"),
+
+  TotalFetchRequestsPerSec("TotalFetchRequestsPerSec"),
+
+  FailedProduceRequestsPerSec("FailedProduceRequestsPerSec"),
+
+  /** Byte in rate from clients */
+  BytesInPerSec("BytesInPerSec"),
+
+  TotalProduceRequestsPerSec("TotalProduceRequestsPerSec"),
+
+  /** Message validation failure rate due to invalid magic number */
+  InvalidMagicNumberRecordsPerSec("InvalidMagicNumberRecordsPerSec"),
+
+  /** Outgoing byte rate of reassignment traffic */
+  ReassignmentBytesOutPerSec("ReassignmentBytesOutPerSec"),
+
+  /** Bytes in rate from other brokers */
+  ReplicationBytesOutPerSec("ReplicationBytesOutPerSec"),
+
+  ProduceMessageConversionsPerSec("ProduceMessageConversionsPerSec"),
+
+  /** Byte out rate to clients. */
+  BytesOutPerSec("BytesOutPerSec");
+
+  private final String metricName;
+
+  BrokerTopicMetrics(String name) {
+    this.metricName = name;
+  }
+
+  public String metricName() {
+    return metricName;
+  }
+}
diff --git a/app/src/main/java/org/astraea/metrics/kafka/metrics/BrokerTopicMetricsResult.java b/app/src/main/java/org/astraea/metrics/kafka/metrics/BrokerTopicMetricsResult.java
new file mode 100644
index 00000000..f3c77410
--- /dev/null
+++ b/app/src/main/java/org/astraea/metrics/kafka/metrics/BrokerTopicMetricsResult.java
@@ -0,0 +1,35 @@
+package org.astraea.metrics.kafka.metrics;
+
+import java.util.Map;
+import java.util.Objects;
+import org.astraea.metrics.jmx.BeanObject;
+import org.astraea.metrics.kafka.metrics.modifiers.*;
+
+public class BrokerTopicMetricsResult implements HasCount, HasEventType, HasRate {
+
+  private final BeanObject beanObject;
+  private final BrokerTopicMetrics metric;
+
+  public BrokerTopicMetricsResult(BrokerTopicMetrics metric, BeanObject beanObject) {
+    this.beanObject = Objects.requireNonNull(beanObject);
+    this.metric = Objects.requireNonNull(metric);
+  }
+
+  @Override
+  public String toString() {
+    StringBuilder sb = new StringBuilder();
+    for (Map.Entry<String, Object> e : beanObject().getAttributes().entrySet()) {
+      sb.append(System.lineSeparator())
+          .append("  ")
+          .append(e.getKey())
+          .append("=")
+          .append(e.getValue());
+    }
+    return metric.name() + "{" + sb + "}";
+  }
+
+  @Override
+  public BeanObject beanObject() {
+    return beanObject;
+  }
+}
diff --git a/app/src/main/java/org/astraea/metrics/kafka/metrics/modifiers/HasBeanObject.java b/app/src/main/java/org/astraea/metrics/kafka/metrics/modifiers/HasBeanObject.java
new file mode 100644
index 00000000..608e6aff
--- /dev/null
+++ b/app/src/main/java/org/astraea/metrics/kafka/metrics/modifiers/HasBeanObject.java
@@ -0,0 +1,7 @@
+package org.astraea.metrics.kafka.metrics.modifiers;
+
+import org.astraea.metrics.jmx.BeanObject;
+
+public interface HasBeanObject {
+  BeanObject beanObject();
+}
diff --git a/app/src/main/java/org/astraea/metrics/kafka/metrics/modifiers/HasCount.java b/app/src/main/java/org/astraea/metrics/kafka/metrics/modifiers/HasCount.java
new file mode 100644
index 00000000..236bbfff
--- /dev/null
+++ b/app/src/main/java/org/astraea/metrics/kafka/metrics/modifiers/HasCount.java
@@ -0,0 +1,7 @@
+package org.astraea.metrics.kafka.metrics.modifiers;
+
+public interface HasCount extends HasBeanObject {
+  default long count() {
+    return (long) beanObject().getAttributes().getOrDefault("Count", 0);
+  }
+}
diff --git a/app/src/main/java/org/astraea/metrics/kafka/metrics/modifiers/HasEventType.java b/app/src/main/java/org/astraea/metrics/kafka/metrics/modifiers/HasEventType.java
new file mode 100644
index 00000000..827209a4
--- /dev/null
+++ b/app/src/main/java/org/astraea/metrics/kafka/metrics/modifiers/HasEventType.java
@@ -0,0 +1,7 @@
+package org.astraea.metrics.kafka.metrics.modifiers;
+
+public interface HasEventType extends HasBeanObject {
+  default String eventType() {
+    return (String) beanObject().getAttributes().get("EventType");
+  }
+}
diff --git a/app/src/main/java/org/astraea/metrics/kafka/metrics/modifiers/HasRate.java b/app/src/main/java/org/astraea/metrics/kafka/metrics/modifiers/HasRate.java
new file mode 100644
index 00000000..91627ae1
--- /dev/null
+++ b/app/src/main/java/org/astraea/metrics/kafka/metrics/modifiers/HasRate.java
@@ -0,0 +1,25 @@
+package org.astraea.metrics.kafka.metrics.modifiers;
+
+import java.util.concurrent.TimeUnit;
+
+public interface HasRate extends HasBeanObject {
+  default double meanRate() {
+    return (double) beanObject().getAttributes().getOrDefault("MeanRate", 0);
+  }
+
+  default double oneMinuteRate() {
+    return (double) beanObject().getAttributes().getOrDefault("OneMinuteRate", 0);
+  }
+
+  default double fiveMinuteRate() {
+    return (double) beanObject().getAttributes().getOrDefault("FiveMinuteRate", 0);
+  }
+
+  default double fifteenMinuteRate() {
+    return (double) beanObject().getAttributes().getOrDefault("FifteenMinuteRate", 0);
+  }
+
+  default TimeUnit rateUnit() {
+    return (TimeUnit) beanObject().getAttributes().get("RateUnit");
+  }
+}


Test patch:
diff --git a/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricClientAppTest.java b/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricClientAppTest.java
new file mode 100644
index 00000000..6d2db880
--- /dev/null
+++ b/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricClientAppTest.java
@@ -0,0 +1,31 @@
+package org.astraea.metrics.kafka;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.io.OutputStream;
+import java.io.PrintStream;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+class KafkaMetricClientAppTest {
+
+  private final PrintStream stderr = System.err;
+
+  @BeforeEach
+  void beforeEach() {
+    // swallow stderr output to null stream, avoid help() info from KafkaMetricClientApp
+    System.setErr(new PrintStream(OutputStream.nullOutputStream()));
+  }
+
+  @AfterEach
+  void afterEach() {
+    // restore stderr stream
+    System.setErr(stderr);
+  }
+
+  @Test
+  void testBadArguments() {
+    assertThrows(IllegalArgumentException.class, () -> KafkaMetricClientApp.main(new String[0]));
+  }
+}
diff --git a/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricClientTest.java b/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricClientTest.java
new file mode 100644
index 00000000..d4c58782
--- /dev/null
+++ b/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricClientTest.java
@@ -0,0 +1,103 @@
+package org.astraea.metrics.kafka;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.io.IOException;
+import java.lang.management.ManagementFactory;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import javax.management.*;
+import javax.management.remote.JMXConnectorServer;
+import javax.management.remote.JMXConnectorServerFactory;
+import javax.management.remote.JMXServiceURL;
+import org.astraea.metrics.jmx.Utility;
+import org.astraea.metrics.kafka.metrics.BrokerTopicMetrics;
+import org.astraea.metrics.kafka.metrics.BrokerTopicMetricsResult;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+class KafkaMetricClientTest {
+
+  private MBeanServer mBeanServer;
+  private JMXConnectorServer jmxServer;
+  private Map<ObjectName, Object> registeredBeans = new HashMap<>();
+  private KafkaMetricClient sut;
+
+  private void register(ObjectName name, Object mBean) {
+    registeredBeans.put(name, mBean);
+    try {
+      mBeanServer.registerMBean(mBean, name);
+    } catch (InstanceAlreadyExistsException
+        | MBeanRegistrationException
+        | NotCompliantMBeanException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  private void clearRegisteredMBeans() {
+    registeredBeans.forEach(
+        (name, mbeans) -> {
+          try {
+            mBeanServer.unregisterMBean(name);
+          } catch (InstanceNotFoundException | MBeanRegistrationException e) {
+            throw new RuntimeException(e);
+          }
+        });
+    registeredBeans.clear();
+  }
+
+  @BeforeEach
+  void setUp() throws IOException {
+    JMXServiceURL serviceURL = new JMXServiceURL("service:jmx:rmi://127.0.0.1");
+
+    mBeanServer = ManagementFactory.getPlatformMBeanServer();
+
+    jmxServer = JMXConnectorServerFactory.newJMXConnectorServer(serviceURL, null, mBeanServer);
+    jmxServer.start();
+
+    sut = new KafkaMetricClient(jmxServer.getAddress());
+  }
+
+  @AfterEach
+  void tearDown() throws Exception {
+    jmxServer.stop();
+    clearRegisteredMBeans();
+    mBeanServer = null;
+    sut.close();
+  }
+
+  @ParameterizedTest
+  @EnumSource(value = BrokerTopicMetrics.class)
+  void testRequestBrokerTopicMetrics(BrokerTopicMetrics metric)
+      throws MalformedObjectNameException {
+    // arrange
+    Object mbean =
+        Utility.createReadOnlyDynamicMBean(
+            Map.of(
+                "Count", 1L,
+                "EventType", "bytes",
+                "FifteenMinuteRate", 2.0,
+                "FiveMinuteRate", 4.0,
+                "MeanRate", 8.0,
+                "OneMinuteRate", 16.0,
+                "RateUnit", TimeUnit.SECONDS));
+    register(
+        ObjectName.getInstance("kafka.server:type=BrokerTopicMetrics,name=" + metric.name()),
+        mbean);
+
+    // act
+    BrokerTopicMetricsResult result = sut.requestMetric(metric);
+
+    // assert
+    assertEquals(1L, result.count());
+    assertEquals("bytes", result.eventType());
+    assertEquals(2.0, result.fifteenMinuteRate());
+    assertEquals(4.0, result.fiveMinuteRate());
+    assertEquals(8.0, result.meanRate());
+    assertEquals(16.0, result.oneMinuteRate());
+    assertEquals(TimeUnit.SECONDS, result.rateUnit());
+  }
+}
diff --git a/app/src/test/java/org/astraea/metrics/kafka/metrics/BrokerTopicMetricsTest.java b/app/src/test/java/org/astraea/metrics/kafka/metrics/BrokerTopicMetricsTest.java
new file mode 100644
index 00000000..a38c37d7
--- /dev/null
+++ b/app/src/test/java/org/astraea/metrics/kafka/metrics/BrokerTopicMetricsTest.java
@@ -0,0 +1,23 @@
+package org.astraea.metrics.kafka.metrics;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.util.Arrays;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.junit.jupiter.api.Test;
+
+class BrokerTopicMetricsTest {
+
+  @Test
+  void testAllEnumNameUnique() {
+    // arrange act
+    Set<String> collectedName =
+        Arrays.stream(BrokerTopicMetrics.values())
+            .map(BrokerTopicMetrics::metricName)
+            .collect(Collectors.toSet());
+
+    // assert
+    assertEquals(BrokerTopicMetrics.values().length, collectedName.size());
+  }
+}