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

Fix flaky test in `BalancerHandlerTest`
Source: https://github.com/skiptests/astraea/actions/runs/3060745599/jobs/4939630645#step:5:509

`BalancerHandlerTest` 裡面有一段邏輯在確保被改變的 topics 數量是 2 個，但是經常發生數量是 0 的情況。

被改變的數量是0的源頭是找不到可用的計劃，目前測試裡面是使用 `ReplicaSizeCost` 來找計劃，而且目前邏輯直接拿第一個計劃來當做結果，加上計劃完全隨機產生，直覺上這樣測試不太能總是通過。

這個 PR 採用的修正是，使用一個總是回傳更好分數的 ClusterCost，確保更好的計劃被提出這件事總是成立。這樣做對確認 BalancerHandler 回傳的結果是否正確不影響。

chia7712: @garyparrot 你貼的連結裡面失敗的設施是 TopicHandlerTest ，跟你的描述好像不太一樣？
garyparrot: 抱歉，應該是這個 https://github.com/skiptests/astraea/actions/runs/3054025985/jobs/4925397855#step:5:505

Patch:
diff --git a/app/src/main/java/org/astraea/app/performance/ConsumerThread.java b/app/src/main/java/org/astraea/app/performance/ConsumerThread.java
index 3974f9c9..fd5e742a 100644
--- a/app/src/main/java/org/astraea/app/performance/ConsumerThread.java
+++ b/app/src/main/java/org/astraea/app/performance/ConsumerThread.java
@@ -58,7 +58,6 @@ public interface ConsumerThread extends AbstractThread {
               @SuppressWarnings("resource")
               var consumer = consumerSupplier.apply(ps -> {});
               var closed = new AtomicBoolean(false);
-              var report = Report.of(consumer.clientId(), closed::get);
               var closeLatch = closeLatches.get(index);
               var subscribed = new AtomicBoolean(true);
               executors.execute(
@@ -71,15 +70,7 @@ public interface ConsumerThread extends AbstractThread {
                           Utils.sleep(Duration.ofSeconds(1));
                           continue;
                         }
-                        consumer
-                            .poll(Duration.ofSeconds(1))
-                            .forEach(
-                                record ->
-                                    // record ene-to-end latency, and record input byte (header and
-                                    // timestamp size excluded)
-                                    report.record(
-                                        System.currentTimeMillis() - record.timestamp(),
-                                        record.serializedKeySize() + record.serializedValueSize()));
+                        consumer.poll(Duration.ofSeconds(1));
                       }
                     } catch (WakeupException ignore) {
                       // Stop polling and being ready to clean up
@@ -111,11 +102,6 @@ public interface ConsumerThread extends AbstractThread {
                   subscribed.set(false);
                 }
 
-                @Override
-                public Report report() {
-                  return report;
-                }
-
                 @Override
                 public void close() {
                   closed.set(true);
@@ -129,7 +115,4 @@ public interface ConsumerThread extends AbstractThread {
   void resubscribe();
 
   void unsubscribe();
-
-  /** @return report of this thread */
-  Report report();
 }
diff --git a/app/src/main/java/org/astraea/app/performance/Performance.java b/app/src/main/java/org/astraea/app/performance/Performance.java
index c461df08..d58168d3 100644
--- a/app/src/main/java/org/astraea/app/performance/Performance.java
+++ b/app/src/main/java/org/astraea/app/performance/Performance.java
@@ -104,16 +104,11 @@ public class Performance {
                     .consumerRebalanceListener(listener)
                     .build());
 
-    Supplier<List<Report>> consumerReporter =
-        () ->
-            consumerThreads.stream()
-                .map(ConsumerThread::report)
-                .collect(Collectors.toUnmodifiableList());
-
     System.out.println("creating tracker");
     var tracker =
         TrackerThread.create(
-            () -> consumerThreads.stream().allMatch(AbstractThread::closed), consumerReporter);
+            () -> producerThreads.stream().allMatch(AbstractThread::closed),
+            () -> consumerThreads.stream().allMatch(AbstractThread::closed));
 
     Optional<Runnable> fileWriter =
         param.CSVPath == null
@@ -123,8 +118,7 @@ public class Performance {
                     param.reportFormat,
                     param.CSVPath,
                     () -> consumerThreads.stream().allMatch(AbstractThread::closed),
-                    () -> producerThreads.stream().allMatch(AbstractThread::closed),
-                    consumerReporter));
+                    () -> producerThreads.stream().allMatch(AbstractThread::closed)));
 
     var fileWriterFuture =
         fileWriter.map(CompletableFuture::runAsync).orElse(CompletableFuture.completedFuture(null));
diff --git a/app/src/main/java/org/astraea/app/performance/Report.java b/app/src/main/java/org/astraea/app/performance/Report.java
index 7bc2da4d..80d17fa8 100644
--- a/app/src/main/java/org/astraea/app/performance/Report.java
+++ b/app/src/main/java/org/astraea/app/performance/Report.java
@@ -17,7 +17,6 @@
 package org.astraea.app.performance;
 
 import java.util.List;
-import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import org.astraea.common.metrics.MBeanClient;
 import org.astraea.common.metrics.client.consumer.ConsumerMetrics;
@@ -34,6 +33,40 @@ public interface Report {
             .sum();
   }
 
+  static List<Report> consumers() {
+
+    return ConsumerMetrics.fetches(MBeanClient.local()).stream()
+        .map(
+            m ->
+                new Report() {
+                  @Override
+                  public long records() {
+                    return (long) m.recordsConsumedTotal();
+                  }
+
+                  @Override
+                  public long maxLatency() {
+                    return (long) m.fetchLatencyMax();
+                  }
+
+                  @Override
+                  public double avgLatency() {
+                    return (long) m.fetchLatencyAvg();
+                  }
+
+                  @Override
+                  public long totalBytes() {
+                    return (long) m.bytesConsumedTotal();
+                  }
+
+                  @Override
+                  public String clientId() {
+                    return m.clientId();
+                  }
+                })
+        .collect(Collectors.toList());
+  }
+
   static List<Report> producers() {
     return ProducerMetrics.of(MBeanClient.local()).stream()
         .map(
@@ -59,18 +92,10 @@ public interface Report {
                     return (long) m.outgoingByteTotal();
                   }
 
-                  @Override
-                  public boolean isClosed() {
-                    return false;
-                  }
-
                   @Override
                   public String clientId() {
                     return m.clientId();
                   }
-
-                  @Override
-                  public void record(long latency, int bytes) {}
                 })
         .collect(Collectors.toList());
   }
@@ -86,60 +111,5 @@ public interface Report {
   /** @return total send/received bytes */
   long totalBytes();
 
-  boolean isClosed();
-
   String clientId();
-
-  void record(long latency, int bytes);
-
-  static Report of(String clientId, Supplier<Boolean> isClosed) {
-    return new Report() {
-
-      private double avgLatency = 0;
-      private long records = 0;
-      private long max = 0;
-      private long totalBytes = 0;
-
-      @Override
-      public synchronized void record(long latency, int bytes) {
-        ++records;
-        max = Math.max(max, latency);
-        avgLatency += (((double) latency) - avgLatency) / (double) records;
-        totalBytes += bytes;
-      }
-
-      /** @return Get the number of records. */
-      @Override
-      public synchronized long records() {
-        return records;
-      }
-      /** @return Get the maximum of latency put. */
-      @Override
-      public synchronized long maxLatency() {
-        return max;
-      }
-
-      /** @return Get the average latency. */
-      @Override
-      public synchronized double avgLatency() {
-        return avgLatency;
-      }
-
-      /** @return total send/received bytes */
-      @Override
-      public synchronized long totalBytes() {
-        return totalBytes;
-      }
-
-      @Override
-      public boolean isClosed() {
-        return isClosed.get();
-      }
-
-      @Override
-      public String clientId() {
-        return clientId;
-      }
-    };
-  }
 }
