Repository: skiptests/astraea
Pull request #371 (2022-05-27 05:36:21)

Create `ClusterInfo` from `Admin`
新增 `ClusterInfo#of(Admin)`，從 `Admin` 建立 `ClusterInfo`，這是目前唯一有包含到 data directory 的 `ClusterInfo` 實作。

實作內有用到 `Admin#replicas()`，目前實作下會遇到 offline replica 無法索取的問題，這部分在等 #308 修正。

chia7712: @garyparrot  抱歉#392造成了許多衝突，麻煩你解決一下衝突
chia7712: @garyparrot 不好意思，license header已經合併，要請你rebase

Patch:
diff --git a/.gitignore b/.gitignore
index 3f50cc0e..2dc26129 100644
--- a/.gitignore
+++ b/.gitignore
@@ -13,7 +13,7 @@ build
 # generated dockerfiles
 zookeeper.dockerfile
 broker.dockerfile
-kafka_tool.dockerfile
+app.dockerfile
 spark.dockerfile
 deps.dockerfile
 # we don't put binary file to git repo
diff --git a/README.md b/README.md
index aa976303..9724a9db 100644
--- a/README.md
+++ b/README.md
@@ -219,7 +219,7 @@ Run the tool from source code
 
 Run the tool from release
 ```shell
-./docker/start_kafka_tool.sh offset --bootstrap.servers 192.168.50.178:19993
+./docker/start_app.sh offset --bootstrap.servers 192.168.50.178:19993
 ```
 
 ### Offset Explorer Configurations
@@ -253,17 +253,17 @@ Run the tool from source code
 Run the tool from release
 ```shell
 # fetch every Mbeans from specific JMX server.
-./docker/start_kafka_tool.sh metrics --jmx.server 192.168.50.178:1099
+./docker/start_app.sh metrics --jmx.server 192.168.50.178:1099
 
 # fetch any Mbean that its object name contains property "type=Memory".
-./docker/start_kafka_tool.sh metrics --jmx.server 192.168.50.178:1099 --property type=Memory
+./docker/start_app.sh metrics --jmx.server 192.168.50.178:1099 --property type=Memory
 
 # fetch any Mbean that belongs to "kafka.network" domain name,
 # and it's object name contains two properties "request=Metadata" and "name=LocalTimeMs".
-./docker/start_kafka_tool.sh metrics --jmx.server 192.168.50.178:1099 --domain kafka.network --property request=Metadata --property name=LocalTimeMs
+./docker/start_app.sh metrics --jmx.server 192.168.50.178:1099 --domain kafka.network --property request=Metadata --property name=LocalTimeMs
 
 # list all Mbeans' object name on specific JMX server.
-./docker/start_kafka_tool.sh metrics --jmx.server 192.168.50.178:1099 --view-object-name-list
+./docker/start_app.sh metrics --jmx.server 192.168.50.178:1099 --view-object-name-list
 ```
 
 ### Metric Explorer Configurations
@@ -390,7 +390,7 @@ $ ./gradlew run --args="monitor --bootstrap.servers 192.168.103.39:9092"
 2. --port: the port used by web server
 
 ```shell
-./docker/start_kafka_tool.sh web --bootstrap.servers 192.168.50.178:19993 --port 12345"
+./docker/start_app.sh web --bootstrap.servers 192.168.50.178:19993 --port 12345"
 ```
 
 ## Query all topics 
