Repository: skiptests/astraea
Pull request #521 (2022-07-28 17:36:46)

Refactor broker topic metrics
as title


Patch:
diff --git a/README.md b/README.md
index a5022ed3..ffee3952 100644
--- a/README.md
+++ b/README.md
@@ -15,9 +15,8 @@ This project offers many kafka tools to simplify the life for kafka users.
 1. [Kafka quick start](#kafka-cluster-quick-start): set up a true kafka cluster in one minute
 2. [Kafka performance](#Performance-Benchmark): check producing/consuming performance.
 3. [Kafka metric explorer](#kafka-metric-explorer): utility for accessing kafka Mbean metrics via JMX.
-4. [Kafka partition score](#Kafka-partition-score): score all broker's partitions. 
-5. [Kafka replica syncing monitor](#Kafka-replica-syncing-monitor): Tracking replica syncing progress.
-6. [Astraea Web Server 中文文件連結](./docs/web_server/README.md)
+4. [Kafka replica syncing monitor](#Kafka-replica-syncing-monitor): Tracking replica syncing progress.
+5. [Astraea Web Server 中文文件連結](./docs/web_server/README.md)
 
 [Github packages](https://github.com/orgs/skiptests/packages?repo_name=astraea) offers the docker image to run mentioned tools
 ```shell
@@ -254,22 +253,6 @@ Run the tool from release
 
 ---
 
-## Kafka Partition Score
-
-This tool will score the partition on brokers, the higher score the heavier load.
-
-### Start scoring partitions on broker address "192.168.103.39:9092"
-
-```shell
-./gradlew run --args="score --bootstrap.servers 192.168.103.39:9092"
-```
-
-### Partition Score Configurations
-
-1. --bootstrap.servers: the server to connect to
-2. --exclude.internal.topics: True if you want to ignore internal topics like _consumer_offsets while counting score. 
-3. --hide.balanced: True if you want to hide topics and partitions thar already balanced.:q
-
 ## Kafka Replica Syncing Monitor
 
 This tool will track partition replica syncing progress. This tool can be used to observe 
@@ -316,4 +299,4 @@ $ ./gradlew run --args="monitor --bootstrap.servers 192.168.103.39:9092"
 2. --interval: the frequency(time interval in second) to check replica state, support floating point value. (default: 1 second)
 3. --prop.file: the path to a file that containing the properties to be passed to kafka admin.
 4. --topic: topics to track (default: track all non-synced partition by default)
-5. --track: keep track even if all the replicas are synced. Also attempts to discover any non-synced replicas. (default: false)
\ No newline at end of file
+5. --track: keep track even if all the replicas are synced. Also attempts to discover any non-synced replicas. (default: false)
diff --git a/app/src/main/java/org/astraea/app/App.java b/app/src/main/java/org/astraea/app/App.java
index b6e23e32..c1177822 100644
--- a/app/src/main/java/org/astraea/app/App.java
+++ b/app/src/main/java/org/astraea/app/App.java
@@ -23,7 +23,6 @@ import java.util.List;
 import java.util.Map;
 import org.astraea.app.admin.ReplicaSyncingMonitor;
 import org.astraea.app.automation.Automation;
-import org.astraea.app.cost.topic.PartitionScore;
 import org.astraea.app.metrics.MetricExplorer;
 import org.astraea.app.performance.Performance;
 import org.astraea.app.web.WebService;
@@ -32,7 +31,6 @@ public class App {
   private static final Map<String, Class<?>> MAIN_CLASSES =
       Map.of(
           "metrics", MetricExplorer.class,
-          "score", PartitionScore.class,
           "performance", Performance.class,
           "monitor", ReplicaSyncingMonitor.class,
           "automation", Automation.class,
diff --git a/app/src/main/java/org/astraea/app/cost/BrokerInputCost.java b/app/src/main/java/org/astraea/app/cost/BrokerInputCost.java
index b93d2cce..fd8aa7e4 100644
--- a/app/src/main/java/org/astraea/app/cost/BrokerInputCost.java
+++ b/app/src/main/java/org/astraea/app/cost/BrokerInputCost.java
@@ -22,8 +22,7 @@ import java.util.Optional;
 import java.util.stream.Collectors;
 import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
-import org.astraea.app.metrics.KafkaMetrics;
-import org.astraea.app.metrics.broker.BrokerTopicMetricsResult;
+import org.astraea.app.metrics.broker.ServerMetrics;
 import org.astraea.app.metrics.collector.Fetcher;
 
 public class BrokerInputCost implements HasBrokerCost {
@@ -35,14 +34,14 @@ public class BrokerInputCost implements HasBrokerCost {
                 Collectors.toMap(
                     Map.Entry::getKey,
                     entry ->
-                        KafkaMetrics.BrokerTopic.BytesInPerSec.of(entry.getValue()).stream()
-                            .mapToDouble(BrokerTopicMetricsResult::oneMinuteRate)
+                        ServerMetrics.Topic.BYTES_IN_PER_SEC.of(entry.getValue()).stream()
+                            .mapToDouble(ServerMetrics.Topic.Meter::oneMinuteRate)
                             .sum()));
     return () -> brokerCost;
   }
 
   @Override
   public Optional<Fetcher> fetcher() {
-    return Optional.of(client -> List.of(KafkaMetrics.BrokerTopic.BytesInPerSec.fetch(client)));
+    return Optional.of(client -> List.of(ServerMetrics.Topic.BYTES_IN_PER_SEC.fetch(client)));
   }
 }
diff --git a/app/src/main/java/org/astraea/app/cost/BrokerOutputCost.java b/app/src/main/java/org/astraea/app/cost/BrokerOutputCost.java
index c4a7b351..14acdd85 100644
--- a/app/src/main/java/org/astraea/app/cost/BrokerOutputCost.java
+++ b/app/src/main/java/org/astraea/app/cost/BrokerOutputCost.java
@@ -22,8 +22,7 @@ import java.util.Optional;
 import java.util.stream.Collectors;
 import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
-import org.astraea.app.metrics.KafkaMetrics;
-import org.astraea.app.metrics.broker.BrokerTopicMetricsResult;
+import org.astraea.app.metrics.broker.ServerMetrics;
 import org.astraea.app.metrics.collector.Fetcher;
 
 public class BrokerOutputCost implements HasBrokerCost {
@@ -36,14 +35,14 @@ public class BrokerOutputCost implements HasBrokerCost {
                 Collectors.toMap(
                     Map.Entry::getKey,
                     entry ->
-                        KafkaMetrics.BrokerTopic.BytesOutPerSec.of(entry.getValue()).stream()
-                            .mapToDouble(BrokerTopicMetricsResult::oneMinuteRate)
+                        ServerMetrics.Topic.BYTES_OUT_PER_SEC.of(entry.getValue()).stream()
+                            .mapToDouble(ServerMetrics.Topic.Meter::oneMinuteRate)
                             .sum()));
     return () -> brokerCost;
   }
 
   @Override
   public Optional<Fetcher> fetcher() {
-    return Optional.of(client -> List.of(KafkaMetrics.BrokerTopic.BytesOutPerSec.fetch(client)));
+    return Optional.of(client -> List.of(ServerMetrics.Topic.BYTES_OUT_PER_SEC.fetch(client)));
   }
 }
diff --git a/app/src/main/java/org/astraea/app/cost/LoadCost.java b/app/src/main/java/org/astraea/app/cost/LoadCost.java
index 469e5889..ed44e03d 100644
--- a/app/src/main/java/org/astraea/app/cost/LoadCost.java
+++ b/app/src/main/java/org/astraea/app/cost/LoadCost.java
@@ -26,8 +26,7 @@ import java.util.stream.IntStream;
 import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.metrics.HasBeanObject;
-import org.astraea.app.metrics.KafkaMetrics;
-import org.astraea.app.metrics.broker.BrokerTopicMetricsResult;
+import org.astraea.app.metrics.broker.ServerMetrics;
 import org.astraea.app.metrics.collector.Fetcher;
 import org.astraea.app.partitioner.PartitionerUtils;
 
@@ -35,9 +34,9 @@ public class LoadCost implements HasBrokerCost {
   private final Map<Integer, BrokerMetric> brokersMetric = new HashMap<>();
   private final Map<String, Double> metricNameAndWeight =
       Map.of(
-          KafkaMetrics.BrokerTopic.BytesInPerSec.metricName(),
+          ServerMetrics.Topic.BYTES_IN_PER_SEC.metricName(),
           0.5,
-          KafkaMetrics.BrokerTopic.BytesOutPerSec.metricName(),
+          ServerMetrics.Topic.BYTES_OUT_PER_SEC.metricName(),
           0.5);
 
   /** Do "Poisson" and "weightPoisson" calculation on "load". And change output to double. */
@@ -86,8 +85,8 @@ public class LoadCost implements HasBrokerCost {
         (brokerID, value) -> {
           if (!brokersMetric.containsKey(brokerID)) brokersMetric.put(brokerID, new BrokerMetric());
           value.stream()
-              .filter(hasBeanObject -> hasBeanObject instanceof BrokerTopicMetricsResult)
-              .map(hasBeanObject -> (BrokerTopicMetricsResult) hasBeanObject)
+              .filter(hasBeanObject -> hasBeanObject instanceof ServerMetrics.Topic.Meter)
+              .map(hasBeanObject -> (ServerMetrics.Topic.Meter) hasBeanObject)
               .forEach(
                   result ->
                       brokersMetric
@@ -151,8 +150,8 @@ public class LoadCost implements HasBrokerCost {
     return Optional.of(
         client ->
             List.of(
-                KafkaMetrics.BrokerTopic.BytesInPerSec.fetch(client),
-                KafkaMetrics.BrokerTopic.BytesOutPerSec.fetch(client)));
+                ServerMetrics.Topic.BYTES_IN_PER_SEC.fetch(client),
+                ServerMetrics.Topic.BYTES_OUT_PER_SEC.fetch(client)));
   }
 
   private static class BrokerMetric {
diff --git a/app/src/main/java/org/astraea/app/cost/topic/CalculateUtils.java b/app/src/main/java/org/astraea/app/cost/topic/CalculateUtils.java
deleted file mode 100644
index 54ee0dc8..00000000
--- a/app/src/main/java/org/astraea/app/cost/topic/CalculateUtils.java
+++ /dev/null
@@ -1,131 +0,0 @@
-/*
- * 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.app.cost.topic;
-
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-import java.util.TreeMap;
-import java.util.function.Function;
-import java.util.stream.Collectors;
-import org.astraea.app.admin.TopicPartition;
-
-public class CalculateUtils {
-  public static Map<Integer, Map<TopicPartition, Double>> getLoad(
-      Map<Integer, Map<TopicPartition, Integer>> brokerPartitionSize,
-      Map<String, Integer> retentionMillis) {
-    Map<Integer, Map<TopicPartition, Double>> brokerPartitionLoad = new HashMap<>();
-
-    brokerPartitionSize
-        .keySet()
-        .forEach(
-            (broker) -> {
-              Map<TopicPartition, Double> partitionLoad =
-                  brokerPartitionSize.get(broker).keySet().stream()
-                      .filter(partition -> retentionMillis.containsKey(partition.topic()))
-                      .collect(
-                          Collectors.toMap(
-                              Function.identity(),
-                              partition ->
-                                  (double) brokerPartitionSize.get(broker).get(partition)
-                                      / retentionMillis.get(partition.topic())));
-              brokerPartitionLoad.put(broker, partitionLoad);
-            });
-    return brokerPartitionLoad;
-  }
-
-  private static Double countSum(Set<Double> in) {
-    return in.stream().mapToDouble(i -> i).sum();
-  }
-
-  public static Map<Integer, Map<TopicPartition, Double>> getScore(
-      Map<Integer, Map<TopicPartition, Double>> load) {
-    Map<Integer, Double> brokerLoad = new HashMap<>();
-    Map<TopicPartition, Double> partitionLoad = new HashMap<>();
-    Map<Integer, Double> partitionSD = new HashMap<>();
-    Map<Integer, Double> partitionMean = new HashMap<>();
-    Map<Integer, Map<TopicPartition, Double>> brokerPartitionScore = new HashMap<>();
-    double brokerSD;
-    Set<Double> loadSet = new HashSet<>();
-    Set<Double> LoadSQR = new HashSet<>();
-
-    load.keySet()
-        .forEach(
-            broker -> {
-              load.get(broker)
-                  .keySet()
-                  .forEach(
-                      tp -> {
-                        loadSet.add(load.get(broker).get(tp));
-                        partitionLoad.put(tp, load.get(broker).get(tp));
-                        LoadSQR.add(Math.pow(load.get(broker).get(tp), 2));
-                      });
-              var loadSum = countSum(loadSet);
-              var partitionNum = load.get(broker).keySet().size();
-              brokerLoad.put(broker, loadSum);
-              var mean = loadSum / load.get(broker).size();
-              partitionMean.put(broker, loadSum / load.get(broker).size());
-              var SD =
-                  Math.pow((countSum(LoadSQR) - mean * mean * partitionNum) / partitionNum, 0.5);
-              partitionSD.put(broker, SD);
-              loadSet.clear();
-              LoadSQR.clear();
-            });
-
-    brokerLoad
-        .keySet()
-        .forEach(
-            broker -> {
-              loadSet.add(brokerLoad.get(broker));
-              LoadSQR.add(Math.pow(brokerLoad.get(broker), 2));
-            });
-    var brokerLoadMean = countSum(loadSet) / brokerLoad.keySet().size();
-    var brokerLoadSQR = countSum(LoadSQR);
-    brokerSD =
-        Math.pow(
-            (brokerLoadSQR - brokerLoadMean * brokerLoadMean * brokerLoad.keySet().size())
-                / brokerLoad.keySet().size(),
-            0.5);
-    load.keySet()
-        .forEach(
-            broker -> {
-              var partitionScore = new TreeMap<TopicPartition, Double>();
-              load.get(broker)
-                  .keySet()
-                  .forEach(
-                      topicPartition -> {
-                        if (brokerLoad.get(broker) - brokerLoadMean > 0) {
-                          partitionScore.put(
-                              topicPartition,
-                              Math.round(
-                                      (((brokerLoad.get(broker) - brokerLoadMean) / brokerSD)
-                                              * ((partitionLoad.get(topicPartition)
-                                                      - partitionMean.get(broker))
-                                                  / partitionSD.get(broker))
-                                              * 60.0)
-                                          * 100.0)
-                                  / 100.0);
-                        } else {
-                          partitionScore.put(topicPartition, 0.0);
-                        }
-                        brokerPartitionScore.put(broker, partitionScore);
-                      });
-            });
-    return brokerPartitionScore;
-  }
-}
diff --git a/app/src/main/java/org/astraea/app/cost/topic/GetPartitionInf.java b/app/src/main/java/org/astraea/app/cost/topic/GetPartitionInf.java
deleted file mode 100644
index a8b39918..00000000
--- a/app/src/main/java/org/astraea/app/cost/topic/GetPartitionInf.java
+++ /dev/null
@@ -1,63 +0,0 @@
-/*
- * 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.app.cost.topic;
-
-import java.util.HashMap;
-import java.util.Map;
-import java.util.NoSuchElementException;
-import java.util.TreeMap;
-import java.util.stream.Collectors;
-import org.astraea.app.admin.Admin;
-import org.astraea.app.admin.TopicPartition;
-
-public class GetPartitionInf {
-  static Map<Integer, Map<TopicPartition, Integer>> getSize(Admin client) {
-    Map<Integer, Map<TopicPartition, Integer>> brokerPartitionSize = new HashMap<>();
-    client
-        .brokerIds()
-        .forEach(
-            (broker) -> {
-              var partitionSize = new TreeMap<TopicPartition, Integer>();
-              client
-                  .replicas(client.topicNames())
-                  .forEach(
-                      (tp, assignment) -> {
-                        assignment.forEach(
-                            partition -> {
-                              if (partition.broker() == broker)
-                                partitionSize.put(tp, (int) partition.size());
-                            });
-                        brokerPartitionSize.put(broker, partitionSize);
-                      });
-            });
-    return brokerPartitionSize;
-  }
-
-  static Map<String, Integer> getRetentionMillis(Admin client) {
-    return client.topics().entrySet().stream()
-        .collect(
-            Collectors.toMap(
-                Map.Entry::getKey,
-                entry ->
-                    entry
-                        .getValue()
-                        .value("retention.ms")
-                        .map(Integer::parseInt)
-                        .orElseThrow(
-                            () -> new NoSuchElementException("retention.ms does not exist"))));
-  }
-}
diff --git a/app/src/main/java/org/astraea/app/cost/topic/PartitionScore.java b/app/src/main/java/org/astraea/app/cost/topic/PartitionScore.java
deleted file mode 100644
index 59fd0582..00000000
--- a/app/src/main/java/org/astraea/app/cost/topic/PartitionScore.java
+++ /dev/null
@@ -1,118 +0,0 @@
-/*
- * 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.app.cost.topic;
-
-import com.beust.jcommander.Parameter;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.stream.Collectors;
-import org.astraea.app.admin.Admin;
-import org.astraea.app.admin.TopicPartition;
-import org.astraea.app.argument.BooleanField;
-
-public class PartitionScore {
-  public static void printScore(
-      Map<Integer, Map<TopicPartition, Double>> score, Argument argument) {
-    List<TopicPartition> partitionGood = new ArrayList<>();
-    Map<Integer, Boolean> brokerGood = new HashMap<>();
-    score
-        .keySet()
-        .forEach(
-            broker -> {
-              brokerGood.put(broker, true);
-              score
-                  .get(broker)
-                  .keySet()
-                  .forEach(
-                      tp -> {
-                        if (score.get(broker).get(tp) > 0) brokerGood.put(broker, false);
-                      });
-
-              if (!brokerGood.get(broker)) {
-                System.out.println("\nbroker: " + broker);
-                score
-                    .get(broker)
-                    .keySet()
-                    .forEach(
-                        tp -> {
-                          if (score.get(broker).get(tp) > 0) {
-                            System.out.println(tp + ": " + score.get(broker).get(tp));
-                          } else {
-                            partitionGood.add(tp);
-                          }
-                        });
-              }
-            });
-    if (!argument.hideBalanced) {
-      System.out.println(
-          "\nThe following brokers are balanced: "
-              + brokerGood.entrySet().stream()
-                  .filter(Map.Entry::getValue)
-                  .map(Map.Entry::getKey)
-                  .collect(Collectors.toSet()));
-
-      System.out.println(
-          "The following partitions are balanced: "
-              + partitionGood.stream()
-                  .sorted()
-                  .map(String::valueOf)
-                  .collect(Collectors.joining(", ", "[", "]")));
-    }
-  }
-
-  public static Map<Integer, Map<TopicPartition, Double>> execute(Argument argument, Admin admin) {
-    var internalTopic =
-        Set.of(
-            "__consumer_offsets",
-            "_confluent-command",
-            "_confluent-metrics",
-            "_confluent-telemetry-metrics",
-            "__transaction_state");
-    var brokerPartitionSize = GetPartitionInf.getSize(admin);
-    var retentionMillis = GetPartitionInf.getRetentionMillis(admin);
-    if (argument.excludeInternalTopic) internalTopic.forEach(retentionMillis::remove);
-    var load = CalculateUtils.getLoad(brokerPartitionSize, retentionMillis);
-    return CalculateUtils.getScore(load);
-  }
-
-  public static void main(String[] args) {
-    var argument = org.astraea.app.argument.Argument.parse(new Argument(), args);
-    var admin = Admin.of(argument.bootstrapServers());
-    var score = execute(argument, admin);
-    printScore(score, argument);
-  }
-
-  static class Argument extends org.astraea.app.argument.Argument {
-    @Parameter(
-        names = {"--exclude.internal.topics"},
-        description =
-            "True if you want to ignore internal topics like _consumer_offsets while counting score.",
-        validateWith = BooleanField.class,
-        converter = BooleanField.class)
-    boolean excludeInternalTopic = false;
-
-    @Parameter(
-        names = {"--hide.balanced"},
-        description = "True if you want to hide topics and partitions thar already balanced.",
-        validateWith = BooleanField.class,
-        converter = BooleanField.class)
-    boolean hideBalanced = false;
-  }
-}
diff --git a/app/src/main/java/org/astraea/app/metrics/KafkaMetrics.java b/app/src/main/java/org/astraea/app/metrics/KafkaMetrics.java
index ba3dc573..8fd0cf74 100644
--- a/app/src/main/java/org/astraea/app/metrics/KafkaMetrics.java
+++ b/app/src/main/java/org/astraea/app/metrics/KafkaMetrics.java
@@ -21,7 +21,6 @@ import java.util.Collection;
 import java.util.Map;
 import java.util.function.Function;
 import java.util.stream.Collectors;
-import org.astraea.app.metrics.broker.BrokerTopicMetricsResult;
 import org.astraea.app.metrics.broker.HasValue;
 import org.astraea.app.metrics.broker.TotalTimeMs;
 import org.astraea.app.metrics.producer.HasProducerNodeMetrics;
@@ -31,129 +30,6 @@ public final class KafkaMetrics {
 
   private KafkaMetrics() {}
 
-  public enum BrokerTopic {
-    /** 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;
-
-    BrokerTopic(String name) {
-      this.metricName = name;
-    }
-
-    public String metricName() {
-      return metricName;
-    }
-
-    /**
-     * find out the objects related to this metrics.
-     *
-     * @param objects to search
-     * @return collection of BrokerTopicMetricsResult, or empty if all objects are not related to
-     *     this metrics
-     */
-    public Collection<BrokerTopicMetricsResult> of(Collection<HasBeanObject> objects) {
-      return objects.stream()
-          .filter(o -> o instanceof BrokerTopicMetricsResult)
-          .filter(o -> metricName().equals(o.beanObject().properties().get("name")))
-          .map(o -> (BrokerTopicMetricsResult) o)
-          .collect(Collectors.toUnmodifiableList());
-    }
-
-    public BrokerTopicMetricsResult fetch(MBeanClient mBeanClient) {
-      return new BrokerTopicMetricsResult(
-          mBeanClient.queryBean(
-              BeanQuery.builder()
-                  .domainName("kafka.server")
-                  .property("type", "BrokerTopicMetrics")
-                  .property("name", this.metricName())
-                  .build()));
-    }
-
-    /**
-     * resolve specific {@link BrokerTopic} by the given metric string, compare by case insensitive
-     *
-     * @param metricName the metric to resolve
-     * @return a {@link BrokerTopic} match to give metric name
-     */
-    public static BrokerTopic of(String metricName) {
-      return Arrays.stream(BrokerTopic.values())
-          .filter(metric -> metric.metricName().equalsIgnoreCase(metricName))
-          .findFirst()
-          .orElseThrow(() -> new IllegalArgumentException("No such metric: " + metricName));
-    }
-
-    public static long linuxDiskReadBytes(MBeanClient mBeanClient) {
-      return (long)
-          mBeanClient
-              .queryBean(
-                  BeanQuery.builder()
-                      .domainName("kafka.server")
-                      .property("type", "KafkaServer")
-                      .property("name", "linux-disk-read-bytes")
-                      .build())
-              .attributes()
-              .get("Value");
-    }
-
-    public static long linuxDiskWriteBytes(MBeanClient mBeanClient) {
-      return (long)
-          mBeanClient
-              .queryBean(
-                  BeanQuery.builder()
-                      .domainName("kafka.server")
-                      .property("type", "KafkaServer")
-                      .property("name", "linux-disk-write-bytes")
-                      .build())
-              .attributes()
-              .get("Value");
-    }
-  }
-
   public enum Request {
     Produce,
     FetchConsumer,
diff --git a/app/src/main/java/org/astraea/app/metrics/broker/BrokerTopicMetricsResult.java b/app/src/main/java/org/astraea/app/metrics/broker/BrokerTopicMetricsResult.java
deleted file mode 100644
index 74bfb32e..00000000
--- a/app/src/main/java/org/astraea/app/metrics/broker/BrokerTopicMetricsResult.java
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * 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.app.metrics.broker;
-
-import java.util.Map;
-import java.util.Objects;
-import org.astraea.app.metrics.BeanObject;
-
-public class BrokerTopicMetricsResult implements HasCount, HasEventType, HasRate {
-
-  private final BeanObject beanObject;
-
-  public BrokerTopicMetricsResult(BeanObject beanObject) {
-    this.beanObject = Objects.requireNonNull(beanObject);
-  }
-
-  @Override
-  public String toString() {
-    StringBuilder sb = new StringBuilder();
-    for (Map.Entry<String, Object> e : beanObject().attributes().entrySet()) {
-      sb.append(System.lineSeparator())
-          .append("  ")
-          .append(e.getKey())
-          .append("=")
-          .append(e.getValue());
-    }
-    return beanObject().properties().get("name") + "{" + sb + "}";
-  }
-
-  @Override
-  public BeanObject beanObject() {
-    return beanObject;
-  }
-}
diff --git a/app/src/main/java/org/astraea/app/metrics/broker/ServerMetrics.java b/app/src/main/java/org/astraea/app/metrics/broker/ServerMetrics.java
index 303b55b9..305726d5 100644
--- a/app/src/main/java/org/astraea/app/metrics/broker/ServerMetrics.java
+++ b/app/src/main/java/org/astraea/app/metrics/broker/ServerMetrics.java
@@ -18,21 +18,24 @@ package org.astraea.app.metrics.broker;
 
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.Map;
+import java.util.Objects;
 import java.util.stream.Collectors;
 import org.astraea.app.metrics.BeanObject;
 import org.astraea.app.metrics.BeanQuery;
+import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.MBeanClient;
 
 public final class ServerMetrics {
 
   public enum DelayedOperationPurgatory {
-    AlterAcls("AlterAcls"),
-    DeleteRecords("DeleteRecords"),
-    ElectLeader("ElectLeader"),
-    Fetch("Fetch"),
-    Heartbeat("Heartbeat"),
-    Produce("Produce"),
-    Rebalance("Rebalance");
+    ALTER_ACLS("AlterAcls"),
+    DELETE_RECORDS("DeleteRecords"),
+    ELECT_LEADER("ElectLeader"),
+    FETCH("Fetch"),
+    HEARTBEAT("Heartbeat"),
+    PRODUCE("Produce"),
+    REBALANCE("Rebalance");
 
     private final String metricName;
 
@@ -86,4 +89,136 @@ public final class ServerMetrics {
       }
     }
   }
+
+  public enum Topic {
+    /** Message validation failure rate due to non-continuous offset or sequence number in batch */
+    INVALID_OFFSET_OR_SEQUENCE_RECORDS_PER_SEC("InvalidOffsetOrSequenceRecordsPerSec"),
+
+    /** Message validation failure rate due to incorrect crc checksum */
+    INVALID_MESSAGE_CRC_RECORDS_PER_SEC("InvalidMessageCrcRecordsPerSec"),
+
+    FETCH_MESSAGE_CONVERSIONS_PER_SEC("FetchMessageConversionsPerSec"),
+
+    BYTES_REJECTED_PER_SEC("BytesRejectedPerSec"),
+
+    /** Message in rate */
+    MESSAGES_IN_PER_SEC("MessagesInPerSec"),
+
+    /** Incoming byte rate of reassignment traffic */
+    REASSIGNMENT_BYTES_IN_PER_SEC("ReassignmentBytesInPerSec"),
+
+    FAILED_FETCH_REQUESTS_PER_SEC("FailedFetchRequestsPerSec"),
+
+    /** Byte in rate from other brokers */
+    REPLICATION_BYTES_IN_PER_SEC("ReplicationBytesInPerSec"),
+
+    /** Message validation failure rate due to no key specified for compacted topic */
+    NO_KEY_COMPACTED_TOPIC_RECORDS_PER_SEC("NoKeyCompactedTopicRecordsPerSec"),
+
+    TOTAL_FETCH_REQUESTS_PER_SEC("TotalFetchRequestsPerSec"),
+
+    FAILED_PRODUCE_REQUESTS_PER_SEC("FailedProduceRequestsPerSec"),
+
+    /** Byte in rate from clients */
+    BYTES_IN_PER_SEC("BytesInPerSec"),
+
+    TOTAL_PRODUCE_REQUESTS_PER_SEC("TotalProduceRequestsPerSec"),
+
+    /** Message validation failure rate due to invalid magic number */
+    INVALID_MAGIC_NUMBER_RECORDS_PER_SEC("InvalidMagicNumberRecordsPerSec"),
+
+    /** Outgoing byte rate of reassignment traffic */
+    REASSIGNMENT_BYTES_OUT_PER_SEC("ReassignmentBytesOutPerSec"),
+
+    /** Bytes in rate from other brokers */
+    REPLICATION_BYTES_OUT_PER_SEC("ReplicationBytesOutPerSec"),
+
+    PRODUCE_MESSAGE_CONVERSIONS_PER_SEC("ProduceMessageConversionsPerSec"),
+
+    /** Byte out rate to clients. */
+    BYTES_OUT_PER_SEC("BytesOutPerSec");
+
+    private final String metricName;
+
+    Topic(String name) {
+      this.metricName = name;
+    }
+
+    public String metricName() {
+      return metricName;
+    }
+
+    /**
+     * find out the objects related to this metrics.
+     *
+     * @param objects to search
+     * @return collection of BrokerTopicMetricsResult, or empty if all objects are not related to
+     *     this metrics
+     */
+    public Collection<Meter> of(Collection<HasBeanObject> objects) {
+      return objects.stream()
+          .filter(o -> o instanceof Meter)
+          .filter(o -> metricName().equals(o.beanObject().properties().get("name")))
+          .map(o -> (Meter) o)
+          .collect(Collectors.toUnmodifiableList());
+    }
+
+    public Meter fetch(MBeanClient mBeanClient) {
+      return new Meter(
+          mBeanClient.queryBean(
+              BeanQuery.builder()
+                  .domainName("kafka.server")
+                  .property("type", "BrokerTopicMetrics")
+                  .property("name", this.metricName())
+                  .build()));
+    }
+
+    /**
+     * resolve specific {@link Topic} by the given metric string, compare by case-insensitive
+     *
+     * @param metricName the metric to resolve
+     * @return a {@link Topic} match to give metric name
+     */
+    public static Topic of(String metricName) {
+      return Arrays.stream(Topic.values())
+          .filter(metric -> metric.metricName().equalsIgnoreCase(metricName))
+          .findFirst()
+          .orElseThrow(() -> new IllegalArgumentException("No such metric: " + metricName));
+    }
+
+    public static class Meter implements HasCount, HasEventType, HasRate {
+
+      private final BeanObject beanObject;
+
+      public Meter(BeanObject beanObject) {
+        this.beanObject = Objects.requireNonNull(beanObject);
+      }
+
+      public String metricsName() {
+        return beanObject().properties().get("name");
+      }
+
+      public Topic type() {
+        return Topic.of(metricsName());
+      }
+
+      @Override
+      public String toString() {
+        StringBuilder sb = new StringBuilder();
+        for (Map.Entry<String, Object> e : beanObject().attributes().entrySet()) {
+          sb.append(System.lineSeparator())
+              .append("  ")
+              .append(e.getKey())
+              .append("=")
+              .append(e.getValue());
+        }
+        return beanObject().properties().get("name") + "{" + sb + "}";
+      }
+
+      @Override
+      public BeanObject beanObject() {
+        return beanObject;
+      }
+    }
+  }
 }
diff --git a/app/src/main/java/org/astraea/app/metrics/platform/HostMetrics.java b/app/src/main/java/org/astraea/app/metrics/platform/HostMetrics.java
index 1b7937fa..2aaf0c1f 100644
--- a/app/src/main/java/org/astraea/app/metrics/platform/HostMetrics.java
+++ b/app/src/main/java/org/astraea/app/metrics/platform/HostMetrics.java
@@ -18,6 +18,7 @@ package org.astraea.app.metrics.platform;
 
 import org.astraea.app.metrics.BeanQuery;
 import org.astraea.app.metrics.MBeanClient;
+import org.astraea.app.metrics.broker.HasValue;
 
 public final class HostMetrics {
 
@@ -36,5 +37,27 @@ public final class HostMetrics {
             BeanQuery.builder().domainName("java.lang").property("type", "Memory").build()));
   }
 
+  public static HasValue linuxDiskReadBytes(MBeanClient mBeanClient) {
+    var bean =
+        mBeanClient.queryBean(
+            BeanQuery.builder()
+                .domainName("kafka.server")
+                .property("type", "KafkaServer")
+                .property("name", "linux-disk-read-bytes")
+                .build());
+    return () -> bean;
+  }
+
+  public static HasValue linuxDiskWriteBytes(MBeanClient mBeanClient) {
+    var bean =
+        mBeanClient.queryBean(
+            BeanQuery.builder()
+                .domainName("kafka.server")
+                .property("type", "KafkaServer")
+                .property("name", "linux-disk-write-bytes")
+                .build());
+    return () -> bean;
+  }
+
   private HostMetrics() {}
 }
diff --git a/app/src/main/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobin.java b/app/src/main/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobin.java
index 62a9b73d..c4699306 100644
--- a/app/src/main/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobin.java
+++ b/app/src/main/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobin.java
@@ -21,9 +21,9 @@ import java.util.Comparator;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 import org.astraea.app.admin.ClusterInfo;
-import org.astraea.app.cost.CostUtils;
 import org.astraea.app.cost.Periodic;
 
 /**
@@ -87,7 +87,7 @@ public final class SmoothWeightRoundRobin
               // If the average offset of all brokers from the cluster is greater than 0.1, it is
               // unbalanced.
               var balance =
-                  CostUtils.standardDeviationImperative(avgScore, brokerScore)
+                  standardDeviationImperative(avgScore, brokerScore)
                       > upperLimitOffsetRatio * avgScore;
 
               var effectiveWeights =
@@ -163,4 +163,15 @@ public final class SmoothWeightRoundRobin
           .sum();
     }
   }
+
+  private static double standardDeviationImperative(
+      double avgMetrics, Map<Integer, Double> metrics) {
+    var variance = new AtomicReference<>(0.0);
+    metrics
+        .values()
+        .forEach(
+            metric ->
+                variance.updateAndGet(v -> v + (metric - avgMetrics) * (metric - avgMetrics)));
+    return Math.sqrt(variance.get() / metrics.size());
+  }
 }
diff --git a/app/src/main/java/org/astraea/app/cost/CostUtils.java b/app/src/test/java/org/astraea/app/metrics/platform/LinuxDiskMetricsTest.java
similarity index 52%
rename from app/src/main/java/org/astraea/app/cost/CostUtils.java
rename to app/src/test/java/org/astraea/app/metrics/platform/LinuxDiskMetricsTest.java
index 9b70c4d4..652f68f8 100644
--- a/app/src/main/java/org/astraea/app/cost/CostUtils.java
+++ b/app/src/test/java/org/astraea/app/metrics/platform/LinuxDiskMetricsTest.java
@@ -14,20 +14,27 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.astraea.app.cost;
+package org.astraea.app.metrics.platform;
 
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicReference;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.condition.OS.LINUX;
 
-public class CostUtils {
-  public static double standardDeviationImperative(
-      double avgMetrics, Map<Integer, Double> metrics) {
-    var variance = new AtomicReference<>(0.0);
-    metrics
-        .values()
-        .forEach(
-            metric ->
-                variance.updateAndGet(v -> v + (metric - avgMetrics) * (metric - avgMetrics)));
-    return Math.sqrt(variance.get() / metrics.size());
+import org.astraea.app.metrics.MBeanClient;
+import org.astraea.app.service.RequireSingleBrokerCluster;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledOnOs;
+
+public class LinuxDiskMetricsTest extends RequireSingleBrokerCluster {
+
+  @Test
+  @EnabledOnOs(LINUX)
+  void linuxDiskReadBytes() {
+    assertDoesNotThrow(() -> HostMetrics.linuxDiskReadBytes(MBeanClient.local()));
+  }
+
+  @Test
+  @EnabledOnOs(LINUX)
+  void linuxDiskWriteBytes() {
+    assertDoesNotThrow(() -> HostMetrics.linuxDiskWriteBytes(MBeanClient.local()));
   }
 }


Test patch:
diff --git a/app/src/test/java/org/astraea/app/cost/BrokerInputCostTest.java b/app/src/test/java/org/astraea/app/cost/BrokerInputCostTest.java
index 95c43e3e..eb085537 100644
--- a/app/src/test/java/org/astraea/app/cost/BrokerInputCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/BrokerInputCostTest.java
@@ -21,8 +21,7 @@ import java.util.Map;
 import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.metrics.BeanObject;
-import org.astraea.app.metrics.KafkaMetrics;
-import org.astraea.app.metrics.broker.BrokerTopicMetricsResult;
+import org.astraea.app.metrics.broker.ServerMetrics;
 import org.astraea.app.metrics.collector.BeanCollector;
 import org.astraea.app.metrics.collector.Receiver;
 import org.astraea.app.service.RequireBrokerCluster;
@@ -41,11 +40,11 @@ public class BrokerInputCostTest extends RequireBrokerCluster {
                 ClusterBean.of(
                     Map.of(
                         1,
-                        List.of(brokerTopicMetricsResult(10000D)),
+                        List.of(meter(10000D)),
                         2,
-                        List.of(brokerTopicMetricsResult(20000D)),
+                        List.of(meter(20000D)),
                         3,
-                        List.of(brokerTopicMetricsResult(5000D)))))
+                        List.of(meter(5000D)))))
             .value();
     Assertions.assertEquals(10000D, scores.get(1));
     Assertions.assertEquals(20000D, scores.get(2));
@@ -69,23 +68,24 @@ public class BrokerInputCostTest extends RequireBrokerCluster {
           receiver.current().stream()
               .allMatch(
                   o ->
-                      (o instanceof BrokerTopicMetricsResult)
-                          && (KafkaMetrics.BrokerTopic.BytesInPerSec.metricName()
+                      (o instanceof ServerMetrics.Topic.Meter)
+                          && (ServerMetrics.Topic.BYTES_IN_PER_SEC
+                              .metricName()
                               .equals(o.beanObject().properties().get("name")))));
 
       // Test the fetched object's value.
       Assertions.assertTrue(
           receiver.current().stream()
-              .map(o -> (BrokerTopicMetricsResult) o)
+              .map(o -> (ServerMetrics.Topic.Meter) o)
               .allMatch(result -> result.count() == 0));
     }
   }
 
-  private static BrokerTopicMetricsResult brokerTopicMetricsResult(double value) {
-    return new BrokerTopicMetricsResult(
+  private static ServerMetrics.Topic.Meter meter(double value) {
+    return new ServerMetrics.Topic.Meter(
         new BeanObject(
             "object",
-            Map.of("name", KafkaMetrics.BrokerTopic.BytesInPerSec.metricName()),
+            Map.of("name", ServerMetrics.Topic.BYTES_IN_PER_SEC.metricName()),
             Map.of("OneMinuteRate", value)));
   }
 }
diff --git a/app/src/test/java/org/astraea/app/cost/BrokerOutputCostTest.java b/app/src/test/java/org/astraea/app/cost/BrokerOutputCostTest.java
index 0e736aa3..092aaa52 100644
--- a/app/src/test/java/org/astraea/app/cost/BrokerOutputCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/BrokerOutputCostTest.java
@@ -21,8 +21,7 @@ import java.util.Map;
 import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.metrics.BeanObject;
-import org.astraea.app.metrics.KafkaMetrics;
-import org.astraea.app.metrics.broker.BrokerTopicMetricsResult;
+import org.astraea.app.metrics.broker.ServerMetrics;
 import org.astraea.app.metrics.collector.BeanCollector;
 import org.astraea.app.metrics.collector.Receiver;
 import org.astraea.app.service.RequireBrokerCluster;
@@ -41,11 +40,11 @@ public class BrokerOutputCostTest extends RequireBrokerCluster {
                 ClusterBean.of(
                     Map.of(
                         1,
-                        List.of(brokerTopicMetricsResult(10000D)),
+                        List.of(meter(10000D)),
                         2,
-                        List.of(brokerTopicMetricsResult(20000D)),
+                        List.of(meter(20000D)),
                         3,
-                        List.of(brokerTopicMetricsResult(5000D)))))
+                        List.of(meter(5000D)))))
             .value();
     Assertions.assertEquals(10000D, scores.get(1));
     Assertions.assertEquals(20000D, scores.get(2));
@@ -69,23 +68,24 @@ public class BrokerOutputCostTest extends RequireBrokerCluster {
           receiver.current().stream()
               .allMatch(
                   o ->
-                      (o instanceof BrokerTopicMetricsResult)
-                          && (KafkaMetrics.BrokerTopic.BytesOutPerSec.metricName()
+                      (o instanceof ServerMetrics.Topic.Meter)
+                          && (ServerMetrics.Topic.BYTES_OUT_PER_SEC
+                              .metricName()
                               .equals(o.beanObject().properties().get("name")))));
 
       // Test the fetched object's value.
       Assertions.assertTrue(
           receiver.current().stream()
-              .map(o -> (BrokerTopicMetricsResult) o)
+              .map(o -> (ServerMetrics.Topic.Meter) o)
               .allMatch(result -> result.count() == 0));
     }
   }
 
-  private static BrokerTopicMetricsResult brokerTopicMetricsResult(double value) {
-    return new BrokerTopicMetricsResult(
+  private static ServerMetrics.Topic.Meter meter(double value) {
+    return new ServerMetrics.Topic.Meter(
         new BeanObject(
             "object",
-            Map.of("name", KafkaMetrics.BrokerTopic.BytesOutPerSec.metricName()),
+            Map.of("name", ServerMetrics.Topic.BYTES_OUT_PER_SEC.metricName()),
             Map.of("OneMinuteRate", value)));
   }
 }
diff --git a/app/src/test/java/org/astraea/app/cost/LoadCostTest.java b/app/src/test/java/org/astraea/app/cost/LoadCostTest.java
index 508d1654..8ec340c2 100644
--- a/app/src/test/java/org/astraea/app/cost/LoadCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/LoadCostTest.java
@@ -22,8 +22,7 @@ import java.util.Map;
 import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.metrics.BeanObject;
 import org.astraea.app.metrics.HasBeanObject;
-import org.astraea.app.metrics.KafkaMetrics;
-import org.astraea.app.metrics.broker.BrokerTopicMetricsResult;
+import org.astraea.app.metrics.broker.ServerMetrics;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
@@ -48,12 +47,12 @@ public class LoadCostTest {
   }
 
   private ClusterBean clusterBean() {
-    var BytesInPerSec1 = mockResult(KafkaMetrics.BrokerTopic.BytesInPerSec.metricName(), 50000L);
-    var BytesInPerSec2 = mockResult(KafkaMetrics.BrokerTopic.BytesInPerSec.metricName(), 100000L);
-    var BytesInPerSec3 = mockResult(KafkaMetrics.BrokerTopic.BytesInPerSec.metricName(), 200000L);
-    var BytesOutPerSec1 = mockResult(KafkaMetrics.BrokerTopic.BytesOutPerSec.metricName(), 210L);
-    var BytesOutPerSec2 = mockResult(KafkaMetrics.BrokerTopic.BytesOutPerSec.metricName(), 20L);
-    var BytesOutPerSec3 = mockResult(KafkaMetrics.BrokerTopic.BytesOutPerSec.metricName(), 10L);
+    var BytesInPerSec1 = mockResult(ServerMetrics.Topic.BYTES_IN_PER_SEC.metricName(), 50000L);
+    var BytesInPerSec2 = mockResult(ServerMetrics.Topic.BYTES_IN_PER_SEC.metricName(), 100000L);
+    var BytesInPerSec3 = mockResult(ServerMetrics.Topic.BYTES_IN_PER_SEC.metricName(), 200000L);
+    var BytesOutPerSec1 = mockResult(ServerMetrics.Topic.BYTES_OUT_PER_SEC.metricName(), 210L);
+    var BytesOutPerSec2 = mockResult(ServerMetrics.Topic.BYTES_OUT_PER_SEC.metricName(), 20L);
+    var BytesOutPerSec3 = mockResult(ServerMetrics.Topic.BYTES_OUT_PER_SEC.metricName(), 10L);
 
     Collection<HasBeanObject> broker1 = List.of(BytesInPerSec1, BytesOutPerSec1);
     Collection<HasBeanObject> broker2 = List.of(BytesInPerSec2, BytesOutPerSec2);
@@ -61,8 +60,8 @@ public class LoadCostTest {
     return ClusterBean.of(Map.of(1, broker1, 2, broker2, 3, broker3));
   }
 
-  private BrokerTopicMetricsResult mockResult(String name, long count) {
-    var result = Mockito.mock(BrokerTopicMetricsResult.class);
+  private ServerMetrics.Topic.Meter mockResult(String name, long count) {
+    var result = Mockito.mock(ServerMetrics.Topic.Meter.class);
     var bean = Mockito.mock(BeanObject.class);
     Mockito.when(result.beanObject()).thenReturn(bean);
     Mockito.when(bean.properties()).thenReturn(Map.of("name", name));
diff --git a/app/src/test/java/org/astraea/app/cost/topic/CalculateUtilsTest.java b/app/src/test/java/org/astraea/app/cost/topic/CalculateUtilsTest.java
deleted file mode 100644
index ce8ca664..00000000
--- a/app/src/test/java/org/astraea/app/cost/topic/CalculateUtilsTest.java
+++ /dev/null
@@ -1,101 +0,0 @@
-/*
- * 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.app.cost.topic;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-
-import java.util.HashMap;
-import java.util.Map;
-import org.astraea.app.admin.TopicPartition;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.Test;
-
-public class CalculateUtilsTest {
-  static Map<Integer, Map<TopicPartition, Integer>> fakeBrokerPartitionSize = new HashMap<>();
-  static Map<String, Integer> fakeRetentionMillis = new HashMap<>();
-  static Map<Integer, Map<TopicPartition, Double>> fakePartitionLoad = new HashMap<>();
-
-  @BeforeAll
-  static void setup() {
-    // set partition size and retention time
-    Map<TopicPartition, Integer> fakePartitionSize = new HashMap<>();
-    fakePartitionSize.put(TopicPartition.of("test0", 0), 100000000);
-    fakePartitionSize.put(TopicPartition.of("test0", 1), 200000000);
-    fakePartitionSize.put(TopicPartition.of("test0", 2), 300000000);
-    fakePartitionSize.put(TopicPartition.of("test0", 3), 400000000);
-    fakeBrokerPartitionSize.put(0, fakePartitionSize);
-    fakePartitionSize = new HashMap<>();
-    fakePartitionSize.put(TopicPartition.of("test1", 0), 500000000);
-    fakePartitionSize.put(TopicPartition.of("test1", 1), 600000000);
-    fakePartitionSize.put(TopicPartition.of("test1", 2), 700000000);
-    fakePartitionSize.put(TopicPartition.of("test1", 3), 800000000);
-    fakeBrokerPartitionSize.put(1, fakePartitionSize);
-    fakeRetentionMillis.put("test0", 604800000);
-    fakeRetentionMillis.put("test1", 604800000);
-
-    // set partition load
-    fakePartitionLoad = new HashMap<>();
-    Map<TopicPartition, Double> fakeTopicLoad = new HashMap<>();
-    fakeTopicLoad.put(TopicPartition.of("test0", 0), 0.1);
-    fakeTopicLoad.put(TopicPartition.of("test0", 1), 0.2);
-    fakeTopicLoad.put(TopicPartition.of("test0", 2), 0.3);
-    fakeTopicLoad.put(TopicPartition.of("test0", 3), 0.4);
-    fakePartitionLoad.put(0, fakeTopicLoad);
-    fakeTopicLoad = new HashMap<>();
-    fakeTopicLoad.put(TopicPartition.of("test1", 0), 1.5);
-    fakeTopicLoad.put(TopicPartition.of("test1", 1), 1.6);
-    fakeTopicLoad.put(TopicPartition.of("test1", 2), 1.7);
-    fakeTopicLoad.put(TopicPartition.of("test1", 3), 1.8);
-    fakePartitionLoad.put(1, fakeTopicLoad);
-  }
-
-  @Test
-  void testGetLoad() {
-    var Load = CalculateUtils.getLoad(fakeBrokerPartitionSize, fakeRetentionMillis);
-    assertEquals(2, Load.size());
-    assertEquals(4, Load.get(0).size());
-    assertEquals(4, Load.get(1).size());
-    assertEquals(0.17, round(Load.get(0).get(TopicPartition.of("test0", 0))));
-    assertEquals(0.33, round(Load.get(0).get(TopicPartition.of("test0", 1))));
-    assertEquals(0.50, round(Load.get(0).get(TopicPartition.of("test0", 2))));
-    assertEquals(0.66, round(Load.get(0).get(TopicPartition.of("test0", 3))));
-    assertEquals(0.83, round(Load.get(1).get(TopicPartition.of("test1", 0))));
-    assertEquals(0.99, round(Load.get(1).get(TopicPartition.of("test1", 1))));
-    assertEquals(1.16, round(Load.get(1).get(TopicPartition.of("test1", 2))));
-    assertEquals(1.32, round(Load.get(1).get(TopicPartition.of("test1", 3))));
-  }
-
-  @Test
-  void testGetScore() {
-    var Score = CalculateUtils.getScore(fakePartitionLoad);
-    assertEquals(2, Score.size());
-    assertEquals(4, Score.get(0).size());
-    assertEquals(4, Score.get(1).size());
-    assertEquals(0.0, round(Score.get(0).get(TopicPartition.of("test0", 0))));
-    assertEquals(0.0, round(Score.get(0).get(TopicPartition.of("test0", 1))));
-    assertEquals(0.0, round(Score.get(0).get(TopicPartition.of("test0", 2))));
-    assertEquals(0.0, round(Score.get(0).get(TopicPartition.of("test0", 3))));
-    assertEquals(-80.50, round(Score.get(1).get(TopicPartition.of("test1", 0))));
-    assertEquals(-26.83, round(Score.get(1).get(TopicPartition.of("test1", 1))));
-    assertEquals(26.83, round(Score.get(1).get(TopicPartition.of("test1", 2))));
-    assertEquals(80.50, round(Score.get(1).get(TopicPartition.of("test1", 3))));
-  }
-
-  double round(double score) {
-    return Math.round(100 * score) / 100.0;
-  }
-}
diff --git a/app/src/test/java/org/astraea/app/cost/topic/GetPartitionInfTest.java b/app/src/test/java/org/astraea/app/cost/topic/GetPartitionInfTest.java
deleted file mode 100644
index 48f1508c..00000000
--- a/app/src/test/java/org/astraea/app/cost/topic/GetPartitionInfTest.java
+++ /dev/null
@@ -1,103 +0,0 @@
-/*
- * 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.app.cost.topic;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-
-import java.time.Duration;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.concurrent.ExecutionException;
-import org.astraea.app.admin.Admin;
-import org.astraea.app.common.Utils;
-import org.astraea.app.producer.Producer;
-import org.astraea.app.producer.Serializer;
-import org.astraea.app.service.RequireBrokerCluster;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.Test;
-
-public class GetPartitionInfTest extends RequireBrokerCluster {
-  static Admin admin;
-
-  @BeforeAll
-  static void setup() throws ExecutionException, InterruptedException {
-    admin = Admin.of(bootstrapServers());
-    Map<Integer, String> topicName = new HashMap<>();
-    topicName.put(0, "testPartitionScore0");
-    topicName.put(1, "testPartitionScore1");
-    topicName.put(2, "testPartitionScore2");
-
-    try (var admin = Admin.of(bootstrapServers())) {
-      admin
-          .creator()
-          .topic(topicName.get(0))
-          .numberOfPartitions(4)
-          .numberOfReplicas((short) 1)
-          .create();
-      admin
-          .creator()
-          .topic(topicName.get(1))
-          .numberOfPartitions(4)
-          .numberOfReplicas((short) 1)
-          .create();
-      admin
-          .creator()
-          .topic(topicName.get(2))
-          .numberOfPartitions(4)
-          .numberOfReplicas((short) 1)
-          .create();
-      // wait for topic
-      Utils.sleep(Duration.ofSeconds(5));
-    }
-    var producer =
-        Producer.builder()
-            .bootstrapServers(bootstrapServers())
-            .keySerializer(Serializer.STRING)
-            .build();
-    int size = 10000;
-    for (int t = 0; t <= 2; t++) {
-      for (int p = 0; p <= 3; p++) {
-        producer
-            .sender()
-            .topic(topicName.get(t))
-            .partition(p)
-            .value(new byte[size])
-            .run()
-            .toCompletableFuture()
-            .get();
-      }
-      size += 10000;
-    }
-  }
-
-  @Test
-  void testGetSize() {
-    var brokerPartitionSize = GetPartitionInf.getSize(admin);
-    assertEquals(3, brokerPartitionSize.size());
-    assertEquals(
-        3 * 4,
-        brokerPartitionSize.get(0).size()
-            + brokerPartitionSize.get(1).size()
-            + brokerPartitionSize.get(2).size());
-  }
-
-  @Test
-  void testGetRetentionMillis() {
-    var brokerPartitionRetentionMillis = GetPartitionInf.getRetentionMillis(admin);
-    assertEquals(3, brokerPartitionRetentionMillis.size());
-  }
-}
diff --git a/app/src/test/java/org/astraea/app/cost/topic/PartitionScoreTest.java b/app/src/test/java/org/astraea/app/cost/topic/PartitionScoreTest.java
deleted file mode 100644
index 3298c591..00000000
--- a/app/src/test/java/org/astraea/app/cost/topic/PartitionScoreTest.java
+++ /dev/null
@@ -1,99 +0,0 @@
-/*
- * 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.app.cost.topic;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-
-import java.time.Duration;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.concurrent.ExecutionException;
-import org.astraea.app.admin.Admin;
-import org.astraea.app.common.Utils;
-import org.astraea.app.producer.Producer;
-import org.astraea.app.producer.Serializer;
-import org.astraea.app.service.RequireBrokerCluster;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.Test;
-
-public class PartitionScoreTest extends RequireBrokerCluster {
-  static Admin admin;
-
-  @BeforeAll
-  static void setup() throws ExecutionException, InterruptedException {
-    admin = Admin.of(bootstrapServers());
-    Map<Integer, String> topicName = new HashMap<>();
-    topicName.put(0, "testPartitionScore0");
-    topicName.put(1, "testPartitionScore1");
-    topicName.put(2, "__consumer_offsets");
-    try (var admin = Admin.of(bootstrapServers())) {
-      admin
-          .creator()
-          .topic(topicName.get(0))
-          .numberOfPartitions(4)
-          .numberOfReplicas((short) 1)
-          .create();
-      admin
-          .creator()
-          .topic(topicName.get(1))
-          .numberOfPartitions(4)
-          .numberOfReplicas((short) 1)
-          .create();
-      admin
-          .creator()
-          .topic(topicName.get(2))
-          .numberOfPartitions(4)
-          .numberOfReplicas((short) 1)
-          .create();
-      // wait for topic
-      Utils.sleep(Duration.ofSeconds(5));
-    }
-    var producer =
-        Producer.builder()
-            .bootstrapServers(bootstrapServers())
-            .keySerializer(Serializer.STRING)
-            .build();
-    int size = 10000;
-    for (int t = 0; t <= 1; t++) {
-      for (int p = 0; p <= 3; p++) {
-        producer
-            .sender()
-            .topic(topicName.get(t))
-            .partition(p)
-            .value(new byte[size])
-            .run()
-            .toCompletableFuture()
-            .get();
-      }
-      size += 10000;
-    }
-    producer.close();
-  }
-
-  @Test
-  void testGetScore() {
-    PartitionScore.Argument argument = new PartitionScore.Argument();
-    argument.excludeInternalTopic = false;
-    var score = PartitionScore.execute(argument, admin);
-    assertEquals(3, score.size());
-    assertEquals(3 * 4, score.get(0).size() + score.get(1).size() + score.get(2).size());
-    argument.excludeInternalTopic = true;
-    score = PartitionScore.execute(argument, admin);
-    assertEquals(3, score.size());
-    assertEquals(2 * 4, score.get(0).size() + score.get(1).size() + score.get(2).size());
-  }
-}
diff --git a/app/src/test/java/org/astraea/app/metrics/KafkaMetricsTest.java b/app/src/test/java/org/astraea/app/metrics/KafkaMetricsTest.java
index a75fc346..726a5ba4 100644
--- a/app/src/test/java/org/astraea/app/metrics/KafkaMetricsTest.java
+++ b/app/src/test/java/org/astraea/app/metrics/KafkaMetricsTest.java
@@ -17,34 +17,22 @@
 package org.astraea.app.metrics;
 
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
-import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.condition.OS.LINUX;
 
 import java.io.IOException;
 import java.lang.management.ManagementFactory;
 import java.time.Duration;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.stream.Collectors;
 import javax.management.MBeanServer;
 import javax.management.remote.JMXConnectorServer;
 import javax.management.remote.JMXConnectorServerFactory;
 import javax.management.remote.JMXServiceURL;
 import org.astraea.app.admin.Admin;
 import org.astraea.app.common.Utils;
-import org.astraea.app.metrics.broker.BrokerTopicMetricsResult;
 import org.astraea.app.metrics.broker.LogMetrics;
 import org.astraea.app.metrics.broker.TotalTimeMs;
 import org.astraea.app.service.RequireBrokerCluster;
 import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.EnabledOnOs;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.EnumSource;
 
@@ -73,35 +61,6 @@ class KafkaMetricsTest extends RequireBrokerCluster {
     mBeanClient.close();
   }
 
-  @Test
-  void testAllEnumNameUnique() {
-    // arrange act
-    Set<String> collectedName =
-        Arrays.stream(KafkaMetrics.BrokerTopic.values())
-            .map(KafkaMetrics.BrokerTopic::metricName)
-            .collect(Collectors.toSet());
-
-    // assert
-    assertEquals(KafkaMetrics.BrokerTopic.values().length, collectedName.size());
-  }
-
-  @ParameterizedTest
-  @EnumSource(value = KafkaMetrics.BrokerTopic.class)
-  void testRequestBrokerTopicMetrics(KafkaMetrics.BrokerTopic metric) {
-    // act
-    BrokerTopicMetricsResult result = metric.fetch(mBeanClient);
-
-    // assert access attribute will not throw casting error
-    // assert attribute actually exists
-    assertDoesNotThrow(result::count);
-    assertDoesNotThrow(result::eventType);
-    assertDoesNotThrow(result::fifteenMinuteRate);
-    assertDoesNotThrow(result::fiveMinuteRate);
-    assertDoesNotThrow(result::meanRate);
-    assertDoesNotThrow(result::oneMinuteRate);
-    assertDoesNotThrow(result::rateUnit);
-  }
-
   @ParameterizedTest()
   @EnumSource(value = KafkaMetrics.Request.class)
   void testRequestTotalTimeMs(KafkaMetrics.Request request) {
@@ -136,46 +95,4 @@ class KafkaMetricsTest extends RequireBrokerCluster {
     var beans = request.fetch(mBeanClient);
     assertNotEquals(0, beans.size());
   }
-
-  @Test
-  void testKafkaMetricsOf() {
-    assertEquals(
-        KafkaMetrics.BrokerTopic.BytesInPerSec, KafkaMetrics.BrokerTopic.of("ByTeSiNpErSeC"));
-    assertEquals(
-        KafkaMetrics.BrokerTopic.BytesOutPerSec, KafkaMetrics.BrokerTopic.of("bytesoutpersec"));
-    assertEquals(
-        KafkaMetrics.BrokerTopic.MessagesInPerSec, KafkaMetrics.BrokerTopic.of("MessagesInPERSEC"));
-    assertThrows(IllegalArgumentException.class, () -> KafkaMetrics.BrokerTopic.of("nothing"));
-  }
-
-  @Test
-  @EnabledOnOs(LINUX)
-  void linuxDiskReadBytes() {
-    assertDoesNotThrow(() -> KafkaMetrics.BrokerTopic.linuxDiskReadBytes(mBeanClient));
-  }
-
-  @Test
-  @EnabledOnOs(LINUX)
-  void linuxDiskWriteBytes() {
-    assertDoesNotThrow(() -> KafkaMetrics.BrokerTopic.linuxDiskWriteBytes(mBeanClient));
-  }
-
-  @ParameterizedTest
-  @EnumSource(KafkaMetrics.BrokerTopic.class)
-  void testBrokerTopic(KafkaMetrics.BrokerTopic brokerTopic) {
-    var object =
-        new BrokerTopicMetricsResult(
-            new BeanObject("object", Map.of("name", brokerTopic.metricName()), Map.of()));
-    Assertions.assertEquals(1, brokerTopic.of(List.of(object)).size());
-
-    Assertions.assertEquals(
-        0,
-        brokerTopic
-            .of(
-                List.of(
-                    new BrokerTopicMetricsResult(
-                        new BeanObject(
-                            "object", Map.of("name", Utils.randomString(10)), Map.of()))))
-            .size());
-  }
 }
diff --git a/app/src/test/java/org/astraea/app/metrics/broker/ServerMetricsTest.java b/app/src/test/java/org/astraea/app/metrics/broker/ServerMetricsTest.java
index d1362fb5..9d0d808f 100644
--- a/app/src/test/java/org/astraea/app/metrics/broker/ServerMetricsTest.java
+++ b/app/src/test/java/org/astraea/app/metrics/broker/ServerMetricsTest.java
@@ -16,8 +16,19 @@
  */
 package org.astraea.app.metrics.broker;
 
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.astraea.app.common.Utils;
+import org.astraea.app.metrics.BeanObject;
 import org.astraea.app.metrics.MBeanClient;
 import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.EnumSource;
 
@@ -28,4 +39,50 @@ public class ServerMetricsTest {
   void testPurgatorySize(ServerMetrics.DelayedOperationPurgatory request) {
     request.fetch(MBeanClient.local()).forEach(s -> Assertions.assertTrue(s.value() >= 0));
   }
+
+  @Test
+  void testKafkaMetricsOf() {
+    Arrays.stream(ServerMetrics.Topic.values())
+        .forEach(
+            t ->
+                Assertions.assertEquals(
+                    t, ServerMetrics.Topic.of(t.metricName().toLowerCase(Locale.ROOT))));
+    Arrays.stream(ServerMetrics.Topic.values())
+        .forEach(
+            t ->
+                Assertions.assertEquals(
+                    t, ServerMetrics.Topic.of(t.metricName().toUpperCase(Locale.ROOT))));
+    assertThrows(IllegalArgumentException.class, () -> ServerMetrics.Topic.of("nothing"));
+  }
+
+  @ParameterizedTest
+  @EnumSource(ServerMetrics.Topic.class)
+  void testBrokerTopic(ServerMetrics.Topic brokerTopic) {
+    var object =
+        new ServerMetrics.Topic.Meter(
+            new BeanObject("object", Map.of("name", brokerTopic.metricName()), Map.of()));
+    Assertions.assertEquals(1, brokerTopic.of(List.of(object)).size());
+
+    Assertions.assertEquals(
+        0,
+        brokerTopic
+            .of(
+                List.of(
+                    new ServerMetrics.Topic.Meter(
+                        new BeanObject(
+                            "object", Map.of("name", Utils.randomString(10)), Map.of()))))
+            .size());
+  }
+
+  @Test
+  void testAllEnumNameUnique() {
+    // arrange act
+    Set<String> collectedName =
+        Arrays.stream(ServerMetrics.Topic.values())
+            .map(ServerMetrics.Topic::metricName)
+            .collect(Collectors.toSet());
+
+    // assert
+    Assertions.assertEquals(ServerMetrics.Topic.values().length, collectedName.size());
+  }
 }
diff --git a/app/src/test/java/org/astraea/app/metrics/broker/ServerTopicMetricsTest.java b/app/src/test/java/org/astraea/app/metrics/broker/ServerTopicMetricsTest.java
new file mode 100644
index 00000000..e8c69442
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/metrics/broker/ServerTopicMetricsTest.java
@@ -0,0 +1,43 @@
+/*
+ * 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.app.metrics.broker;
+
+import org.astraea.app.metrics.MBeanClient;
+import org.astraea.app.service.RequireSingleBrokerCluster;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+public class ServerTopicMetricsTest extends RequireSingleBrokerCluster {
+
+  @ParameterizedTest
+  @EnumSource(value = ServerMetrics.Topic.class)
+  void testRequestBrokerTopicMetrics(ServerMetrics.Topic metric) {
+    // act
+    var result = metric.fetch(MBeanClient.local());
+
+    // assert access attribute will not throw casting error
+    // assert attribute actually exists
+    Assertions.assertDoesNotThrow(result::count);
+    Assertions.assertDoesNotThrow(result::eventType);
+    Assertions.assertDoesNotThrow(result::fifteenMinuteRate);
+    Assertions.assertDoesNotThrow(result::fiveMinuteRate);
+    Assertions.assertDoesNotThrow(result::meanRate);
+    Assertions.assertDoesNotThrow(result::oneMinuteRate);
+    Assertions.assertDoesNotThrow(result::rateUnit);
+  }
+}