diff --git a/app/src/main/java/org/astraea/app/performance/ReportFormat.java b/app/src/main/java/org/astraea/app/performance/ReportFormat.java
index c6d402b7..d8b46925 100644
--- a/app/src/main/java/org/astraea/app/performance/ReportFormat.java
+++ b/app/src/main/java/org/astraea/app/performance/ReportFormat.java
@@ -69,8 +69,7 @@ public enum ReportFormat implements EnumInfo {
       ReportFormat reportFormat,
       Path path,
       Supplier<Boolean> consumerDone,
-      Supplier<Boolean> producerDone,
-      Supplier<List<Report>> consumerReporter)
+      Supplier<Boolean> producerDone)
       throws IOException {
     var filePath =
         FileSystems.getDefault()
@@ -81,7 +80,7 @@ public enum ReportFormat implements EnumInfo {
                     + "."
                     + reportFormat);
     var writer = new BufferedWriter(new FileWriter(filePath.toFile()));
-    var elements = latencyAndIO(consumerReporter);
+    var elements = latencyAndIO();
     switch (reportFormat) {
       case CSV:
         initCSVFormat(writer, elements);
@@ -161,9 +160,9 @@ public enum ReportFormat implements EnumInfo {
     }
   }
 
-  private static List<CSVContentElement> latencyAndIO(Supplier<List<Report>> consumerReporter) {
+  private static List<CSVContentElement> latencyAndIO() {
     var producerReports = Report.producers();
-    var consumerReports = consumerReporter.get();
+    var consumerReports = Report.consumers();
     var elements = new ArrayList<CSVContentElement>();
     elements.add(
         CSVContentElement.create(
diff --git a/app/src/main/java/org/astraea/app/performance/TrackerThread.java b/app/src/main/java/org/astraea/app/performance/TrackerThread.java
index 732bc9ad..0b15391a 100644
--- a/app/src/main/java/org/astraea/app/performance/TrackerThread.java
+++ b/app/src/main/java/org/astraea/app/performance/TrackerThread.java
@@ -97,6 +97,10 @@ public interface TrackerThread extends AbstractThread {
     private final Supplier<List<Report>> reportSupplier;
     private long lastRecords = 0;
 
+    ConsumerPrinter() {
+      this(Report::consumers);
+    }
+
     ConsumerPrinter(Supplier<List<Report>> reportSupplier) {
       this.reportSupplier = reportSupplier;
     }
@@ -154,11 +158,10 @@ public interface TrackerThread extends AbstractThread {
     }
   }
 
-  static TrackerThread create(
-      Supplier<Boolean> producersDone, Supplier<List<Report>> consumerReporter) {
+  static TrackerThread create(Supplier<Boolean> producersDone, Supplier<Boolean> consumersDone) {
     var closed = new AtomicBoolean(false);
     var latch = new CountDownLatch(1);
-    CompletableFuture.runAsync(trackerLoop(closed::get, producersDone, consumerReporter))
+    CompletableFuture.runAsync(trackerLoop(closed::get, producersDone, consumersDone))
         .whenComplete((m, e) -> latch.countDown());
 
     return new TrackerThread() {
@@ -182,13 +185,11 @@ public interface TrackerThread extends AbstractThread {
   }
 
   static Runnable trackerLoop(
-      Supplier<Boolean> closed,
-      Supplier<Boolean> producersDone,
-      Supplier<List<Report>> consumerReports) {
+      Supplier<Boolean> closed, Supplier<Boolean> producersDone, Supplier<Boolean> consumersDone) {
     var start = System.currentTimeMillis();
     return () -> {
       var producerPrinter = new ProducerPrinter();
-      var consumerPrinter = new ConsumerPrinter(consumerReports);
+      var consumerPrinter = new ConsumerPrinter();
       while (!closed.get()) {
         var duration = Duration.ofMillis(System.currentTimeMillis() - start);
         System.out.println();
@@ -202,8 +203,7 @@ public interface TrackerThread extends AbstractThread {
                 + "sec");
         producerPrinter.tryToPrint(duration);
         consumerPrinter.tryToPrint(duration);
-        if (producersDone.get() && consumerReports.get().stream().allMatch(Report::isClosed))
-          return;
+        if (producersDone.get() && consumersDone.get()) return;
         // Log after waiting for one second
         Utils.sleep(Duration.ofSeconds(1));
       }
diff --git a/common/src/main/java/org/astraea/common/cost/ReplicaLeaderCost.java b/common/src/main/java/org/astraea/common/cost/ReplicaLeaderCost.java
index 8cef827e..913262aa 100644
--- a/common/src/main/java/org/astraea/common/cost/ReplicaLeaderCost.java
+++ b/common/src/main/java/org/astraea/common/cost/ReplicaLeaderCost.java
@@ -68,7 +68,7 @@ public class ReplicaLeaderCost implements HasBrokerCost, HasClusterCost {
                         .map(x -> (ServerMetrics.ReplicaManager.Gauge) x)
                         .sorted(Comparator.comparing(HasBeanObject::createdTimestamp).reversed())
                         .limit(1)
-                        .mapToInt(v -> (int) v.value())
+                        .mapToInt(v -> v.value().intValue())
                         .sum()));
   }
 
diff --git a/common/src/main/java/org/astraea/common/metrics/broker/ControllerMetrics.java b/common/src/main/java/org/astraea/common/metrics/broker/ControllerMetrics.java
index dc75f419..90c94a72 100644
--- a/common/src/main/java/org/astraea/common/metrics/broker/ControllerMetrics.java
+++ b/common/src/main/java/org/astraea/common/metrics/broker/ControllerMetrics.java
@@ -70,7 +70,7 @@ public class ControllerMetrics {
                   .build()));
     }
 
-    public static class Gauge implements HasGauge {
+    public static class Gauge implements HasGauge<Long> {
       private final BeanObject beanObject;
 
       public Gauge(BeanObject beanObject) {
diff --git a/common/src/main/java/org/astraea/common/metrics/broker/HasGauge.java b/common/src/main/java/org/astraea/common/metrics/broker/HasGauge.java
index 45a7e1ee..bb6dd37c 100644
--- a/common/src/main/java/org/astraea/common/metrics/broker/HasGauge.java
+++ b/common/src/main/java/org/astraea/common/metrics/broker/HasGauge.java
@@ -16,20 +16,23 @@
  */
 package org.astraea.common.metrics.broker;
 
+import java.util.Objects;
 import org.astraea.common.metrics.BeanObject;
 import org.astraea.common.metrics.HasBeanObject;
 
 /**
  * You can find some default metric in {@link kafka.metrics.KafkaMetricsGroup}. This object is
- * mapped to {@link com.yammer.metrics.core.Gauge}
+ * mapped to {@link com.yammer.metrics.core.Gauge}.Kafka gauges are String, Integer, Long and
+ * Double.
  */
-public interface HasGauge extends HasBeanObject {
-  default long value() {
-    var value = beanObject().attributes().getOrDefault("Value", 0);
-    return ((Number) value).longValue();
-  }
+public interface HasGauge<T> extends HasBeanObject {
 
-  static HasGauge of(BeanObject beanObject) {
+  static HasGauge<Long> ofLong(BeanObject beanObject) {
     return () -> beanObject;
   }
+
+  @SuppressWarnings("unchecked")
+  default T value() {
+    return (T) Objects.requireNonNull(beanObject().attributes().get("Value"));
+  }
 }
diff --git a/common/src/main/java/org/astraea/common/metrics/broker/LogMetrics.java b/common/src/main/java/org/astraea/common/metrics/broker/LogMetrics.java
index 405a4af4..6a64e5e5 100644
--- a/common/src/main/java/org/astraea/common/metrics/broker/LogMetrics.java
+++ b/common/src/main/java/org/astraea/common/metrics/broker/LogMetrics.java
@@ -16,7 +16,6 @@
  */
 package org.astraea.common.metrics.broker;
 
-import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
 import java.util.stream.Collectors;
@@ -58,13 +57,6 @@ public final class LogMetrics {
       return alias();
     }
 
-    public static LogMetrics.Log of(String metricName) {
-      return Arrays.stream(LogMetrics.Log.values())
-          .filter(metric -> metric.metricName().equalsIgnoreCase(metricName))
-          .findFirst()
-          .orElseThrow(() -> new IllegalArgumentException("No such metric: " + metricName));
-    }
-
     public static Collection<Gauge> gauges(Collection<HasBeanObject> beans, Log type) {
       return beans.stream()
           .filter(m -> m instanceof Gauge)
@@ -88,7 +80,7 @@ public final class LogMetrics {
           .collect(Collectors.toUnmodifiableList());
     }
 
-    public static class Gauge implements HasGauge {
+    public static class Gauge implements HasGauge<Long> {
       private final BeanObject beanObject;
 
       public Gauge(BeanObject beanObject) {
@@ -108,7 +100,7 @@ public final class LogMetrics {
       }
 
       public Log type() {
-        return Log.of(metricsName());
+        return ofAlias(metricsName());
       }
 
       @Override
diff --git a/common/src/main/java/org/astraea/common/metrics/broker/NetworkMetrics.java b/common/src/main/java/org/astraea/common/metrics/broker/NetworkMetrics.java
index 14cc30c7..3963ea1b 100644
--- a/common/src/main/java/org/astraea/common/metrics/broker/NetworkMetrics.java
+++ b/common/src/main/java/org/astraea/common/metrics/broker/NetworkMetrics.java
@@ -16,7 +16,6 @@
  */
 package org.astraea.common.metrics.broker;
 
-import java.util.Arrays;
 import org.astraea.common.EnumInfo;
 import org.astraea.common.metrics.BeanObject;
 import org.astraea.common.metrics.BeanQuery;
@@ -138,7 +137,7 @@ public class NetworkMetrics {
       }
 
       public Request type() {
-        return Request.of(beanObject.properties().get("request"));
+        return ofAlias(beanObject.properties().get("request"));
       }
 
       @Override
@@ -151,13 +150,6 @@ public class NetworkMetrics {
         return beanObject;
       }
     }
-
-    public static Request of(String metricName) {
-      return Arrays.stream(Request.values())
-          .filter(metric -> metric.metricName().equalsIgnoreCase(metricName))
-          .findFirst()
-          .orElseThrow(() -> new IllegalArgumentException("No such metric: " + metricName));
-    }
   }
 
   private NetworkMetrics() {}
diff --git a/common/src/main/java/org/astraea/common/metrics/broker/ServerMetrics.java b/common/src/main/java/org/astraea/common/metrics/broker/ServerMetrics.java
index cf3f679c..2ae12e47 100644
--- a/common/src/main/java/org/astraea/common/metrics/broker/ServerMetrics.java
+++ b/common/src/main/java/org/astraea/common/metrics/broker/ServerMetrics.java
@@ -16,8 +16,8 @@
  */
 package org.astraea.common.metrics.broker;
 
-import java.util.Arrays;
 import java.util.Collection;
+import java.util.List;
 import java.util.Objects;
 import java.util.stream.Collectors;
 import org.astraea.common.EnumInfo;
@@ -28,6 +28,81 @@ import org.astraea.common.metrics.MBeanClient;
 
 public final class ServerMetrics {
 
+  public enum KafkaServer implements EnumInfo {
+    YAMMER_METRICS_COUNT("yammer-metrics-count"),
+    BROKER_STATE("BrokerState"),
+    LINUX_DISK_READ_BYTES("linux-disk-read-bytes"),
+    LINUX_DISK_WRITE_BYTES("linux-disk-write-bytes");
+
+    /** Others are Gauge-Number , this is Gauge-String */
+    public static final String CLUSTER_ID = "ClusterId";
+
+    private final String metricName;
+
+    public static HasGauge<String> clusterId(MBeanClient mBeanClient) {
+      return () ->
+          mBeanClient.queryBean(
+              BeanQuery.builder()
+                  .domainName("kafka.server")
+                  .property("type", "KafkaServer")
+                  .property("name", CLUSTER_ID)
+                  .build());
+    }
+
+    KafkaServer(String name) {
+      this.metricName = name;
+    }
+
+    public String metricName() {
+      return metricName;
+    }
+
+    public Gauge fetch(MBeanClient mBeanClient) {
+      return new Gauge(
+          mBeanClient.queryBean(
+              BeanQuery.builder()
+                  .domainName("kafka.server")
+                  .property("type", "KafkaServer")
+                  .property("name", metricName)
+                  .build()));
+    }
+
+    public static KafkaServer ofAlias(String alias) {
+      return EnumInfo.ignoreCaseEnum(KafkaServer.class, alias);
+    }
+
+    @Override
+    public String alias() {
+      return metricName();
+    }
+
+    @Override
+    public String toString() {
+      return alias();
+    }
+
+    public static class Gauge implements HasGauge<Long> {
+      private final BeanObject beanObject;
+
+      public Gauge(BeanObject beanObject) {
+        this.beanObject = beanObject;
+      }
+
+      public String metricsName() {
+        return beanObject().properties().get("name");
+      }
+
+      public KafkaServer type() {
+        return ofAlias(metricsName());
+      }
+
+      @Override
+      public BeanObject beanObject() {
+        return beanObject;
+      }
+    }
+  }
+
   public enum DelayedOperationPurgatory implements EnumInfo {
     ALTER_ACLS("AlterAcls"),
     DELETE_RECORDS("DeleteRecords"),
@@ -75,7 +150,7 @@ public final class ServerMetrics {
           .collect(Collectors.toUnmodifiableList());
     }
 
-    public static class Gauge implements HasGauge {
+    public static class Gauge implements HasGauge<Long> {
       private final BeanObject beanObject;
 
       public Gauge(BeanObject beanObject) {
@@ -87,7 +162,7 @@ public final class ServerMetrics {
       }
 
       public DelayedOperationPurgatory type() {
-        return DelayedOperationPurgatory.ofAlias(metricsName());
+        return ofAlias(metricsName());
       }
 
       @Override
@@ -194,19 +269,6 @@ public final class ServerMetrics {
                   .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 HasMeter {
 
       private final BeanObject beanObject;
@@ -220,7 +282,7 @@ public final class ServerMetrics {
       }
 
       public Topic type() {
-        return Topic.of(metricsName());
+        return ofAlias(metricsName());
       }
 
       @Override
@@ -278,7 +340,7 @@ public final class ServerMetrics {
       return alias();
     }
 
-    public static class Gauge implements HasGauge {
+    public static class Gauge implements HasGauge<Long> {
 
       private final BeanObject beanObject;
 
@@ -305,4 +367,389 @@ public final class ServerMetrics {
       }
     }
   }
+
+  public static class Socket {
+    private static final String METRIC_TYPE = "socket-server-metrics";
+    private static final String PROP_LISTENER = "listener";
+    private static final String PROP_NETWORK_PROCESSOR = "networkProcessor";
+    private static final String PROP_CLIENT_SOFTWARE_NAME = "clientSoftwareName";
+    private static final String PROP_CLIENT_SOFTWARE_VERSION = "clientSoftwareVersion";
+
+    public static SocketMetric socket(MBeanClient mBeanClient) {
+      return new SocketMetric(
+          mBeanClient.queryBean(
+              BeanQuery.builder()
+                  .domainName("kafka.server")
+                  .property("type", METRIC_TYPE)
+                  .build()));
+    }
+
+    public static List<SocketListenerMetric> socketListener(MBeanClient mBeanClient) {
+      return mBeanClient
+          .queryBeans(
+              BeanQuery.builder()
+                  .domainName("kafka.server")
+                  .property("type", METRIC_TYPE)
+                  .property(PROP_LISTENER, "*")
+                  .build())
+          .stream()
+          .map(SocketListenerMetric::new)
+          .collect(Collectors.toList());
+    }
+
+    public static List<SocketNetworkProcessorMetric> socketNetworkProcessor(
+        MBeanClient mBeanClient) {
+      return mBeanClient
+          .queryBeans(
+              BeanQuery.builder()
+                  .domainName("kafka.server")
+                  .property("type", METRIC_TYPE)
+                  .property(PROP_LISTENER, "*")
+                  .property(PROP_NETWORK_PROCESSOR, "*")
+                  .build())
+          .stream()
+          .map(SocketNetworkProcessorMetric::new)
+          .collect(Collectors.toList());
+    }
+
+    public static List<Client> client(MBeanClient mBeanClient) {
+      return mBeanClient
+          .queryBeans(
+              BeanQuery.builder()
+                  .domainName("kafka.server")
+                  .property("type", METRIC_TYPE)
+                  .property(PROP_LISTENER, "*")
+                  .property(PROP_NETWORK_PROCESSOR, "*")
+                  .property(PROP_CLIENT_SOFTWARE_NAME, "*")
+                  .property(PROP_CLIENT_SOFTWARE_VERSION, "*")
+                  .build())
+          .stream()
+          .map(Client::new)
+          .collect(Collectors.toList());
+    }
+
+    public static class SocketMetric implements HasBeanObject {
+      private static final String MEMORY_POOL_DEPLETED_TIME_TOTAL = "MemoryPoolDepletedTimeTotal";
+      private static final String MEMORY_POOL_AVG_DEPLETED_PERCENT = "MemoryPoolAvgDepletedPercent";
+      private static final String BROKER_CONNECTION_ACCEPT_RATE = "broker-connection-accept-rate";
+
+      private final BeanObject beanObject;
+
+      public SocketMetric(BeanObject beanObject) {
+        this.beanObject = Objects.requireNonNull(beanObject);
+      }
+
+      @Override
+      public BeanObject beanObject() {
+        return beanObject;
+      }
+
+      public double memoryPoolDepletedTimeTotal() {
+        return (double) beanObject().attributes().get(MEMORY_POOL_DEPLETED_TIME_TOTAL);
+      }
+
+      public double memoryPoolAvgDepletedPercent() {
+        return (double) beanObject().attributes().get(MEMORY_POOL_AVG_DEPLETED_PERCENT);
+      }
+
+      public double brokerConnectionAcceptRate() {
+        return (double) beanObject().attributes().get(BROKER_CONNECTION_ACCEPT_RATE);
+      }
+    }
+
+    /** property : listener */
+    public static class SocketListenerMetric implements HasBeanObject {
+      private static final String CONNECTION_ACCEPT_THROTTLE_TIME =
+          "connection-accept-throttle-time";
+      private static final String CONNECTION_ACCEPT_RATE = "connection-accept-rate";
+      private static final String IP_CONNECTION_ACCEPT_THROTTLE_TIME =
+          "ip-connection-accept-throttle-time";
+
+      private final BeanObject beanObject;
+
+      public SocketListenerMetric(BeanObject beanObject) {
+        this.beanObject = Objects.requireNonNull(beanObject);
+      }
+
+      @Override
+      public BeanObject beanObject() {
+        return beanObject;
+      }
+
+      public String listener() {
+        return beanObject().properties().get(PROP_LISTENER);
+      }
+
+      public double connectionAcceptThrottleTime() {
+        return (double) beanObject().attributes().get(CONNECTION_ACCEPT_THROTTLE_TIME);
+      }
+
+      public double connectionAcceptRate() {
+        return (double) beanObject().attributes().get(CONNECTION_ACCEPT_RATE);
+      }
+
+      public double ipConnectionAcceptThrottleTime() {
+        return (double) beanObject().attributes().get(IP_CONNECTION_ACCEPT_THROTTLE_TIME);
+      }
+    }
+
+    /** property : listener and networkProcessor */
+    public static class SocketNetworkProcessorMetric implements HasBeanObject {
+      private static final String INCOMING_BYTE_TOTAL = "incoming-byte-total";
+      private static final String SELECT_TOTAL = "select-total";
+      private static final String SUCCESSFUL_AUTHENTICATION_RATE = "successful-authentication-rate";
+      private static final String REAUTHENTICATION_LATENCY_AVG = "reauthentication-latency-avg";
+      private static final String NETWORK_IO_RATE = "network-io-rate";
+      private static final String CONNECTION_CREATION_TOTAL = "connection-creation-total";
+      private static final String SUCCESSFUL_REAUTHENTICATION_RATE =
+          "successful-reauthentication-rate";
+      private static final String REQUEST_SIZE_MAX = "request-size-max";
+      private static final String CONNECTION_CLOSE_RATE = "connection-close-rate";
+      private static final String SUCCESSFUL_AUTHENTICATION_TOTAL =
+          "successful-authentication-total";
+      private static final String IO_TIME_NS_TOTAL = "io-time-ns-total";
+      private static final String CONNECTION_COUNT = "connection-count";
+      private static final String FAILED_REAUTHENTICATION_TOTAL = "failed-reauthentication-total";
+      private static final String REQUEST_RATE = "request-rate";
+      private static final String SUCCESSFUL_REAUTHENTICATION_TOTAL =
+          "successful-reauthentication-total";
+      private static final String RESPONSE_RATE = "response-rate";
+      private static final String CONNECTION_CREATION_RATE = "connection-creation-rate";
+      private static final String IO_WAIT_TIME_NS_AVG = "io-wait-time-ns-avg";
+      private static final String IO_WAIT_TIME_NS_TOTAL = "io-wait-time-ns-total";
+      private static final String OUTGOING_BYTE_RATE = "outgoing-byte-rate";
+      private static final String IOTIME_TOTAL = "iotime-total";
+      private static final String IO_RATIO = "io-ratio";
+      private static final String REQUEST_SIZE_AVG = "request-size-avg";
+      private static final String OUTGOING_BYTE_TOTAL = "outgoing-byte-total";
+      private static final String EXPIRED_CONNECTIONS_KILLED_COUNT =
+          "expired-connections-killed-count";
+      private static final String CONNECTION_CLOSE_TOTAL = "connection-close-total";
+      private static final String FAILED_REAUTHENTICATION_RATE = "failed-reauthentication-rate";
+      private static final String NETWORK_IO_TOTAL = "network-io-total";
+      private static final String FAILED_AUTHENTICATION_TOTAL = "failed-authentication-total";
+      private static final String INCOMING_BYTE_RATE = "incoming-byte-rate";
+      private static final String SELECT_RATE = "select-rate";
+      private static final String IO_TIME_NS_AVG = "io-time-ns-avg";
+      private static final String REAUTHENTICATION_LATENCY_MAX = "reauthentication-latency-max";
+      private static final String RESPONSE_TOTAL = "response-total";
+      private static final String FAILED_AUTHENTICATION_RATE = "failed-authentication-rate";
+      private static final String IO_WAIT_RATIO = "io-wait-ratio";
+      private static final String SUCCESSFUL_AUTHENTICATION_NO_REAUTH_TOTAL =
+          "successful-authentication-no-reauth-total";
+      private static final String REQUEST_TOTAL = "request-total";
+      private static final String IO_WAITTIME_TOTAL = "io-waittime-total";
+
+      private final BeanObject beanObject;
+
+      public SocketNetworkProcessorMetric(BeanObject beanObject) {
+        this.beanObject = Objects.requireNonNull(beanObject);
+      }
+
+      @Override
+      public BeanObject beanObject() {
+        return beanObject;
+      }
+
+      public String listener() {
+        return beanObject().properties().get(PROP_LISTENER);
+      }
+
+      public String networkProcessor() {
+        return beanObject().properties().get(PROP_NETWORK_PROCESSOR);
+      }
+
+      public double incomingByteTotal() {
+        return (double) beanObject().attributes().get(INCOMING_BYTE_TOTAL);
+      }
+
+      public double selectTotal() {
+        return (double) beanObject().attributes().get(SELECT_TOTAL);
+      }
+
+      public double successfulAuthenticationRate() {
+        return (double) beanObject().attributes().get(SUCCESSFUL_AUTHENTICATION_RATE);
+      }
+
+      public double reauthenticationLatencyAvg() {
+        return (double) beanObject().attributes().get(REAUTHENTICATION_LATENCY_AVG);
+      }
+
+      public double networkIoRate() {
+        return (double) beanObject().attributes().get(NETWORK_IO_RATE);
+      }
+
+      public double connectionCreationTotal() {
+        return (double) beanObject().attributes().get(CONNECTION_CREATION_TOTAL);
+      }
+
+      public double successfulReauthenticationRate() {
+        return (double) beanObject().attributes().get(SUCCESSFUL_REAUTHENTICATION_RATE);
+      }
+
+      public double requestSizeMax() {
+        return (double) beanObject().attributes().get(REQUEST_SIZE_MAX);
+      }
+
+      public double connectionCloseRate() {
+        return (double) beanObject().attributes().get(CONNECTION_CLOSE_RATE);
+      }
+
+      public double successfulAuthenticationTotal() {
+        return (double) beanObject().attributes().get(SUCCESSFUL_AUTHENTICATION_TOTAL);
+      }
+
+      public double ioTimeNsTotal() {
+        return (double) beanObject().attributes().get(IO_TIME_NS_TOTAL);
+      }
+
+      public double connectionCount() {
+        return (double) beanObject().attributes().get(CONNECTION_COUNT);
+      }
+
+      public double failedReauthenticationTotal() {
+        return (double) beanObject().attributes().get(FAILED_REAUTHENTICATION_TOTAL);
+      }
+
+      public double requestRate() {
+        return (double) beanObject().attributes().get(REQUEST_RATE);
+      }
+
+      public double successfulReauthenticationTotal() {
+        return (double) beanObject().attributes().get(SUCCESSFUL_REAUTHENTICATION_TOTAL);
+      }
+
+      public double responseRate() {
+        return (double) beanObject().attributes().get(RESPONSE_RATE);
+      }
+
+      public double connectionCreationRate() {
+        return (double) beanObject().attributes().get(CONNECTION_CREATION_RATE);
+      }
+
+      public double ioWaitTimeNsAvg() {
+        return (double) beanObject().attributes().get(IO_WAIT_TIME_NS_AVG);
+      }
+
+      public double ioWaitTimeNsTotal() {
+        return (double) beanObject().attributes().get(IO_WAIT_TIME_NS_TOTAL);
+      }
+
+      public double outgoingByteRate() {
+        return (double) beanObject().attributes().get(OUTGOING_BYTE_RATE);
+      }
+
+      public double iotimeTotal() {
+        return (double) beanObject().attributes().get(IOTIME_TOTAL);
+      }
+
+      public double ioRatio() {
+        return (double) beanObject().attributes().get(IO_RATIO);
+      }
+
+      public double requestSizeAvg() {
+        return (double) beanObject().attributes().get(REQUEST_SIZE_AVG);
+      }
+
+      public double outgoingByteTotal() {
+        return (double) beanObject().attributes().get(OUTGOING_BYTE_TOTAL);
+      }
+
+      public double expiredConnectionsKilledCount() {
+        return (double) beanObject().attributes().get(EXPIRED_CONNECTIONS_KILLED_COUNT);
+      }
+
+      public double connectionCloseTotal() {
+        return (double) beanObject().attributes().get(CONNECTION_CLOSE_TOTAL);
+      }
+
+      public double failedReauthenticationRate() {
+        return (double) beanObject().attributes().get(FAILED_REAUTHENTICATION_RATE);
+      }
+
+      public double networkIoTotal() {
+        return (double) beanObject().attributes().get(NETWORK_IO_TOTAL);
+      }
+
+      public double failedAuthenticationTotal() {
+        return (double) beanObject().attributes().get(FAILED_AUTHENTICATION_TOTAL);
+      }
+
+      public double incomingByteRate() {
+        return (double) beanObject().attributes().get(INCOMING_BYTE_RATE);
+      }
+
+      public double selectRate() {
+        return (double) beanObject().attributes().get(SELECT_RATE);
+      }
+
+      public double ioTimeNsAvg() {
+        return (double) beanObject().attributes().get(IO_TIME_NS_AVG);
+      }
+
+      public double reauthenticationLatencyMax() {
+        return (double) beanObject().attributes().get(REAUTHENTICATION_LATENCY_MAX);
+      }
+
+      public double responseTotal() {
+        return (double) beanObject().attributes().get(RESPONSE_TOTAL);
+      }
+
+      public double failedAuthenticationRate() {
+        return (double) beanObject().attributes().get(FAILED_AUTHENTICATION_RATE);
+      }
+
+      public double ioWaitRatio() {
+        return (double) beanObject().attributes().get(IO_WAIT_RATIO);
+      }
+
+      public double successfulAuthenticationNoReauthTotal() {
+        return (double) beanObject().attributes().get(SUCCESSFUL_AUTHENTICATION_NO_REAUTH_TOTAL);
+      }
+
+      public double requestTotal() {
+        return (double) beanObject().attributes().get(REQUEST_TOTAL);
+      }
+
+      public double ioWaittimeTotal() {
+        return (double) beanObject().attributes().get(IO_WAITTIME_TOTAL);
+      }
+    }
+
+    /** property : listener and networkProcessor and clientSoftwareName */
+    public static class Client implements HasBeanObject {
+
+      private static final String CONNECTIONS = "connections";
+      private final BeanObject beanObject;
+
+      public Client(BeanObject beanObject) {
+        this.beanObject = Objects.requireNonNull(beanObject);
+      }
+
+      @Override
+      public BeanObject beanObject() {
+        return beanObject;
+      }
+
+      public String listener() {
+        return beanObject().properties().get(PROP_LISTENER);
+      }
+
+      public String networkProcessor() {
+        return beanObject().properties().get(PROP_NETWORK_PROCESSOR);
+      }
+
+      public String clientSoftwareName() {
+        return beanObject().properties().get(PROP_CLIENT_SOFTWARE_NAME);
+      }
+
+      public String clientSoftwareVersion() {
+        return beanObject().properties().get(PROP_CLIENT_SOFTWARE_VERSION);
+      }
+
+      public int connections() {
+        return (int) beanObject().attributes().get(CONNECTIONS);
+      }
+    }
+  }
 }
diff --git a/common/src/main/java/org/astraea/common/metrics/client/consumer/ConsumerMetrics.java b/common/src/main/java/org/astraea/common/metrics/client/consumer/ConsumerMetrics.java
index 872b8b58..2d443ca3 100644
--- a/common/src/main/java/org/astraea/common/metrics/client/consumer/ConsumerMetrics.java
+++ b/common/src/main/java/org/astraea/common/metrics/client/consumer/ConsumerMetrics.java
@@ -72,7 +72,7 @@ public class ConsumerMetrics {
         .collect(Collectors.toUnmodifiableList());
   }
 
-  public static Collection<HasConsumerMetrics> consumer(MBeanClient mBeanClient) {
+  public static Collection<HasConsumerMetrics> of(MBeanClient mBeanClient) {
     return mBeanClient
         .queryBeans(
             BeanQuery.builder()
diff --git a/common/src/main/java/org/astraea/common/metrics/platform/HostMetrics.java b/common/src/main/java/org/astraea/common/metrics/platform/HostMetrics.java
index a75dad73..6546616b 100644
--- a/common/src/main/java/org/astraea/common/metrics/platform/HostMetrics.java
+++ b/common/src/main/java/org/astraea/common/metrics/platform/HostMetrics.java
@@ -18,7 +18,6 @@ package org.astraea.common.metrics.platform;
 
 import org.astraea.common.metrics.BeanQuery;
 import org.astraea.common.metrics.MBeanClient;
-import org.astraea.common.metrics.broker.HasGauge;
 
 public final class HostMetrics {
 
@@ -37,27 +36,5 @@ public final class HostMetrics {
             BeanQuery.builder().domainName("java.lang").property("type", "Memory").build()));
   }
 
-  public static HasGauge 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 HasGauge 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() {}
 }


Test patch:
diff --git a/app/src/test/java/org/astraea/app/balancer/BalancerUtilsTest.java b/app/src/test/java/org/astraea/app/balancer/BalancerUtilsTest.java
index bed4c99d..2e43049d 100644
--- a/app/src/test/java/org/astraea/app/balancer/BalancerUtilsTest.java
+++ b/app/src/test/java/org/astraea/app/balancer/BalancerUtilsTest.java
@@ -37,20 +37,20 @@ import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 class BalancerUtilsTest {
-  private static final HasGauge OLD_TP1_0 =
+  private static final HasGauge<Long> OLD_TP1_0 =
       fakePartitionBeanObject("Log", LogMetrics.Log.SIZE.metricName(), "test-1", "0", 1000, 1000L);
-  private static final HasGauge NEW_TP1_0 =
+  private static final HasGauge<Long> NEW_TP1_0 =
       fakePartitionBeanObject(
           "Log", LogMetrics.Log.SIZE.metricName(), "test-1", "0", 500000, 10000L);
-  private static final HasGauge OLD_TP1_1 =
+  private static final HasGauge<Long> OLD_TP1_1 =
       fakePartitionBeanObject("Log", LogMetrics.Log.SIZE.metricName(), "test-1", "1", 500, 1000L);
-  private static final HasGauge NEW_TP1_1 =
+  private static final HasGauge<Long> NEW_TP1_1 =
       fakePartitionBeanObject(
           "Log", LogMetrics.Log.SIZE.metricName(), "test-1", "1", 100000000, 10000L);
-  private static final HasGauge LEADER_BROKER1 =
+  private static final HasGauge<Long> LEADER_BROKER1 =
       fakeBrokerBeanObject(
           "ReplicaManager", ServerMetrics.ReplicaManager.LEADER_COUNT.metricName(), 2, 10000L);
-  private static final HasGauge LEADER_BROKER2 =
+  private static final HasGauge<Long> LEADER_BROKER2 =
       fakeBrokerBeanObject(
           "ReplicaManager", ServerMetrics.ReplicaManager.LEADER_COUNT.metricName(), 4, 10000L);
   private static final Collection<HasBeanObject> broker1 =
diff --git a/app/src/test/java/org/astraea/app/performance/MetricsTest.java b/app/src/test/java/org/astraea/app/performance/MetricsTest.java
deleted file mode 100644
index 60229dd3..00000000
--- a/app/src/test/java/org/astraea/app/performance/MetricsTest.java
+++ /dev/null
@@ -1,51 +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.performance;
-
-import java.util.Random;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Test;
-
-public class MetricsTest {
-
-  @Test
-  void testAverage() {
-    Random rand = new Random();
-    final int num = 1000;
-    double avg = 0.0;
-    var metrics = Report.of("c", () -> false);
-
-    Assertions.assertEquals(0, metrics.avgLatency());
-
-    for (int i = 0; i < num; ++i) {
-      long next = rand.nextInt();
-      avg += ((double) next - avg) / (i + 1);
-      metrics.record(next, 0);
-    }
-
-    Assertions.assertEquals(avg, metrics.avgLatency());
-  }
-
-  @Test
-  void testBytes() {
-    var metrics = Report.of("c", () -> false);
-
-    Assertions.assertEquals(0, metrics.totalBytes());
-    metrics.record(0L, 1000);
-    Assertions.assertEquals(1000, metrics.totalBytes());
-  }
-}
diff --git a/app/src/test/java/org/astraea/app/performance/TrackerTest.java b/app/src/test/java/org/astraea/app/performance/TrackerTest.java
index c2014ad2..83e6758c 100644
--- a/app/src/test/java/org/astraea/app/performance/TrackerTest.java
+++ b/app/src/test/java/org/astraea/app/performance/TrackerTest.java
@@ -20,6 +20,7 @@ import java.time.Duration;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
 import org.astraea.common.Utils;
 import org.astraea.common.metrics.client.HasNodeMetrics;
 import org.junit.jupiter.api.Assertions;
@@ -28,17 +29,10 @@ import org.mockito.Mockito;
 
 public class TrackerTest {
 
-  @Test
-  void testEmptyReports() {
-    var tracker = TrackerThread.create(() -> true, List::of);
-    Utils.sleep(Duration.ofSeconds(2));
-    Assertions.assertTrue(tracker.closed());
-  }
-
   @Test
   void testCloseConsumer() {
     var closed = new AtomicBoolean(false);
-    var tracker = TrackerThread.create(() -> true, () -> List.of(Report.of("c", closed::get)));
+    var tracker = TrackerThread.create(() -> true, closed::get);
     Assertions.assertFalse(tracker.closed());
     closed.set(true);
     Utils.sleep(Duration.ofSeconds(2));
@@ -49,7 +43,7 @@ public class TrackerTest {
   void testPartialClose() {
     var closed0 = new AtomicBoolean(false);
     var closed1 = new AtomicBoolean(false);
-    var tracker = TrackerThread.create(closed0::get, () -> List.of(Report.of("c", closed1::get)));
+    var tracker = TrackerThread.create(closed0::get, closed1::get);
     Assertions.assertFalse(tracker.closed());
     closed0.set(true);
     Utils.sleep(Duration.ofSeconds(2));
@@ -80,10 +74,7 @@ public class TrackerTest {
     var closed = new AtomicBoolean(false);
     var f =
         CompletableFuture.runAsync(
-            TrackerThread.trackerLoop(
-                closed::get,
-                producerClosed::get,
-                () -> List.of(Report.of("c", consumerClosed::get))));
+            TrackerThread.trackerLoop(closed::get, producerClosed::get, consumerClosed::get));
     Assertions.assertFalse(f.isDone());
     producerClosed.set(true);
     Utils.sleep(Duration.ofSeconds(2));
@@ -95,19 +86,25 @@ public class TrackerTest {
 
   @Test
   void testProducerPrinter() {
-    var report = Report.of("c", () -> false);
+    var report = Mockito.mock(Report.class);
+    var records = new AtomicLong(0);
+    Mockito.when(report.records()).thenAnswer(a -> records.get());
     var printer = new TrackerThread.ProducerPrinter(() -> List.of(report));
     Assertions.assertFalse(printer.tryToPrint(Duration.ofSeconds(1)));
-    report.record(10, 100);
+    records.set(100);
     Assertions.assertTrue(printer.tryToPrint(Duration.ofSeconds(1)));
+    Assertions.assertFalse(printer.tryToPrint(Duration.ofSeconds(1)));
   }
 
   @Test
   void testConsumerPrinter() {
-    var report = Report.of("c", () -> false);
+    var report = Mockito.mock(Report.class);
+    var records = new AtomicLong(0);
+    Mockito.when(report.records()).thenAnswer(a -> records.get());
     var printer = new TrackerThread.ConsumerPrinter(() -> List.of(report));
     Assertions.assertFalse(printer.tryToPrint(Duration.ofSeconds(1)));
-    report.record(10, 100);
+    records.set(100);
     Assertions.assertTrue(printer.tryToPrint(Duration.ofSeconds(1)));
+    Assertions.assertFalse(printer.tryToPrint(Duration.ofSeconds(1)));
   }
 }
diff --git a/app/src/test/java/org/astraea/app/web/BalancerHandlerTest.java b/app/src/test/java/org/astraea/app/web/BalancerHandlerTest.java
index f1a11769..0972460c 100644
--- a/app/src/test/java/org/astraea/app/web/BalancerHandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/BalancerHandlerTest.java
@@ -28,11 +28,14 @@ import org.astraea.app.balancer.RebalancePlanProposal;
 import org.astraea.app.balancer.log.ClusterLogAllocation;
 import org.astraea.common.Utils;
 import org.astraea.common.admin.Admin;
+import org.astraea.common.admin.ClusterBean;
 import org.astraea.common.admin.ClusterInfo;
 import org.astraea.common.admin.NodeInfo;
 import org.astraea.common.admin.Replica;
+import org.astraea.common.cost.ClusterCost;
 import org.astraea.common.cost.HasClusterCost;
 import org.astraea.common.cost.HasMoveCost;
+import org.astraea.common.cost.ReplicaSizeCost;
 import org.astraea.common.producer.Producer;
 import org.astraea.it.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
@@ -43,7 +46,7 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
   @Test
   void testReport() {
     try (var admin = Admin.of(bootstrapServers())) {
-      var handler = new BalancerHandler(admin);
+      var handler = new BalancerHandler(admin, new DegradeCost(), new ReplicaSizeCost());
       var report =
           Assertions.assertInstanceOf(
               BalancerHandler.Report.class,
@@ -74,7 +77,7 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
   void testTopic() {
     var topicNames = createAndProduceTopic(3);
     try (var admin = Admin.of(bootstrapServers())) {
-      var handler = new BalancerHandler(admin);
+      var handler = new BalancerHandler(admin, new DegradeCost(), new ReplicaSizeCost());
       var report =
           Assertions.assertInstanceOf(
               BalancerHandler.Report.class,
@@ -98,11 +101,24 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
     }
   }
 
+  /** The score will getter better after each call, pretend we find a better plan */
+  private static class DegradeCost implements HasClusterCost {
+
+    private double value0 = 1.0;
+
+    @Override
+    public synchronized ClusterCost clusterCost(
+        ClusterInfo<Replica> clusterInfo, ClusterBean clusterBean) {
+      value0 = value0 * 0.998;
+      return () -> value0;
+    }
+  }
+
   @Test
   void testTopics() {
     var topicNames = createAndProduceTopic(3);
     try (var admin = Admin.of(bootstrapServers())) {
-      var handler = new BalancerHandler(admin);
+      var handler = new BalancerHandler(admin, new DegradeCost(), new ReplicaSizeCost());
       var report =
           Assertions.assertInstanceOf(
               BalancerHandler.Report.class,
diff --git a/common/src/test/java/org/astraea/common/admin/ClusterBeanTest.java b/common/src/test/java/org/astraea/common/admin/ClusterBeanTest.java
index 1c99cacd..bd596b83 100644
--- a/common/src/test/java/org/astraea/common/admin/ClusterBeanTest.java
+++ b/common/src/test/java/org/astraea/common/admin/ClusterBeanTest.java
@@ -82,12 +82,12 @@ class ClusterBeanTest {
         ClusterBean.of(
             Map.of(
                 1,
-                List.of(HasGauge.of(testBeanObjectWithPartition1)),
+                List.of(HasGauge.ofLong(testBeanObjectWithPartition1)),
                 2,
                 List.of(
-                    HasGauge.of(testBeanObjectWithoutPartition),
-                    HasGauge.of(testBeanObjectWithPartition2),
-                    HasGauge.of(testBeanObjectWithPartition3))));
+                    HasGauge.ofLong(testBeanObjectWithoutPartition),
+                    HasGauge.ofLong(testBeanObjectWithPartition2),
+                    HasGauge.ofLong(testBeanObjectWithPartition3))));
     // test all
     Assertions.assertEquals(2, clusterBean.all().size());
     Assertions.assertEquals(1, clusterBean.all().get(1).size());
diff --git a/common/src/test/java/org/astraea/common/cost/ReplicaDiskInCostTest.java b/common/src/test/java/org/astraea/common/cost/ReplicaDiskInCostTest.java
index 447dd031..d54c21d6 100644
--- a/common/src/test/java/org/astraea/common/cost/ReplicaDiskInCostTest.java
+++ b/common/src/test/java/org/astraea/common/cost/ReplicaDiskInCostTest.java
@@ -32,17 +32,17 @@ import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 class ReplicaDiskInCostTest extends RequireBrokerCluster {
-  private static final HasGauge OLD_TP1_0 =
+  private static final HasGauge<Long> OLD_TP1_0 =
       fakeBeanObject("Log", LogMetrics.Log.SIZE.metricName(), "test-1", "0", 1000, 1000L);
-  private static final HasGauge NEW_TP1_0 =
+  private static final HasGauge<Long> NEW_TP1_0 =
       fakeBeanObject("Log", LogMetrics.Log.SIZE.metricName(), "test-1", "0", 50000000, 5000L);
-  private static final HasGauge OLD_TP1_1 =
+  private static final HasGauge<Long> OLD_TP1_1 =
       fakeBeanObject("Log", LogMetrics.Log.SIZE.metricName(), "test-1", "1", 500, 1000L);
-  private static final HasGauge NEW_TP1_1 =
+  private static final HasGauge<Long> NEW_TP1_1 =
       fakeBeanObject("Log", LogMetrics.Log.SIZE.metricName(), "test-1", "1", 100000000, 5000L);
-  private static final HasGauge OLD_TP2_0 =
+  private static final HasGauge<Long> OLD_TP2_0 =
       fakeBeanObject("Log", LogMetrics.Log.SIZE.metricName(), "test-2", "0", 200, 1000L);
-  private static final HasGauge NEW_TP2_0 =
+  private static final HasGauge<Long> NEW_TP2_0 =
       fakeBeanObject("Log", LogMetrics.Log.SIZE.metricName(), "test-2", "0", 40000000, 5000L);
 
   /*
diff --git a/common/src/test/java/org/astraea/common/metrics/MetricsTestUtil.java b/common/src/test/java/org/astraea/common/metrics/MetricsTestUtil.java
index a8940415..89f92696 100644
--- a/common/src/test/java/org/astraea/common/metrics/MetricsTestUtil.java
+++ b/common/src/test/java/org/astraea/common/metrics/MetricsTestUtil.java
@@ -46,14 +46,14 @@ public class MetricsTestUtil {
     } else if (hasBeanObject instanceof HasHistogram) {
       testHistogram((HasHistogram) hasBeanObject);
     } else if (hasBeanObject instanceof HasGauge) {
-      testGauge((HasGauge) hasBeanObject);
+      testGauge((HasGauge<?>) hasBeanObject);
     } else {
       throw new UnsupportedOperationException(
           String.format("Not implement. %s", hasBeanObject.getClass()));
     }
   }
 
-  private static void testGauge(HasGauge hasGauge) {
+  private static void testGauge(HasGauge<?> hasGauge) {
     Assertions.assertDoesNotThrow(hasGauge::value);
   }
 
diff --git a/common/src/test/java/org/astraea/common/metrics/broker/ServerMetricsTest.java b/common/src/test/java/org/astraea/common/metrics/broker/ServerMetricsTest.java
index 6eb203eb..dd337c36 100644
--- a/common/src/test/java/org/astraea/common/metrics/broker/ServerMetricsTest.java
+++ b/common/src/test/java/org/astraea/common/metrics/broker/ServerMetricsTest.java
@@ -16,7 +16,10 @@
  */
 package org.astraea.common.metrics.broker;
 
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.util.Arrays;
 import java.util.List;
@@ -26,17 +29,122 @@ import org.astraea.common.Utils;
 import org.astraea.common.metrics.BeanObject;
 import org.astraea.common.metrics.MBeanClient;
 import org.astraea.common.metrics.MetricsTestUtil;
+import org.astraea.it.RequireSingleBrokerCluster;
 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;
 
-public class ServerMetricsTest {
+public class ServerMetricsTest extends RequireSingleBrokerCluster {
 
   @ParameterizedTest()
   @EnumSource(value = ServerMetrics.DelayedOperationPurgatory.class)
   void testPurgatorySize(ServerMetrics.DelayedOperationPurgatory request) {
-    request.fetch(MBeanClient.local()).forEach(MetricsTestUtil::validate);
+    var gauges = request.fetch(MBeanClient.local());
+    Assertions.assertFalse(gauges.isEmpty());
+    gauges.forEach(MetricsTestUtil::validate);
+  }
+
+  @ParameterizedTest()
+  @EnumSource(value = ServerMetrics.KafkaServer.class)
+  void testKafkaServer(ServerMetrics.KafkaServer request) {
+    MetricsTestUtil.validate(request.fetch(MBeanClient.local()));
+  }
+
+  @Test
+  void testKafkaServerOtherMetrics() {
+    MetricsTestUtil.validate(ServerMetrics.KafkaServer.clusterId(MBeanClient.local()));
+  }
+
+  @Test
+  void testSocketMetrics() {
+    var socketMetric = ServerMetrics.Socket.socket(MBeanClient.local());
+
+    assertDoesNotThrow(socketMetric::brokerConnectionAcceptRate);
+    assertDoesNotThrow(socketMetric::memoryPoolAvgDepletedPercent);
+    assertDoesNotThrow(socketMetric::memoryPoolDepletedTimeTotal);
+  }
+
+  @Test
+  void testSocketListenerMetrics() {
+    var socketListenerMetrics = ServerMetrics.Socket.socketListener(MBeanClient.local());
+    assertTrue(socketListenerMetrics.size() > 0);
+    socketListenerMetrics.forEach(
+        x -> {
+          assertNotNull(x.listener());
+
+          assertDoesNotThrow(x::connectionAcceptRate);
+          assertDoesNotThrow(x::connectionAcceptThrottleTime);
+          assertDoesNotThrow(x::ipConnectionAcceptThrottleTime);
+        });
+  }
+
+  @Test
+  void testSocketNetworkProcessorMetrics() {
+    var socketNetworkProcessorMetrics =
+        ServerMetrics.Socket.socketNetworkProcessor(MBeanClient.local());
+    assertTrue(socketNetworkProcessorMetrics.size() > 0);
+    socketNetworkProcessorMetrics.forEach(
+        x -> {
+          assertNotNull(x.listener());
+          assertNotNull(x.networkProcessor());
+
+          assertDoesNotThrow(x::connectionCloseRate);
+          assertDoesNotThrow(x::incomingByteTotal);
+          assertDoesNotThrow(x::selectTotal);
+          assertDoesNotThrow(x::successfulAuthenticationRate);
+          assertDoesNotThrow(x::reauthenticationLatencyAvg);
+          assertDoesNotThrow(x::networkIoRate);
+          assertDoesNotThrow(x::connectionCreationTotal);
+          assertDoesNotThrow(x::successfulReauthenticationRate);
+          assertDoesNotThrow(x::requestSizeMax);
+          assertDoesNotThrow(x::connectionCloseRate);
+          assertDoesNotThrow(x::successfulAuthenticationTotal);
+          assertDoesNotThrow(x::ioTimeNsTotal);
+          assertDoesNotThrow(x::connectionCount);
+          assertDoesNotThrow(x::failedReauthenticationTotal);
+          assertDoesNotThrow(x::requestRate);
+          assertDoesNotThrow(x::successfulReauthenticationTotal);
+          assertDoesNotThrow(x::responseRate);
+          assertDoesNotThrow(x::connectionCreationRate);
+          assertDoesNotThrow(x::ioWaitTimeNsAvg);
+          assertDoesNotThrow(x::ioWaitTimeNsTotal);
+          assertDoesNotThrow(x::outgoingByteRate);
+          assertDoesNotThrow(x::iotimeTotal);
+          assertDoesNotThrow(x::ioRatio);
+          assertDoesNotThrow(x::requestSizeAvg);
+          assertDoesNotThrow(x::outgoingByteTotal);
+          assertDoesNotThrow(x::expiredConnectionsKilledCount);
+          assertDoesNotThrow(x::connectionCloseTotal);
+          assertDoesNotThrow(x::failedReauthenticationRate);
+          assertDoesNotThrow(x::networkIoTotal);
+          assertDoesNotThrow(x::failedAuthenticationTotal);
+          assertDoesNotThrow(x::incomingByteRate);
+          assertDoesNotThrow(x::selectRate);
+          assertDoesNotThrow(x::ioTimeNsAvg);
+          assertDoesNotThrow(x::reauthenticationLatencyMax);
+          assertDoesNotThrow(x::responseTotal);
+          assertDoesNotThrow(x::failedAuthenticationRate);
+          assertDoesNotThrow(x::ioWaitRatio);
+          assertDoesNotThrow(x::successfulAuthenticationNoReauthTotal);
+          assertDoesNotThrow(x::requestTotal);
+          assertDoesNotThrow(x::ioWaittimeTotal);
+        });
+  }
+
+  @Test
+  void testSocketClientMetrics() {
+    var clientMetrics = ServerMetrics.Socket.client(MBeanClient.local());
+    assertTrue(clientMetrics.size() > 0);
+    clientMetrics.forEach(
+        x -> {
+          assertNotNull(x.listener());
+          assertNotNull(x.networkProcessor());
+          assertNotNull(x.clientSoftwareName());
+          assertNotNull(x.clientSoftwareVersion());
+
+          assertDoesNotThrow(x::connections);
+        });
   }
 
   @Test
@@ -45,13 +153,13 @@ public class ServerMetricsTest {
         .forEach(
             t ->
                 Assertions.assertEquals(
-                    t, ServerMetrics.Topic.of(t.metricName().toLowerCase(Locale.ROOT))));
+                    t, ServerMetrics.Topic.ofAlias(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"));
+                    t, ServerMetrics.Topic.ofAlias(t.metricName().toUpperCase(Locale.ROOT))));
+    assertThrows(IllegalArgumentException.class, () -> ServerMetrics.Topic.ofAlias("nothing"));
   }
 
   @ParameterizedTest
diff --git a/common/src/test/java/org/astraea/common/metrics/client/consumer/HasConsumerMetricsTest.java b/common/src/test/java/org/astraea/common/metrics/client/consumer/HasConsumerMetricsTest.java
index 864172c7..cbcbaaec 100644
--- a/common/src/test/java/org/astraea/common/metrics/client/consumer/HasConsumerMetricsTest.java
+++ b/common/src/test/java/org/astraea/common/metrics/client/consumer/HasConsumerMetricsTest.java
@@ -51,7 +51,7 @@ public class HasConsumerMetricsTest extends RequireSingleBrokerCluster {
             .build()) {
       Assertions.assertEquals(10, consumer.poll(10, Duration.ofSeconds(5)).size());
       consumer.commitOffsets(Duration.ofSeconds(2));
-      var metrics = ConsumerMetrics.consumer(MBeanClient.local());
+      var metrics = ConsumerMetrics.of(MBeanClient.local());
       Assertions.assertEquals(1, metrics.size());
       var m = metrics.iterator().next();
       Assertions.assertNotNull(m.clientId());
diff --git a/common/src/test/java/org/astraea/common/metrics/platform/LinuxDiskMetricsTest.java b/common/src/test/java/org/astraea/common/metrics/platform/LinuxDiskMetricsTest.java
deleted file mode 100644
index e751bc4c..00000000
--- a/common/src/test/java/org/astraea/common/metrics/platform/LinuxDiskMetricsTest.java
+++ /dev/null
@@ -1,40 +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.common.metrics.platform;
-
-import static org.junit.jupiter.api.condition.OS.LINUX;
-
-import org.astraea.common.metrics.MBeanClient;
-import org.astraea.common.metrics.MetricsTestUtil;
-import org.astraea.it.RequireSingleBrokerCluster;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.EnabledOnOs;
-
-public class LinuxDiskMetricsTest extends RequireSingleBrokerCluster {
-
-  @Test
-  @EnabledOnOs(LINUX)
-  void linuxDiskReadBytes() {
-    MetricsTestUtil.validate(HostMetrics.linuxDiskReadBytes(MBeanClient.local()));
-  }
-
-  @Test
-  @EnabledOnOs(LINUX)
-  void linuxDiskWriteBytes() {
-    MetricsTestUtil.validate(HostMetrics.linuxDiskWriteBytes(MBeanClient.local()));
-  }
-}