@@ -411,4 +411,6 @@ GET http://localhost:12345/groups
 ## Query single group
 ```shell
 GET http://localhost:12345/groups/g1
-```
\ No newline at end of file
+```
+
+## [Astraea Web Server 中文文件連結](./docs/web_server/README.md)
diff --git a/app/build.gradle b/app/build.gradle
index b5ab15ae..c4386d41 100644
--- a/app/build.gradle
+++ b/app/build.gradle
@@ -26,7 +26,7 @@ repositories {
 }
 
 def versions = [
-    kafka: project.properties['kafka.version'] ?: "3.1.1",
+    kafka: project.properties['kafka.version'] ?: "3.2.0",
     junit: project.properties['junit.version'] ?: "5.8.2",
     mockito: project.properties['mockito.version'] ?: "4.2.0",
     jcommander: project.properties['jcommander.version'] ?: "1.81",
diff --git a/app/src/main/java/org/astraea/app/admin/Admin.java b/app/src/main/java/org/astraea/app/admin/Admin.java
index 04fd2702..28311065 100644
--- a/app/src/main/java/org/astraea/app/admin/Admin.java
+++ b/app/src/main/java/org/astraea/app/admin/Admin.java
@@ -21,6 +21,9 @@ import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.stream.Collectors;
+import org.astraea.app.cost.ClusterInfo;
+import org.astraea.app.cost.NodeInfo;
 
 public interface Admin extends Closeable {
 
@@ -108,8 +111,13 @@ public interface Admin extends Closeable {
    */
   Map<Integer, Config> brokers(Set<Integer> brokerIds);
 
-  /** @return all brokers' ids */
-  Set<Integer> brokerIds();
+  /** @return all alive brokers' ids */
+  default Set<Integer> brokerIds() {
+    return nodes().stream().map(NodeInfo::id).collect(Collectors.toUnmodifiableSet());
+  }
+
+  /** @return all alive node information in the cluster */
+  Set<NodeInfo> nodes();
 
   /**
    * list all partitions belongs to input brokers
@@ -182,6 +190,17 @@ public interface Admin extends Closeable {
   /** @return all quotas */
   Collection<Quota> quotas();
 
+  /** @return a snapshot object of cluster state at the moment */
+  default ClusterInfo clusterInfo() {
+    return clusterInfo(topicNames());
+  }
+
+  /**
+   * @param topics query only this subset of topics
+   * @return a snapshot object of cluster state at the moment
+   */
+  ClusterInfo clusterInfo(Set<String> topics);
+
   @Override
   void close();
 }
diff --git a/app/src/main/java/org/astraea/app/admin/Builder.java b/app/src/main/java/org/astraea/app/admin/Builder.java
index 5558d6af..800dc055 100644
--- a/app/src/main/java/org/astraea/app/admin/Builder.java
+++ b/app/src/main/java/org/astraea/app/admin/Builder.java
@@ -24,6 +24,7 @@ import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.NoSuchElementException;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
@@ -39,7 +40,6 @@ import org.apache.kafka.clients.admin.NewPartitionReassignment;
 import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.clients.admin.OffsetSpec;
 import org.apache.kafka.common.ElectionType;
-import org.apache.kafka.common.Node;
 import org.apache.kafka.common.TopicPartitionReplica;
 import org.apache.kafka.common.config.ConfigResource;
 import org.apache.kafka.common.errors.ElectionNotNeededException;
@@ -48,6 +48,10 @@ import org.apache.kafka.common.quota.ClientQuotaEntity;
 import org.apache.kafka.common.quota.ClientQuotaFilter;
 import org.apache.kafka.common.quota.ClientQuotaFilterComponent;
 import org.astraea.app.common.Utils;
+import org.astraea.app.cost.ClusterInfo;
+import org.astraea.app.cost.NodeInfo;
+import org.astraea.app.cost.ReplicaInfo;
+import org.astraea.app.metrics.HasBeanObject;
 
 public class Builder {
 
@@ -83,12 +87,12 @@ public class Builder {
     }
 
     @Override
-    public Set<Integer> brokerIds() {
+    public Set<NodeInfo> nodes() {
       return Utils.packException(
           () ->
               admin.describeCluster().nodes().get().stream()
-                  .map(Node::id)
-                  .collect(Collectors.toSet()));
+                  .map(NodeInfo::of)
+                  .collect(Collectors.toUnmodifiableSet()));
     }
 
     @Override
@@ -426,6 +430,96 @@ public class Builder {
       return Quota.of(
           Utils.packException(() -> admin.describeClientQuotas(filter).entities().get()));
     }
+
+    @Override
+    public ClusterInfo clusterInfo(Set<String> topics) {
+      final var nodeInfo = this.nodes().stream().collect(Collectors.toUnmodifiableList());
+
+      final Map<String, List<ReplicaInfo>> topicToReplicasMap =
+          Utils.packException(() -> this.replicas(topics)).entrySet().stream()
+              .flatMap(
+                  entry -> {
+                    final var topicPartition = entry.getKey();
+                    final var replicas = entry.getValue();
+
+                    return replicas.stream()
+                        .map(
+                            replica ->
+                                ReplicaInfo.of(
+                                    topicPartition.topic(),
+                                    topicPartition.partition(),
+                                    nodeInfo.stream()
+                                        .filter(x -> x.id() == replica.broker())
+                                        .findFirst()
+                                        .orElse(NodeInfo.ofOfflineNode(replica.broker())),
+                                    replica.leader(),
+                                    replica.inSync(),
+                                    replica.isOffline(),
+                                    replica.path()));
+                  })
+              .collect(Collectors.groupingBy(ReplicaInfo::topic));
+      final var dataDirectories =
+          this.brokerFolders(
+                  nodeInfo.stream().map(NodeInfo::id).collect(Collectors.toUnmodifiableSet()))
+              .entrySet()
+              .stream()
+              .collect(
+                  Collectors.toUnmodifiableMap(Map.Entry::getKey, x -> Set.copyOf(x.getValue())));
+
+      return new ClusterInfo() {
+        @Override
+        public List<NodeInfo> nodes() {
+          return nodeInfo;
+        }
+
+        @Override
+        public Set<String> dataDirectories(int brokerId) {
+          final var set = dataDirectories.get(brokerId);
+          if (set == null) throw new NoSuchElementException("Unknown broker \"" + brokerId + "\"");
+          else return set;
+        }
+
+        @Override
+        public List<ReplicaInfo> availableReplicaLeaders(String topic) {
+          return replicas(topic).stream()
+              .filter(ReplicaInfo::isLeader)
+              .filter(ReplicaInfo::isOnlineReplica)
+              .collect(Collectors.toUnmodifiableList());
+        }
+
+        @Override
+        public List<ReplicaInfo> availableReplicas(String topic) {
+          return replicas(topic).stream()
+              .filter(ReplicaInfo::isOnlineReplica)
+              .collect(Collectors.toUnmodifiableList());
+        }
+
+        @Override
+        public Set<String> topics() {
+          return topics;
+        }
+
+        @Override
+        public List<ReplicaInfo> replicas(String topic) {
+          final var replicaInfos = topicToReplicasMap.get(topic);
+          if (replicaInfos == null)
+            throw new NoSuchElementException(
+                "This ClusterInfo have no information about topic \"" + topic + "\"");
+          else return replicaInfos;
+        }
+
+        @Override
+        public Collection<HasBeanObject> beans(int brokerId) {
+          return List.of();
+        }
+
+        @Override
+        public Map<Integer, Collection<HasBeanObject>> allBeans() {
+          return nodeInfo.stream()
+              .collect(Collectors.toUnmodifiableMap(NodeInfo::id, ignore -> List.of()));
+        }
+      };
+    }
   }
 
   private static class ConfigImpl implements Config {
diff --git a/app/src/main/java/org/astraea/app/cost/BrokerCost.java b/app/src/main/java/org/astraea/app/cost/BrokerCost.java
index 6b712c2f..78574636 100644
--- a/app/src/main/java/org/astraea/app/cost/BrokerCost.java
+++ b/app/src/main/java/org/astraea/app/cost/BrokerCost.java
@@ -20,6 +20,22 @@ import java.util.Map;
 
 /** Return type of cost function, `HasBrokerCost`. It returns the score of brokers. */
 public interface BrokerCost {
+  /**
+   * A new BrokerCost has been normalized.
+   *
+   * @param normalizer
+   * @return A normalized BrokerCost.
+   */
+  default BrokerCost normalize(Normalizer normalizer) {
+    var map = this.value();
+    var keys = map.keySet().iterator();
+    var normalization = normalizer.normalize(map.values()).iterator();
+    while (normalization.hasNext()) {
+      map.replace(keys.next(), normalization.next());
+    }
+    return () -> map;
+  }
+
   /** @return broker-id and its "cost" */
   Map<Integer, Double> value();
 }
diff --git a/app/src/main/java/org/astraea/app/cost/ClusterInfo.java b/app/src/main/java/org/astraea/app/cost/ClusterInfo.java
index 261ae2c3..d2781be1 100644
--- a/app/src/main/java/org/astraea/app/cost/ClusterInfo.java
+++ b/app/src/main/java/org/astraea/app/cost/ClusterInfo.java
@@ -48,28 +48,43 @@ public interface ClusterInfo {
 
       @Override
       public List<ReplicaInfo> availableReplicaLeaders(String topic) {
-        return cluster.availablePartitionsForTopic(topic).stream()
-            .map(ReplicaInfo::of)
-            .map(
-                replicas ->
-                    replicas.stream().filter(ReplicaInfo::isLeader).findFirst().orElseThrow())
-            .collect(Collectors.toUnmodifiableList());
+        var info =
+            cluster.availablePartitionsForTopic(topic).stream()
+                .map(ReplicaInfo::of)
+                .map(
+                    replicas ->
+                        replicas.stream().filter(ReplicaInfo::isLeader).findFirst().orElseThrow())
+                .collect(Collectors.toUnmodifiableList());
+        if (info.isEmpty())
+          throw new NoSuchElementException(
+              "This ClusterInfo have no information about topic \"" + topic + "\"");
+        return info;
       }
 
       @Override
       public List<ReplicaInfo> availableReplicas(String topic) {
-        return cluster.availablePartitionsForTopic(topic).stream()
-            .map(ReplicaInfo::of)
-            .flatMap(Collection::stream)
-            .collect(Collectors.toUnmodifiableList());
+        var info =
+            cluster.availablePartitionsForTopic(topic).stream()
+                .map(ReplicaInfo::of)
+                .flatMap(Collection::stream)
+                .collect(Collectors.toUnmodifiableList());
+        if (info.isEmpty())
+          throw new NoSuchElementException(
+              "This ClusterInfo have no information about topic \"" + topic + "\"");
+        return info;
       }
 
       @Override
       public List<ReplicaInfo> replicas(String topic) {
-        return cluster.partitionsForTopic(topic).stream()
-            .map(ReplicaInfo::of)
-            .flatMap(Collection::stream)
-            .collect(Collectors.toUnmodifiableList());
+        var info =
+            cluster.partitionsForTopic(topic).stream()
+                .map(ReplicaInfo::of)
+                .flatMap(Collection::stream)
+                .collect(Collectors.toUnmodifiableList());
+        if (info.isEmpty())
+          throw new NoSuchElementException(
+              "This ClusterInfo have no information about topic \"" + topic + "\"");
+        return info;
       }
 
       @Override
@@ -168,21 +183,24 @@ public interface ClusterInfo {
     return nodes().stream()
         .filter(n -> n.id() == id)
         .findAny()
-        .orElseThrow(() -> new IllegalArgumentException(id + " is nonexistent"));
+        .orElseThrow(() -> new NoSuchElementException(id + " is nonexistent"));
   }
 
   /** @return The known set of nodes */
   List<NodeInfo> nodes();
 
-  /** @return return the data directories on specific broker */
+  /**
+   * @return return the data directories on specific broker. It throws NoSuchElementException if
+   *     specify node id is not associated to any node.
+   */
   Set<String> dataDirectories(int brokerId);
-  // TODO: provide a ClusterInfo implementation with this info
 
   /**
    * Get the list of replica leader information of each available partition for the given topic
    *
    * @param topic The Topic name
-   * @return A list of {@link ReplicaInfo}
+   * @return A list of {@link ReplicaInfo}. It throws NoSuchElementException if the replica info of
+   *     the given topic is unknown to this ClusterInfo
    */
   List<ReplicaInfo> availableReplicaLeaders(String topic);
 
@@ -191,7 +209,8 @@ public interface ClusterInfo {
    * topic
    *
    * @param topic The topic name
-   * @return A list of {@link ReplicaInfo}
+   * @return A list of {@link ReplicaInfo}. It throws NoSuchElementException if the replica info of
+   *     the given topic is unknown to this ClusterInfo
    */
   List<ReplicaInfo> availableReplicas(String topic);
 
@@ -206,7 +225,8 @@ public interface ClusterInfo {
    * Get the list of replica information of each partition/replica pair for the given topic
    *
    * @param topic The topic name
-   * @return A list of {@link ReplicaInfo}
+   * @return A list of {@link ReplicaInfo}. It throws NoSuchElementException if the replica info of
+   *     the given topic is unknown to this ClusterInfo
    */
   List<ReplicaInfo> replicas(String topic);
 
diff --git a/app/src/main/java/org/astraea/app/cost/HasBrokerCost.java b/app/src/main/java/org/astraea/app/cost/HasBrokerCost.java
index 9207d5e6..0eab831c 100644
--- a/app/src/main/java/org/astraea/app/cost/HasBrokerCost.java
+++ b/app/src/main/java/org/astraea/app/cost/HasBrokerCost.java
@@ -18,10 +18,10 @@ package org.astraea.app.cost;
 
 public interface HasBrokerCost extends CostFunction {
   /**
-   * score all nodes according to passed beans and cluster information.
+   * score all nodes for a particular metrics according to passed beans and cluster information.
    *
    * @param clusterInfo cluster information
-   * @return the score of each broker. The score ranges in [0 - 1].
+   * @return the score of each broker.
    */
   BrokerCost brokerCost(ClusterInfo clusterInfo);
 }
diff --git a/app/src/main/java/org/astraea/app/cost/NodeInfo.java b/app/src/main/java/org/astraea/app/cost/NodeInfo.java
index 9cfb1f53..efe9204a 100644
--- a/app/src/main/java/org/astraea/app/cost/NodeInfo.java
+++ b/app/src/main/java/org/astraea/app/cost/NodeInfo.java
@@ -24,7 +24,19 @@ public interface NodeInfo extends Comparable<NodeInfo> {
     return of(node.id(), node.host(), node.port());
   }
 
+  /**
+   * Represent a location unknown Kafka node. The node used to be in the cluster but probably become
+   * offline and unreachable for some reason.
+   */
+  static NodeInfo ofOfflineNode(int id) {
+    return of(id, "", -1, true);
+  }
+
   static NodeInfo of(int id, String host, int port) {
+    return of(id, host, port, false);
+  }
+
+  static NodeInfo of(int id, String host, int port, boolean isOffline) {
     return new NodeInfo() {
       @Override
       public String host() {
@@ -36,6 +48,11 @@ public interface NodeInfo extends Comparable<NodeInfo> {
         return id;
       }
 
+      @Override
+      public boolean isOffline() {
+        return isOffline;
+      }
+
       @Override
       public int port() {
         return port;
@@ -71,4 +88,12 @@ public interface NodeInfo extends Comparable<NodeInfo> {
 
   /** @return id of broker node. it must be unique. */
   int id();
+
+  /** @return true if the node is considered offline */
+  boolean isOffline();
+
+  /** @return true if the node is considered online */
+  default boolean isOnline() {
+    return !isOffline();
+  }
 }
diff --git a/app/src/main/java/org/astraea/app/cost/Normalizer.java b/app/src/main/java/org/astraea/app/cost/Normalizer.java
index 2ce2e99c..ff669406 100644
--- a/app/src/main/java/org/astraea/app/cost/Normalizer.java
+++ b/app/src/main/java/org/astraea/app/cost/Normalizer.java
@@ -27,7 +27,11 @@ public interface Normalizer {
 
   /** @return all normalizers */
   static List<Normalizer> all() {
-    return List.of(Normalizer.proportion(), Normalizer.minMax(true), Normalizer.minMax(false));
+    return List.of(
+        Normalizer.proportion(),
+        Normalizer.minMax(true),
+        Normalizer.minMax(false),
+        Normalizer.TScore());
   }
 
   /**
@@ -70,6 +74,35 @@ public interface Normalizer {
     };
   }
 
+  /**
+   * implement the TScore normalization
+   *
+   * <p>(value -avg/standardDeviation)*10+50
+   *
+   * @return TScore normalizer
+   */
+  static Normalizer TScore() {
+    return values -> {
+      var avg = values.stream().mapToDouble(i -> i).sum() / values.size();
+      var standardDeviation =
+          Math.sqrt(values.stream().mapToDouble(i -> (i - avg) * (i - avg)).sum() / values.size());
+
+      return values.stream()
+          .map(i -> (i - avg) / standardDeviation)
+          .map(
+              i -> {
+                var score = (i * 10 + 50) / 100.0;
+                if (score > 1) {
+                  score = 1.0;
+                } else if (score < 0) {
+                  score = 0.0;
+                }
+                return Math.round(score * 100) / 100.0;
+              })
+          .collect(Collectors.toUnmodifiableList());
+    };
+  }
+
   /**
    * rescales the values into a range of [0,1]
    *
diff --git a/app/src/main/java/org/astraea/app/cost/ReplicaInfo.java b/app/src/main/java/org/astraea/app/cost/ReplicaInfo.java
index 06b26a04..73e0b894 100644
--- a/app/src/main/java/org/astraea/app/cost/ReplicaInfo.java
+++ b/app/src/main/java/org/astraea/app/cost/ReplicaInfo.java
@@ -156,6 +156,11 @@ public interface ReplicaInfo {
   /** @return true if this replica is offline */
   boolean isOfflineReplica();
 
+  /** @return true if this replica is online */
+  default boolean isOnlineReplica() {
+    return !isOfflineReplica();
+  }
+
   /**
    * The path to the data folder which hosts this replica. Since this information is not very openly
    * available. An application might find it hard to retrieve this information(for example, the
diff --git a/app/src/main/java/org/astraea/app/cost/broker/BrokerInputCost.java b/app/src/main/java/org/astraea/app/cost/broker/BrokerInputCost.java
index 53c9d774..9140a46d 100644
--- a/app/src/main/java/org/astraea/app/cost/broker/BrokerInputCost.java
+++ b/app/src/main/java/org/astraea/app/cost/broker/BrokerInputCost.java
@@ -73,8 +73,7 @@ public class BrokerInputCost implements HasBrokerCost {
                       broker.accumulateCount = inBean.count();
                       return count;
                     }));
-
-    CostUtils.TScore(costMetrics).forEach((broker, v) -> brokersMetric.get(broker).updateLoad(v));
+    costMetrics.forEach((broker, v) -> brokersMetric.get(broker).updateLoad(v));
 
     return this::computeLoad;
   }
diff --git a/app/src/main/java/org/astraea/app/cost/broker/BrokerOutputCost.java b/app/src/main/java/org/astraea/app/cost/broker/BrokerOutputCost.java
index c8a268f7..73e5c82f 100644
--- a/app/src/main/java/org/astraea/app/cost/broker/BrokerOutputCost.java
+++ b/app/src/main/java/org/astraea/app/cost/broker/BrokerOutputCost.java
@@ -73,8 +73,7 @@ public class BrokerOutputCost implements HasBrokerCost {
                       broker.accumulateCount = inBean.count();
                       return count;
                     }));
-
-    CostUtils.TScore(costMetrics).forEach((broker, v) -> brokersMetric.get(broker).updateLoad(v));
+    costMetrics.forEach((broker, v) -> brokersMetric.get(broker).updateLoad(v));
 
     return this::computeLoad;
   }
diff --git a/app/src/main/java/org/astraea/app/cost/broker/MemoryCost.java b/app/src/main/java/org/astraea/app/cost/broker/MemoryCost.java
index b7488a0e..557659ce 100644
--- a/app/src/main/java/org/astraea/app/cost/broker/MemoryCost.java
+++ b/app/src/main/java/org/astraea/app/cost/broker/MemoryCost.java
@@ -68,8 +68,7 @@ public class MemoryCost extends Periodic<Map<Integer, Double>> implements HasBro
                       return (jvmBean.heapMemoryUsage().getUsed() + 0.0)
                           / (jvmBean.heapMemoryUsage().getMax() + 1);
                     }));
-
-    CostUtils.TScore(costMetrics).forEach((broker, v) -> brokersMetric.get(broker).updateLoad(v));
+    costMetrics.forEach((broker, v) -> brokersMetric.get(broker).updateLoad(v));
 
     return this::computeLoad;
   }
diff --git a/app/src/main/java/org/astraea/app/cost/broker/NeutralIntegratedCost.java b/app/src/main/java/org/astraea/app/cost/broker/NeutralIntegratedCost.java
index a447d86c..ba972f49 100644
--- a/app/src/main/java/org/astraea/app/cost/broker/NeutralIntegratedCost.java
+++ b/app/src/main/java/org/astraea/app/cost/broker/NeutralIntegratedCost.java
@@ -60,21 +60,25 @@ public class NeutralIntegratedCost implements HasBrokerCost {
           if (hasBrokerCost instanceof BrokerInputCost) {
             hasBrokerCost
                 .brokerCost(clusterInfo)
+                .normalize(Normalizer.TScore())
                 .value()
                 .forEach((brokerID, value) -> brokersMetric.get(brokerID).inputScore = value);
           } else if (hasBrokerCost instanceof BrokerOutputCost) {
             hasBrokerCost
                 .brokerCost(clusterInfo)
+                .normalize(Normalizer.TScore())
                 .value()
                 .forEach((brokerID, value) -> brokersMetric.get(brokerID).outputScore = value);
           } else if (hasBrokerCost instanceof CpuCost) {
             hasBrokerCost
                 .brokerCost(clusterInfo)
+                .normalize(Normalizer.TScore())
                 .value()
                 .forEach((brokerID, value) -> brokersMetric.get(brokerID).cpuScore = value);
           } else if (hasBrokerCost instanceof MemoryCost) {
             hasBrokerCost
                 .brokerCost(clusterInfo)
+                .normalize(Normalizer.TScore())
                 .value()
                 .forEach((brokerID, value) -> brokersMetric.get(brokerID).memoryScore = value);
           }
diff --git a/docker/start_kafka_tool.sh b/docker/start_app.sh
similarity index 95%
rename from docker/start_kafka_tool.sh
rename to docker/start_app.sh
index 60636d6f..add5b565 100755
--- a/docker/start_kafka_tool.sh
+++ b/docker/start_app.sh
@@ -19,9 +19,9 @@ source $DOCKER_FOLDER/docker_build_common.sh
 
 # ===============================[global variables]===============================
 declare -r VERSION=${REVISION:-${VERSION:-main}}
-declare -r REPO=${REPO:-ghcr.io/skiptests/astraea/kafka-tool}
+declare -r REPO=${REPO:-ghcr.io/skiptests/astraea/app}
 declare -r IMAGE_NAME="$REPO:$VERSION"
-declare -r DOCKERFILE=$DOCKER_FOLDER/kafka_tool.dockerfile
+declare -r DOCKERFILE=$DOCKER_FOLDER/app.dockerfile
 declare -r JMX_PORT=${JMX_PORT:-"$(getRandomPort)"}
 # for web service
 declare -r WEB_PORT=${WEB_PORT:-"$(getRandomPort)"}
@@ -31,9 +31,9 @@ declare -r HEAP_OPTS="${HEAP_OPTS:-"-Xmx2G -Xms2G"}"
 # ===================================[functions]===================================
 
 function showHelp() {
-  echo "Usage: [ENV] start_kafka_tool.sh"
+  echo "Usage: [ENV] start_app.sh"
   echo "ENV: "
-  echo "    REPO=astraea/kafka-tool    set the docker repo"
+  echo "    REPO=astraea/app    set the docker repo"
   echo "    BUILD=false                set true if you want to build image locally"
   echo "    RUN=false                  set false if you want to build/pull image only"
 }
diff --git a/docker/start_broker.sh b/docker/start_broker.sh
index 7d26a199..bbbde237 100755
--- a/docker/start_broker.sh
+++ b/docker/start_broker.sh
@@ -18,7 +18,7 @@ declare -r DOCKER_FOLDER=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" &>/dev/null
 source $DOCKER_FOLDER/docker_build_common.sh
 
 # ===============================[global variables]===============================
-declare -r VERSION=${REVISION:-${VERSION:-3.1.1}}
+declare -r VERSION=${REVISION:-${VERSION:-3.2.0}}
 declare -r DOCKERFILE=$DOCKER_FOLDER/broker.dockerfile
 declare -r CONFLUENT_BROKER=${CONFLUENT_BROKER:-false}
 declare -r CONFLUENT_VERSION=${CONFLUENT_VERSION:-7.0.1}
@@ -68,7 +68,7 @@ function showHelp() {
   echo "    REPO=astraea/broker                      set the docker repo"
   echo "    HEAP_OPTS=\"-Xmx2G -Xms2G\"                set broker JVM memory"
   echo "    REVISION=trunk                           set revision of kafka source code to build container"
-  echo "    VERSION=3.1.1                            set version of kafka distribution"
+  echo "    VERSION=3.2.0                            set version of kafka distribution"
   echo "    BUILD=false                              set true if you want to build image locally"
   echo "    RUN=false                                set false if you want to build/pull image only"
   echo "    DATA_FOLDERS=/tmp/folder1                set host folders used by broker"
diff --git a/docs/web_server/README.md b/docs/web_server/README.md
new file mode 100644
index 00000000..331fb21e
--- /dev/null
+++ b/docs/web_server/README.md
@@ -0,0 +1,20 @@
+Astraea Web Server 中文文件
+===
+
+Astraea 建立了一套 Web Server 服務，使用者可以透過簡易好上手的 [Web APIs](#Web-APIs) 來操作 Kafka 各項指令
+
+## 如何啟動 Web Server ?
+
+```shell
+./docker/start_kafka_tool.sh web --bootstrap.servers 192.168.50.5:9092 --port 12345
+```
+
+- `--bootstrap.servers`: broker 位址與連接埠，用以索取叢集資訊，可填寫多台 brokers，並以 `,` 做區隔
+- `--port`: web server 連接埠，此參數為選填值，不填寫的話會選一個連接埠隨機綁定
+
+## Web APIs
+- [/topics](./web_api_topics_chinese.md)
+- [/groups](./web_api_groups_chinese.md)
+- [/brokers](./web_api_brokers_chinese.md)
+- [/producers](./web_api_producers_chinese.md)
+- [/quotas](./web_api_quotas_chinese.md)
diff --git a/docs/web_server/web_api_brokers_chinese.md b/docs/web_server/web_api_brokers_chinese.md
new file mode 100644
index 00000000..6e9532ae
--- /dev/null
+++ b/docs/web_server/web_api_brokers_chinese.md
@@ -0,0 +1,72 @@
+/brokers
+===
+
+- [查詢所有 brokers](#查詢所有-brokers)
+- [查詢指定 broker](#查詢指定-broker)
+
+## 查詢所有 brokers
+```shell
+GET /brokers
+```
+
+cURL 範例
+```shell
+curl -X GET http://localhost:8001/brokers
+```
+
+JSON Response 範例
+- `id`: broker 在叢集中的識別碼
+- `topics`: 此 broker 下所有 topic
+  - `topic`: topic 名稱
+  - `partitionCount`: 此 topic 有多少個 partitions 在這個節點身上
+- `configs`: 此 broker 正在使用的參數。這個參數列表為下列項目的組合，啟動節點時所用的參數檔、系統內預設值以及使用者動態修改的參數
+```json
+{
+  "brokers": [
+    {
+      "id": 1001,
+      "topics": [
+        {
+          "topic": "test1",
+          "partitionCount": 1
+        }
+      ],
+      "configs": {
+        "log.cleaner.min.compaction.lag.ms": "0",
+        "metric.reporters": "",
+        ...
+      }
+    }
+  ]
+}
+```
+
+## 查詢指定 broker
+```shell
+GET /brokers/{brokerId}
+```
+
+cURL 範例
+
+查詢 id 為 1001 的 broker 資訊
+```shell
+curl -X GET http://localhost:8001/brokers/1001
+```
+
+JSON Response 範例
+```json
+{
+  "id": 1001,
+  "topics": [
+    {
+      "topic": "test1",
+      "partitionCount": 1
+    }
+  ],
+  "configs": {
+    "log.cleaner.min.compaction.lag.ms": "0",
+    "metric.reporters": "",
+    ...
+  }
+}
+```
\ No newline at end of file
diff --git a/docs/web_server/web_api_groups_chinese.md b/docs/web_server/web_api_groups_chinese.md
new file mode 100644
index 00000000..1c2b569b
--- /dev/null
+++ b/docs/web_server/web_api_groups_chinese.md
@@ -0,0 +1,89 @@
+/groups
+===
+
+- [查詢所有 groups](#查詢所有-groups)
+- [查詢指定 group](#查詢指定-group)
+
+## 查詢所有 groups
+```shell
+GET /groups
+```
+
+cURL 範例
+```shell
+curl -X GET http://localhost:8001/groups
+```
+
+JSON Response 範例
+- `memberId`: consumer 加入群組時由 (broker) coordinator所給予的唯一值
+- `clientId`: consumer 送出請求時使用的名稱，使用者可以自定義，預設值為隨機字串
+- `host`: consumer 運行的位址
+- `offsetProgress`: consumer 目前消化資料的進度
+  - `topicName`: consumer 提取資料的 topic 名稱
+  - `partitionId`: consumer 提取資料的 partition id
+  - `earliest`: partition 現存的資料最早的紀錄位置
+  - `current`: consumer 於 partition 目前處理的數據位置
+  - `latest`: partition 最新紀錄位置
+```json
+{
+  "groups": [
+    {
+      "groupId": "groupId-1653924298342",
+      "members": []
+    },
+    {
+      "groupId": "group-1",
+      "members": [
+        {
+          "memberId": "consumer-group-1-1-22b93f1b-0eae-48ef-95f6-f562bf6769c0",
+          "clientId": "consumer-group-1-1",
+          "host": "/172.17.0.1",
+          "offsetProgress": [
+            {
+              "topicName": "test1",
+              "partitionId": 0,
+              "earliest": 0,
+              "current": 2,
+              "latest": 8
+            }
+          ]
+        }
+      ]
+    }
+  ]
+}
+```
+
+## 查詢指定 group
+```shell
+GET /groups/{groupId}
+```
+
+cURL 範例
+查詢名為 group-1 的 group 資訊
+```shell
+curl -X GET http://localhost:8001/groups/group-1
+```
+
+JSON Response 範例
+```json
+{
+  "groupId": "group-1",
+  "members": [
+    {
+      "memberId": "consumer-group-1-1-d0370b05-99e8-4d15-88ef-6da2d57e84fb",
+      "clientId": "consumer-group-1-1",
+      "host": "/172.17.0.1",
+      "offsetProgress": [
+        {
+          "topicName": "test1",
+          "partitionId": 0,
+          "earliest": 0,
+          "current": 0,
+          "latest": 0
+        }
+      ]
+    }
+  ]
+}
+```
diff --git a/docs/web_server/web_api_producers_chinese.md b/docs/web_server/web_api_producers_chinese.md
new file mode 100644
index 00000000..31f9e39d
--- /dev/null
+++ b/docs/web_server/web_api_producers_chinese.md
@@ -0,0 +1,68 @@
+/producers
+===
+
+- [查詢 producers](#查詢-producers)
+
+## 查詢 producers
+
+```shell
+GET /producers
+```
+
+參數
+
+| 名稱        | 說明                |
+|-----------|-------------------|
+| topic     | (選填) topic 名稱     |
+| partition | (選填) partition id |
+
+- 僅能指定查詢 topic 或者 topic + partition，其他狀況均會回傳全部結果
+
+cURL 範例
+
+查詢所有 producers
+
+```shell
+curl -X GET http://localhost:8001/producers
+```
+
+查詢所有 producers 其 topic = test1 且 partition = 0
+
+```shell
+curl -X GET http://localhost:8001/producers?topic=test1&partition=0
+```
+
+JSON Response 範例
+
+- `topic`: topic 名稱
+- `partition`: partition id，從 0 開始遞增
+- `states`: 當前活躍（未達到connections.max.idle.ms所設置的時間）的 producers 狀態
+    - `producerId`: producer id，從 0 開始遞增
+    - `producerEpoch`: producer epoch，從 0 開始遞增，用以保證 producer transaction 的唯一性
+    - `lastSequence`: producer 提交於此 partition 最新 sequence number，從 0 開始遞增，搭配 producer id 來處理冪等性，保證單個 partition 不會重複寫入資料
+    - `lastTimestamp`: producer 提交到此 partition 最新一筆資料的時間戳記
+
+```json
+{
+  "partitions": [
+    {
+      "topic": "test1",
+      "partition": 0,
+      "states": [
+        {
+          "producerId": 0,
+          "producerEpoch": 0,
+          "lastSequence": 0,
+          "lastTimestamp": 1653578001967
+        },
+        {
+          "producerId": 1,
+          "producerEpoch": 0,
+          "lastSequence": 0,
+          "lastTimestamp": 1653578023368
+        }
+      ]
+    }
+  ]
+}
+```
\ No newline at end of file
diff --git a/docs/web_server/web_api_quotas_chinese.md b/docs/web_server/web_api_quotas_chinese.md
new file mode 100644
index 00000000..54659269
--- /dev/null
+++ b/docs/web_server/web_api_quotas_chinese.md
@@ -0,0 +1,170 @@
+/quotas
+===
+
+- [建立/變更 quotas](#建立或變更-quotas)
+  - [建立/變更 ip quotas](#建立或變更-ip-quotas)
+  - [建立/變更 client quotas](#建立或變更-client-quotas)
+- [查詢 quotas](#查詢-quotas)
+
+## 建立或變更 quotas
+```shell
+POST /quotas
+```
+以 ip 或是 client 為目標對象來增設 quotas
+
+### 建立或變更 ip quotas
+
+參數
+
+| 名稱                       | 說明               | 預設  |
+|--------------------------|------------------|-----|
+| ip                       | (必填) ip 地址       | 無   |
+| connection_creation_rate | (選填) 每秒建立的最大連線數量 | 無上限 |
+- 除了 ip 以外的參數都不填寫的話，quotas 不會建立
+
+cURL 範例
+
+將 ip 為 192.168.1.102 的 connection_creation_rate 設為 100
+```shell
+curl -X POST http://localhost:8001/quotas \
+    -H "Content-Type: application/json" \
+    -d '{
+    "ip": "192.168.1.102", 
+    "connection_creation_rate": 100 
+    }' 
+```
+
+JSON Response 範例
+```json
+{
+  "quotas": [
+    {
+      "target": {
+        "name": "ip",
+        "value": "192.168.1.102"
+      },
+      "limit": {
+        "name": "connection_creation_rate",
+        "value": 100
+      }
+    }
+  ]
+}
+```
+
+### 建立或變更 client quotas
+參數
+
+| 名稱                 | 說明                           | 預設  |
+|--------------------|------------------------------|-----|
+| client-id          | (必填) client id               | 無   |
+| producer_byte_rate | (選填) producer 每秒發佈的最大 byte 數 | 無上限 |
+| consumer_byte_rate | (選填) consumer 每秒提取的最大 byte 數 | 無上限 |
+- 除了 client-id 以外的參數都不填寫的話，quotas 不會建立
+
+cURL 範例
+
+將 client-id 為 my-id 的 producer_byte_rate 設為 10，且 consumer_byte_rate 設為 100
+```shell
+curl -X POST http://localhost:8001/quotas \
+    -H "Content-Type: application/json" \
+    -d '{
+    "client-id": "my-id", 
+    "consumer_byte_rate": 100
+    "producer_byte_rate": 10
+    }' 
+```
+
+JSON Response 範例
+```json
+{
+  "quotas": [
+    {
+      "target": {
+        "name": "client-id",
+        "value": "my-id"
+      },
+      "limit": {
+        "name": "consumer_byte_rate",
+        "value": 100
+      }
+    },
+    {
+      "target": {
+        "name": "client-id",
+        "value": "my-id"
+      },
+      "limit": {
+        "name": "producer_byte_rate",
+        "value": 10
+      }
+    }
+  ]
+}
+```
+
+## 查詢 quotas
+```shell
+GET /quotas
+```
+參數
+
+| 名稱        | 說明             |
+|-----------|----------------|
+| ip        | (選填) ip 位址     |
+| client-id | (選填) client id |
+
+cURL 範例
+
+查詢所有 quotas
+```shell
+curl -X GET http://localhost:8001/quotas
+```
+
+查詢 ip 為 192.168.1.102 的 quotas
+```shell
+curl -X GET http://localhost:8001/quotas?ip=192.168.1.102
+```
+
+查詢 client-id 為 my-id 的 quotas
+```shell
+curl -X GET http://localhost:8001/quotas?client-id=my-id
+```
+
+JSON Response 範例
+```json
+{
+  "quotas": [
+    {
+      "target": {
+        "name": "client-id",
+        "value": "my-id"
+      },
+      "limit": {
+        "name": "consumer_byte_rate",
+        "value": 200
+      }
+    },
+    {
+      "target": {
+        "name": "client-id",
+        "value": "my-id"
+      },
+      "limit": {
+        "name": "producer_byte_rate",
+        "value": 30
+      }
+    },
+    {
+      "target": {
+        "name": "ip",
+        "value": "192.168.1.102"
+      },
+      "limit": {
+        "name": "connection_creation_rate",
+        "value": 200
+      }
+    }
+  ]
+}
+```
\ No newline at end of file
diff --git a/docs/web_server/web_api_topics_chinese.md b/docs/web_server/web_api_topics_chinese.md
new file mode 100644
index 00000000..efc9323a
--- /dev/null
+++ b/docs/web_server/web_api_topics_chinese.md
@@ -0,0 +1,244 @@
+/topics
+===
+
+- [建立 topic](#建立-topic)
+- [查詢所有 topics](#查詢所有-topics)
+- [查詢指定 topic](#查詢指定-topic)
+
+## 建立 topic
+```shell
+POST /topics
+```
+
+參數
+
+| 名稱         | 說明                  | 預設值 |
+|------------|---------------------|-----|
+| name       | (必填) topic 名稱       | 無   |
+| partitions | (選填) partition 數量   | 1   |
+| replicas   | (選填) replication 數量 | 1   |
+- replicas 數量須 <= brokers 數量
+
+cURL 範例
+
+建立名為 test1 且 partitions 為 1，replicas 為 1 的 topic
+```shell
+curl -X POST http://localhost:8001/topics \
+    -H "Content-Type: application/json" \
+    -d '{
+    "name": "test1",
+    "partitions": 1,
+    "replicas": 1
+    }'
+```
+
+所有在 JSON Response `configs` 裡頭的參數也可以透過此 api 來設定。範例如下
+```shell
+curl -X POST http://localhost:8001/topics \
+    -H "Content-Type: application/json" \
+    -d '{
+    "name": "test1",
+    "max.message.bytes": 1000
+    }'
+```
+
+JSON Response 範例
+- `name`: topic 名稱
+- `partitions`: 該 topic 下所有 partitions
+  - `id`: partition id，從 0 開始遞增
+  - `earliest`: partition 現存的資料最早的紀錄位置
+  - `latest`: partition 最新紀錄位置
+  - `replicas`: partition 副本資料
+    - `broker`: 儲存此副本的 broker 在叢集中的識別碼
+    - `lag`: 副本同步狀況，計算公式為副本 LEO (Log End offset) - HW (High Watermark)，用以判斷副本最新資料與所有 ISR 資料水位比較
+    - `size`: 副本資料大小，單位為 bytes
+    - `leader`: 此副本所在的節點是否為 leader
+    - `inSync`: 此副本所在的節點是否位於 ISR (In-Sync Replicas) 列表裡頭。每個 partition 都會有一個 ISR，由能夠和 leader 保持同步的 follower 與 leader 組成的集合 
+    - `isFuture`: 此副本是否正準備由另一份新的副本取代，可能的觸發時機點為 reassign partitions，副本準備要搬移到不同節點，或者副本準備移到同節點的不同路徑下
+    - `path`: 副本於節點中的路徑
+- `configs`: 此 topic 正在使用的參數，這些參數都可以在建立 topic 時設定
+
+```json
+{
+  "name": "test1",
+  "partitions": [
+    {
+      "id": 0,
+      "earliest": 0,
+      "latest": 0,
+      "replicas": [
+        {
+          "broker": 1001,
+          "lag": 0,
+          "size": 0,
+          "leader": true,
+          "inSync": true,
+          "isFuture": false,
+          "path": "/tmp/log-folder-0"
+        }
+      ]
+    }
+  ],
+  "configs": {
+    "compression.type": "producer",
+    "leader.replication.throttled.replicas": "",
+    "message.downconversion.enable": "true",
+    "min.insync.replicas": "1",
+    "segment.jitter.ms": "0",
+    "cleanup.policy": "delete",
+    "flush.ms": "9223372036854775807",
+    "follower.replication.throttled.replicas": "",
+    "segment.bytes": "1073741824",
+    "retention.ms": "604800000",
+    "flush.messages": "9223372036854775807",
+    "message.format.version": "3.0-IV1",
+    "max.compaction.lag.ms": "9223372036854775807",
+    "file.delete.delay.ms": "60000",
+    "max.message.bytes": "1048588",
+    "min.compaction.lag.ms": "0",
+    "message.timestamp.type": "CreateTime",
+    "preallocate": "false",
+    "min.cleanable.dirty.ratio": "0.5",
+    "index.interval.bytes": "4096",
+    "unclean.leader.election.enable": "false",
+    "retention.bytes": "-1",
+    "delete.retention.ms": "86400000",
+    "segment.ms": "604800000",
+    "message.timestamp.difference.max.ms": "9223372036854775807",
+    "segment.index.bytes": "10485760"
+  }
+}
+```
+
+## 查詢所有 topics
+```shell
+GET /topics
+```
+
+cURL 範例
+```shell
+curl -X GET http://localhost:8001/topics
+```
+
+JSON Response 範例
+```json
+{
+  "topics": [
+    {
+      "name": "test1",
+      "partitions": [
+        {
+          "id": 0,
+          "earliest": 0,
+          "latest": 0,
+          "replicas": [
+            {
+              "broker": 1001,
+              "lag": 0,
+              "size": 0,
+              "leader": true,
+              "inSync": true,
+              "isFuture": false,
+              "path": "/tmp/log-folder-0"
+            }
+          ]
+        }
+      ],
+      "configs": {
+        "compression.type": "producer",
+        "leader.replication.throttled.replicas": "",
+        "message.downconversion.enable": "true",
+        "min.insync.replicas": "1",
+        "segment.jitter.ms": "0",
+        "cleanup.policy": "delete",
+        "flush.ms": "9223372036854775807",
+        "follower.replication.throttled.replicas": "",
+        "segment.bytes": "1073741824",
+        "retention.ms": "604800000",
+        "flush.messages": "9223372036854775807",
+        "message.format.version": "3.0-IV1",
+        "max.compaction.lag.ms": "9223372036854775807",
+        "file.delete.delay.ms": "60000",
+        "max.message.bytes": "1048588",
+        "min.compaction.lag.ms": "0",
+        "message.timestamp.type": "CreateTime",
+        "preallocate": "false",
+        "min.cleanable.dirty.ratio": "0.5",
+        "index.interval.bytes": "4096",
+        "unclean.leader.election.enable": "false",
+        "retention.bytes": "-1",
+        "delete.retention.ms": "86400000",
+        "segment.ms": "604800000",
+        "message.timestamp.difference.max.ms": "9223372036854775807",
+        "segment.index.bytes": "10485760"
+      }
+    }
+  ]
+}
+```
+
+## 查詢指定 topic
+
+```shell
+GET /topics/{topicName}
+```
+
+cURL 範例
+
+查詢名為 test1 的 topic 資訊
+```shell
+curl -X GET http://localhost:8001/topics/test1
+```
+
+JSON Response 範例
+ ```json
+{
+  "name": "test1",
+  "partitions": [
+    {
+      "id": 0,
+      "earliest": 0,
+      "latest": 0,
+      "replicas": [
+        {
+          "broker": 1001,
+          "lag": 0,
+          "size": 0,
+          "leader": true,
+          "inSync": true,
+          "isFuture": false,
+          "path": "/tmp/log-folder-0"
+        }
+      ]
+    }
+  ],
+  "configs": {
+    "compression.type": "producer",
+    "leader.replication.throttled.replicas": "",
+    "message.downconversion.enable": "true",
+    "min.insync.replicas": "1",
+    "segment.jitter.ms": "0",
+    "cleanup.policy": "delete",
+    "flush.ms": "9223372036854775807",
+    "follower.replication.throttled.replicas": "",
+    "segment.bytes": "1073741824",
+    "retention.ms": "604800000",
+    "flush.messages": "9223372036854775807",
+    "message.format.version": "3.0-IV1",
+    "max.compaction.lag.ms": "9223372036854775807",
+    "file.delete.delay.ms": "60000",
+    "max.message.bytes": "1048588",
+    "min.compaction.lag.ms": "0",
+    "message.timestamp.type": "CreateTime",
+    "preallocate": "false",
+    "min.cleanable.dirty.ratio": "0.5",
+    "index.interval.bytes": "4096",
+    "unclean.leader.election.enable": "false",
+    "retention.bytes": "-1",
+    "delete.retention.ms": "86400000",
+    "segment.ms": "604800000",
+    "message.timestamp.difference.max.ms": "9223372036854775807",
+    "segment.index.bytes": "10485760"
+  }
+}
+ ```


Test patch:
diff --git a/app/src/test/java/org/astraea/app/admin/AdminTest.java b/app/src/test/java/org/astraea/app/admin/AdminTest.java
index dfb09493..7542f6dd 100644
--- a/app/src/test/java/org/astraea/app/admin/AdminTest.java
+++ b/app/src/test/java/org/astraea/app/admin/AdminTest.java
@@ -25,17 +25,22 @@ import java.time.Duration;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
+import java.util.NoSuchElementException;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
+import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
 import org.apache.kafka.common.config.TopicConfig;
 import org.astraea.app.common.Utils;
 import org.astraea.app.consumer.Consumer;
 import org.astraea.app.consumer.Deserializer;
+import org.astraea.app.cost.NodeInfo;
+import org.astraea.app.cost.ReplicaInfo;
 import org.astraea.app.producer.Producer;
 import org.astraea.app.producer.Serializer;
 import org.astraea.app.service.RequireBrokerCluster;
@@ -536,6 +541,133 @@ public class AdminTest extends RequireBrokerCluster {
     restartCluster();
   }
 
+  @Test
+  void testNodes() {
+    try (var admin = Admin.of(bootstrapServers())) {
+      final Set<NodeInfo> nodes = admin.nodes();
+      Assertions.assertEquals(
+          brokerIds(), nodes.stream().map(NodeInfo::id).collect(Collectors.toUnmodifiableSet()));
+    }
+  }
+
+  @Test
+  void testClusterInfo() throws InterruptedException {
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      String topic0 = "testClusterInfoFromAdmin_" + Utils.randomString(8);
+      String topic1 = "testClusterInfoFromAdmin_" + Utils.randomString(8);
+      String topic2 = "testClusterInfoFromAdmin_" + Utils.randomString(8);
+      int partitionCount = 10;
+      short replicaCount = 2;
+
+      Stream.of(topic0, topic1, topic2)
+          .forEach(
+              topicName ->
+                  admin
+                      .creator()
+                      .topic(topicName)
+                      .numberOfPartitions(partitionCount)
+                      .numberOfReplicas(replicaCount)
+                      .create());
+      TimeUnit.SECONDS.sleep(2);
+
+      final var clusterInfo = admin.clusterInfo(Set.of(topic0, topic1, topic2));
+
+      // ClusterInfo#nodes
+      Assertions.assertEquals(
+          brokerIds(),
+          clusterInfo.nodes().stream().map(NodeInfo::id).collect(Collectors.toUnmodifiableSet()));
+      // ClusterInfo#topics
+      Assertions.assertEquals(Set.of(topic0, topic1, topic2), clusterInfo.topics());
+      // ClusterInfo#replicas
+      Assertions.assertEquals(partitionCount * replicaCount, clusterInfo.replicas(topic0).size());
+      Assertions.assertEquals(partitionCount * replicaCount, clusterInfo.replicas(topic1).size());
+      Assertions.assertEquals(partitionCount * replicaCount, clusterInfo.replicas(topic2).size());
+      // ClusterInfo#dataDirectories
+      brokerIds()
+          .forEach(
+              id -> Assertions.assertEquals(logFolders().get(id), clusterInfo.dataDirectories(id)));
+      // ClusterInfo#availableReplicas
+      Assertions.assertEquals(
+          partitionCount * replicaCount, clusterInfo.availableReplicas(topic0).size());
+      Assertions.assertEquals(
+          partitionCount * replicaCount, clusterInfo.availableReplicas(topic1).size());
+      Assertions.assertEquals(
+          partitionCount * replicaCount, clusterInfo.availableReplicas(topic2).size());
+      // ClusterInfo#availableReplicaLeaders
+      Assertions.assertEquals(partitionCount, clusterInfo.availableReplicaLeaders(topic0).size());
+      Assertions.assertEquals(partitionCount, clusterInfo.availableReplicaLeaders(topic1).size());
+      Assertions.assertEquals(partitionCount, clusterInfo.availableReplicaLeaders(topic2).size());
+      // No resource match found will raise exception
+      Assertions.assertThrows(
+          NoSuchElementException.class, () -> clusterInfo.replicas("Unknown Topic"));
+      Assertions.assertThrows(
+          NoSuchElementException.class, () -> clusterInfo.availableReplicas("Unknown Topic"));
+      Assertions.assertThrows(
+          NoSuchElementException.class, () -> clusterInfo.availableReplicaLeaders("Unknown Topic"));
+      Assertions.assertThrows(NoSuchElementException.class, () -> clusterInfo.dataDirectories(-1));
+      Assertions.assertThrows(NoSuchElementException.class, () -> clusterInfo.node(-1));
+      Assertions.assertThrows(
+          NoSuchElementException.class, () -> clusterInfo.node("unknown", 1024));
+    }
+  }
+
+  @Test
+  void testClusterInfoWithOfflineNode() throws InterruptedException {
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var topicName = "ClusterInfo_Offline_" + Utils.randomString();
+      var partitionCount = 30;
+      var replicaCount = (short) 3;
+      admin
+          .creator()
+          .topic(topicName)
+          .numberOfPartitions(partitionCount)
+          .numberOfReplicas(replicaCount)
+          .create();
+      TimeUnit.SECONDS.sleep(3);
+
+      // before node offline
+      var before = admin.clusterInfo(Set.of(topicName));
+      Assertions.assertEquals(
+          partitionCount * replicaCount,
+          before.replicas(topicName).stream().filter(x -> !x.isOfflineReplica()).count());
+      Assertions.assertEquals(
+          partitionCount * replicaCount, before.availableReplicas(topicName).size());
+      Assertions.assertEquals(partitionCount, before.availableReplicaLeaders(topicName).size());
+
+      // act
+      int brokerToClose = ThreadLocalRandom.current().nextInt(0, 3);
+      closeBroker(brokerToClose);
+      TimeUnit.SECONDS.sleep(1);
+
+      // after node offline
+      var after = admin.clusterInfo(Set.of(topicName));
+      Assertions.assertEquals(
+          partitionCount * (replicaCount - 1),
+          after.replicas(topicName).stream().filter(x -> !x.isOfflineReplica()).count());
+      Assertions.assertEquals(
+          partitionCount * (replicaCount - 1), after.availableReplicas(topicName).size());
+      Assertions.assertEquals(
+          partitionCount,
+          after.availableReplicaLeaders(topicName).size(),
+          "One of the rest replicas should take over the leadership");
+      Assertions.assertTrue(
+          after.availableReplicas(topicName).stream()
+              .allMatch(x -> x.nodeInfo().id() != brokerToClose));
+      Assertions.assertTrue(
+          after.availableReplicaLeaders(topicName).stream()
+              .allMatch(x -> x.nodeInfo().id() != brokerToClose));
+      Assertions.assertTrue(
+          after.replicas(topicName).stream()
+              .filter(ReplicaInfo::isOfflineReplica)
+              .allMatch(x -> x.nodeInfo().id() == brokerToClose));
+      Assertions.assertTrue(
+          after.replicas(topicName).stream()
+              .filter(x -> !x.isOfflineReplica())
+              .allMatch(x -> x.nodeInfo().id() != brokerToClose));
+    }
+    restartCluster();
+  }
+
   @ParameterizedTest
   @ValueSource(shorts = {1, 2, 3})
   void preferredLeaderElection(short replicaSize) throws InterruptedException {
diff --git a/app/src/test/java/org/astraea/app/cost/BrokerCostTest.java b/app/src/test/java/org/astraea/app/cost/BrokerCostTest.java
new file mode 100644
index 00000000..7cfe08f5
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/cost/BrokerCostTest.java
@@ -0,0 +1,45 @@
+/*
+ * 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;
+
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class BrokerCostTest {
+  @Test
+  void testNormalize() {
+    var testMap =
+        IntStream.range(0, 100).boxed().collect(Collectors.toMap(i -> i, i -> i * 2 + 0.0));
+    var oldKey = -1;
+    var oldValue = -1.0;
+    brokerCost(testMap)
+        .normalize(Normalizer.TScore())
+        .value()
+        .forEach(
+            (key, value) -> {
+              Assertions.assertTrue(key > oldKey);
+              Assertions.assertTrue(value > oldValue);
+            });
+  }
+
+  BrokerCost brokerCost(Map<Integer, Double> map) {
+    return () -> map;
+  }
+}
diff --git a/app/src/test/java/org/astraea/app/cost/ClusterInfoTest.java b/app/src/test/java/org/astraea/app/cost/ClusterInfoTest.java
index 14dbe4f8..102cb206 100644
--- a/app/src/test/java/org/astraea/app/cost/ClusterInfoTest.java
+++ b/app/src/test/java/org/astraea/app/cost/ClusterInfoTest.java
@@ -19,6 +19,7 @@ package org.astraea.app.cost;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
+import java.util.NoSuchElementException;
 import org.apache.kafka.common.Cluster;
 import org.astraea.app.metrics.HasBeanObject;
 import org.junit.jupiter.api.Assertions;
@@ -70,4 +71,19 @@ public class ClusterInfoTest {
     Assertions.assertEquals(0, clusterInfo.beans(19).size());
     Assertions.assertEquals(1, clusterInfo.beans(1).size());
   }
+
+  @Test
+  void testIllegalQuery() {
+    var clusterInfo = ClusterInfo.of(Cluster.empty());
+
+    Assertions.assertThrows(NoSuchElementException.class, () -> clusterInfo.replicas("unknown"));
+    Assertions.assertThrows(NoSuchElementException.class, () -> clusterInfo.node(0));
+    Assertions.assertThrows(NoSuchElementException.class, () -> clusterInfo.node("", -1));
+    Assertions.assertThrows(
+        NoSuchElementException.class, () -> clusterInfo.availableReplicas("unknown"));
+    Assertions.assertThrows(
+        NoSuchElementException.class, () -> clusterInfo.availableReplicaLeaders("unknown"));
+    Assertions.assertThrows(
+        UnsupportedOperationException.class, () -> clusterInfo.dataDirectories(0));
+  }
 }
diff --git a/app/src/test/java/org/astraea/app/cost/broker/BrokerInputCostTest.java b/app/src/test/java/org/astraea/app/cost/broker/BrokerInputCostTest.java
index df25c5d8..7304dca1 100644
--- a/app/src/test/java/org/astraea/app/cost/broker/BrokerInputCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/broker/BrokerInputCostTest.java
@@ -21,6 +21,7 @@ import java.util.List;
 import java.util.Map;
 import org.astraea.app.cost.ClusterInfo;
 import org.astraea.app.cost.FakeClusterInfo;
+import org.astraea.app.cost.Normalizer;
 import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.jmx.BeanObject;
 import org.astraea.app.metrics.kafka.BrokerTopicMetricsResult;
@@ -31,20 +32,33 @@ import org.mockito.Mockito;
 
 public class BrokerInputCostTest {
   @Test
-  void testCost() {
+  void testTScoreCost() {
+    ClusterInfo clusterInfo = exampleClusterInfo(50000L, 20000L, 5000L);
+
+    var brokerInputCost = new BrokerInputCost();
+    var scores = brokerInputCost.brokerCost(clusterInfo).normalize(Normalizer.TScore()).value();
+    Assertions.assertEquals(0.63, scores.get(1));
+    Assertions.assertEquals(0.47, scores.get(2));
+    Assertions.assertEquals(0.39, scores.get(3));
+
+    ClusterInfo clusterInfo2 = exampleClusterInfo(55555L, 25352L, 25000L);
+    scores = brokerInputCost.brokerCost(clusterInfo2).normalize(Normalizer.TScore()).value();
+    Assertions.assertEquals(0.64, scores.get(1));
+    Assertions.assertEquals(0.43, scores.get(2));
+    Assertions.assertEquals(0.43, scores.get(3));
+  }
+
+  @Test
+  void testNoNormalize() {
     ClusterInfo clusterInfo = exampleClusterInfo(10000L, 20000L, 5000L);
 
     var brokerInputCost = new BrokerInputCost();
     var scores = brokerInputCost.brokerCost(clusterInfo).value();
-    Assertions.assertEquals(0.47, scores.get(1));
-    Assertions.assertEquals(0.63, scores.get(2));
-    Assertions.assertEquals(0.39, scores.get(3));
+    Assertions.assertEquals(10000.0, scores.get(1));
+    Assertions.assertEquals(20000.0, scores.get(2));
+    Assertions.assertEquals(5000.0, scores.get(3));
 
     ClusterInfo clusterInfo2 = exampleClusterInfo(55555L, 25352L, 25000L);
-    scores = brokerInputCost.brokerCost(clusterInfo2).value();
-    Assertions.assertEquals(0.55, scores.get(1));
-    Assertions.assertEquals(0.51, scores.get(2));
-    Assertions.assertEquals(0.44, scores.get(3));
   }
 
   private ClusterInfo exampleClusterInfo(long in1, long in2, long in3) {
diff --git a/app/src/test/java/org/astraea/app/cost/broker/BrokerOutPutCostTest.java b/app/src/test/java/org/astraea/app/cost/broker/BrokerOutPutCostTest.java
index 5db1a14f..ef53ffaf 100644
--- a/app/src/test/java/org/astraea/app/cost/broker/BrokerOutPutCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/broker/BrokerOutPutCostTest.java
@@ -21,6 +21,7 @@ import java.util.List;
 import java.util.Map;
 import org.astraea.app.cost.ClusterInfo;
 import org.astraea.app.cost.FakeClusterInfo;
+import org.astraea.app.cost.Normalizer;
 import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.jmx.BeanObject;
 import org.astraea.app.metrics.kafka.BrokerTopicMetricsResult;
@@ -35,16 +36,16 @@ public class BrokerOutPutCostTest {
     ClusterInfo clusterInfo = exampleClusterInfo(10000L, 20000L, 5000L);
 
     var brokerOutputCost = new BrokerOutputCost();
-    var scores = brokerOutputCost.brokerCost(clusterInfo).value();
+    var scores = brokerOutputCost.brokerCost(clusterInfo).normalize(Normalizer.TScore()).value();
     Assertions.assertEquals(0.47, scores.get(1));
     Assertions.assertEquals(0.63, scores.get(2));
     Assertions.assertEquals(0.39, scores.get(3));
 
     ClusterInfo clusterInfo2 = exampleClusterInfo(55555L, 25352L, 25000L);
-    scores = brokerOutputCost.brokerCost(clusterInfo2).value();
-    Assertions.assertEquals(0.55, scores.get(1));
-    Assertions.assertEquals(0.51, scores.get(2));
-    Assertions.assertEquals(0.44, scores.get(3));
+    scores = brokerOutputCost.brokerCost(clusterInfo2).normalize(Normalizer.TScore()).value();
+    Assertions.assertEquals(0.64, scores.get(1));
+    Assertions.assertEquals(0.43, scores.get(2));
+    Assertions.assertEquals(0.43, scores.get(3));
   }
 
   private ClusterInfo exampleClusterInfo(long out1, long out2, long out3) {
diff --git a/app/src/test/java/org/astraea/app/cost/broker/CpuCostTest.java b/app/src/test/java/org/astraea/app/cost/broker/CpuCostTest.java
index 1496a584..c8edd31e 100644
--- a/app/src/test/java/org/astraea/app/cost/broker/CpuCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/broker/CpuCostTest.java
@@ -23,6 +23,7 @@ import java.util.Set;
 import org.astraea.app.cost.ClusterInfo;
 import org.astraea.app.cost.FakeClusterInfo;
 import org.astraea.app.cost.NodeInfo;
+import org.astraea.app.cost.Normalizer;
 import org.astraea.app.cost.ReplicaInfo;
 import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.java.OperatingSystemInfo;
@@ -62,7 +63,7 @@ public class CpuCostTest {
         };
 
     var cpuCost = new CpuCost();
-    var scores = cpuCost.brokerCost(clusterInfo).value();
+    var scores = cpuCost.brokerCost(clusterInfo).normalize(Normalizer.TScore()).value();
     Assertions.assertEquals(0.39, scores.get(1));
     Assertions.assertEquals(0.63, scores.get(2));
     Assertions.assertEquals(0.48, scores.get(3));
@@ -95,10 +96,10 @@ public class CpuCostTest {
                 ReplicaInfo.of("t", 0, NodeInfo.of(3, "host3", 9092), false, true, false));
           }
         };
-    scores = cpuCost.brokerCost(clusterInfo2).value();
-    Assertions.assertEquals(0.51, scores.get(1));
-    Assertions.assertEquals(0.55, scores.get(2));
-    Assertions.assertEquals(0.44, scores.get(3));
+    scores = cpuCost.brokerCost(clusterInfo2).normalize(Normalizer.TScore()).value();
+    Assertions.assertEquals(0.52, scores.get(1));
+    Assertions.assertEquals(0.61, scores.get(2));
+    Assertions.assertEquals(0.37, scores.get(3));
   }
 
   private OperatingSystemInfo mockResult(double usage) {
diff --git a/app/src/test/java/org/astraea/app/cost/broker/MemoryCostTest.java b/app/src/test/java/org/astraea/app/cost/broker/MemoryCostTest.java
index 9c7f3fb6..0521718c 100644
--- a/app/src/test/java/org/astraea/app/cost/broker/MemoryCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/broker/MemoryCostTest.java
@@ -24,6 +24,7 @@ import java.util.Set;
 import org.astraea.app.cost.ClusterInfo;
 import org.astraea.app.cost.FakeClusterInfo;
 import org.astraea.app.cost.NodeInfo;
+import org.astraea.app.cost.Normalizer;
 import org.astraea.app.cost.ReplicaInfo;
 import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.java.HasJvmMemory;
@@ -63,7 +64,7 @@ public class MemoryCostTest {
         };
 
     var memoryCost = new MemoryCost();
-    var scores = memoryCost.brokerCost(clusterInfo).value();
+    var scores = memoryCost.brokerCost(clusterInfo).normalize(Normalizer.TScore()).value();
     Assertions.assertEquals(0.39, scores.get(1));
     Assertions.assertEquals(0.63, scores.get(2));
     Assertions.assertEquals(0.47, scores.get(3));
@@ -96,10 +97,10 @@ public class MemoryCostTest {
                 ReplicaInfo.of("t", 0, NodeInfo.of(3, "host3", 9092), false, true, false));
           }
         };
-    scores = memoryCost.brokerCost(clusterInfo2).value();
-    Assertions.assertEquals(0.42, scores.get(1));
-    Assertions.assertEquals(0.52, scores.get(2));
-    Assertions.assertEquals(0.55, scores.get(3));
+    scores = memoryCost.brokerCost(clusterInfo2).normalize(Normalizer.TScore()).value();
+    Assertions.assertEquals(0.36, scores.get(1));
+    Assertions.assertEquals(0.58, scores.get(2));
+    Assertions.assertEquals(0.56, scores.get(3));
   }
 
   private HasJvmMemory mockResult(long used, long max) {