Repository: skiptests/astraea
Pull request #633 (2022-08-25 09:25:00)

WebService `POST /throttles` 和 `DELETE /throttles` 的文件
這個 PR：

1. 實作 `POST /throttles` 和 `DELETE /throttles` 的文件
2. 嘗試實作和 `/throttles` 相關物件的反序列化程式碼，過程中有遇到一些問題

爲了避免 PR 太肥大，實作之後才會開工，這邊先推文件。


Patch:
diff --git a/README.md b/README.md
index ad969102..a06c1726 100644
--- a/README.md
+++ b/README.md
@@ -20,55 +20,4 @@
 3. [Performance Tool ](./docs/performance_benchmark.md): 可模擬多種使用情境來驗證`Kafka`叢集的吞吐量和延遲
 4. [快速啟動 Prometheus ](./docs/run_prometheus.md):  建構`Kafka`叢集資訊收集系統
 5. [快速啟動 Grafana ](./docs/run_grafana.md): 建置圖形化介面監控`kafka`叢集使用狀況
-6. [Web Server](./docs/web_server/README.md): 可透過`Restful APIs`操作`Kafka`叢集
-7. [Kafka replica syncing monitor](#Kafka-replica-syncing-monitor): Tracking replica syncing progress. (deprecated)
-
----
-
-## Kafka Replica Syncing Monitor
-
-This tool will track partition replica syncing progress. This tool can be used to observe 
-the partition migration process.
-
-### Start monitor syncing progress
-
-```shell
-$ ./gradlew run --args="monitor --bootstrap.servers 192.168.103.39:9092"
-
-[2021-11-23T16:00:26.282676667]
-  Topic "my-topic":
-  | Partition 0:
-  | | replica on broker   0 => [####################] 100.00% [leader, synced]
-  | | replica on broker   2 => [                    ]   1.37% 0.00 B/s (unknown) []
-  | Partition 1:
-  | | replica on broker   0 => [####################] 100.00% [leader, synced]
-  | | replica on broker   2 => [                    ]   1.35% 0.00 B/s (unknown) []
-
-[2021-11-23T16:00:26.862637796]
-  Topic "my-topic":
-  | Partition 0:
-  | | replica on broker   0 => [####################] 100.00% [leader, synced]
-  | | replica on broker   2 => [#                   ]   5.62% 240.54 MB/s (11s estimated) []
-  | Partition 1:
-  | | replica on broker   0 => [####################] 100.00% [leader, synced]
-  | | replica on broker   2 => [#                   ]   5.25% 242.53 MB/s (12s estimated) []
-
-[2021-11-23T16:00:27.400814839]
-  Topic "my-topic":
-  | Partition 0:
-  | | replica on broker   0 => [####################] 100.00% [leader, synced]
-  | | replica on broker   2 => [##                  ]   9.90% 242.53 MB/s (10s estimated) []
-  | Partition 1:
-  | | replica on broker   0 => [####################] 100.00% [leader, synced]
-  | | replica on broker   2 => [##                  ]   9.13% 240.54 MB/s (11s estimated) []
-
-...
-```
-
-### Replica Syncing Monitor Configurations
-
-1. --bootstrap.servers: the server to connect to
-2. --interval: the frequency(time interval in second) to check replica state, support floating point value. (default: 1 second)
-3. --prop.file: the path to a file that containing the properties to be passed to kafka admin.
-4. --topic: topics to track (default: track all non-synced partition by default)
-5. --track: keep track even if all the replicas are synced. Also attempts to discover any non-synced replicas. (default: false)
+6. [Web Server](./docs/web_server/README.md): 可透過`Restful APIs`操作`Kafka`叢集
\ No newline at end of file
diff --git a/app/src/main/java/org/astraea/app/App.java b/app/src/main/java/org/astraea/app/App.java
index 0d149fbf..defd2e10 100644
--- a/app/src/main/java/org/astraea/app/App.java
+++ b/app/src/main/java/org/astraea/app/App.java
@@ -21,7 +21,6 @@ import java.lang.reflect.InvocationTargetException;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
-import org.astraea.app.admin.ReplicaSyncingMonitor;
 import org.astraea.app.automation.Automation;
 import org.astraea.app.balancer.BalanceProcessDemo;
 import org.astraea.app.performance.Performance;
@@ -31,7 +30,6 @@ public class App {
   private static final Map<String, Class<?>> MAIN_CLASSES =
       Map.of(
           "performance", Performance.class,
-          "monitor", ReplicaSyncingMonitor.class,
           "automation", Automation.class,
           "web", WebService.class,
           "balance-demo", BalanceProcessDemo.class);
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 f5149e6d..059e95a9 100644
--- a/app/src/main/java/org/astraea/app/admin/Admin.java
+++ b/app/src/main/java/org/astraea/app/admin/Admin.java
@@ -37,8 +37,16 @@ public interface Admin extends Closeable {
     return builder().configs(configs).build();
   }
 
-  /** @return names of all topics */
-  Set<String> topicNames();
+  /**
+   * @param listInternal should list internal topics or not
+   * @return names of topics
+   */
+  Set<String> topicNames(boolean listInternal);
+
+  /** @return names of all topics (include internal topics). */
+  default Set<String> topicNames() {
+    return topicNames(true);
+  }
 
   /** @return the topic name and its configurations. */
   default Map<String, Config> topics() {
@@ -282,6 +290,12 @@ public interface Admin extends Closeable {
    */
   void clearReplicationThrottle(TopicPartitionReplica log);
 
+  /** Clear the ingress bandwidth of replication throttle for the specified brokers. */
+  void clearIngressReplicationThrottle(Set<Integer> brokerIds);
+
+  /** Clear the egress bandwidth of replication throttle for the specified brokers. */
+  void clearEgressReplicationThrottle(Set<Integer> brokerIds);
+
   @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 105e1318..9a015675 100644
--- a/app/src/main/java/org/astraea/app/admin/Builder.java
+++ b/app/src/main/java/org/astraea/app/admin/Builder.java
@@ -273,9 +273,9 @@ public class Builder {
     }
 
     @Override
-    public Set<String> topicNames() {
+    public Set<String> topicNames(boolean listInternal) {
       return Utils.packException(
-          () -> admin.listTopics(new ListTopicsOptions().listInternal(true)).names().get());
+          () -> admin.listTopics(new ListTopicsOptions().listInternal(listInternal)).names().get());
     }
 
     @Override
@@ -333,9 +333,9 @@ public class Builder {
         Set<String> topics, Set<Integer> brokerIds) {
       return replicas(topics).entrySet().stream()
           .flatMap(
-              e -> e.getValue().stream().map(replica -> Map.entry(replica.broker(), e.getKey())))
-          .filter(e -> brokerIds.contains(e.getKey()))
-          .collect(Collectors.groupingBy(Map.Entry::getKey))
+              e -> e.getValue().stream().map(replica -> Map.entry(replica.nodeInfo(), e.getKey())))
+          .filter(e -> brokerIds.contains(e.getKey().id()))
+          .collect(Collectors.groupingBy(e -> e.getKey().id()))
           .entrySet()
           .stream()
           .collect(
@@ -346,80 +346,68 @@ public class Builder {
 
     @Override
     public Map<TopicPartition, List<Replica>> replicas(Set<String> topics) {
+      var logInfo =
+          Utils.packException(() -> admin.describeLogDirs(brokerIds()).allDescriptions().get());
+
+      BiFunction<
+              org.apache.kafka.common.TopicPartition,
+              org.apache.kafka.common.TopicPartitionInfo,
+              List<Replica>>
+          toReplicas =
+              (tp, tpi) ->
+                  tpi.replicas().stream()
+                      .map(
+                          node -> {
+                            var dataPath =
+                                node.isEmpty()
+                                    ? null
+                                    : logInfo.get(node.id()).entrySet().stream()
+                                        .filter(e -> e.getValue().replicaInfos().containsKey(tp))
+                                        .map(Map.Entry::getKey)
+                                        .findFirst()
+                                        .orElse(null);
+                            var replicaInfo =
+                                node.isEmpty() || dataPath == null
+                                    ? null
+                                    : logInfo.get(node.id()).get(dataPath).replicaInfos().get(tp);
+                            return Replica.of(
+                                tp.topic(),
+                                tp.partition(),
+                                NodeInfo.of(node),
+                                replicaInfo != null ? replicaInfo.offsetLag() : -1L,
+                                replicaInfo != null ? replicaInfo.size() : -1L,
+                                tpi.leader() != null
+                                    && !tpi.leader().isEmpty()
+                                    && tpi.leader().id() == node.id(),
+                                tpi.isr().contains(node),
+                                replicaInfo != null && replicaInfo.isFuture(),
+                                node.isEmpty(),
+                                // The first replica in the return result is the
+                                // preferred leader. This only works with Kafka broker version
+                                // after 0.11.
+                                // Version before 0.11 returns the replicas in unspecified order
+                                // due to a bug.
+                                tpi.replicas().get(0).id() == node.id(),
+                                dataPath);
+                          })
+                      .collect(Collectors.toList());
+
       return Utils.packException(
-          () -> {
-            var logInfo = admin.describeLogDirs(brokerIds()).allDescriptions().get();
-            var tpPathMap = new HashMap<TopicPartition, Map<Integer, String>>();
-            for (var entry0 : logInfo.entrySet()) {
-              for (var entry1 : entry0.getValue().entrySet()) {
-                for (var entry2 : entry1.getValue().replicaInfos().entrySet()) {
-                  var tp = TopicPartition.from(entry2.getKey());
-                  var broker = entry0.getKey();
-                  var dataPath = entry1.getKey();
-                  tpPathMap.computeIfAbsent(tp, (ignore) -> new HashMap<>()).put(broker, dataPath);
-                }
-              }
-            }
-            return admin.describeTopics(topics).allTopicNames().get().entrySet().stream()
-                .flatMap(
-                    entry ->
-                        entry.getValue().partitions().stream()
-                            .map(
-                                tpInfo -> {
-                                  var topicName = entry.getKey();
-                                  var partition = tpInfo.partition();
-                                  var topicPartition = TopicPartition.of(topicName, partition);
-                                  return Map.entry(topicPartition, tpInfo);
-                                }))
-                .collect(
-                    Collectors.toUnmodifiableMap(
-                        Map.Entry::getKey,
-                        (entry) -> {
-                          var topicPartition = entry.getKey();
-                          var tpInfo = entry.getValue();
-                          var replicaLeaderId = tpInfo.leader() != null ? tpInfo.leader().id() : -1;
-                          var isrSet = tpInfo.isr();
-                          // The first replica in the return result is the preferred leader. This
-                          // only works with Kafka broker version after 0.11. Version before 0.11
-                          // returns the replicas in unspecified order due to a bug.
-                          var preferredLeader = entry.getValue().replicas().get(0);
-                          return entry.getValue().replicas().stream()
-                              .map(
-                                  node -> {
-                                    int broker = node.id();
-                                    var dataPath =
-                                        tpPathMap
-                                            .getOrDefault(topicPartition, Map.of())
-                                            .getOrDefault(broker, null);
-                                    var replicaInfo =
-                                        node.isEmpty() || dataPath == null
-                                            ? null
-                                            : logInfo
-                                                .get(broker)
-                                                .get(dataPath)
-                                                .replicaInfos()
-                                                .get(TopicPartition.to(topicPartition));
-                                    boolean isLeader = node.id() == replicaLeaderId;
-                                    boolean inSync = isrSet.contains(node);
-                                    long lag = replicaInfo != null ? replicaInfo.offsetLag() : -1L;
-                                    long size = replicaInfo != null ? replicaInfo.size() : -1L;
-                                    boolean future = replicaInfo != null && replicaInfo.isFuture();
-                                    boolean offline = node.isEmpty();
-                                    boolean isPreferredLeader = preferredLeader.id() == broker;
-                                    return new Replica(
-                                        broker,
-                                        lag,
-                                        size,
-                                        isLeader,
-                                        inSync,
-                                        future,
-                                        offline,
-                                        isPreferredLeader,
-                                        dataPath);
-                                  })
-                              .collect(Collectors.toList());
-                        }));
-          });
+              () ->
+                  admin.describeTopics(topics).allTopicNames().get().entrySet().stream()
+                      .flatMap(
+                          e ->
+                              e.getValue().partitions().stream()
+                                  .map(
+                                      tpInfo ->
+                                          Map.entry(
+                                              new org.apache.kafka.common.TopicPartition(
+                                                  e.getKey(), tpInfo.partition()),
+                                              tpInfo))))
+          .collect(
+              Collectors.toUnmodifiableMap(
+                  e -> TopicPartition.from(e.getKey()),
+                  entry -> toReplicas.apply(entry.getKey(), entry.getValue())));
     }
 
     @Override
@@ -461,29 +449,13 @@ public class Builder {
     public ClusterInfo clusterInfo(Set<String> topics) {
       final var nodeInfo = this.nodes().stream().collect(Collectors.toUnmodifiableList());
 
-      final var 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));
+      var replicas =
+          Utils.packException(
+              () ->
+                  replicas(topics).values().stream()
+                      .flatMap(Collection::stream)
+                      .map(r -> (ReplicaInfo) r)
+                      .collect(Collectors.toUnmodifiableList()));
 
       return new ClusterInfo() {
         @Override
@@ -492,13 +464,8 @@ public class Builder {
         }
 
         @Override
-        public Set<String> topics() {
-          return topics;
-        }
-
-        @Override
-        public List<ReplicaInfo> replicas(String topic) {
-          return topicToReplicasMap.getOrDefault(topic, List.of());
+        public List<ReplicaInfo> replicas() {
+          return replicas;
         }
       };
     }
@@ -590,7 +557,7 @@ public class Builder {
                                                   new org.apache.kafka.common.TopicPartitionReplica(
                                                       e.getKey().topic(),
                                                       e.getKey().partition(),
-                                                      r.broker())))
+                                                      r.nodeInfo().id())))
                               .collect(Collectors.toUnmodifiableList()))
                       .all()
                       .get());
@@ -690,14 +657,14 @@ public class Builder {
                   (tp, replicas) -> {
                     replicas.forEach(
                         replica -> {
-                          if (replica.leader())
+                          if (replica.isLeader())
                             leaders.add(
                                 TopicPartitionReplica.of(
-                                    tp.topic(), tp.partition(), replica.broker()));
+                                    tp.topic(), tp.partition(), replica.nodeInfo().id()));
                           else
                             followers.add(
                                 TopicPartitionReplica.of(
-                                    tp.topic(), tp.partition(), replica.broker()));
+                                    tp.topic(), tp.partition(), replica.nodeInfo().id()));
                         });
                   });
           return this;
@@ -709,14 +676,18 @@ public class Builder {
               replicas(Set.of(topicPartition.topic())).getOrDefault(topicPartition, List.of());
           replicas.forEach(
               replica -> {
-                if (replica.leader())
+                if (replica.isLeader())
                   leaders.add(
                       TopicPartitionReplica.of(
-                          topicPartition.topic(), topicPartition.partition(), replica.broker()));
+                          topicPartition.topic(),
+                          topicPartition.partition(),
+                          replica.nodeInfo().id()));
                 else
                   followers.add(
                       TopicPartitionReplica.of(
-                          topicPartition.topic(), topicPartition.partition(), replica.broker()));
+                          topicPartition.topic(),
+                          topicPartition.partition(),
+                          replica.nodeInfo().id()));
               });
           return this;
         }
@@ -889,7 +860,7 @@ public class Builder {
     public void clearReplicationThrottle(TopicPartition topicPartition) {
       var configValue =
           replicas(Set.of(topicPartition.topic())).get(topicPartition).stream()
-              .map(replica -> topicPartition.partition() + ":" + replica.broker())
+              .map(replica -> topicPartition.partition() + ":" + replica.nodeInfo().id())
               .collect(Collectors.joining(","));
       var configEntry0 = new ConfigEntry("leader.replication.throttled.replicas", configValue);
       var configEntry1 = new ConfigEntry("follower.replication.throttled.replicas", configValue);
@@ -915,6 +886,43 @@ public class Builder {
               admin.incrementalAlterConfigs(
                   Map.of(configResource, List.of(alterConfigOp0, alterConfigOp1))));
     }
+
+    @Override
+    public void clearIngressReplicationThrottle(Set<Integer> brokerIds) {
+      deleteBrokerConfigs(
+          brokerIds.stream()
+              .collect(
+                  Collectors.toUnmodifiableMap(
+                      id -> id, id -> Set.of("follower.replication.throttled.rate"))));
+    }
+
+    @Override
+    public void clearEgressReplicationThrottle(Set<Integer> brokerIds) {
+      deleteBrokerConfigs(
+          brokerIds.stream()
+              .collect(
+                  Collectors.toUnmodifiableMap(
+                      id -> id, id -> Set.of("leader.replication.throttled.rate"))));
+    }
+
+    private void deleteBrokerConfigs(Map<Integer, Set<String>> brokerAndConfigKeys) {
+      Function<String, AlterConfigOp> deleteConfig =
+          (key) -> new AlterConfigOp(new ConfigEntry(key, ""), AlterConfigOp.OpType.DELETE);
+      var map =
+          brokerAndConfigKeys.entrySet().stream()
+              .map(
+                  entry ->
+                      Map.entry(
+                          String.valueOf(entry.getKey()),
+                          entry.getValue().stream()
+                              .map(deleteConfig)
+                              .collect(Collectors.toUnmodifiableList())))
+              .collect(
+                  Collectors.toUnmodifiableMap(
+                      entry -> new ConfigResource(ConfigResource.Type.BROKER, entry.getKey()),
+                      entry -> (Collection<AlterConfigOp>) entry.getValue()));
+      Utils.packException(() -> admin.incrementalAlterConfigs(map).all().get());
+    }
   }
 
   private static class ConfigImpl implements Config {
diff --git a/app/src/main/java/org/astraea/app/admin/ClusterInfo.java b/app/src/main/java/org/astraea/app/admin/ClusterInfo.java
index 0fbfb7c8..4289df4d 100644
--- a/app/src/main/java/org/astraea/app/admin/ClusterInfo.java
+++ b/app/src/main/java/org/astraea/app/admin/ClusterInfo.java
@@ -19,6 +19,7 @@ package org.astraea.app.admin;
 import java.util.List;
 import java.util.Map;
 import java.util.NoSuchElementException;
+import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
 
@@ -32,45 +33,43 @@ public interface ClusterInfo {
         }
 
         @Override
-        public Set<String> topics() {
-          return Set.of();
-        }
-
-        @Override
-        public List<ReplicaInfo> replicas(String topic) {
+        public List<ReplicaInfo> replicas() {
           return List.of();
         }
       };
 
   /**
-   * convert the kafka Cluster to our ClusterInfo. All data structure are converted immediately, so
-   * you should cache the result if the performance is critical
+   * convert the kafka Cluster to our ClusterInfo. Noted: this method is used by {@link
+   * org.astraea.app.cost.HasBrokerCost} normally, so all data structure are converted immediately
    *
    * @param cluster kafka ClusterInfo
-   * @return astraea ClusterInfo
+   * @return ClusterInfo
    */
   static ClusterInfo of(org.apache.kafka.common.Cluster cluster) {
-    var nodes = cluster.nodes().stream().map(NodeInfo::of).collect(Collectors.toUnmodifiableList());
-    var topics = cluster.topics();
-    var allReplicas =
-        topics.stream()
+    return of(
+        cluster.nodes().stream().map(NodeInfo::of).collect(Collectors.toUnmodifiableList()),
+        cluster.topics().stream()
             .flatMap(t -> cluster.partitionsForTopic(t).stream())
             .flatMap(p -> ReplicaInfo.of(p).stream())
-            .collect(Collectors.toUnmodifiableList());
-    var replicasForTopic = allReplicas.stream().collect(Collectors.groupingBy(ReplicaInfo::topic));
+            .collect(Collectors.toUnmodifiableList()));
+  }
+
+  static ClusterInfo of(List<NodeInfo> nodes, List<ReplicaInfo> replicas) {
+    var topics = replicas.stream().map(ReplicaInfo::topic).collect(Collectors.toUnmodifiableSet());
+    var replicasForTopic = replicas.stream().collect(Collectors.groupingBy(ReplicaInfo::topic));
     var availableReplicasForTopic =
-        allReplicas.stream()
-            .filter(ReplicaInfo::isOnlineReplica)
+        replicas.stream()
+            .filter(ReplicaInfo::isOnline)
             .collect(Collectors.groupingBy(ReplicaInfo::topic));
     var availableReplicaLeadersForTopics =
-        allReplicas.stream()
-            .filter(ReplicaInfo::isOnlineReplica)
+        replicas.stream()
+            .filter(ReplicaInfo::isOnline)
             .filter(ReplicaInfo::isLeader)
             .collect(Collectors.groupingBy(ReplicaInfo::topic));
     // This group is used commonly, so we cache it.
     var availableLeaderReplicasForBrokersTopics =
-        allReplicas.stream()
-            .filter(ReplicaInfo::isOnlineReplica)
+        replicas.stream()
+            .filter(ReplicaInfo::isOnline)
             .filter(ReplicaInfo::isLeader)
             .collect(Collectors.groupingBy(r -> Map.entry(r.nodeInfo().id(), r.topic())));
 
@@ -104,6 +103,11 @@ public interface ClusterInfo {
       public List<ReplicaInfo> replicas(String topic) {
         return replicasForTopic.getOrDefault(topic, List.of());
       }
+
+      @Override
+      public List<ReplicaInfo> replicas() {
+        return replicas;
+      }
     };
   }
 
@@ -156,7 +160,7 @@ public interface ClusterInfo {
    */
   default List<ReplicaInfo> availableReplicas(String topic) {
     return replicas(topic).stream()
-        .filter(ReplicaInfo::isOnlineReplica)
+        .filter(ReplicaInfo::isOnline)
         .collect(Collectors.toUnmodifiableList());
   }
 
@@ -168,7 +172,9 @@ public interface ClusterInfo {
    *
    * @return return a set of topic names
    */
-  Set<String> topics();
+  default Set<String> topics() {
+    return replicas().stream().map(ReplicaInfo::topic).collect(Collectors.toUnmodifiableSet());
+  }
 
   /**
    * Get the list of replica information of each partition/replica pair for the given topic
@@ -176,5 +182,20 @@ public interface ClusterInfo {
    * @param topic The topic name
    * @return A list of {@link ReplicaInfo}.
    */
-  List<ReplicaInfo> replicas(String topic);
+  default List<ReplicaInfo> replicas(String topic) {
+    return replicas().stream()
+        .filter(r -> r.topic().equals(topic))
+        .collect(Collectors.toUnmodifiableList());
+  }
+
+  /**
+   * @param replica to search
+   * @return the replica matched to input replica
+   */
+  default Optional<ReplicaInfo> replica(TopicPartitionReplica replica) {
+    return replicas().stream().filter(r -> r.topicPartitionReplica().equals(replica)).findFirst();
+  }
+
+  /** @return all replicas cached by this cluster info. */
+  List<ReplicaInfo> replicas();
 }
diff --git a/app/src/main/java/org/astraea/app/admin/NodeInfo.java b/app/src/main/java/org/astraea/app/admin/NodeInfo.java
index cd7ad6a2..364deabf 100644
--- a/app/src/main/java/org/astraea/app/admin/NodeInfo.java
+++ b/app/src/main/java/org/astraea/app/admin/NodeInfo.java
@@ -24,19 +24,7 @@ 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() {
       // NodeInfo is used to be key of Map commonly, so creating hash can reduce the memory pressure
       private final int hashCode = Objects.hash(id, host, port);
@@ -51,11 +39,6 @@ public interface NodeInfo extends Comparable<NodeInfo> {
         return id;
       }
 
-      @Override
-      public boolean isOffline() {
-        return isOffline;
-      }
-
       @Override
       public int port() {
         return port;
@@ -91,12 +74,4 @@ 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/admin/Reassignment.java b/app/src/main/java/org/astraea/app/admin/Reassignment.java
index 358dbbbc..94a78fc1 100644
--- a/app/src/main/java/org/astraea/app/admin/Reassignment.java
+++ b/app/src/main/java/org/astraea/app/admin/Reassignment.java
@@ -58,19 +58,19 @@ public class Reassignment {
 
   public static class Location {
     private final int broker;
-    private final String path;
+    private final String dataFolder;
 
-    public Location(int broker, String path) {
+    public Location(int broker, String dataFolder) {
       this.broker = broker;
-      this.path = Objects.requireNonNull(path);
+      this.dataFolder = Objects.requireNonNull(dataFolder);
     }
 
     public int broker() {
       return broker;
     }
 
-    public String path() {
-      return path;
+    public String dataFolder() {
+      return dataFolder;
     }
 
     @Override
@@ -78,17 +78,17 @@ public class Reassignment {
       if (this == o) return true;
       if (o == null || getClass() != o.getClass()) return false;
       Location location = (Location) o;
-      return broker == location.broker && path.equals(location.path);
+      return broker == location.broker && dataFolder.equals(location.dataFolder);
     }
 
     @Override
     public int hashCode() {
-      return Objects.hash(broker, path);
+      return Objects.hash(broker, dataFolder);
     }
 
     @Override
     public String toString() {
-      return "Location{" + "broker=" + broker + ", path='" + path + '\'' + '}';
+      return "Location{" + "broker=" + broker + ", dataFolder='" + dataFolder + '\'' + '}';
     }
   }
 }
diff --git a/app/src/main/java/org/astraea/app/admin/Replica.java b/app/src/main/java/org/astraea/app/admin/Replica.java
index 8aee24d3..d3d97145 100644
--- a/app/src/main/java/org/astraea/app/admin/Replica.java
+++ b/app/src/main/java/org/astraea/app/admin/Replica.java
@@ -18,19 +18,12 @@ package org.astraea.app.admin;
 
 import java.util.Objects;
 
-public final class Replica {
-  private final int broker;
-  private final long lag;
-  private final long size;
-  private final boolean leader;
-  private final boolean inSync;
-  private final boolean isFuture;
-  private final boolean isPreferredLeader;
-  private final boolean offline;
-  private final String path;
-
-  Replica(
-      int broker,
+public interface Replica extends ReplicaInfo {
+
+  static Replica of(
+      String topic,
+      int partition,
+      NodeInfo nodeInfo,
       long lag,
       long size,
       boolean leader,
@@ -38,83 +31,127 @@ public final class Replica {
       boolean isFuture,
       boolean offline,
       boolean isPreferredLeader,
-      String path) {
-    this.broker = broker;
-    this.lag = lag;
-    this.size = size;
-    this.leader = leader;
-    this.inSync = inSync;
-    this.isFuture = isFuture;
-    this.isPreferredLeader = isPreferredLeader;
-    this.offline = offline;
-    this.path = path;
-  }
-
-  @Override
-  public boolean equals(Object o) {
-    if (this == o) return true;
-    if (o == null || getClass() != o.getClass()) return false;
-    Replica replica = (Replica) o;
-    return broker == replica.broker
-        && lag == replica.lag
-        && size == replica.size
-        && leader == replica.leader
-        && inSync == replica.inSync
-        && isFuture == replica.isFuture
-        && isPreferredLeader == replica.isPreferredLeader
-        && offline == replica.offline
-        && path.equals(replica.path);
-  }
-
-  @Override
-  public int hashCode() {
-    return Objects.hash(
-        broker, lag, size, leader, inSync, isFuture, isPreferredLeader, offline, path);
-  }
-
-  @Override
-  public String toString() {
-    return "Replica{"
-        + "broker="
-        + broker
-        + ", lag="
-        + lag
-        + ", size="
-        + size
-        + ", leader="
-        + leader
-        + ", inSync="
-        + inSync
-        + ", isFuture="
-        + isFuture
-        + ", isPreferredLeader="
-        + isPreferredLeader
-        + ", offline="
-        + offline
-        + ", path='"
-        + path
-        + '\''
-        + '}';
-  }
-
-  public int broker() {
-    return broker;
-  }
-
-  public long lag() {
-    return lag;
-  }
-
-  public long size() {
-    return size;
-  }
-
-  public boolean leader() {
-    return leader;
-  }
-
-  public boolean inSync() {
-    return inSync;
+      String dataFolder) {
+    return new Replica() {
+      @Override
+      public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        var replica = (Replica) o;
+        return partition == replica.partition()
+            && Objects.equals(nodeInfo, replica.nodeInfo())
+            && lag == replica.lag()
+            && size == replica.size()
+            && leader == replica.isLeader()
+            && inSync == replica.inSync()
+            && isFuture == replica.isFuture()
+            && isPreferredLeader == replica.isPreferredLeader()
+            && offline == replica.isOffline()
+            && Objects.equals(topic, replica.topic())
+            && Objects.equals(dataFolder, replica.dataFolder());
+      }
+
+      @Override
+      public int hashCode() {
+        return Objects.hash(
+            topic,
+            partition,
+            nodeInfo,
+            lag,
+            size,
+            leader,
+            inSync,
+            isFuture,
+            isPreferredLeader,
+            offline,
+            dataFolder);
+      }
+
+      @Override
+      public String toString() {
+        return "Replica{"
+            + "topic='"
+            + topic
+            + '\''
+            + ", partition="
+            + partition
+            + ", broker="
+            + nodeInfo
+            + ", lag="
+            + lag
+            + ", size="
+            + size
+            + ", leader="
+            + leader
+            + ", inSync="
+            + inSync
+            + ", isFuture="
+            + isFuture
+            + ", isPreferredLeader="
+            + isPreferredLeader
+            + ", offline="
+            + offline
+            + ", path='"
+            + dataFolder
+            + '\''
+            + '}';
+      }
+
+      @Override
+      public String topic() {
+        return topic;
+      }
+
+      @Override
+      public int partition() {
+        return partition;
+      }
+
+      @Override
+      public NodeInfo nodeInfo() {
+        return nodeInfo;
+      }
+
+      @Override
+      public long lag() {
+        return lag;
+      }
+
+      public long size() {
+        return size;
+      }
+
+      @Override
+      public boolean isLeader() {
+        return leader;
+      }
+
+      @Override
+      public boolean inSync() {
+        return inSync;
+      }
+
+      @Override
+      public boolean isFuture() {
+        return isFuture;
+      }
+
+      @Override
+      public boolean isPreferredLeader() {
+        return isPreferredLeader;
+      }
+
+      /** @return true if the replica on the broker is offline. */
+      @Override
+      public boolean isOffline() {
+        return offline;
+      }
+
+      @Override
+      public String dataFolder() {
+        return dataFolder;
+      }
+    };
   }
 
   /**
@@ -124,26 +161,22 @@ public final class Replica {
    * @return true if this log is created by AlterReplicaLogDirsRequest and will replace the current
    *     log of the replica at some time in the future.
    */
-  public boolean isFuture() {
-    return isFuture;
-  }
+  boolean isFuture();
 
   /** @return true if this is current log of replica. */
-  public boolean isCurrent() {
-    return !isFuture;
+  default boolean isCurrent() {
+    return !isFuture();
   }
 
   /** @return true if the replica is the preferred leader */
-  public boolean isPreferredLeader() {
-    return isPreferredLeader;
-  }
+  boolean isPreferredLeader();
 
-  /** @return true if the replica on the broker is offline. */
-  public boolean isOffline() {
-    return offline;
-  }
+  long lag();
 
-  public String path() {
-    return path;
-  }
+  long size();
+
+  /**
+   * @return that indicates the data folder path which stored this replica on a specific Kafka node.
+   */
+  String dataFolder();
 }
diff --git a/app/src/main/java/org/astraea/app/admin/ReplicaInfo.java b/app/src/main/java/org/astraea/app/admin/ReplicaInfo.java
index c40e76d1..02536e5d 100644
--- a/app/src/main/java/org/astraea/app/admin/ReplicaInfo.java
+++ b/app/src/main/java/org/astraea/app/admin/ReplicaInfo.java
@@ -17,7 +17,6 @@
 package org.astraea.app.admin;
 
 import java.util.List;
-import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
 
@@ -48,18 +47,7 @@ public interface ReplicaInfo {
       NodeInfo nodeInfo,
       boolean isLeader,
       boolean isSynced,
-      boolean isOfflineReplica) {
-    return of(topic, partition, nodeInfo, isLeader, isSynced, isOfflineReplica, null);
-  }
-
-  static ReplicaInfo of(
-      String topic,
-      int partition,
-      NodeInfo nodeInfo,
-      boolean isLeader,
-      boolean isSynced,
-      boolean isOfflineReplica,
-      String dataFolder) {
+      boolean isOffline) {
     return new ReplicaInfo() {
       @Override
       public String topic() {
@@ -87,13 +75,8 @@ public interface ReplicaInfo {
       }
 
       @Override
-      public boolean isOfflineReplica() {
-        return isOfflineReplica;
-      }
-
-      @Override
-      public Optional<String> dataFolder() {
-        return Optional.ofNullable(dataFolder);
+      public boolean isOffline() {
+        return isOffline;
       }
 
       @Override
@@ -105,12 +88,10 @@ public interface ReplicaInfo {
             + partition
             + " replicaAtBroker="
             + nodeInfo.id()
-            + " dataFolder="
-            + (dataFolder().map(x -> "\"" + x + "\"").orElse("unknown"))
             + (isLeader() ? " leader" : "")
             + (isFollower() ? " follower" : "")
             + (inSync() ? ":synced" : "")
-            + (isOfflineReplica() ? ":offline" : "")
+            + (isOffline() ? ":offline" : "")
             + "}";
       }
 
@@ -125,14 +106,22 @@ public interface ReplicaInfo {
               && this.isLeader() == that.isLeader()
               && this.isFollower() == that.isFollower()
               && this.inSync() == that.inSync()
-              && this.isOfflineReplica() == that.isOfflineReplica()
-              && this.dataFolder().equals(that.dataFolder());
+              && this.isOffline() == that.isOffline();
         }
         return false;
       }
     };
   }
 
+  /**
+   * a helper to build TopicPartitionReplica quickly
+   *
+   * @return TopicPartitionReplica
+   */
+  default TopicPartitionReplica topicPartitionReplica() {
+    return TopicPartitionReplica.of(topic(), partition(), nodeInfo().id());
+  }
+
   /** @return topic name */
   String topic();
 
@@ -154,21 +143,10 @@ public interface ReplicaInfo {
   boolean inSync();
 
   /** @return true if this replica is offline */
-  boolean isOfflineReplica();
+  boolean isOffline();
 
   /** @return true if this replica is online */
-  default boolean isOnlineReplica() {
-    return !isOfflineReplica();
+  default boolean isOnline() {
+    return !isOffline();
   }
-
-  /**
-   * 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
-   * producer client might need to initialize an AdminClient to access this information). To provide
-   * this information or not is totally up to the caller.
-   *
-   * @return a {@link Optional<String>} that indicates the data folder path which stored this
-   *     replica on a specific Kafka node.
-   */
-  Optional<String> dataFolder();
 }
diff --git a/app/src/main/java/org/astraea/app/admin/ReplicaSyncingMonitor.java b/app/src/main/java/org/astraea/app/admin/ReplicaSyncingMonitor.java
deleted file mode 100644
index 7d3dfd69..00000000
--- a/app/src/main/java/org/astraea/app/admin/ReplicaSyncingMonitor.java
+++ /dev/null
@@ -1,376 +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.admin;
-
-import com.beust.jcommander.Parameter;
-import java.time.Duration;
-import java.time.LocalDateTime;
-import java.time.format.DateTimeFormatter;
-import java.time.temporal.ChronoUnit;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Set;
-import java.util.function.Supplier;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-import org.apache.kafka.common.TopicPartitionReplica;
-import org.astraea.app.argument.DurationField;
-import org.astraea.app.argument.NonEmptyStringField;
-import org.astraea.app.common.DataRate;
-import org.astraea.app.common.DataSize;
-import org.astraea.app.common.DataUnit;
-import org.astraea.app.common.Utils;
-
-public class ReplicaSyncingMonitor {
-
-  public static void main(String[] args) {
-    Argument argument = org.astraea.app.argument.Argument.parse(new Argument(), args);
-    try (Admin topicAdmin = Admin.of(argument.bootstrapServers())) {
-      execute(topicAdmin, argument);
-    }
-  }
-
-  static void execute(final Admin topicAdmin, final Argument argument) {
-
-    // this supplier will give you all the topic names that the client is interested in.
-    // discover any newly happened non-synced replica can be a quiet useful scenario, so here we use
-    // a supplier to do the query for us.
-    Supplier<Set<String>> topicToTrack =
-        () -> argument.topics.isEmpty() ? topicAdmin.topicNames() : argument.topics;
-
-    // the non-synced topic-partition we want to monitor
-    var topicPartitionToTrack = findNonSyncedTopicPartition(topicAdmin, topicToTrack.get());
-
-    // keep tracking the previous replica size of a topic-partition-replica tuple
-    final var previousCheckedSize = new HashMap<TopicPartitionReplica, Long>();
-
-    while (!topicPartitionToTrack.isEmpty() || argument.keepTrack) {
-
-      long startTime = System.nanoTime();
-
-      System.out.printf(
-          "[%s]%n", LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_DATE_TIME));
-      var replicaProgress =
-          topicAdmin.replicas(
-              topicPartitionToTrack.stream()
-                  .map(TopicPartition::topic)
-                  .collect(Collectors.toUnmodifiableSet()));
-
-      var topicPartitionLeaderReplicaTable =
-          replicaProgress.entrySet().stream()
-              .map(
-                  x ->
-                      Map.entry(
-                          x.getKey(),
-                          x.getValue().stream().filter(Replica::leader).findFirst().orElseThrow()))
-              .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
-
-      topicPartitionToTrack.stream()
-          .map(TopicPartition::topic)
-          .distinct()
-          .sorted()
-          .forEachOrdered(
-              topic -> {
-                var partitionReplicas =
-                    replicaProgress.entrySet().stream()
-                        .filter(tpr -> tpr.getKey().topic().equals(topic))
-                        .filter(tpr -> topicPartitionToTrack.contains(tpr.getKey()))
-                        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
-
-                System.out.printf("  Topic \"%s\":%n", topic);
-
-                partitionReplicas.keySet().stream()
-                    .map(TopicPartition::partition)
-                    .distinct()
-                    .sorted()
-                    .forEachOrdered(
-                        partition -> {
-                          TopicPartition tp = TopicPartition.of(topic, partition);
-                          Replica leaderReplica = topicPartitionLeaderReplicaTable.get(tp);
-                          List<Replica> thisReplicas = partitionReplicas.get(tp);
-
-                          // printing info
-                          System.out.printf("  | Partition %d:%n", partition);
-                          thisReplicas.stream()
-                              .map(
-                                  replica ->
-                                      Map.entry(
-                                          new TopicPartitionReplica(
-                                              topic, partition, replica.broker()),
-                                          replica))
-                              .map(
-                                  entry ->
-                                      Map.entry(
-                                          entry.getValue(),
-                                          new ProgressInfo(
-                                              DataSize.Byte.of(leaderReplica.size()),
-                                              DataSize.Byte.of(
-                                                  previousCheckedSize.getOrDefault(
-                                                      entry.getKey(), entry.getValue().size())),
-                                              DataSize.Byte.of(entry.getValue().size()),
-                                              argument.interval)))
-                              .map(
-                                  entry ->
-                                      String.format(
-                                          "replica on broker %3d => %s",
-                                          entry.getKey().broker(),
-                                          formatString(entry.getKey(), entry.getValue())))
-                              .map(s -> String.format("  | | %s", s))
-                              .forEachOrdered(System.out::println);
-
-                          // update previous size
-                          thisReplicas.stream()
-                              .map(
-                                  replica ->
-                                      Map.entry(
-                                          new TopicPartitionReplica(
-                                              topic, partition, replica.broker()),
-                                          replica))
-                              .forEach(
-                                  entry ->
-                                      previousCheckedSize.put(
-                                          entry.getKey(), entry.getValue().size()));
-                        });
-              });
-
-      // remove synced topic-partition-replica
-      var topicPartitionFinished =
-          replicaProgress.entrySet().stream()
-              .filter(x -> x.getValue().stream().allMatch(Replica::inSync))
-              .map(Map.Entry::getKey)
-              .collect(Collectors.toSet());
-      topicPartitionToTrack.removeAll(topicPartitionFinished);
-
-      if (topicPartitionToTrack.isEmpty()) {
-        System.out.println("  Every replica is synced.");
-      }
-      System.out.println();
-
-      // attempts to discover any non-synced replica if flag --keep-track is used
-      if (argument.keepTrack) {
-        // find new non-synced topic-partition
-        Set<TopicPartition> nonSyncedTopicPartition =
-            findNonSyncedTopicPartition(topicAdmin, topicToTrack.get());
-        // add all the non-synced topic-partition into tracking
-        topicPartitionToTrack.addAll(nonSyncedTopicPartition);
-        // remove previous progress from size map
-        previousCheckedSize.entrySet().stream()
-            .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))
-            .entrySet()
-            .stream()
-            .filter(
-                tpr ->
-                    !nonSyncedTopicPartition.contains(
-                        TopicPartition.of(tpr.getKey().topic(), tpr.getKey().partition())))
-            .distinct()
-            .forEach(tpr -> previousCheckedSize.remove(tpr.getKey()));
-      }
-
-      long expectedWaitNs = argument.interval.toNanos();
-      long elapsedNs = (System.nanoTime() - startTime);
-      Utils.sleep(Duration.ofNanos(Math.max(expectedWaitNs - elapsedNs, 0)));
-    }
-  }
-
-  private static String formatString(Replica replica, ProgressInfo value) {
-    return Stream.of(
-            Optional.ofNullable(value.progressBar()),
-            Optional.ofNullable(replica.leader() ? null : value.dataRateString()),
-            Optional.ofNullable(
-                replica.leader() || replica.inSync()
-                    ? null
-                    : "(" + value.estimateFinishTimeString() + ")"),
-            Optional.ofNullable(replicaDescriptor(replica, value)))
-        .flatMap(Optional::stream)
-        .collect(Collectors.joining(" "));
-  }
-
-  static Set<TopicPartition> findNonSyncedTopicPartition(
-      Admin topicAdmin, Set<String> topicToTrack) {
-    return topicAdmin.replicas(topicToTrack).entrySet().stream()
-        .filter(tpr -> tpr.getValue().stream().anyMatch(replica -> !replica.inSync()))
-        .map(Map.Entry::getKey)
-        .collect(Collectors.toSet());
-  }
-
-  static class ProgressInfo {
-    private final DataSize leaderSize;
-    private final DataSize previousSize;
-    private final DataSize currentSize;
-    private final Duration interval;
-
-    ProgressInfo(
-        DataSize leaderSize, DataSize previousSize, DataSize currentSize, Duration interval) {
-      this.leaderSize = leaderSize;
-      this.previousSize = previousSize;
-      this.currentSize = currentSize;
-      this.interval = interval;
-    }
-
-    public double progress() {
-      var currentBit = currentSize.measurement(DataUnit.Bit).doubleValue();
-      var leaderBit = leaderSize.measurement(DataUnit.Bit).doubleValue();
-      if (currentBit == leaderBit) return 100;
-      // The Admin API doesn't offer any concurrency guarantee for these log size numbers. It is
-      // hard ensure the relationship between these two numbers, So we just assume the follower log
-      // size might somehow be larger than the leader log.
-      // TODO: prove the below condition is impossible to occur so it is safe to remove it.
-      if (currentBit > leaderBit) return Double.NaN;
-      return currentBit / leaderBit * 100.0;
-    }
-
-    public String progressBar() {
-      final int totalBlocks = 20;
-      final int filledBlocks =
-          (int) Math.min(totalBlocks, Math.floor(0.2 + progress() / (100.0 / totalBlocks)));
-      final int emptyBlocks = totalBlocks - filledBlocks;
-
-      return String.format(
-          "[%s%s] %6.2f%%",
-          String.join("", Collections.nCopies(filledBlocks, "#")),
-          String.join("", Collections.nCopies(emptyBlocks, " ")),
-          progress());
-    }
-
-    public DataRate dataRate() {
-      if (isProgressFallback()) {
-        // log retention/compaction occurred, we don't know the actual data rate at this moment.
-        return DataRate.ZERO;
-      } else {
-        return currentSize.subtract(previousSize).dataRate(interval);
-      }
-    }
-
-    public double dataRate(DataUnit dataUnit, ChronoUnit chronoUnit) {
-      return dataRate().toBigDecimal(dataUnit, chronoUnit).doubleValue();
-    }
-
-    boolean isStalled() {
-      return currentSize.equals(previousSize);
-    }
-
-    /**
-     * For some reason the log size decreased. This might occur due to various reasons, for example:
-     * log retention/compaction.
-     */
-    boolean isProgressFallback() {
-      return currentSize.compareTo(previousSize) < 0;
-    }
-
-    /**
-     * For some reason the current log size is smaller than leader size. This might occur due to
-     * various reasons, for example: race condition between API fetching data.
-     */
-    boolean isProgressOverflow() {
-      return currentSize.compareTo(leaderSize) > 0;
-    }
-
-    /**
-     * Estimated the finish time, return a negative duration if we can't infer the actual finished
-     * time. This might happen due to the following reason:
-     *
-     * <ol>
-     *   <li>The replica syncing progress stalled.
-     *   <li>Log retention/compaction: cause the actual log size changed.
-     *   <li>Significant log size changed between replica data RPC call. The size of a log is a
-     *       moving target. It is hard to find the actual size without strong synchronization. But
-     *       doing that probably hurts performance and offers not much value to the user.
-     * </ol>
-     *
-     * @return a {@link Duration} represent the estimated finish time, negative if progress is
-     *     stalled.
-     */
-    public Duration estimateFinishTime() {
-      if (isStalled() || isProgressFallback() || isProgressOverflow())
-        return Duration.ofSeconds(-1);
-      else {
-        final double leaderByte = leaderSize.measurement(DataUnit.Byte).doubleValue();
-        final double currentByte = currentSize.measurement(DataUnit.Byte).doubleValue();
-        final double estimateFinishTime =
-            ((leaderByte - currentByte) / dataRate(DataUnit.Byte, ChronoUnit.SECONDS));
-        return Duration.ofSeconds((long) estimateFinishTime);
-      }
-    }
-
-    public String estimateFinishTimeString() {
-      Duration estimatedTime = estimateFinishTime();
-      return estimatedTime.isNegative()
-          ? "unknown"
-          : estimatedTime.toSeconds() == 0
-              ? "about now"
-              : Stream.of(
-                      Map.entry(estimatedTime.toHoursPart(), "h"),
-                      Map.entry(estimatedTime.toMinutesPart(), "m"),
-                      Map.entry(estimatedTime.toSecondsPart(), "s"))
-                  .dropWhile(x -> x.getKey() == 0)
-                  .map(x -> x.getKey().toString() + x.getValue())
-                  .collect(Collectors.joining(" ", "", " estimated"));
-    }
-
-    public String dataRateString() {
-      if (isProgressFallback()) {
-        return dataRate().toString(ChronoUnit.SECONDS).replaceAll("[+\\-\\d.]+", "?");
-      } else {
-        return dataRate().toString(ChronoUnit.SECONDS);
-      }
-    }
-
-    @Override
-    public String toString() {
-      if (estimateFinishTime().isZero() && isStalled()) return dataRateString();
-      else return String.format("%s (%s)", dataRateString(), estimateFinishTimeString());
-    }
-  }
-
-  static String replicaDescriptor(Replica replica, ProgressInfo progressInfo) {
-    var logShrinkage = progressInfo.isProgressFallback() | progressInfo.isProgressOverflow();
-    return Stream.of(
-            Optional.ofNullable(replica.leader() ? "leader" : null),
-            Optional.ofNullable(replica.inSync() ? "synced" : null),
-            Optional.ofNullable(replica.lag() > 0 ? "lagged" : null),
-            Optional.ofNullable(logShrinkage ? "log-shrunk" : null))
-        .flatMap(Optional::stream)
-        .collect(Collectors.joining(", ", "[", "]"));
-  }
-
-  static class Argument extends org.astraea.app.argument.Argument {
-
-    @Parameter(
-        names = {"--topics"},
-        description = "String: topics to track, use all non-synced topics by default",
-        validateWith = NonEmptyStringField.class)
-    public Set<String> topics = Set.of();
-
-    @Parameter(
-        names = {"--track"},
-        description =
-            "Boolean: keep track even if all the replicas are synced, also attempts to discover any non-synced replicas")
-    public boolean keepTrack = false;
-
-    @Parameter(
-        names = {"--interval"},
-        description =
-            "Time: the time interval between replica state check, support multiple time unit like 10s, 500ms and 100us. "
-                + "If no time unit specified, second unit will be used.",
-        validateWith = DurationField.class,
-        converter = DurationField.class)
-    public Duration interval = Duration.ofSeconds(1);
-  }
-}
diff --git a/app/src/main/java/org/astraea/app/balancer/BalancerUtils.java b/app/src/main/java/org/astraea/app/balancer/BalancerUtils.java
index 070d0e17..7f8ecdd6 100644
--- a/app/src/main/java/org/astraea/app/balancer/BalancerUtils.java
+++ b/app/src/main/java/org/astraea/app/balancer/BalancerUtils.java
@@ -20,7 +20,6 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Function;
@@ -30,6 +29,7 @@ import java.util.stream.IntStream;
 import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.admin.NodeInfo;
+import org.astraea.app.admin.Replica;
 import org.astraea.app.admin.ReplicaInfo;
 import org.astraea.app.admin.TopicPartition;
 import org.astraea.app.balancer.log.ClusterLogAllocation;
@@ -38,6 +38,55 @@ import org.astraea.app.metrics.HasBeanObject;
 
 public class BalancerUtils {
 
+  /**
+   * Update the replicas of ClusterInfo according to ClusterLogAllocation. Noted that only "broker"
+   * and "data folder" get updated. The replicas matched to nothing from ClusterLogAllocation won't
+   * get any update.
+   *
+   * @param clusterInfo to get updated
+   * @param allocation offers new host and data folder
+   * @return new cluster info
+   */
+  public static ClusterInfo update(ClusterInfo clusterInfo, ClusterLogAllocation allocation) {
+    var newReplicas =
+        clusterInfo.replicas().stream()
+            .collect(Collectors.groupingBy(r -> TopicPartition.of(r.topic(), r.partition())))
+            .entrySet()
+            .stream()
+            .flatMap(
+                entry -> {
+                  var lps = allocation.logPlacements(entry.getKey());
+                  var replicas =
+                      entry.getValue().stream()
+                          .map(r -> (Replica) r)
+                          .collect(Collectors.toUnmodifiableList());
+                  return IntStream.range(0, replicas.size())
+                      .mapToObj(
+                          index -> {
+                            var previous = replicas.get(index);
+                            // return previous replica due to no new information
+                            if (index >= lps.size()) return previous;
+                            var lp = lps.get(index);
+                            return (ReplicaInfo)
+                                Replica.of(
+                                    previous.topic(),
+                                    previous.partition(),
+                                    clusterInfo.node(lp.broker()),
+                                    previous.lag(),
+                                    previous.size(),
+                                    index == 0,
+                                    previous.inSync(),
+                                    previous.isFuture(),
+                                    previous.isOffline(),
+                                    previous.isPreferredLeader(),
+                                    lp.dataFolder());
+                          });
+                })
+            .collect(Collectors.toUnmodifiableList());
+
+    return ClusterInfo.of(clusterInfo.nodes(), newReplicas);
+  }
+
   /**
    * Create a {@link ClusterInfo} with its log placement replaced by {@link ClusterLogAllocation}.
    * Every log will be marked as online & synced. Based on the given content in {@link
@@ -54,6 +103,35 @@ public class BalancerUtils {
   public static ClusterInfo merge(ClusterInfo clusterInfo, ClusterLogAllocation allocation) {
     return new ClusterInfo() {
       // TODO: maybe add a field to tell if this cluster info is mocked.
+      private final Map<Integer, NodeInfo> nodeIdMap =
+          nodes().stream().collect(Collectors.toUnmodifiableMap(NodeInfo::id, Function.identity()));
+      private final List<ReplicaInfo> replicas =
+          allocation.topicPartitions().stream()
+              .map(tp -> Map.entry(tp, allocation.logPlacements(tp)))
+              .flatMap(
+                  entry -> {
+                    var tp = entry.getKey();
+                    var logs = entry.getValue();
+
+                    return IntStream.range(0, logs.size())
+                        .mapToObj(
+                            i ->
+                                // TODO: too many fake data!!! we should use another data structure
+                                // https://github.com/skiptests/astraea/issues/526
+                                Replica.of(
+                                    tp.topic(),
+                                    tp.partition(),
+                                    nodeIdMap.get(logs.get(i).broker()),
+                                    0,
+                                    -1,
+                                    i == 0,
+                                    true,
+                                    false,
+                                    false,
+                                    false,
+                                    logs.get(i).dataFolder()));
+                  })
+              .collect(Collectors.toUnmodifiableList());
 
       @Override
       public List<NodeInfo> nodes() {
@@ -61,38 +139,8 @@ public class BalancerUtils {
       }
 
       @Override
-      public Set<String> topics() {
-        return allocation.topicPartitions().stream()
-            .map(TopicPartition::topic)
-            .collect(Collectors.toUnmodifiableSet());
-      }
-
-      @Override
-      public List<ReplicaInfo> replicas(String topic) {
-        var nodeIdMap =
-            nodes().stream()
-                .collect(Collectors.toUnmodifiableMap(NodeInfo::id, Function.identity()));
-        return allocation.topicPartitions().stream()
-            .filter(tp -> tp.topic().equals(topic))
-            .map(tp -> Map.entry(tp, allocation.logPlacements(tp)))
-            .flatMap(
-                entry -> {
-                  var tp = entry.getKey();
-                  var logs = entry.getValue();
-
-                  return IntStream.range(0, logs.size())
-                      .mapToObj(
-                          i ->
-                              ReplicaInfo.of(
-                                  tp.topic(),
-                                  tp.partition(),
-                                  nodeIdMap.get(logs.get(i).broker()),
-                                  i == 0,
-                                  true,
-                                  false,
-                                  logs.get(i).logDirectory().orElse(null)));
-                })
-            .collect(Collectors.toUnmodifiableList());
+      public List<ReplicaInfo> replicas() {
+        return replicas;
       }
     };
   }
diff --git a/app/src/main/java/org/astraea/app/balancer/executor/RebalanceAdminImpl.java b/app/src/main/java/org/astraea/app/balancer/executor/RebalanceAdminImpl.java
index 1f9d32ac..61980a11 100644
--- a/app/src/main/java/org/astraea/app/balancer/executor/RebalanceAdminImpl.java
+++ b/app/src/main/java/org/astraea/app/balancer/executor/RebalanceAdminImpl.java
@@ -21,8 +21,8 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Predicate;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import org.astraea.app.admin.Admin;
 import org.astraea.app.admin.ClusterInfo;
@@ -56,7 +56,7 @@ class RebalanceAdminImpl implements RebalanceAdmin {
   private List<LogPlacement> fetchCurrentPlacement(TopicPartition topicPartition) {
     ensureTopicPermitted(topicPartition.topic());
     return admin.replicas(Set.of(topicPartition.topic())).get(topicPartition).stream()
-        .map(replica -> LogPlacement.of(replica.broker(), replica.path()))
+        .map(replica -> LogPlacement.of(replica.nodeInfo().id(), replica.dataFolder()))
         .collect(Collectors.toUnmodifiableList());
   }
 
@@ -89,10 +89,7 @@ class RebalanceAdminImpl implements RebalanceAdmin {
     final var declareMap =
         preferredPlacements.stream()
             .filter(futurePlacement -> !currentBrokerAllocation.contains(futurePlacement.broker()))
-            .filter(futurePlacement -> futurePlacement.logDirectory().isPresent())
-            .collect(
-                Collectors.toUnmodifiableMap(
-                    LogPlacement::broker, x -> x.logDirectory().orElseThrow()));
+            .collect(Collectors.toUnmodifiableMap(LogPlacement::broker, LogPlacement::dataFolder));
 
     admin
         .migrator()
@@ -129,10 +126,7 @@ class RebalanceAdminImpl implements RebalanceAdmin {
     var forCrossDirMigration =
         expectedPlacement.stream()
             .filter(placement -> currentReplicaBrokers.contains(placement.broker()))
-            .filter(placement -> placement.logDirectory().isPresent())
-            .collect(
-                Collectors.toUnmodifiableMap(
-                    LogPlacement::broker, placement -> placement.logDirectory().orElseThrow()));
+            .collect(Collectors.toUnmodifiableMap(LogPlacement::broker, LogPlacement::dataFolder));
     admin
         .migrator()
         .partition(topicPartition.topic(), topicPartition.partition())
@@ -160,34 +154,17 @@ class RebalanceAdminImpl implements RebalanceAdmin {
     ensureTopicPermitted(log.topic());
 
     return CompletableFuture.supplyAsync(
-        () -> {
-          // due to the state consistency issue in Kafka broker design. the cluster state returned
-          // from the API might bounce between the `old state` and the `new state` during the very
-          // beginning and accomplishment of the cluster state alteration API. to fix this we use
-          // debounce technique, to ensure the target condition is held over a few successive tries,
-          // which mean the cluster state alteration is considered stable.
-          var debounce = 2;
-          var endTime = getEndTime(timeout);
-          while (!Thread.currentThread().isInterrupted()) {
-            boolean synced =
+        debounceCheck(
+            timeout,
+            () ->
                 admin.replicas(Set.of(log.topic())).entrySet().stream()
                     .filter(x -> x.getKey().partition() == log.partition())
                     .filter(x -> x.getKey().topic().equals(log.topic()))
                     .flatMap(x -> x.getValue().stream())
-                    .filter(x -> x.broker() == log.brokerId())
+                    .filter(x -> x.nodeInfo().id() == log.brokerId())
                     .findFirst()
                     .map(x -> x.inSync() && !x.isFuture())
-                    .orElse(false);
-            // debounce & retrial interval
-            Utils.sleep(retrialTime.get());
-            debounce = synced ? (debounce - 1) : 2;
-            // synced
-            if (synced && debounce <= 0) return true;
-            // timeout
-            if (System.currentTimeMillis() > endTime) return false;
-          }
-          return false;
-        });
+                    .orElse(false)));
   }
 
   @Override
@@ -196,16 +173,9 @@ class RebalanceAdminImpl implements RebalanceAdmin {
     ensureTopicPermitted(topicPartition.topic());
 
     return CompletableFuture.supplyAsync(
-        () -> {
-          // due to the state consistency issue in Kafka broker design. the cluster state returned
-          // from the API might bounce between the `old state` and the `new state` during the very
-          // beginning and accomplishment of the cluster state alteration API. to fix this we use
-          // debounce technique, to ensure the target condition is held over a few successive tries,
-          // which mean the cluster state alteration is considered stable.
-          var debounce = 2;
-          var endTime = getEndTime(timeout);
-          while (!Thread.currentThread().isInterrupted()) {
-            var synced =
+        debounceCheck(
+            timeout,
+            () ->
                 admin.replicas(Set.of(topicPartition.topic())).entrySet().stream()
                     .filter(x -> x.getKey().equals(topicPartition))
                     .findFirst()
@@ -217,19 +187,33 @@ class RebalanceAdminImpl implements RebalanceAdmin {
                                   .filter(Replica::isPreferredLeader)
                                   .findFirst()
                                   .orElseThrow();
-                          return preferred.leader();
+                          return preferred.isLeader();
                         })
-                    .orElseThrow();
-            // debounce & retrial interval
-            Utils.sleep(retrialTime.get());
-            debounce = synced ? (debounce - 1) : 2;
-            // synced
-            if (synced && debounce <= 0) return true;
-            // timeout
-            if (System.currentTimeMillis() > endTime) return false;
-          }
-          return false;
-        });
+                    .orElseThrow()));
+  }
+
+  private Supplier<Boolean> debounceCheck(Duration timeout, Supplier<Boolean> testDone) {
+    var debounceInitialCount = 10;
+    return () -> {
+      // due to the state consistency issue in Kafka broker design. the cluster state returned
+      // from the API might bounce between the `old state` and the `new state` during the very
+      // beginning and accomplishment of the cluster state alteration API. to fix this we use
+      // debounce technique, to ensure the target condition is held over a few successive tries,
+      // which mean the cluster state alteration is considered stable.
+      var debounce = debounceInitialCount;
+      var endTime = getEndTime(timeout);
+      while (!Thread.currentThread().isInterrupted()) {
+        // debounce & retrial interval
+        Utils.sleep(Duration.ofMillis(100));
+        var isDone = testDone.get();
+        debounce = isDone ? (debounce - 1) : debounceInitialCount;
+        // synced
+        if (isDone && debounce <= 0) return true;
+        // timeout
+        if (System.currentTimeMillis() > endTime) return false;
+      }
+      return false;
+    };
   }
 
   @Override
@@ -251,12 +235,4 @@ class RebalanceAdminImpl implements RebalanceAdmin {
   public Predicate<String> topicFilter() {
     return topicFilter;
   }
-
-  private static final AtomicReference<Duration> retrialTime =
-      new AtomicReference<>(Duration.ofSeconds(1));
-
-  // visible for test
-  static void changeRetrialTime(Duration newDebounceTime) {
-    retrialTime.set(newDebounceTime);
-  }
 }
diff --git a/app/src/main/java/org/astraea/app/balancer/log/ClusterLogAllocation.java b/app/src/main/java/org/astraea/app/balancer/log/ClusterLogAllocation.java
index fe96cd54..e4c649a1 100644
--- a/app/src/main/java/org/astraea/app/balancer/log/ClusterLogAllocation.java
+++ b/app/src/main/java/org/astraea/app/balancer/log/ClusterLogAllocation.java
@@ -16,7 +16,6 @@
  */
 package org.astraea.app.balancer.log;
 
-import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.HashMap;
@@ -27,6 +26,7 @@ import java.util.Set;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import org.astraea.app.admin.ClusterInfo;
+import org.astraea.app.admin.Replica;
 import org.astraea.app.admin.ReplicaInfo;
 import org.astraea.app.admin.TopicPartition;
 
@@ -39,35 +39,27 @@ public interface ClusterLogAllocation {
 
   static ClusterLogAllocation of(ClusterInfo clusterInfo) {
     return of(
-        clusterInfo.topics().stream()
-            .map(clusterInfo::replicas)
-            .flatMap(Collection::stream)
-            .collect(
-                Collectors.groupingBy(
-                    replica ->
-                        TopicPartition.of(replica.topic(), Integer.toString(replica.partition()))))
+        clusterInfo.replicas().stream()
+            .filter(r -> r instanceof Replica)
+            .map(r -> (Replica) r)
+            .collect(Collectors.groupingBy(r -> TopicPartition.of(r.topic(), r.partition())))
             .entrySet()
             .stream()
-            .map(
-                (entry) -> {
-                  // validate if the given log placements are valid
-                  if (entry.getValue().stream().filter(ReplicaInfo::isLeader).count() != 1)
-                    throw new IllegalArgumentException(
-                        "The " + entry.getKey() + " leader count mismatch 1.");
-
-                  final var topicPartition = entry.getKey();
-                  final var logPlacements =
-                      entry.getValue().stream()
+            .collect(
+                Collectors.toMap(
+                    Map.Entry::getKey,
+                    entry -> {
+                      // validate if the given log placements are valid
+                      if (entry.getValue().stream().filter(ReplicaInfo::isLeader).count() != 1)
+                        throw new IllegalArgumentException(
+                            "The " + entry.getKey() + " leader count mismatch 1.");
+                      return entry.getValue().stream()
                           .sorted(Comparator.comparingInt(replica -> replica.isLeader() ? 0 : 1))
                           .map(
                               replica ->
-                                  LogPlacement.of(
-                                      replica.nodeInfo().id(), replica.dataFolder().orElse(null)))
+                                  LogPlacement.of(replica.nodeInfo().id(), replica.dataFolder()))
                           .collect(Collectors.toList());
-
-                  return Map.entry(topicPartition, logPlacements);
-                })
-            .collect(Collectors.toConcurrentMap(Map.Entry::getKey, Map.Entry::getValue)));
+                    })));
   }
 
   static ClusterLogAllocation of(Map<TopicPartition, List<LogPlacement>> allocation) {
@@ -150,10 +142,7 @@ public interface ClusterLogAllocation {
                   .forEach(
                       log ->
                           stringBuilder.append(
-                              String.format(
-                                  "(%s, %s) ",
-                                  log.broker(),
-                                  log.logDirectory().orElse("log dir not specified"))));
+                              String.format("(%s, %s) ", log.broker(), log.dataFolder())));
 
               stringBuilder.append(System.lineSeparator());
             });
diff --git a/app/src/main/java/org/astraea/app/balancer/log/LogPlacement.java b/app/src/main/java/org/astraea/app/balancer/log/LogPlacement.java
index 9425707f..ed063f8d 100644
--- a/app/src/main/java/org/astraea/app/balancer/log/LogPlacement.java
+++ b/app/src/main/java/org/astraea/app/balancer/log/LogPlacement.java
@@ -17,7 +17,6 @@
 package org.astraea.app.balancer.log;
 
 import java.util.List;
-import java.util.Optional;
 import java.util.stream.IntStream;
 
 /** This class describe the placement state of one kafka log. */
@@ -25,7 +24,7 @@ public interface LogPlacement {
 
   int broker();
 
-  Optional<String> logDirectory();
+  String dataFolder();
 
   static boolean isMatch(List<LogPlacement> sourcePlacements, List<LogPlacement> targetPlacements) {
     if (sourcePlacements.size() != targetPlacements.size()) return false;
@@ -42,34 +41,17 @@ public interface LogPlacement {
         IntStream.range(0, sourcePlacements.size())
             .allMatch(
                 index ->
-                    meetLogDirectoryMigrationRequirement(
-                        sourcePlacements.get(index).logDirectory(),
-                        targetPlacements.get(index).logDirectory()));
+                    sourcePlacements
+                        .get(index)
+                        .dataFolder()
+                        .equals(targetPlacements.get(index).dataFolder()));
     //noinspection RedundantIfStatement
     if (!logDirectoryMatch) return false;
 
     return true;
   }
 
-  @SuppressWarnings("OptionalUsedAsFieldOrParameterType")
-  private static boolean meetLogDirectoryMigrationRequirement(
-      Optional<String> sourceDir, Optional<String> targetDir) {
-    // don't care which log directory will eventually be in the destination.
-    if (targetDir.isEmpty()) return true;
-
-    // we care which data directory the target will eventually be, but we don't know.
-    if (sourceDir.isEmpty()) return false;
-
-    // both candidate is specified, if and only if two paths match, will consider as a requirement
-    // meet.
-    return sourceDir.get().equals(targetDir.get());
-  }
-
-  static LogPlacement of(int broker) {
-    return of(broker, null);
-  }
-
-  static LogPlacement of(int broker, String logDirectory) {
+  static LogPlacement of(int broker, String dataFolder) {
     return new LogPlacement() {
       @Override
       public int broker() {
@@ -77,22 +59,22 @@ public interface LogPlacement {
       }
 
       @Override
-      public Optional<String> logDirectory() {
-        return Optional.ofNullable(logDirectory);
+      public String dataFolder() {
+        return dataFolder;
       }
 
       @Override
       public boolean equals(Object obj) {
         if (obj instanceof LogPlacement) {
           final var that = (LogPlacement) obj;
-          return this.broker() == that.broker() && this.logDirectory().equals(that.logDirectory());
+          return this.broker() == that.broker() && this.dataFolder().equals(that.dataFolder());
         }
         return false;
       }
 
       @Override
       public String toString() {
-        return "LogPlacement{broker=" + broker() + " logDir=" + logDirectory() + "}";
+        return "LogPlacement{broker=" + broker() + " dataFolder=" + dataFolder() + "}";
       }
     };
   }
diff --git a/app/src/main/java/org/astraea/app/common/json/OptionalIntSerializer.java b/app/src/main/java/org/astraea/app/common/json/OptionalIntSerializer.java
deleted file mode 100644
index 6c96ca6d..00000000
--- a/app/src/main/java/org/astraea/app/common/json/OptionalIntSerializer.java
+++ /dev/null
@@ -1,30 +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.common.json;
-
-import com.google.gson.JsonElement;
-import com.google.gson.JsonSerializationContext;
-import com.google.gson.JsonSerializer;
-import java.lang.reflect.Type;
-import java.util.OptionalInt;
-
-public class OptionalIntSerializer implements JsonSerializer<OptionalInt> {
-  @Override
-  public JsonElement serialize(OptionalInt src, Type typeOfSrc, JsonSerializationContext context) {
-    return src.isPresent() ? context.serialize(src.getAsInt()) : null;
-  }
-}
diff --git a/app/src/main/java/org/astraea/app/common/json/OptionalSerializer.java b/app/src/main/java/org/astraea/app/common/json/OptionalSerializer.java
deleted file mode 100644
index ddcdf8c9..00000000
--- a/app/src/main/java/org/astraea/app/common/json/OptionalSerializer.java
+++ /dev/null
@@ -1,30 +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.common.json;
-
-import com.google.gson.JsonElement;
-import com.google.gson.JsonSerializationContext;
-import com.google.gson.JsonSerializer;
-import java.lang.reflect.Type;
-import java.util.Optional;
-
-public class OptionalSerializer implements JsonSerializer<Optional<?>> {
-  @Override
-  public JsonElement serialize(Optional<?> src, Type typeOfSrc, JsonSerializationContext context) {
-    return src.map(context::serialize).orElse(null);
-  }
-}
diff --git a/app/src/main/java/org/astraea/app/cost/NodeTopicSizeCost.java b/app/src/main/java/org/astraea/app/cost/NodeTopicSizeCost.java
index 07686a86..0db27e09 100644
--- a/app/src/main/java/org/astraea/app/cost/NodeTopicSizeCost.java
+++ b/app/src/main/java/org/astraea/app/cost/NodeTopicSizeCost.java
@@ -21,10 +21,14 @@ import java.util.Optional;
 import java.util.stream.Collectors;
 import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
+import org.astraea.app.admin.NodeInfo;
+import org.astraea.app.admin.Replica;
 import org.astraea.app.metrics.broker.LogMetrics;
 import org.astraea.app.metrics.collector.Fetcher;
 
-public class NodeTopicSizeCost implements HasBrokerCost {
+public class NodeTopicSizeCost implements HasBrokerCost, HasClusterCost {
+  private final Dispersion dispersion = Dispersion.correlationCoefficient();
+
   @Override
   public Optional<Fetcher> fetcher() {
     return Optional.of(LogMetrics.Log.SIZE::fetch);
@@ -47,4 +51,21 @@ public class NodeTopicSizeCost implements HasBrokerCost {
                             .sum()));
     return () -> result;
   }
+
+  @Override
+  public ClusterCost clusterCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
+    var brokerCost =
+        clusterInfo.nodes().stream()
+            .collect(
+                Collectors.toMap(
+                    NodeInfo::id,
+                    nodeInfo ->
+                        clusterInfo.replicas().stream()
+                            .filter(r -> r.nodeInfo().id() == nodeInfo.id())
+                            .mapToLong(r -> r instanceof Replica ? ((Replica) r).size() : 0)
+                            .sum()));
+    return () ->
+        dispersion.calculate(
+            brokerCost.values().stream().map(v -> (double) v).collect(Collectors.toList()));
+  }
 }
diff --git a/app/src/main/java/org/astraea/app/web/BalancerHandler.java b/app/src/main/java/org/astraea/app/web/BalancerHandler.java
index ed8347d4..fe8d699f 100644
--- a/app/src/main/java/org/astraea/app/web/BalancerHandler.java
+++ b/app/src/main/java/org/astraea/app/web/BalancerHandler.java
@@ -16,31 +16,40 @@
  */
 package org.astraea.app.web;
 
+import java.util.Arrays;
 import java.util.Comparator;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 import org.astraea.app.admin.Admin;
 import org.astraea.app.admin.ClusterBean;
+import org.astraea.app.admin.Replica;
+import org.astraea.app.admin.TopicPartitionReplica;
 import org.astraea.app.balancer.BalancerUtils;
 import org.astraea.app.balancer.RebalancePlanProposal;
 import org.astraea.app.balancer.generator.RebalancePlanGenerator;
 import org.astraea.app.balancer.log.ClusterLogAllocation;
 import org.astraea.app.balancer.log.LogPlacement;
 import org.astraea.app.cost.HasClusterCost;
-import org.astraea.app.cost.ReplicaLeaderCost;
+import org.astraea.app.cost.NodeTopicSizeCost;
 
 class BalancerHandler implements Handler {
 
   static String LIMIT_KEY = "limit";
 
+  static String TOPICS_KEY = "topics";
+
   static int LIMIT_DEFAULT = 10000;
   private final Admin admin;
   private final RebalancePlanGenerator generator = RebalancePlanGenerator.random(30);
-  private final HasClusterCost costFunction;
+  final HasClusterCost costFunction;
 
   BalancerHandler(Admin admin) {
-    this(admin, new ReplicaLeaderCost());
+    this(admin, new NodeTopicSizeCost());
   }
 
   BalancerHandler(Admin admin, HasClusterCost costFunction) {
@@ -50,14 +59,19 @@ class BalancerHandler implements Handler {
 
   @Override
   public Response get(Channel channel) {
+    var topics =
+        Optional.ofNullable(channel.queries().get(TOPICS_KEY))
+            .map(s -> (Set<String>) new HashSet<>(Arrays.asList(s.split(","))))
+            .orElseGet(() -> admin.topicNames(false));
     var clusterInfo = admin.clusterInfo();
-    var clusterAllocation = ClusterLogAllocation.of(clusterInfo);
     var cost = costFunction.clusterCost(clusterInfo, ClusterBean.EMPTY).value();
     var limit =
         Integer.parseInt(channel.queries().getOrDefault(LIMIT_KEY, String.valueOf(LIMIT_DEFAULT)));
+    // generate migration plans only for specify topics
+    var targetAllocations = ClusterLogAllocation.of(admin.clusterInfo(topics));
     var planAndCost =
         generator
-            .generate(admin.brokerFolders(), clusterAllocation)
+            .generate(admin.brokerFolders(), targetAllocations)
             .limit(limit)
             .map(RebalancePlanProposal::rebalancePlan)
             .map(
@@ -65,7 +79,7 @@ class BalancerHandler implements Handler {
                     Map.entry(
                         cla,
                         costFunction
-                            .clusterCost(BalancerUtils.merge(clusterInfo, cla), ClusterBean.EMPTY)
+                            .clusterCost(BalancerUtils.update(clusterInfo, cla), ClusterBean.EMPTY)
                             .value()))
             .filter(e -> e.getValue() <= cost)
             .min(Comparator.comparingDouble(Map.Entry::getValue));
@@ -79,21 +93,32 @@ class BalancerHandler implements Handler {
             .map(
                 entry ->
                     ClusterLogAllocation.findNonFulfilledAllocation(
-                            clusterAllocation, entry.getKey())
+                            targetAllocations, entry.getKey())
                         .stream()
                         .map(
                             tp ->
                                 new Change(
                                     tp.topic(),
                                     tp.partition(),
-                                    placements(clusterAllocation.logPlacements(tp)),
-                                    placements(entry.getKey().logPlacements(tp))))
+                                    // only log the size from source replicas
+                                    placements(
+                                        targetAllocations.logPlacements(tp),
+                                        l ->
+                                            clusterInfo
+                                                .replica(
+                                                    TopicPartitionReplica.of(
+                                                        tp.topic(), tp.partition(), l.broker()))
+                                                .map(r -> ((Replica) r).size())
+                                                .orElse(null)),
+                                    placements(entry.getKey().logPlacements(tp), ignored -> null)))
                         .collect(Collectors.toUnmodifiableList()))
             .orElse(List.of()));
   }
 
-  static List<Placement> placements(List<LogPlacement> lps) {
-    return lps.stream().map(Placement::new).collect(Collectors.toUnmodifiableList());
+  static List<Placement> placements(List<LogPlacement> lps, Function<LogPlacement, Long> size) {
+    return lps.stream()
+        .map(p -> new Placement(p, size.apply(p)))
+        .collect(Collectors.toUnmodifiableList());
   }
 
   static class Placement {
@@ -101,9 +126,12 @@ class BalancerHandler implements Handler {
     final int brokerId;
     final String directory;
 
-    Placement(LogPlacement lp) {
+    final Long size;
+
+    Placement(LogPlacement lp, Long size) {
       this.brokerId = lp.broker();
-      this.directory = lp.logDirectory().orElse(null);
+      this.directory = lp.dataFolder();
+      this.size = size;
     }
   }
 
diff --git a/app/src/main/java/org/astraea/app/web/ReassignmentHandler.java b/app/src/main/java/org/astraea/app/web/ReassignmentHandler.java
index 23acb8c1..45895372 100644
--- a/app/src/main/java/org/astraea/app/web/ReassignmentHandler.java
+++ b/app/src/main/java/org/astraea/app/web/ReassignmentHandler.java
@@ -81,7 +81,7 @@ public class ReassignmentHandler implements Handler {
 
     Location(org.astraea.app.admin.Reassignment.Location location) {
       this.broker = location.broker();
-      this.path = location.path();
+      this.path = location.dataFolder();
     }
   }
 
diff --git a/app/src/main/java/org/astraea/app/web/Response.java b/app/src/main/java/org/astraea/app/web/Response.java
index 08623d9f..c2e600d1 100644
--- a/app/src/main/java/org/astraea/app/web/Response.java
+++ b/app/src/main/java/org/astraea/app/web/Response.java
@@ -17,12 +17,7 @@
 package org.astraea.app.web;
 
 import com.google.gson.Gson;
-import com.google.gson.GsonBuilder;
 import java.util.NoSuchElementException;
-import java.util.Optional;
-import java.util.OptionalInt;
-import org.astraea.app.common.json.OptionalIntSerializer;
-import org.astraea.app.common.json.OptionalSerializer;
 
 interface Response {
 
@@ -57,11 +52,7 @@ interface Response {
   default String json() {
     // TODO: manage the customized serialize/deserialize logic, see
     // https://github.com/skiptests/astraea/issues/626
-    return new GsonBuilder()
-        .registerTypeAdapter(Optional.class, new OptionalSerializer())
-        .registerTypeAdapter(OptionalInt.class, new OptionalIntSerializer())
-        .create()
-        .toJson(this);
+    return new Gson().toJson(this);
   }
 
   /**
diff --git a/app/src/main/java/org/astraea/app/web/ThrottleHandler.java b/app/src/main/java/org/astraea/app/web/ThrottleHandler.java
index c9018229..96daac6a 100644
--- a/app/src/main/java/org/astraea/app/web/ThrottleHandler.java
+++ b/app/src/main/java/org/astraea/app/web/ThrottleHandler.java
@@ -18,10 +18,7 @@ package org.astraea.app.web;
 
 import java.util.Arrays;
 import java.util.Collection;
-import java.util.Map;
 import java.util.Objects;
-import java.util.Optional;
-import java.util.OptionalInt;
 import java.util.Set;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -41,29 +38,25 @@ public class ThrottleHandler implements Handler {
   }
 
   private Response get() {
-    final var bandwidths =
+    final var brokers =
         admin.brokers().entrySet().stream()
-            .collect(
-                Collectors.toUnmodifiableMap(
-                    Map.Entry::getKey,
-                    entry -> {
-                      final var egress =
-                          entry
-                              .getValue()
-                              .value("leader.replication.throttled.rate")
-                              .map(Long::valueOf)
-                              .map(rate -> Map.entry(ThrottleBandwidths.egress, rate));
-                      final var ingress =
-                          entry
-                              .getValue()
-                              .value("follower.replication.throttled.rate")
-                              .map(Long::valueOf)
-                              .map(rate -> Map.entry(ThrottleBandwidths.ingress, rate));
-                      return Stream.of(egress, ingress)
-                          .flatMap(Optional::stream)
-                          .collect(
-                              Collectors.toUnmodifiableMap(Map.Entry::getKey, Map.Entry::getValue));
-                    }));
+            .map(
+                entry -> {
+                  final var egress =
+                      entry
+                          .getValue()
+                          .value("leader.replication.throttled.rate")
+                          .map(Long::valueOf)
+                          .orElse(null);
+                  final var ingress =
+                      entry
+                          .getValue()
+                          .value("follower.replication.throttled.rate")
+                          .map(Long::valueOf)
+                          .orElse(null);
+                  return new BrokerThrottle(entry.getKey(), ingress, egress);
+                })
+            .collect(Collectors.toUnmodifiableSet());
     final var topicConfigs = admin.topics();
     final var leaderTargets =
         topicConfigs.entrySet().stream()
@@ -87,7 +80,7 @@ public class ThrottleHandler implements Handler {
             .flatMap(Collection::stream)
             .collect(Collectors.toUnmodifiableSet());
 
-    return new ThrottleSetting(bandwidths, simplify(leaderTargets, followerTargets));
+    return new ThrottleSetting(brokers, simplify(leaderTargets, followerTargets));
   }
 
   /**
@@ -115,7 +108,7 @@ public class ThrottleHandler implements Handler {
    * the simplest form by merging any targets with a common topic/partition/replica scope throttle
    * target.
    */
-  private Set<ThrottleTarget> simplify(
+  private Set<TopicThrottle> simplify(
       Set<TopicPartitionReplica> leaders, Set<TopicPartitionReplica> followers) {
     var commonReplicas =
         leaders.stream().filter(followers::contains).collect(Collectors.toUnmodifiableSet());
@@ -124,13 +117,14 @@ public class ThrottleHandler implements Handler {
         commonReplicas.stream()
             .map(
                 replica ->
-                    new ThrottleTarget(replica.topic(), replica.partition(), replica.brokerId()));
+                    new TopicThrottle(
+                        replica.topic(), replica.partition(), replica.brokerId(), null));
     var leaderReplicas =
         leaders.stream()
             .filter(replica -> !commonReplicas.contains(replica))
             .map(
                 replica ->
-                    new ThrottleTarget(
+                    new TopicThrottle(
                         replica.topic(),
                         replica.partition(),
                         replica.brokerId(),
@@ -140,7 +134,7 @@ public class ThrottleHandler implements Handler {
             .filter(replica -> !commonReplicas.contains(replica))
             .map(
                 replica ->
-                    new ThrottleTarget(
+                    new TopicThrottle(
                         replica.topic(),
                         replica.partition(),
                         replica.brokerId(),
@@ -152,27 +146,65 @@ public class ThrottleHandler implements Handler {
 
   static class ThrottleSetting implements Response {
 
-    final Map<Integer, Map<ThrottleBandwidths, Long>> brokers;
-    final Collection<ThrottleTarget> topics;
+    final Collection<BrokerThrottle> brokers;
+    final Collection<TopicThrottle> topics;
 
-    private ThrottleSetting(
-        Map<Integer, Map<ThrottleBandwidths, Long>> brokers, Collection<ThrottleTarget> topics) {
+    ThrottleSetting(Collection<BrokerThrottle> brokers, Collection<TopicThrottle> topics) {
       this.brokers = brokers;
       this.topics = topics;
     }
   }
 
-  static class ThrottleTarget {
+  static class BrokerThrottle {
+    final int id;
+    final Long ingress;
+    final Long egress;
+
+    BrokerThrottle(int id, Long ingress, Long egress) {
+      this.id = id;
+      this.ingress = ingress;
+      this.egress = egress;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) return true;
+      if (o == null || getClass() != o.getClass()) return false;
+      BrokerThrottle that = (BrokerThrottle) o;
+      return id == that.id
+          && Objects.equals(ingress, that.ingress)
+          && Objects.equals(egress, that.egress);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(id, ingress, egress);
+    }
+
+    @Override
+    public String toString() {
+      return "BrokerThrottle{"
+          + "broker="
+          + id
+          + ", ingress="
+          + ingress
+          + ", egress="
+          + egress
+          + '}';
+    }
+  }
+
+  static class TopicThrottle {
     final String name;
-    final OptionalInt partition;
-    final OptionalInt broker;
-    final Optional<LogIdentity> type;
+    final Integer partition;
+    final Integer broker;
+    final String type;
 
     @Override
     public boolean equals(Object o) {
       if (this == o) return true;
       if (o == null || getClass() != o.getClass()) return false;
-      ThrottleTarget that = (ThrottleTarget) o;
+      TopicThrottle that = (TopicThrottle) o;
       return Objects.equals(name, that.name)
           && Objects.equals(partition, that.partition)
           && Objects.equals(broker, that.broker)
@@ -184,28 +216,31 @@ public class ThrottleHandler implements Handler {
       return Objects.hash(name, partition, broker, type);
     }
 
-    ThrottleTarget(String name, int partition, int broker) {
+    TopicThrottle(String name, Integer partition, Integer broker, LogIdentity identity) {
       this.name = name;
-      this.partition = OptionalInt.of(partition);
-      this.broker = OptionalInt.of(broker);
-      this.type = Optional.empty();
+      this.partition = partition;
+      this.broker = broker;
+      this.type = (identity == null) ? null : identity.name();
     }
 
-    ThrottleTarget(String name, int partition, int broker, LogIdentity type) {
-      this.name = name;
-      this.partition = OptionalInt.of(partition);
-      this.broker = OptionalInt.of(broker);
-      this.type = Optional.of(type);
+    @Override
+    public String toString() {
+      return "ThrottleTarget{"
+          + "name='"
+          + name
+          + '\''
+          + ", partition="
+          + partition
+          + ", broker="
+          + broker
+          + ", type="
+          + type
+          + '}';
     }
   }
 
-  enum ThrottleBandwidths {
-    ingress,
-    egress;
-  }
-
   enum LogIdentity {
     leader,
-    follower;
+    follower
   }
 }
diff --git a/app/src/main/java/org/astraea/app/web/TopicHandler.java b/app/src/main/java/org/astraea/app/web/TopicHandler.java
index 3dade568..3c85f832 100644
--- a/app/src/main/java/org/astraea/app/web/TopicHandler.java
+++ b/app/src/main/java/org/astraea/app/web/TopicHandler.java
@@ -36,6 +36,7 @@ class TopicHandler implements Handler {
   static final String NUMBER_OF_PARTITIONS_KEY = "partitions";
   static final String NUMBER_OF_REPLICAS_KEY = "replicas";
   static final String PARTITION_KEY = "partition";
+  static final String LIST_INTERNAL = "listInternal";
 
   private final Admin admin;
 
@@ -43,14 +44,18 @@ class TopicHandler implements Handler {
     this.admin = admin;
   }
 
-  Set<String> topicNames(Optional<String> target) {
-    return Handler.compare(admin.topicNames(), target);
+  Set<String> topicNames(Optional<String> target, boolean listInternal) {
+    return Handler.compare(admin.topicNames(listInternal), target);
   }
 
   @Override
   public Response get(Channel channel) {
     return get(
-        topicNames(channel.target()),
+        topicNames(
+            channel.target(),
+            Optional.ofNullable(channel.queries().get(LIST_INTERNAL))
+                .map(Boolean::parseBoolean)
+                .orElse(true)),
         partition ->
             !channel.queries().containsKey(PARTITION_KEY)
                 || partition == Integer.parseInt(channel.queries().get(PARTITION_KEY)));
@@ -89,7 +94,7 @@ class TopicHandler implements Handler {
     var configs =
         new HashMap<>(
             request.raw().entrySet().stream()
-                .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().toString())));
+                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)));
     configs.remove(TOPIC_NAME_KEY);
     configs.remove(NUMBER_OF_PARTITIONS_KEY);
     configs.remove(NUMBER_OF_REPLICAS_KEY);
@@ -185,13 +190,13 @@ class TopicHandler implements Handler {
 
     Replica(org.astraea.app.admin.Replica replica) {
       this(
-          replica.broker(),
+          replica.nodeInfo().id(),
           replica.lag(),
           replica.size(),
-          replica.leader(),
+          replica.isLeader(),
           replica.inSync(),
           replica.isFuture(),
-          replica.path());
+          replica.dataFolder());
     }
 
     Replica(
diff --git a/docker/start_app.sh b/docker/start_app.sh
index e74f1fb6..59378515 100755
--- a/docker/start_app.sh
+++ b/docker/start_app.sh
@@ -19,8 +19,8 @@ source $DOCKER_FOLDER/docker_build_common.sh
 
 # ===============================[global variables]===============================
 declare -r VERSION=${REVISION:-${VERSION:-main}}
-declare -r REPO=${REPO:-ghcr.io/skiptests/astraea/app}
-declare -r IMAGE_NAME="$REPO:$VERSION"
+declare -r ACCOUNT=${ACCOUNT:-skiptests}
+declare -r IMAGE_NAME="ghcr.io/${ACCOUNT}/astraea/app:$VERSION"
 declare -r DOCKERFILE=$DOCKER_FOLDER/app.dockerfile
 declare -r JMX_PORT=${JMX_PORT:-"$(getRandomPort)"}
 # for web service
@@ -33,7 +33,7 @@ declare -r HEAP_OPTS="${HEAP_OPTS:-"-Xmx2G -Xms2G"}"
 function showHelp() {
   echo "Usage: [ENV] start_app.sh"
   echo "ENV: "
-  echo "    REPO=astraea/app    set the docker repo"
+  echo "    ACCOUNT=skiptests          set the account to clone from"
   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"
 }
@@ -44,7 +44,7 @@ FROM ghcr.io/skiptests/astraea/deps AS build
 
 # clone repo
 WORKDIR /tmp
-RUN git clone https://github.com/skiptests/astraea
+RUN git clone https://github.com/$ACCOUNT/astraea
 
 # pre-build project to collect all dependencies
 WORKDIR /tmp/astraea
@@ -110,15 +110,15 @@ function runContainer() {
   local need_to_bind_file=""
   local defined_file="false"
   for word in "${sentence[@]}"; do
-    # user has pre-defined port, so we will replace the random port by this one in next loop
-    if [[ "$word" == "--prop.file" ]]; then
+    # user has pre-defined directories/files, so we will mount directories/files
+    if [[ "$word" == "--prop.file" || "$word" == "--report.path" ]]; then
       defined_file="true"
       continue
     fi
-    # this element must be port
+    # this element must be something to mount
     if [[ "$defined_file" == "true" ]]; then
-      need_to_bind_file="-v $word:$word"
-      break
+      need_to_bind_file="${need_to_bind_file} -v $word:$word"
+      defined_file="false"
     fi
   done
 
diff --git a/docs/web_server/web_api_balancer_chinese.md b/docs/web_server/web_api_balancer_chinese.md
index 7913d6eb..f88124b3 100644
--- a/docs/web_server/web_api_balancer_chinese.md
+++ b/docs/web_server/web_api_balancer_chinese.md
@@ -10,9 +10,10 @@ GET /balancer
 
 參數
 
-| 名稱    | 說明           | 預設值   |
-|-------|--------------|-------|
-| limit | (選填) 要嘗試幾種組合 | 10000 |
+| 名稱     | 說明                   | 預設值                      |
+|--------|----------------------|--------------------------|
+| limit  | (選填) 要嘗試幾種組合         | 10000                    |
+| topics | (選填) 只嘗試搬移指定的 topics | 無，除了內部 topics 以外的都作為候選對象 |
 
 cURL 範例
 ```shell
@@ -28,6 +29,9 @@ JSON Response 範例
   - `topic`: topic 名稱
   - `partition`: partition id
   - `before`: 原本的配置
+    - `brokerId`: 有掌管 replica 的節點 id
+    - `directory`: replica 存在資料的路徑
+    - `size`: replica 在硬碟上的資料大小
   - `after`: 比較好的配置
 ```json
 {
@@ -42,7 +46,8 @@ JSON Response 範例
       "before": [
         {
           "brokerId": 1006,
-          "directory": "/tmp/log-folder-0"
+          "directory": "/tmp/log-folder-0",
+          "size": 1234
         }
       ],
       "after": [
@@ -58,7 +63,8 @@ JSON Response 範例
       "before": [
         {
           "brokerId": 1003,
-          "directory": "/tmp/log-folder-0"
+          "directory": "/tmp/log-folder-0",
+          "size": 12355
         }
       ],
       "after": [
diff --git a/docs/web_server/web_api_throttles_chinese.md b/docs/web_server/web_api_throttles_chinese.md
index 902c9e68..baf8e2bb 100644
--- a/docs/web_server/web_api_throttles_chinese.md
+++ b/docs/web_server/web_api_throttles_chinese.md
@@ -4,6 +4,16 @@
 取得與套用限流相關的設定
 
 - [取得叢集的所有 throttle 設定](#取得叢集的所有-throttle-設定)
+- [新增/覆蓋 throttle 設定](#新增/覆蓋-throttle-設定)
+- [刪除 throttle 設定](#刪除-throttle-設定)
+  - [移除整個 Topic 的 replication throttle](#移除整個-Topic-的-replication-throttle)
+  - [移除整個 Topic/Partition 的 replication throttle](#移除整個-Topic/Partition-的-replication-throttle)
+  - [移除整個 Topic/Partition/Replica 的 replication throttle](#移除整個-Topic/Partition/Replica-的-replication-throttle)
+  - [移除特定 Topic/Partition/Replica 針對 leader 或是 follower 的 replication throttle](#移除特定-Topic/Partition/Replica-針對-leader-或是-follower-的-replication-throttle)
+  - [移除特定節點的 replication 輸入限流量](#移除特定節點的-replication-輸入限流)
+  - [移除特定節點的 replication 輸出限流量](#移除特定節點的-replication-輸出限流)
+  - [移除特定節點的 replication 輸出和輸入限流量](#移除特定節點的-replication-輸出和輸入限流)
+
 
 ## 取得叢集的所有 throttle 設定
 
@@ -20,11 +30,11 @@ curl -X GET http://localhost:8001/throttles
 JSON Response 範例
 
 - `brokers`: 所有上線的 Kafka 節點的限流量設定
-  - `1001`: 敘述這個 broker ID 的 kafka 節點設定
-    - `ingress`: 特定 Kafka 節點的 replication 流入流量的限制(單位 bytes/sec)，附註只有 `topics` 中有記錄到的對象，
-      其 replication 流量會受限制，如果此欄位不存在則代表此節點的 replication 流入流量沒有被限制。
-    - `egress`: 特定 Kafka 節點的 replication 流出流量的限制(單位 bytes/sec)，附註只有 `topics` 中有記錄到的對象，
-      其 replication 流量會受限制，如果此欄位不存在則代表此節點的 replication 流出流量沒有被限制。
+  - `id`: Kafka 節點編號
+  - `ingress`: 特定 Kafka 節點的 replication 流入流量的限制(單位 bytes/sec)，附註只有 `topics` 中有記錄到的對象，
+    其 replication 流量會受限制，如果此欄位不存在則代表此節點的 replication 流入流量沒有被限制。
+  - `egress`: 特定 Kafka 節點的 replication 流出流量的限制(單位 bytes/sec)，附註只有 `topics` 中有記錄到的對象，
+    其 replication 流量會受限制，如果此欄位不存在則代表此節點的 replication 流出流量沒有被限制。
 - `topics`: 當前叢集中，被 replication 限流影響的 logs
   - `name`: 套用 replication 限流的 topic
   - `partition`: 套用 replication 限流的 partition
@@ -33,16 +43,157 @@ JSON Response 範例
 
 ```json
 {
-    "brokers": {
-        "1001": { "ingress": 1000, "egress": 1000 },
-        "1002": { "ingress": 1000, "egress": 1000 },
-        "1003": { "ingress": 1000 },
-        "1004": {}
-    },
+    "brokers": [
+      { "id": 1001, "ingress": 1000, "egress": 1000 },
+      { "id": 1002, "ingress": 1000, "egress": 1000 },
+      { "id": 1003, "ingress": 1000 },
+      { "id": 1004 }
+    ],
     "topics": [
-      { "name": "MyTopicB", "partition": 0, "broker": 0 },
-      { "name": "MyTopicC", "partition": 1, "broker": 1, "type": "leader" },
-      { "name": "MyTopicD", "partition": 2, "broker": 2, "type": "follower" }
+      { "name": "MyTopicA", "partition": 0, "broker": 1001 },
+      { "name": "MyTopicB", "partition": 1, "broker": 1002, "type": "leader" },
+      { "name": "MyTopicC", "partition": 2, "broker": 1003, "type": "follower" }
     ]
 }
-```
\ No newline at end of file
+```
+
+## 新增/覆蓋 throttle 設定
+
+```shell
+POST /throttles
+```
+
+cURL 範例
+
+```shell
+curl -X POST http://localhost:8001/throttles \
+    -H "Content-Type: application/json" \
+    -d '{
+      "brokers": [
+        { "id":  1001, "ingress":  1000, "egress":  1000 },
+        { "id":  1002, "ingress":  1000 }
+      ],
+      "topics": [
+        { "name": "MyTopicA" },
+        { "name": "MyTopicB", "partition": 2 },
+        { "name": "MyTopicC", "partition": 3, "broker": 1001 },
+        { "name": "MyTopicD", "partition": 4, "broker": 1001, "type": "leader" }
+      ]
+    }'
+```
+
+JSON Request 格式
+
+| 名稱    | 說明                                     | 預設值 |
+| ------- | ---------------------------------------- | ------ |
+| brokers | (選填) 敘述要更新的 Broker Throttle 設定 | 無     |
+| topics  | (選填) 敘述要更新的 Topic Throttle 設定  | 無     |
+
+brokers 每個資料欄位
+
+| 名稱      | 說明                                                                                                                                    | 預設值 |
+|---------|---------------------------------------------------------------------------------------------------------------------------------------| ------ |
+| id      | (必填) 欲更新節點的 id                                                                                                                        | 無     |
+| ingress | (選填) 特定 Kafka 節點的 replication 流入流量的限制(單位 bytes/sec)，附註只有 `topics` 中有記錄到的對象，其 replication 流量會受限制，如果此欄位不存在則代表此節點的 replication 流入流量沒有被修改。 | 無     |
+| egress  | (選填) 特定 Kafka 節點的 replication 流出流量的限制(單位 bytes/sec)，附註只有 `topics` 中有記錄到的對象，其 replication 流量會受限制。                                      | 無     |
+
+topics 每個資料欄位
+
+| 名稱      | 說明                                                         | 預設值 |
+| --------- | ------------------------------------------------------------ | ------ |
+| name      | (必填) 要套用 repliaction throttle 的 topic 名稱             | 無     |
+| partition | (選填) 要套用 replication throttle 的 partition 編號，如果沒有指定，所有既有 partition 都會被套用 | 無     |
+| broker    | (選填) 要套用 replication throttle 的 replica (所在之 broker)，如果沒有指定，所有既有的 replicas 都會被套用 | 無     |
+| type      | (選填) 要套用 replication throttle 的 replica 身份，此值可以是 `leader` 或是 `follower`，如果沒有指定，則二者都會被套用 | 無     |
+
+JSON Response 範例
+
+- `brokers`: 回傳被此 POST 影響的結果。
+  - `id`: 被影響的節點的編號。
+  - `ingress`: 如果此欄位不存在則代表此節點的 replication 流入流量沒有被修改。
+  - `egress`: 如果此欄位不存在則代表此節點的 replication 流出流量沒有被修改。
+- `topics`: 被此 POST 影響的 logs。
+  - `name`: 套用 replication 限流的 topic。
+  - `partition`: 套用 replication 限流的 partition。
+  - `broker`: 套用 replication 限流的 broker ID。
+  - `type`: 套用的 log 身份，可以是 `leader` 或 `follower`，如果沒有標記這個欄位，意味著二者皆是。
+
+```json
+{
+  "brokers": [
+    { "id":  1001, "ingress":  1000, "egress":  1000 },
+    { "id":  1002, "ingress":  1000 }
+  ],
+  "topics": [
+    { "name":  "MyTopicA", "partition": 0, "broker": 1001, "type": "leader" },
+    { "name":  "MyTopicA", "partition": 0, "broker": 1002, "type": "follower" },
+    { "name":  "MyTopicA", "partition": 0, "broker": 1003, "type": "follower" },
+    { "name":  "MyTopicA", "partition": 1, "broker": 1002, "type": "leader" },
+    { "name":  "MyTopicA", "partition": 1, "broker": 1003, "type": "follower" },
+    { "name":  "MyTopicA", "partition": 1, "broker": 1001, "type": "follower" },
+    { "name":  "MyTopicB", "partition": 2, "broker": 1003, "type": "leader" },
+    { "name":  "MyTopicB", "partition": 2, "broker": 1001, "type": "follower" },
+    { "name":  "MyTopicC", "partition": 3, "broker": 1001 },
+    { "name":  "MyTopicD", "partition": 4, "broker": 1001, "type": "leader" }
+  ]
+}
+```
+
+## 刪除 throttle 設定
+
+```
+DELETE /throttles
+```
+
+cURL 範例
+
+* ##### 移除整個 Topic 的 replication throttle
+
+  ```shell
+  # 移除與 TopicA 相關的所有 replication throttle
+  curl -X DELETE "http://localhost:8001/throttles?topic=MyTopicA"
+  ```
+
+* ##### 移除整個 Topic/Partition 的 replication throttle
+
+  ```shell
+  # 移除與 TopicA/partition 3 相關的所有 replication throttle
+  curl -X DELETE "http://localhost:8001/throttles?topic=MyTopicA&partition=3"
+  ```
+
+* ##### 移除整個 Topic/Partition/Replica 的 replication throttle
+
+  ```shell
+  # 移除與 TopicA/partition 3/broker 4 相關的所有 replication throttle
+  curl -X DELETE "http://localhost:8001/throttles?topic=MyTopicA&partition=3&broker=4"
+  ```
+
+* ##### 移除特定 Topic/Partition/Replica 針對 leader 或是 follower 的 replication throttle
+
+  ```shell
+  # 移除 TopicA/partition 3/broker 4 的 leader replication throttle
+  curl -X DELETE "http://localhost:8001/throttles?topic=MyTopicA&partition=3&broker=4&type=leader"
+  ```
+
+* ##### 移除特定節點的 replication 輸入限流
+
+  ```shell
+  # 移除編號 1001 節點的輸入流量限流
+  curl -X DELETE "http://localhost:8001/throttles?broker=1001&type=ingress"
+  ```
+
+* ##### 移除特定節點的 replication 輸出限流
+
+  ```shell
+  # 移除編號 1001 節點的輸出流量限流
+  curl -X DELETE "http://localhost:8001/throttles?broker=1001&type=egress"
+  ```
+
+* ##### 移除特定節點的 replication 輸出和輸入限流
+
+  ```shell
+  # 移除編號 1001 節點的輸出和輸入流量限流
+  curl -X DELETE "http://localhost:8001/throttles?broker=1001&type=ingress+egress"
+  ```
+
+  
\ 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
index efea6f4e..37bb8d41 100644
--- a/docs/web_server/web_api_topics_chinese.md
+++ b/docs/web_server/web_api_topics_chinese.md
@@ -117,9 +117,10 @@ GET /topics
 
 參數
 
-| 名稱        | 說明                      | 預設值               |
-|-----------|-------------------------|-------------------|
-| partition | (選填) 指定要查看哪一個 partition | 無，代表全部 partitions |
+| 名稱           | 說明                                                                                             | 預設值                |
+|--------------|------------------------------------------------------------------------------------------------|--------------------|
+| partition    | (選填) 指定要查看哪一個 partition                                                                        | 無，代表全部 partitions  |
+| listInternal | (選填) 為 boolean 值。若填寫 true，則會列出 kafka 內部所使用的 topics，如 __commit_offsets。若設為 false，則不會列出此種 topics | true，代表列出所有 topics |
 
 cURL 範例
 ```shell


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 5b536101..30a1f1ff 100644
--- a/app/src/test/java/org/astraea/app/admin/AdminTest.java
+++ b/app/src/test/java/org/astraea/app/admin/AdminTest.java
@@ -16,9 +16,9 @@
  */
 package org.astraea.app.admin;
 
+import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.junit.jupiter.api.condition.OS.WINDOWS;
 
-import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -148,7 +148,7 @@ public class AdminTest extends RequireBrokerCluster {
                   replicas.forEach(
                       replica ->
                           Assertions.assertTrue(
-                              logFolders.stream().anyMatch(replica.path()::contains))));
+                              logFolders.stream().anyMatch(replica.dataFolder()::contains))));
     }
   }
 
@@ -223,11 +223,16 @@ public class AdminTest extends RequireBrokerCluster {
             var partitionReplicas = replicas.entrySet().iterator().next().getValue();
             return replicas.size() == 1
                 && partitionReplicas.size() == 1
-                && partitionReplicas.get(0).broker() == broker;
+                && partitionReplicas.get(0).nodeInfo().id() == broker;
           });
 
       var currentBroker =
-          admin.replicas(Set.of(topicName)).get(TopicPartition.of(topicName, 0)).get(0).broker();
+          admin
+              .replicas(Set.of(topicName))
+              .get(TopicPartition.of(topicName, 0))
+              .get(0)
+              .nodeInfo()
+              .id();
       var allPath = admin.brokerFolders(Set.of(currentBroker));
       var otherPath =
           allPath.get(currentBroker).stream()
@@ -238,7 +243,7 @@ public class AdminTest extends RequireBrokerCluster {
                               .replicas(Set.of(topicName))
                               .get(TopicPartition.of(topicName, 0))
                               .get(0)
-                              .path()))
+                              .dataFolder()))
               .collect(Collectors.toSet());
       admin
           .migrator()
@@ -250,7 +255,7 @@ public class AdminTest extends RequireBrokerCluster {
             var partitionReplicas = replicas.entrySet().iterator().next().getValue();
             return replicas.size() == 1
                 && partitionReplicas.size() == 1
-                && partitionReplicas.get(0).path().equals(otherPath.iterator().next());
+                && partitionReplicas.get(0).dataFolder().equals(otherPath.iterator().next());
           });
     }
   }
@@ -263,7 +268,7 @@ public class AdminTest extends RequireBrokerCluster {
       var topicPartition = TopicPartition.of(topic, 0);
       admin.creator().topic(topic).numberOfPartitions(1).numberOfReplicas((short) 1).create();
       Utils.sleep(Duration.ofSeconds(1));
-      var originalBroker = admin.replicas(Set.of(topic)).get(topicPartition).get(0).broker();
+      var originalBroker = admin.replicas(Set.of(topic)).get(topicPartition).get(0).nodeInfo().id();
       var nextBroker = (originalBroker + 1) % brokerIds().size();
       var nextDir = logFolders().get(nextBroker).stream().findAny().orElseThrow();
       Supplier<Replica> replicaNow = () -> admin.replicas(Set.of(topic)).get(topicPartition).get(0);
@@ -276,16 +281,16 @@ public class AdminTest extends RequireBrokerCluster {
       Utils.sleep(Duration.ofSeconds(1));
 
       // assert, nothing happened until the actual movement
-      Assertions.assertNotEquals(nextBroker, replicaNow.get().broker());
-      Assertions.assertNotEquals(nextDir, replicaNow.get().path());
+      Assertions.assertNotEquals(nextBroker, replicaNow.get().nodeInfo().id());
+      Assertions.assertNotEquals(nextDir, replicaNow.get().dataFolder());
 
       // act, perform the actual movement
       admin.migrator().partition(topic, 0).moveTo(List.of(nextBroker));
       Utils.sleep(Duration.ofSeconds(1));
 
       // assert, everything on the exact broker & dir
-      Assertions.assertEquals(nextBroker, replicaNow.get().broker());
-      Assertions.assertEquals(nextDir, replicaNow.get().path());
+      Assertions.assertEquals(nextBroker, replicaNow.get().nodeInfo().id());
+      Assertions.assertEquals(nextDir, replicaNow.get().dataFolder());
     }
   }
 
@@ -297,7 +302,7 @@ public class AdminTest extends RequireBrokerCluster {
       admin.creator().topic(topic).numberOfPartitions(1).numberOfReplicas((short) 1).create();
       Utils.sleep(Duration.ofSeconds(1));
       var currentReplica = admin.replicas(Set.of(topic)).get(topicPartition).get(0);
-      var currentBroker = currentReplica.broker();
+      var currentBroker = currentReplica.nodeInfo().id();
       var notExistReplica = (currentBroker + 1) % brokerIds().size();
       var nextDir = logFolders().get(notExistReplica).iterator().next();
 
@@ -315,7 +320,7 @@ public class AdminTest extends RequireBrokerCluster {
       admin.creator().topic(topic).numberOfPartitions(1).numberOfReplicas((short) 1).create();
       Utils.sleep(Duration.ofSeconds(1));
       var currentReplica = admin.replicas(Set.of(topic)).get(topicPartition).get(0);
-      var currentBroker = currentReplica.broker();
+      var currentBroker = currentReplica.nodeInfo().id();
       var nextDir = logFolders().get(currentBroker).iterator().next();
 
       Assertions.assertThrows(
@@ -376,7 +381,7 @@ public class AdminTest extends RequireBrokerCluster {
             if (replicas.size() != 3) return false;
             if (!replicas.values().stream().allMatch(rs -> rs.size() == 1)) return false;
             return replicas.values().stream()
-                .allMatch(rs -> rs.stream().allMatch(r -> r.broker() == broker));
+                .allMatch(rs -> rs.stream().allMatch(r -> r.nodeInfo().id() == broker));
           });
     }
   }
@@ -521,7 +526,8 @@ public class AdminTest extends RequireBrokerCluster {
                               entry ->
                                   entry.getValue().stream()
                                       .filter(Replica::isPreferredLeader)
-                                      .map(Replica::broker)
+                                      .map(Replica::nodeInfo)
+                                      .map(NodeInfo::id)
                                       .collect(Collectors.toUnmodifiableList())));
 
       // act, make 0 be the preferred leader of every partition
@@ -755,10 +761,11 @@ public class AdminTest extends RequireBrokerCluster {
                               Map.Entry::getKey,
                               e ->
                                   e.getValue().stream()
-                                      .filter(Replica::leader)
+                                      .filter(Replica::isLeader)
                                       .findFirst()
                                       .orElseThrow()
-                                      .broker()));
+                                      .nodeInfo()
+                                      .id()));
       var expectedReplicaList =
           currentLeaderMap.get().entrySet().stream()
               .collect(
@@ -849,7 +856,7 @@ public class AdminTest extends RequireBrokerCluster {
               index ->
                   producer
                       .sender()
-                      .key(String.valueOf(index).getBytes(StandardCharsets.UTF_8))
+                      .key(String.valueOf(index).getBytes(UTF_8))
                       .topic(topicName)
                       .run());
       producer.flush();
@@ -996,7 +1003,12 @@ public class AdminTest extends RequireBrokerCluster {
       Utils.sleep(Duration.ofSeconds(3));
 
       var currentBroker =
-          admin.replicas(Set.of(topicName)).get(TopicPartition.of(topicName, 0)).get(0).broker();
+          admin
+              .replicas(Set.of(topicName))
+              .get(TopicPartition.of(topicName, 0))
+              .get(0)
+              .nodeInfo()
+              .id();
       var nextBroker = brokerIds().stream().filter(i -> i != currentBroker).findAny().get();
 
       try (var producer = Producer.of(bootstrapServers())) {
@@ -1039,8 +1051,8 @@ public class AdminTest extends RequireBrokerCluster {
 
       var currentReplica =
           admin.replicas(Set.of(topicName)).get(TopicPartition.of(topicName, 0)).get(0);
-      var currentBroker = currentReplica.broker();
-      var currentPath = currentReplica.path();
+      var currentBroker = currentReplica.nodeInfo().id();
+      var currentPath = currentReplica.dataFolder();
       var nextPath =
           logFolders().get(currentBroker).stream()
               .filter(p -> !p.equals(currentPath))
@@ -1058,7 +1070,10 @@ public class AdminTest extends RequireBrokerCluster {
                 });
 
         try {
-          admin.migrator().topic(topicName).moveTo(Map.of(currentReplica.broker(), nextPath));
+          admin
+              .migrator()
+              .topic(topicName)
+              .moveTo(Map.of(currentReplica.nodeInfo().id(), nextPath));
           var reassignment =
               admin.reassignments(Set.of(topicName)).get(TopicPartition.of(topicName, 0));
           // Don't verify the result if the migration is done
@@ -1066,11 +1081,11 @@ public class AdminTest extends RequireBrokerCluster {
             Assertions.assertEquals(1, reassignment.from().size());
             var from = reassignment.from().iterator().next();
             Assertions.assertEquals(currentBroker, from.broker());
-            Assertions.assertEquals(currentPath, from.path());
+            Assertions.assertEquals(currentPath, from.dataFolder());
             Assertions.assertEquals(1, reassignment.to().size());
             var to = reassignment.to().iterator().next();
             Assertions.assertEquals(currentBroker, to.broker());
-            Assertions.assertEquals(nextPath, to.path());
+            Assertions.assertEquals(nextPath, to.dataFolder());
           }
         } finally {
           done.set(true);
@@ -1190,18 +1205,6 @@ public class AdminTest extends RequireBrokerCluster {
     }
   }
 
-  private String fetchBrokerConfig(int brokerId, String configKey) {
-    try (AdminClient adminClient =
-        AdminClient.create(
-            Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers()))) {
-      var resource = new ConfigResource(ConfigResource.Type.BROKER, String.valueOf(brokerId));
-      return Utils.packException(() -> adminClient.describeConfigs(List.of(resource)).all().get())
-          .get(resource)
-          .get(configKey)
-          .value();
-    }
-  }
-
   @Test
   void testReplicationThrottler$egress$ingress$dataRate() {
     try (Admin admin = Admin.of(bootstrapServers())) {
@@ -1214,8 +1217,18 @@ public class AdminTest extends RequireBrokerCluster {
       brokerIds()
           .forEach(
               id -> {
-                var leaderValue = fetchBrokerConfig(id, "leader.replication.throttled.rate");
-                var followerValue = fetchBrokerConfig(id, "follower.replication.throttled.rate");
+                var leaderValue =
+                    admin
+                        .brokers()
+                        .get(id)
+                        .value("leader.replication.throttled.rate")
+                        .orElseThrow();
+                var followerValue =
+                    admin
+                        .brokers()
+                        .get(id)
+                        .value("follower.replication.throttled.rate")
+                        .orElseThrow();
 
                 Assertions.assertEquals(
                     DataSize.MiB.of(72), DataSize.Byte.of(Long.parseLong(leaderValue)));
@@ -1242,8 +1255,18 @@ public class AdminTest extends RequireBrokerCluster {
       Map.of(0, rate0, 1, rate1, 2, rate2)
           .forEach(
               (id, rate) -> {
-                var leaderValue = fetchBrokerConfig(id, "leader.replication.throttled.rate");
-                var followerValue = fetchBrokerConfig(id, "follower.replication.throttled.rate");
+                var leaderValue =
+                    admin
+                        .brokers()
+                        .get(id)
+                        .value("leader.replication.throttled.rate")
+                        .orElseThrow();
+                var followerValue =
+                    admin
+                        .brokers()
+                        .get(id)
+                        .value("follower.replication.throttled.rate")
+                        .orElseThrow();
 
                 Assertions.assertEquals(
                     rate.dataSize(), DataSize.Byte.of(Long.parseLong(leaderValue)));
@@ -1258,13 +1281,13 @@ public class AdminTest extends RequireBrokerCluster {
         .flatMap(
             entry ->
                 entry.getValue().stream()
-                    .filter(Replica::leader)
+                    .filter(Replica::isLeader)
                     .map(
                         replica ->
                             TopicPartitionReplica.of(
                                 entry.getKey().topic(),
                                 entry.getKey().partition(),
-                                replica.broker())))
+                                replica.nodeInfo().id())))
         .collect(Collectors.toUnmodifiableSet());
   }
 
@@ -1273,13 +1296,13 @@ public class AdminTest extends RequireBrokerCluster {
         .flatMap(
             entry ->
                 entry.getValue().stream()
-                    .filter(replica -> !replica.leader())
+                    .filter(replica -> !replica.isLeader())
                     .map(
                         replica ->
                             TopicPartitionReplica.of(
                                 entry.getKey().topic(),
                                 entry.getKey().partition(),
-                                replica.broker())))
+                                replica.nodeInfo().id())))
         .collect(Collectors.toUnmodifiableSet());
   }
 
@@ -1468,23 +1491,22 @@ public class AdminTest extends RequireBrokerCluster {
   }
 
   private Set<TopicPartitionReplica> fetchLogThrottleConfig(String configKey, String topicName) {
-    try (AdminClient adminClient =
-        AdminClient.create(
-            Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers()))) {
-      var resource = new ConfigResource(ConfigResource.Type.TOPIC, topicName);
-      var stringValue =
-          Utils.packException(() -> adminClient.describeConfigs(List.of(resource)).all().get())
-              .get(resource)
-              .get(configKey)
-              .value();
-      if (stringValue.isEmpty()) return Set.of();
-      return Arrays.stream(stringValue.split(","))
-          .map(x -> x.split(":"))
+    try (var admin = Admin.of(bootstrapServers())) {
+      return admin
+          .topics(Set.of(topicName))
+          .get(topicName)
+          .value(configKey)
+          .filter(v -> !v.isEmpty())
           .map(
-              x ->
-                  TopicPartitionReplica.of(
-                      topicName, Integer.parseInt(x[0]), Integer.parseInt(x[1])))
-          .collect(Collectors.toUnmodifiableSet());
+              value ->
+                  Arrays.stream(value.split(","))
+                      .map(x -> x.split(":"))
+                      .map(
+                          x ->
+                              TopicPartitionReplica.of(
+                                  topicName, Integer.parseInt(x[0]), Integer.parseInt(x[1])))
+                      .collect(Collectors.toUnmodifiableSet()))
+          .orElse(Set.of());
     }
   }
 
@@ -1552,4 +1574,79 @@ public class AdminTest extends RequireBrokerCluster {
               .noneMatch(log -> log.partition() == 0 && log.brokerId() == 0));
     }
   }
+
+  @Test
+  void testClearIngressReplicationThrottle() {
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      // apply throttle
+      var dataRate = DataRate.MiB.of(500).perSecond();
+      admin.replicationThrottler().ingress(dataRate).apply();
+      Utils.sleep(Duration.ofSeconds(1));
+
+      // ensure the throttle was applied
+      final var value0 = admin.brokers().get(0).value("follower.replication.throttled.rate");
+      Assertions.assertEquals((long) dataRate.byteRate(), Long.valueOf(value0.orElseThrow()));
+
+      // clear throttle
+      admin.clearIngressReplicationThrottle(Set.of(0, 1, 2));
+      Utils.sleep(Duration.ofSeconds(1));
+
+      // ensure the throttle was removed
+      final var value1 = admin.brokers().get(0).value("follower.replication.throttled.rate");
+      Assertions.assertTrue(value1.isEmpty());
+    }
+  }
+
+  @Test
+  void testClearEgressReplicationThrottle() {
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      // apply throttle
+      var dataRate = DataRate.MiB.of(500).perSecond();
+      admin.replicationThrottler().egress(dataRate).apply();
+      Utils.sleep(Duration.ofSeconds(1));
+
+      // ensure the throttle was applied
+      final var value0 = admin.brokers().get(0).value("leader.replication.throttled.rate");
+      Assertions.assertEquals((long) dataRate.byteRate(), Long.valueOf(value0.orElseThrow()));
+
+      // clear throttle
+      admin.clearEgressReplicationThrottle(Set.of(0, 1, 2));
+      Utils.sleep(Duration.ofSeconds(1));
+
+      // ensure the throttle was removed
+      final var value1 = admin.brokers().get(0).value("leader.replication.throttled.rate");
+      Assertions.assertTrue(value1.isEmpty());
+    }
+  }
+
+  @Test
+  void testTopicNames() {
+    var bootstrapServers = bootstrapServers();
+    var topicName = Utils.randomString(10);
+    try (var admin = Admin.of(bootstrapServers);
+        var producer = Producer.of(bootstrapServers);
+        var consumer =
+            Consumer.forTopics(Set.of(topicName)).bootstrapServers(bootstrapServers).build()) {
+      // producer and consumer here are used to trigger kafka to create internal topic
+      // __consumer_offsets
+      producer
+          .sender()
+          .topic(topicName)
+          .key("foo".getBytes(UTF_8))
+          .run()
+          .toCompletableFuture()
+          .join();
+      consumer.poll(Duration.ofSeconds(1));
+
+      Assertions.assertTrue(
+          admin.topicNames(true).stream().anyMatch(t -> t.equals("__consumer_offsets")));
+      Assertions.assertEquals(
+          1, admin.topicNames(true).stream().filter(t -> t.equals(topicName)).count());
+
+      Assertions.assertFalse(
+          admin.topicNames(false).stream().anyMatch(t -> t.equals("__consumer_offsets")));
+      Assertions.assertEquals(
+          1, admin.topicNames(false).stream().filter(t -> t.equals(topicName)).count());
+    }
+  }
 }
diff --git a/app/src/test/java/org/astraea/app/admin/ClusterInfoWithOfflineNodeTest.java b/app/src/test/java/org/astraea/app/admin/ClusterInfoWithOfflineNodeTest.java
index 41141390..1842f3a9 100644
--- a/app/src/test/java/org/astraea/app/admin/ClusterInfoWithOfflineNodeTest.java
+++ b/app/src/test/java/org/astraea/app/admin/ClusterInfoWithOfflineNodeTest.java
@@ -43,7 +43,7 @@ public class ClusterInfoWithOfflineNodeTest extends RequireBrokerCluster {
       var before = admin.clusterInfo(Set.of(topicName));
       Assertions.assertEquals(
           partitionCount * replicaCount,
-          before.replicas(topicName).stream().filter(x -> !x.isOfflineReplica()).count());
+          before.replicas(topicName).stream().filter(x -> !x.isOffline()).count());
       Assertions.assertEquals(
           partitionCount * replicaCount, before.availableReplicas(topicName).size());
       Assertions.assertEquals(partitionCount, before.availableReplicaLeaders(topicName).size());
@@ -57,7 +57,7 @@ public class ClusterInfoWithOfflineNodeTest extends RequireBrokerCluster {
       var after = admin.clusterInfo(Set.of(topicName));
       Assertions.assertEquals(
           partitionCount * (replicaCount - 1),
-          after.replicas(topicName).stream().filter(x -> !x.isOfflineReplica()).count());
+          after.replicas(topicName).stream().filter(x -> !x.isOffline()).count());
       Assertions.assertEquals(
           partitionCount * (replicaCount - 1), after.availableReplicas(topicName).size());
       Assertions.assertEquals(
@@ -72,11 +72,11 @@ public class ClusterInfoWithOfflineNodeTest extends RequireBrokerCluster {
               .allMatch(x -> x.nodeInfo().id() != brokerToClose));
       Assertions.assertTrue(
           after.replicas(topicName).stream()
-              .filter(ReplicaInfo::isOfflineReplica)
+              .filter(ReplicaInfo::isOffline)
               .allMatch(x -> x.nodeInfo().id() == brokerToClose));
       Assertions.assertTrue(
           after.replicas(topicName).stream()
-              .filter(x -> !x.isOfflineReplica())
+              .filter(x -> !x.isOffline())
               .allMatch(x -> x.nodeInfo().id() != brokerToClose));
     }
   }
diff --git a/app/src/test/java/org/astraea/app/admin/ReplicaSyncingMonitorIntegrationTest.java b/app/src/test/java/org/astraea/app/admin/ReplicaSyncingMonitorIntegrationTest.java
deleted file mode 100644
index a0a99cd4..00000000
--- a/app/src/test/java/org/astraea/app/admin/ReplicaSyncingMonitorIntegrationTest.java
+++ /dev/null
@@ -1,104 +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.admin;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertSame;
-
-import java.time.Duration;
-import java.util.List;
-import java.util.Set;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.TimeUnit;
-import java.util.stream.IntStream;
-import org.astraea.app.argument.Argument;
-import org.astraea.app.common.Utils;
-import org.astraea.app.producer.Producer;
-import org.astraea.app.producer.Sender;
-import org.astraea.app.service.RequireBrokerCluster;
-import org.junit.jupiter.api.Test;
-
-class ReplicaSyncingMonitorIntegrationTest extends RequireBrokerCluster {
-
-  private static final String TOPIC_NAME =
-      ReplicaSyncingMonitorIntegrationTest.class.getSimpleName();
-  private static final byte[] dummyBytes = new byte[1024];
-
-  @Test
-  void execute() throws InterruptedException {
-    // arrange
-    try (Admin topicAdmin = Admin.of(bootstrapServers())) {
-      topicAdmin
-          .creator()
-          .topic(TOPIC_NAME)
-          .numberOfPartitions(1)
-          .numberOfReplicas((short) 1)
-          .create();
-
-      Sender<byte[], byte[]> sender =
-          Producer.of(bootstrapServers()).sender().topic(TOPIC_NAME).partition(0).value(dummyBytes);
-
-      // create 16MB of data
-      IntStream.range(0, 16 * 1024)
-          .mapToObj(i -> sender.run().toCompletableFuture())
-          .forEach(CompletableFuture::join);
-
-      int currentBroker =
-          topicAdmin
-              .replicas(Set.of(TOPIC_NAME))
-              .get(TopicPartition.of(TOPIC_NAME, 0))
-              .get(0)
-              .broker();
-      int moveToBroker = (currentBroker + 1) % logFolders().size();
-
-      Thread executionThread =
-          new Thread(
-              () -> {
-                topicAdmin.migrator().partition(TOPIC_NAME, 0).moveTo(List.of(moveToBroker));
-                ReplicaSyncingMonitor.execute(
-                    topicAdmin,
-                    Argument.parse(
-                        new ReplicaSyncingMonitor.Argument(),
-                        new String[] {
-                          "--bootstrap.servers",
-                          bootstrapServers(),
-                          "--topic",
-                          TOPIC_NAME,
-                          "--interval",
-                          "0.1"
-                        }));
-              });
-
-      // act
-      executionThread.start();
-      TimeUnit.SECONDS.timedJoin(executionThread, 8); // wait until the thread exit
-      Utils.sleep(Duration.ofSeconds(2)); // sleep 2 extra seconds to ensure test run in stable
-
-      // assert
-      assertSame(Thread.State.TERMINATED, executionThread.getState());
-      assertEquals(
-          1, topicAdmin.replicas(Set.of(TOPIC_NAME)).get(TopicPartition.of(TOPIC_NAME, 0)).size());
-      assertEquals(
-          moveToBroker,
-          topicAdmin.replicas(Set.of(TOPIC_NAME)).get(TopicPartition.of(TOPIC_NAME, 0)).stream()
-              .filter(Replica::leader)
-              .findFirst()
-              .orElseThrow()
-              .broker());
-    }
-  }
-}
diff --git a/app/src/test/java/org/astraea/app/admin/ReplicaSyncingMonitorTest.java b/app/src/test/java/org/astraea/app/admin/ReplicaSyncingMonitorTest.java
deleted file mode 100644
index bd72f0b6..00000000
--- a/app/src/test/java/org/astraea/app/admin/ReplicaSyncingMonitorTest.java
+++ /dev/null
@@ -1,416 +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.admin;
-
-import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertNotEquals;
-import static org.junit.jupiter.api.Assertions.assertSame;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.junit.jupiter.api.Assertions.fail;
-import static org.mockito.Mockito.any;
-import static org.mockito.Mockito.anySet;
-import static org.mockito.Mockito.atLeast;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-
-import java.io.ByteArrayOutputStream;
-import java.io.PrintStream;
-import java.time.Duration;
-import java.time.temporal.ChronoUnit;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.TimeUnit;
-import java.util.function.BiFunction;
-import java.util.function.Consumer;
-import java.util.stream.Collectors;
-import java.util.stream.IntStream;
-import org.astraea.app.argument.Argument;
-import org.astraea.app.common.DataSize;
-import org.astraea.app.common.DataUnit;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.CsvSource;
-
-class ReplicaSyncingMonitorTest {
-
-  private ByteArrayOutputStream mockOutput;
-  private final PrintStream stdout = System.out;
-  private final List<Runnable> tearDownTasks = new ArrayList<>();
-
-  @BeforeEach
-  void setUp() {
-    mockOutput = new ByteArrayOutputStream();
-    System.setOut(new PrintStream(mockOutput));
-  }
-
-  @AfterEach
-  void tearDown() {
-    tearDownTasks.forEach(Runnable::run);
-    tearDownTasks.clear();
-
-    System.setOut(stdout);
-  }
-
-  // helper functions
-  private static final BiFunction<String, Integer, TopicPartition> topicPartition =
-      TopicPartition::of;
-  private static final BiFunction<Integer, long[], List<Replica>> replica =
-      (count, size) ->
-          IntStream.range(0, count)
-              .mapToObj(
-                  i ->
-                      new Replica(
-                          i,
-                          0,
-                          size[i],
-                          i == 0,
-                          size[i] == size[0],
-                          false,
-                          false,
-                          i == 0,
-                          "/tmp/log"))
-              .collect(Collectors.toUnmodifiableList());
-
-  @Test
-  void execute() throws InterruptedException {
-    // arrange
-    int interval = 10;
-    Admin mockTopicAdmin = mock(Admin.class);
-    when(mockTopicAdmin.topicNames()).thenReturn(Set.of("topic-1", "topic-2", "topic-3"));
-    when(mockTopicAdmin.replicas(anySet()))
-        .thenReturn(
-            Map.of(
-                /* progress 0% */
-                topicPartition.apply("topic-1", 0), replica.apply(2, new long[] {100, 0}),
-                topicPartition.apply("topic-2", 0), replica.apply(2, new long[] {100, 0}),
-                topicPartition.apply("topic-3", 0), replica.apply(2, new long[] {100, 0})))
-        .thenReturn(
-            Map.of(
-                /* progress 25% */
-                topicPartition.apply("topic-1", 0), replica.apply(2, new long[] {100, 25}),
-                topicPartition.apply("topic-2", 0), replica.apply(2, new long[] {100, 25}),
-                topicPartition.apply("topic-3", 0), replica.apply(2, new long[] {100, 25})))
-        .thenReturn(
-            Map.of(
-                /* progress 50% */
-                topicPartition.apply("topic-1", 0), replica.apply(2, new long[] {100, 50}),
-                topicPartition.apply("topic-2", 0), replica.apply(2, new long[] {100, 50}),
-                topicPartition.apply("topic-3", 0), replica.apply(2, new long[] {100, 50})))
-        .thenReturn(
-            Map.of(
-                /* progress 43.75% (log shrink, progress fall) */
-                topicPartition.apply("topic-1", 0), replica.apply(2, new long[] {80, 35}),
-                topicPartition.apply("topic-2", 0), replica.apply(2, new long[] {80, 35}),
-                topicPartition.apply("topic-3", 0), replica.apply(2, new long[] {80, 35})))
-        .thenReturn(
-            Map.of(
-                /* progress 50% (data catch up) */
-                topicPartition.apply("topic-1", 0), replica.apply(2, new long[] {100, 50}),
-                topicPartition.apply("topic-2", 0), replica.apply(2, new long[] {100, 50}),
-                topicPartition.apply("topic-3", 0), replica.apply(2, new long[] {100, 50})))
-        .thenReturn(
-            Map.of(
-                /* progress NaN% (progress overflow due to non-atomic return result) */
-                topicPartition.apply("topic-1", 0), replica.apply(2, new long[] {50, 80}),
-                topicPartition.apply("topic-2", 0), replica.apply(2, new long[] {50, 80}),
-                topicPartition.apply("topic-3", 0), replica.apply(2, new long[] {50, 80})))
-        .thenReturn(
-            Map.of(
-                /* progress 60% */
-                topicPartition.apply("topic-1", 0), replica.apply(2, new long[] {50, 30}),
-                topicPartition.apply("topic-2", 0), replica.apply(2, new long[] {50, 30}),
-                topicPartition.apply("topic-3", 0), replica.apply(2, new long[] {50, 30})))
-        .thenReturn(
-            Map.of(
-                /* progress 80% */
-                topicPartition.apply("topic-1", 0), replica.apply(2, new long[] {50, 40}),
-                topicPartition.apply("topic-2", 0), replica.apply(2, new long[] {50, 40}),
-                topicPartition.apply("topic-3", 0), replica.apply(2, new long[] {50, 40})))
-        .thenReturn(
-            Map.of(
-                /* progress 90% */
-                topicPartition.apply("topic-1", 0), replica.apply(2, new long[] {50, 45}),
-                topicPartition.apply("topic-2", 0), replica.apply(2, new long[] {50, 45}),
-                topicPartition.apply("topic-3", 0), replica.apply(2, new long[] {50, 45})))
-        .thenReturn(
-            Map.of(
-                /* progress 100% */
-                topicPartition.apply("topic-1", 0), replica.apply(2, new long[] {50, 50}),
-                topicPartition.apply("topic-2", 0), replica.apply(2, new long[] {50, 50}),
-                topicPartition.apply("topic-3", 0), replica.apply(2, new long[] {50, 50})));
-
-    Thread executionThread =
-        new Thread(
-            () -> {
-              ReplicaSyncingMonitor.execute(
-                  mockTopicAdmin,
-                  Argument.parse(
-                      new ReplicaSyncingMonitor.Argument(),
-                      new String[] {
-                        "--bootstrap.servers", "whatever:9092", "--interval", interval + "ms"
-                      }));
-            });
-    tearDownTasks.add(
-        () -> {
-          if (executionThread.isAlive()) {
-            when(mockTopicAdmin.replicas(anySet())).thenThrow(RuntimeException.class);
-            executionThread.interrupt();
-          }
-        });
-
-    // act
-    executionThread.start();
-    TimeUnit.MILLISECONDS.timedJoin(executionThread, 4 * (interval) + 1000);
-
-    // assert execution will exit
-    assertSame(Thread.State.TERMINATED, executionThread.getState());
-
-    // assert important info has been printed
-    System.err.println(mockOutput.toString());
-    assertTrue(mockOutput.toString().contains("topic-1"));
-    assertTrue(mockOutput.toString().contains("topic-2"));
-    assertTrue(mockOutput.toString().contains("topic-3"));
-    assertTrue(mockOutput.toString().contains("Every replica is synced"));
-  }
-
-  @Test
-  void executeWithKeepTrack() throws InterruptedException {
-    // arrange
-    int interval = 10;
-    Admin mockTopicAdmin = mock(Admin.class);
-    when(mockTopicAdmin.topicNames()).thenReturn(Set.of("topic-1"));
-    when(mockTopicAdmin.replicas(anySet()))
-        .thenReturn(
-            Map.of(
-                /* progress 0% */
-                topicPartition.apply("topic-1", 0), replica.apply(3, new long[] {100, 0, 0})))
-        .thenReturn(
-            Map.of(
-                /* progress 50% */
-                topicPartition.apply("topic-1", 0), replica.apply(3, new long[] {100, 50, 50})))
-        .thenReturn(
-            Map.of(
-                /* progress 100% */
-                topicPartition.apply("topic-1", 0), replica.apply(3, new long[] {100, 100, 100})))
-        .thenReturn(
-            Map.of(
-                /* progress 100%, --track should not exit when progress reach 100% */
-                topicPartition.apply("topic-1", 0), replica.apply(3, new long[] {100, 100, 100})))
-        .thenReturn(
-            Map.of(
-                /* progress 0%, another alert happened */
-                topicPartition.apply("topic-1", 0),
-                replica.apply(4, new long[] {100, 100, 100, 0})))
-        .thenReturn(
-            Map.of(
-                /* progress 50% */
-                topicPartition.apply("topic-1", 0),
-                replica.apply(4, new long[] {100, 100, 100, 50})))
-        .thenReturn(
-            Map.of(
-                /* progress 100% */
-                topicPartition.apply("topic-1", 0),
-                replica.apply(4, new long[] {100, 100, 100, 100})));
-
-    Thread executionThread =
-        new Thread(
-            () -> {
-              try {
-                ReplicaSyncingMonitor.execute(
-                    mockTopicAdmin,
-                    Argument.parse(
-                        new ReplicaSyncingMonitor.Argument(),
-                        new String[] {
-                          "--bootstrap.servers",
-                          "whatever:9092",
-                          "--track",
-                          "--interval",
-                          interval + "ms"
-                        }));
-              } catch (Exception e) {
-                // swallow interrupted error
-              }
-            });
-    tearDownTasks.add(
-        () -> {
-          if (executionThread.isAlive()) {
-            when(mockTopicAdmin.replicas(anySet())).thenThrow(RuntimeException.class);
-            executionThread.interrupt();
-          }
-        });
-
-    // act
-    executionThread.start();
-    TimeUnit.MILLISECONDS.timedJoin(executionThread, 7 * (interval) + 1000);
-
-    // assert execution will not exit even if all replicas are synced
-    assertNotEquals(Thread.State.TERMINATED, executionThread.getState());
-
-    // assert TopicAdmin#replicas call multiple times
-    verify(mockTopicAdmin, atLeast(4)).replicas(anySet());
-
-    // assert important info has been printed
-    assertTrue(mockOutput.toString().contains("topic-1"));
-    assertTrue(mockOutput.toString().contains("Every replica is synced"));
-  }
-
-  @Test
-  void executeWithTopic() {
-    // arrange
-    int interval = 10;
-    Admin mockTopicAdmin = mock(Admin.class);
-    when(mockTopicAdmin.replicas(Set.of("target-topic")))
-        .thenReturn(
-            Map.of(
-                topicPartition.apply("target-topic", 0),
-                replica.apply(3, new long[] {100, 100, 100})));
-
-    Runnable execution =
-        () -> {
-          try {
-            ReplicaSyncingMonitor.execute(
-                mockTopicAdmin,
-                Argument.parse(
-                    new ReplicaSyncingMonitor.Argument(),
-                    new String[] {
-                      "--bootstrap.servers",
-                      "whatever:9092",
-                      "--topics",
-                      "target-topic",
-                      "--interval",
-                      interval + "ms"
-                    }));
-          } catch (IllegalStateException e) {
-            // immediate fail due to bad behavior of --topic flag
-            fail();
-          }
-        };
-
-    // act
-    execution.run();
-
-    // assert TopicAdmin#replicas call at least 1 times with Set.of("target-topic")
-    verify(mockTopicAdmin, atLeast(1)).replicas(Set.of("target-topic"));
-  }
-
-  @Test
-  void findNonSyncedTopicPartition() {
-    // arrange
-    final Admin mockTopicAdmin = mock(Admin.class);
-    final Set<String> topics = Set.of("topic1", "topic2");
-    final List<Replica> replicaList1 =
-        List.of(
-            new Replica(0, 0, 0, true, true, false, false, true, "/tmp/broker0/logA"),
-            new Replica(1, 0, 0, true, true, false, false, false, "/tmp/broker1/logA"));
-    final List<Replica> replicaList2 =
-        List.of(
-            new Replica(0, 0, 100, true, false, false, false, true, "/tmp/broker0/logB"),
-            new Replica(1, 0, 100, true, true, false, false, false, "/tmp/broker1/logB"));
-    when(mockTopicAdmin.replicas(any()))
-        .thenReturn(
-            Map.of(
-                TopicPartition.of("topic1", 0), replicaList1,
-                TopicPartition.of("topic2", 0), replicaList2,
-                TopicPartition.of("topic2", 1), replicaList2));
-
-    // act
-    Set<TopicPartition> nonSyncedTopicPartition =
-        ReplicaSyncingMonitor.findNonSyncedTopicPartition(mockTopicAdmin, topics);
-
-    // assert
-    assertTrue(nonSyncedTopicPartition.contains(TopicPartition.of("topic2", 1)));
-    assertTrue(nonSyncedTopicPartition.contains(TopicPartition.of("topic2", 0)));
-    assertFalse(nonSyncedTopicPartition.contains(TopicPartition.of("topic1", 0)));
-  }
-
-  @Test
-  void ensureArgumentFlagExists() {
-    // arrange
-    var correct =
-        Set.of(
-            "--bootstrap.servers localhost:5566",
-            "--bootstrap.servers localhost:5566 --track",
-            "--bootstrap.servers localhost:5566 --topics my-topic --track",
-            "--bootstrap.servers localhost:5566 --interval 1234");
-    var incorrect =
-        Set.of(
-            "--bootstrap.servers localhost:5566 --whatever",
-            "--bootstrap.servers localhost:5566 sad",
-            "wuewuewuewue",
-            "--server");
-
-    // act
-    Consumer<String[]> execution =
-        (String[] args) -> Argument.parse(new ReplicaSyncingMonitor.Argument(), args);
-
-    // assert
-    correct.stream()
-        .map(args -> args.split(" "))
-        .forEach(args -> assertDoesNotThrow(() -> execution.accept(args)));
-    incorrect.stream()
-        .map(args -> args.split(" "))
-        .forEach(args -> assertThrows(Exception.class, () -> execution.accept(args)));
-  }
-
-  @ParameterizedTest
-  @CsvSource(
-      delimiterString = ",",
-      value = {
-        // leader, previous, current, interval, dataRatePerSec, Progress, Remaining, test-purpose
-        "  100   , 0       , 50     , 1000    , 50.0          , 50      , 1        , test",
-        "  200   , 0       , 100    , 1000    , 100.0         , 50      , 1        , test",
-        "  100   , 25      , 50     , 1000    , 25.0          , 50      , 2        , test",
-        "  100   , 0       , 10     , 10000   , 1.0           , 10      , 90       , 10 sec interval",
-        "  100   , 50      , 50     , 1000    , 0.0           , 50      , -1       , stalled progress",
-        "  0     , 50      , 0      , 1000    , 0.0           , 100     , -1       , log shrink(zero)",
-        "  40    , 50      , 10     , 1000    , 0.0           , 25      , -1       , log shrink(progress fall)",
-        "  40    , 50      , 20     , 1000    , 0.0           , 50      , -1       , log shrink(progress fall)",
-        "  100   , 50      , 20     , 1000    , 0.0           , 20      , -1       , log shrink(progress fall)",
-        "  10    , 50      , 30     , 1000    , 0.0           , NaN     , -1       , log shrink(progress fall, overflow)",
-        "  10    , 30      , 50     , 1000    , 20.0          , NaN     , -1       , log shrink(overflow)",
-        "  100   , 50      , 120    , 1000    , 70.0          , NaN     , -1       , log shrink(overflow)",
-      })
-  void dataRate(
-      long leaderSize,
-      long previousSize,
-      long currentSize,
-      int interval,
-      double expectedDataRatePerSec,
-      double expectedProgress,
-      int expectedRemainingTime) {
-    // act
-    var progress =
-        new ReplicaSyncingMonitor.ProgressInfo(
-            DataSize.Byte.of(leaderSize),
-            DataSize.Byte.of(previousSize),
-            DataSize.Byte.of(currentSize),
-            Duration.ofMillis(interval));
-
-    // assert
-    assertEquals(expectedDataRatePerSec, progress.dataRate(DataUnit.Byte, ChronoUnit.SECONDS));
-    assertEquals(expectedProgress, progress.progress());
-    assertEquals(Duration.ofSeconds(expectedRemainingTime), progress.estimateFinishTime());
-  }
-}
diff --git a/app/src/test/java/org/astraea/app/admin/ReplicationThrottlerTest.java b/app/src/test/java/org/astraea/app/admin/ReplicationThrottlerTest.java
index db899e92..f3623496 100644
--- a/app/src/test/java/org/astraea/app/admin/ReplicationThrottlerTest.java
+++ b/app/src/test/java/org/astraea/app/admin/ReplicationThrottlerTest.java
@@ -81,13 +81,11 @@ public class ReplicationThrottlerTest extends RequireBrokerCluster {
       admin.migrator().partition(topicName, 0).moveTo(List.of(1));
       Utils.sleep(Duration.ofMillis(100));
 
-      ReplicaSyncingMonitor.main(new String[] {"--bootstrap.servers", bootstrapServer});
-
-      // 5. wait until it finished
+      // 4. wait until it finished
       Utils.waitFor(
           () ->
               admin.replicas(Set.of(topicName)).get(TopicPartition.of(topicName, 0)).stream()
-                  .filter(x -> x.broker() == 1)
+                  .filter(x -> x.nodeInfo().id() == 1)
                   .findFirst()
                   .map(Replica::inSync)
                   .orElse(false),
diff --git a/app/src/test/java/org/astraea/app/admin/SomePartitionOfflineTest.java b/app/src/test/java/org/astraea/app/admin/SomePartitionOfflineTest.java
index 2b68166c..a74d6dd7 100644
--- a/app/src/test/java/org/astraea/app/admin/SomePartitionOfflineTest.java
+++ b/app/src/test/java/org/astraea/app/admin/SomePartitionOfflineTest.java
@@ -36,7 +36,7 @@ public class SomePartitionOfflineTest extends RequireBrokerCluster {
       Utils.sleep(Duration.ofSeconds(3));
       var replicaOnBroker0 =
           admin.replicas(admin.topicNames()).entrySet().stream()
-              .filter(replica -> replica.getValue().get(0).broker() == 0)
+              .filter(replica -> replica.getValue().get(0).nodeInfo().id() == 0)
               .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
       replicaOnBroker0.forEach((tp, replica) -> Assertions.assertFalse(replica.get(0).isOffline()));
       closeBroker(0);
@@ -45,7 +45,7 @@ public class SomePartitionOfflineTest extends RequireBrokerCluster {
       Assertions.assertNotNull(logFolders().get(2));
       var offlineReplicaOnBroker0 =
           admin.replicas(admin.topicNames()).entrySet().stream()
-              .filter(replica -> replica.getValue().get(0).broker() == 0)
+              .filter(replica -> replica.getValue().get(0).nodeInfo().id() == 0)
               .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
       offlineReplicaOnBroker0.forEach(
           (tp, replica) -> Assertions.assertTrue(replica.get(0).isOffline()));
diff --git a/app/src/test/java/org/astraea/app/balancer/BalancerUtilsIntegratedTest.java b/app/src/test/java/org/astraea/app/balancer/BalancerUtilsIntegratedTest.java
new file mode 100644
index 00000000..a67384e6
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/balancer/BalancerUtilsIntegratedTest.java
@@ -0,0 +1,72 @@
+/*
+ * 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.balancer;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.IntStream;
+import org.astraea.app.admin.Admin;
+import org.astraea.app.admin.Replica;
+import org.astraea.app.admin.TopicPartition;
+import org.astraea.app.balancer.log.ClusterLogAllocation;
+import org.astraea.app.balancer.log.LogPlacement;
+import org.astraea.app.common.Utils;
+import org.astraea.app.producer.Producer;
+import org.astraea.app.service.RequireBrokerCluster;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class BalancerUtilsIntegratedTest extends RequireBrokerCluster {
+
+  @Test
+  void testUpdate() {
+    var topicName = Utils.randomString(5);
+    try (var admin = Admin.of(bootstrapServers())) {
+      admin.creator().topic(topicName).create();
+      Utils.sleep(Duration.ofSeconds(3));
+
+      try (var producer = Producer.of(bootstrapServers())) {
+        IntStream.range(0, 100)
+            .forEach(ignored -> producer.sender().topic(topicName).key(new byte[10]).run());
+      }
+
+      var clusterInfo = admin.clusterInfo(Set.of(topicName));
+      clusterInfo.replicas().forEach(r -> Assertions.assertTrue(((Replica) r).size() > 0));
+
+      var replica = clusterInfo.replicas().iterator().next();
+      var newBrokerId =
+          brokerIds().stream().filter(id -> id != replica.nodeInfo().id()).findFirst().get();
+
+      var merged =
+          BalancerUtils.update(
+              clusterInfo,
+              ClusterLogAllocation.of(
+                  Map.of(
+                      TopicPartition.of(topicName, 0),
+                      // change the broker
+                      List.of(LogPlacement.of(newBrokerId, ((Replica) replica).dataFolder())))));
+
+      Assertions.assertEquals(clusterInfo.replicas().size(), merged.replicas().size());
+      Assertions.assertEquals(clusterInfo.topics().size(), merged.topics().size());
+      merged.replicas().forEach(r -> Assertions.assertTrue(((Replica) r).size() > 0));
+      Assertions.assertEquals(1, merged.replicas(topicName).size());
+      Assertions.assertEquals(newBrokerId, merged.replicas(topicName).get(0).nodeInfo().id());
+    }
+  }
+}
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 f5d6caab..3091acfe 100644
--- a/app/src/test/java/org/astraea/app/balancer/BalancerUtilsTest.java
+++ b/app/src/test/java/org/astraea/app/balancer/BalancerUtilsTest.java
@@ -66,8 +66,8 @@ class BalancerUtilsTest {
   void testMockClusterInfoAllocation() {
     var tp1 = TopicPartition.of("testMockCluster", 1);
     var tp2 = TopicPartition.of("testMockCluster", 0);
-    var logPlacement1 = List.of(LogPlacement.of(0), LogPlacement.of(1));
-    var logPlacement2 = List.of(LogPlacement.of(1), LogPlacement.of(2));
+    var logPlacement1 = List.of(LogPlacement.of(0, "/data"), LogPlacement.of(1, "/data"));
+    var logPlacement2 = List.of(LogPlacement.of(1, "/data1"), LogPlacement.of(2, "/data1"));
     var nodes =
         new Node[] {
           new Node(0, "localhost", 9092),
diff --git a/app/src/test/java/org/astraea/app/balancer/FakeClusterInfo.java b/app/src/test/java/org/astraea/app/balancer/FakeClusterInfo.java
index 942805aa..e926e753 100644
--- a/app/src/test/java/org/astraea/app/balancer/FakeClusterInfo.java
+++ b/app/src/test/java/org/astraea/app/balancer/FakeClusterInfo.java
@@ -26,6 +26,7 @@ import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.admin.NodeInfo;
+import org.astraea.app.admin.Replica;
 import org.astraea.app.admin.ReplicaInfo;
 import org.astraea.app.admin.TopicPartition;
 
@@ -84,38 +85,37 @@ public class FakeClusterInfo implements ClusterInfo {
                     IntStream.range(0, replicaCount)
                         .mapToObj(
                             r ->
-                                ReplicaInfo.of(
-                                    tp.topic(),
-                                    tp.partition(),
-                                    nodes.get(r),
-                                    r == 0,
-                                    true,
-                                    false,
-                                    dataDirectoryList.get(
-                                        tp.partition() % dataDirectories.size()))))
-            .collect(Collectors.groupingBy(ReplicaInfo::topic));
+                                (ReplicaInfo)
+                                    Replica.of(
+                                        tp.topic(),
+                                        tp.partition(),
+                                        nodes.get(r),
+                                        0,
+                                        -1,
+                                        r == 0,
+                                        true,
+                                        false,
+                                        false,
+                                        false,
+                                        dataDirectoryList.get(
+                                            tp.partition() % dataDirectories.size()))))
+            .collect(Collectors.toUnmodifiableList());
 
     return new FakeClusterInfo(
         nodes,
         nodes.stream()
             .collect(Collectors.toMap(NodeInfo::id, n -> new HashSet<String>(dataDirectories))),
-        topics,
         replicas);
   }
 
   private final List<NodeInfo> nodes;
   private final Map<Integer, Set<String>> dataDirectories;
-  private final Set<String> topics;
-  private final Map<String, List<ReplicaInfo>> replicas;
+  private final List<ReplicaInfo> replicas;
 
   FakeClusterInfo(
-      List<NodeInfo> nodes,
-      Map<Integer, Set<String>> dataDirectories,
-      Set<String> topics,
-      Map<String, List<ReplicaInfo>> replicas) {
+      List<NodeInfo> nodes, Map<Integer, Set<String>> dataDirectories, List<ReplicaInfo> replicas) {
     this.nodes = nodes;
     this.dataDirectories = dataDirectories;
-    this.topics = topics;
     this.replicas = replicas;
   }
 
@@ -129,12 +129,7 @@ public class FakeClusterInfo implements ClusterInfo {
   }
 
   @Override
-  public Set<String> topics() {
-    return topics;
-  }
-
-  @Override
-  public List<ReplicaInfo> replicas(String topic) {
-    return replicas.getOrDefault(topic, List.of());
+  public List<ReplicaInfo> replicas() {
+    return replicas;
   }
 }
diff --git a/app/src/test/java/org/astraea/app/balancer/executor/RebalanceAdminTest.java b/app/src/test/java/org/astraea/app/balancer/executor/RebalanceAdminTest.java
index 07211113..ae8bd6b5 100644
--- a/app/src/test/java/org/astraea/app/balancer/executor/RebalanceAdminTest.java
+++ b/app/src/test/java/org/astraea/app/balancer/executor/RebalanceAdminTest.java
@@ -32,6 +32,7 @@ import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
 import org.astraea.app.admin.Admin;
+import org.astraea.app.admin.NodeInfo;
 import org.astraea.app.admin.Replica;
 import org.astraea.app.admin.TopicPartition;
 import org.astraea.app.admin.TopicPartitionReplica;
@@ -53,7 +54,6 @@ class RebalanceAdminTest extends RequireBrokerCluster {
     try (var admin = Admin.of(bootstrapServers())) {
       var topic = prepareTopic(admin, 1, (short) 1);
       var rebalanceAdmin = prepareRebalanceAdmin(admin);
-      RebalanceAdminImpl.changeRetrialTime(Duration.ofMillis(50));
 
       // scale the replica size from 1 to 3, to the following data dir
       var logFolder0 = randomElement(logFolders().get(0));
@@ -76,10 +76,11 @@ class RebalanceAdminTest extends RequireBrokerCluster {
       var replicas = admin.replicas(Set.of(topic)).get(topicPartition);
 
       Assertions.assertEquals(
-          List.of(0, 1, 2), replicas.stream().map(Replica::broker).collect(Collectors.toList()));
-      Assertions.assertEquals(logFolder0, replicas.get(0).path());
-      Assertions.assertEquals(logFolder1, replicas.get(1).path());
-      Assertions.assertEquals(logFolder2, replicas.get(2).path());
+          List.of(0, 1, 2),
+          replicas.stream().map(Replica::nodeInfo).map(NodeInfo::id).collect(Collectors.toList()));
+      Assertions.assertEquals(logFolder0, replicas.get(0).dataFolder());
+      Assertions.assertEquals(logFolder1, replicas.get(1).dataFolder());
+      Assertions.assertEquals(logFolder2, replicas.get(2).dataFolder());
     }
   }
 
@@ -92,16 +93,15 @@ class RebalanceAdminTest extends RequireBrokerCluster {
       var topicPartition = TopicPartition.of(topic, 0);
       var rebalanceAdmin = prepareRebalanceAdmin(admin);
       // decrease the debouncing time so the test has higher chance to fail
-      RebalanceAdminImpl.changeRetrialTime(Duration.ofMillis(150));
       prepareData(topic, 0, DataSize.MiB.of(256));
       Supplier<Replica> replicaNow = () -> admin.replicas(Set.of(topic)).get(topicPartition).get(0);
       var originalReplica = replicaNow.get();
       var nextDir =
-          logFolders().get(originalReplica.broker()).stream()
-              .filter(name -> !name.equals(originalReplica.path()))
+          logFolders().get(originalReplica.nodeInfo().id()).stream()
+              .filter(name -> !name.equals(originalReplica.dataFolder()))
               .findAny()
               .orElseThrow();
-      var expectedPlacement = LogPlacement.of(originalReplica.broker(), nextDir);
+      var expectedPlacement = LogPlacement.of(originalReplica.nodeInfo().id(), nextDir);
 
       // act, change the dir of the only replica
       var task =
@@ -112,7 +112,7 @@ class RebalanceAdminTest extends RequireBrokerCluster {
       var finalReplica = replicaNow.get();
       Assertions.assertTrue(finalReplica.inSync());
       Assertions.assertFalse(finalReplica.isFuture());
-      Assertions.assertEquals(originalReplica.broker(), finalReplica.broker());
+      Assertions.assertEquals(originalReplica.nodeInfo(), finalReplica.nodeInfo());
     }
   }
 
@@ -143,16 +143,18 @@ class RebalanceAdminTest extends RequireBrokerCluster {
       var rebalanceAdmin = prepareRebalanceAdmin(admin);
       var beginReplica = admin.replicas().get(topicPartition).get(0);
       var otherDataDir =
-          admin.brokerFolders().get(beginReplica.broker()).stream()
-              .filter(dir -> !dir.equals(beginReplica.path()))
+          admin.brokerFolders().get(beginReplica.nodeInfo().id()).stream()
+              .filter(dir -> !dir.equals(beginReplica.dataFolder()))
               .findAny()
               .orElseThrow();
-      RebalanceAdminImpl.changeRetrialTime(Duration.ofMillis(150));
       prepareData(topic, 0, DataSize.MiB.of(32));
       // let two brokers join the replica list
       admin.migrator().partition(topic, 0).moveTo(List.of(0, 1, 2));
       // let the existing replica change its directory
-      admin.migrator().partition(topic, 0).moveTo(Map.of(beginReplica.broker(), otherDataDir));
+      admin
+          .migrator()
+          .partition(topic, 0)
+          .moveTo(Map.of(beginReplica.nodeInfo().id(), otherDataDir));
 
       // act
       long time0 = System.currentTimeMillis();
@@ -190,10 +192,11 @@ class RebalanceAdminTest extends RequireBrokerCluster {
                       .filter(x -> x.getKey().topic().equals(topic))
                       .filter(x -> x.getKey().partition() == 0)
                       .flatMap(x -> x.getValue().stream())
-                      .filter(Replica::leader)
+                      .filter(Replica::isLeader)
                       .findFirst()
                       .orElseThrow()
-                      .broker();
+                      .nodeInfo()
+                      .id();
 
       int oldLeader = leaderNow.get();
 
@@ -232,10 +235,11 @@ class RebalanceAdminTest extends RequireBrokerCluster {
                       .filter(x -> x.getKey().topic().equals(topic))
                       .filter(x -> x.getKey().partition() == 0)
                       .flatMap(x -> x.getValue().stream())
-                      .filter(Replica::leader)
+                      .filter(Replica::isLeader)
                       .findFirst()
                       .orElseThrow()
-                      .broker();
+                      .nodeInfo()
+                      .id();
 
       int oldLeader = leaderNow.get();
 
diff --git a/app/src/test/java/org/astraea/app/balancer/executor/StraightPlanExecutorTest.java b/app/src/test/java/org/astraea/app/balancer/executor/StraightPlanExecutorTest.java
index 5c857cce..bb9a2510 100644
--- a/app/src/test/java/org/astraea/app/balancer/executor/StraightPlanExecutorTest.java
+++ b/app/src/test/java/org/astraea/app/balancer/executor/StraightPlanExecutorTest.java
@@ -77,49 +77,4 @@ class StraightPlanExecutorTest extends RequireBrokerCluster {
       System.out.println(ClusterLogAllocation.toString(originalAllocation));
     }
   }
-
-  @Test
-  void testRunNoDirSpecified() {
-    // arrange
-    try (Admin admin = Admin.of(bootstrapServers())) {
-      final var topicName = "StraightPlanExecutorTest_" + Utils.randomString(8);
-      admin.creator().topic(topicName).numberOfPartitions(10).numberOfReplicas((short) 3).create();
-      Utils.sleep(Duration.ofSeconds(2));
-      final var originalAllocation = ClusterLogAllocation.of(admin.clusterInfo(Set.of(topicName)));
-      Utils.sleep(Duration.ofSeconds(3));
-      final var broker0 = 0;
-      final var broker1 = 1;
-      final var broker2 = 2;
-      final var onlyPlacement =
-          List.of(LogPlacement.of(broker0), LogPlacement.of(broker1), LogPlacement.of(broker2));
-      final var allocationMap =
-          IntStream.range(0, 10)
-              .mapToObj(i -> TopicPartition.of(topicName, i))
-              .collect(Collectors.toUnmodifiableMap(tp -> tp, tp -> onlyPlacement));
-      final var expectedAllocation = ClusterLogAllocation.of(allocationMap);
-      final var expectedTopicPartition = expectedAllocation.topicPartitions();
-      final var rebalanceAdmin = RebalanceAdmin.of(admin, (s) -> s.equals(topicName));
-
-      // act
-      new StraightPlanExecutor().run(rebalanceAdmin, expectedAllocation);
-
-      // assert
-      final var currentAllocation = ClusterLogAllocation.of(admin.clusterInfo(Set.of(topicName)));
-      final var currentTopicPartition = currentAllocation.topicPartitions();
-      Assertions.assertEquals(expectedTopicPartition, currentTopicPartition);
-      expectedTopicPartition.forEach(
-          topicPartition ->
-              Assertions.assertTrue(
-                  LogPlacement.isMatch(
-                      currentAllocation.logPlacements(topicPartition),
-                      expectedAllocation.logPlacements(topicPartition))));
-
-      System.out.println("Expected:");
-      System.out.println(ClusterLogAllocation.toString(expectedAllocation));
-      System.out.println("Current:");
-      System.out.println(ClusterLogAllocation.toString(currentAllocation));
-      System.out.println("Original:");
-      System.out.println(ClusterLogAllocation.toString(originalAllocation));
-    }
-  }
 }
diff --git a/app/src/test/java/org/astraea/app/balancer/log/ClusterLogAllocationTest.java b/app/src/test/java/org/astraea/app/balancer/log/ClusterLogAllocationTest.java
index d9ec04e3..7b680e25 100644
--- a/app/src/test/java/org/astraea/app/balancer/log/ClusterLogAllocationTest.java
+++ b/app/src/test/java/org/astraea/app/balancer/log/ClusterLogAllocationTest.java
@@ -38,7 +38,8 @@ class ClusterLogAllocationTest {
         IllegalArgumentException.class, () -> ClusterLogAllocation.of(badAllocation0));
 
     // partial topic/partition
-    var badAllocation1 = Map.of(TopicPartition.of("topic", "999"), List.of(LogPlacement.of(1001)));
+    var badAllocation1 =
+        Map.of(TopicPartition.of("topic", "999"), List.of(LogPlacement.of(1001, "xx")));
     Assertions.assertThrows(
         IllegalArgumentException.class, () -> ClusterLogAllocation.of(badAllocation1));
 
@@ -46,7 +47,10 @@ class ClusterLogAllocationTest {
     var badAllocation2 =
         Map.of(
             TopicPartition.of("topic", "0"),
-            List.of(LogPlacement.of(1001), LogPlacement.of(1001), LogPlacement.of(1001)));
+            List.of(
+                LogPlacement.of(1001, "xx"),
+                LogPlacement.of(1001, "xx"),
+                LogPlacement.of(1001, "xx")));
     Assertions.assertThrows(
         IllegalArgumentException.class, () -> ClusterLogAllocation.of(badAllocation2));
   }
@@ -61,12 +65,6 @@ class ClusterLogAllocationTest {
 
     Assertions.assertEquals(
         1, clusterLogAllocation.logPlacements(sourceTopicPartition).get(0).broker());
-    Assertions.assertNull(
-        clusterLogAllocation
-            .logPlacements(sourceTopicPartition)
-            .get(0)
-            .logDirectory()
-            .orElse(null));
   }
 
   @ParameterizedTest
@@ -85,11 +83,7 @@ class ClusterLogAllocationTest {
         1, clusterLogAllocation.logPlacements(sourceTopicPartition0).get(0).broker());
     Assertions.assertEquals(
         dataDirectory,
-        clusterLogAllocation
-            .logPlacements(sourceTopicPartition0)
-            .get(0)
-            .logDirectory()
-            .orElse(null));
+        clusterLogAllocation.logPlacements(sourceTopicPartition0).get(0).dataFolder());
 
     final var sourceTopicPartition1 = TopicPartition.of("topic", "0");
     clusterLogAllocation =
@@ -98,11 +92,7 @@ class ClusterLogAllocationTest {
         1, clusterLogAllocation.logPlacements(sourceTopicPartition1).get(0).broker());
     Assertions.assertEquals(
         dataDirectory,
-        clusterLogAllocation
-            .logPlacements(sourceTopicPartition1)
-            .get(0)
-            .logDirectory()
-            .orElse(null));
+        clusterLogAllocation.logPlacements(sourceTopicPartition1).get(0).dataFolder());
   }
 
   @Test
@@ -129,12 +119,7 @@ class ClusterLogAllocationTest {
     Assertions.assertEquals(
         0, allocation.logPlacements(TopicPartition.of("topic", "0")).get(0).broker());
     Assertions.assertEquals(
-        "/nowhere",
-        allocation
-            .logPlacements(TopicPartition.of("topic", "0"))
-            .get(0)
-            .logDirectory()
-            .orElseThrow());
+        "/nowhere", allocation.logPlacements(TopicPartition.of("topic", "0")).get(0).dataFolder());
     Assertions.assertEquals(0, allocation.logPlacements(TopicPartition.of("no", "0")).size());
     allocation.logPlacements(TopicPartition.of("no", "0"));
   }
diff --git a/app/src/test/java/org/astraea/app/balancer/log/LogPlacementTest.java b/app/src/test/java/org/astraea/app/balancer/log/LogPlacementTest.java
index 6592b5e6..437e6d1c 100644
--- a/app/src/test/java/org/astraea/app/balancer/log/LogPlacementTest.java
+++ b/app/src/test/java/org/astraea/app/balancer/log/LogPlacementTest.java
@@ -52,26 +52,4 @@ class LogPlacementTest {
         List.of(LogPlacement.of(0, "/Aaa"), LogPlacement.of(1, "/B"), LogPlacement.of(2, "/C"));
     Assertions.assertFalse(LogPlacement.isMatch(placement0, placement1));
   }
-
-  @Test
-  @DisplayName("Optional log dir")
-  void noMatch2() {
-    final var sourcePlacement =
-        List.of(LogPlacement.of(0, "/A"), LogPlacement.of(1, "/B"), LogPlacement.of(2, "/C"));
-    final var targetPlacement =
-        List.of(LogPlacement.of(0), LogPlacement.of(1, "/B"), LogPlacement.of(2, "/C"));
-    // don't care which log dir placement[0] will eventually be.
-    Assertions.assertTrue(LogPlacement.isMatch(sourcePlacement, targetPlacement));
-  }
-
-  @Test
-  @DisplayName("Optional log dir")
-  void noMatch3() {
-    final var sourcePlacement =
-        List.of(LogPlacement.of(0), LogPlacement.of(1, "/B"), LogPlacement.of(2, "/C"));
-    final var targetPlacement =
-        List.of(LogPlacement.of(0, "/A"), LogPlacement.of(1, "/B"), LogPlacement.of(2, "/C"));
-    // do care which log dir placement[0] will eventually be.
-    Assertions.assertFalse(LogPlacement.isMatch(sourcePlacement, targetPlacement));
-  }
 }
diff --git a/app/src/test/java/org/astraea/app/common/json/OptionalIntSerializerTest.java b/app/src/test/java/org/astraea/app/common/json/OptionalIntSerializerTest.java
deleted file mode 100644
index 71e7a3d3..00000000
--- a/app/src/test/java/org/astraea/app/common/json/OptionalIntSerializerTest.java
+++ /dev/null
@@ -1,52 +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.common.json;
-
-import com.google.gson.GsonBuilder;
-import java.util.OptionalInt;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Test;
-
-class OptionalIntSerializerTest {
-
-  @SuppressWarnings("OptionalUsedAsFieldOrParameterType")
-  private static class Dummy {
-
-    private Dummy(int value) {
-      this.value = OptionalInt.of(value);
-    }
-
-    private Dummy() {
-      this.value = OptionalInt.empty();
-    }
-
-    OptionalInt value;
-  }
-
-  @Test
-  void test() {
-    var gson =
-        new GsonBuilder()
-            .registerTypeAdapter(OptionalInt.class, new OptionalIntSerializer())
-            .create();
-    Assertions.assertEquals("1", gson.toJson(OptionalInt.of(1)));
-    Assertions.assertEquals("1024", gson.toJson(OptionalInt.of(1024)));
-    Assertions.assertEquals("null", gson.toJson(OptionalInt.empty()));
-    Assertions.assertEquals("{}", gson.toJson(new Dummy()));
-    Assertions.assertEquals("{\"value\":5}", gson.toJson(new Dummy(5)));
-  }
-}
diff --git a/app/src/test/java/org/astraea/app/common/json/OptionalSerializerTest.java b/app/src/test/java/org/astraea/app/common/json/OptionalSerializerTest.java
deleted file mode 100644
index 78cdae52..00000000
--- a/app/src/test/java/org/astraea/app/common/json/OptionalSerializerTest.java
+++ /dev/null
@@ -1,52 +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.common.json;
-
-import com.google.gson.GsonBuilder;
-import java.util.Optional;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Test;
-
-class OptionalSerializerTest {
-
-  @SuppressWarnings("OptionalUsedAsFieldOrParameterType")
-  private static class Dummy {
-
-    private <T> Dummy(T value) {
-      this.value = Optional.of(value);
-    }
-
-    private Dummy() {
-      this.value = Optional.empty();
-    }
-
-    Optional<?> value;
-  }
-
-  @Test
-  void test() {
-    var gson =
-        new GsonBuilder().registerTypeAdapter(Optional.class, new OptionalSerializer()).create();
-    Assertions.assertEquals("1", gson.toJson(Optional.of(1)));
-    Assertions.assertEquals("1024", gson.toJson(Optional.of(1024)));
-    Assertions.assertEquals("\"Hello\"", gson.toJson(Optional.of("Hello")));
-    Assertions.assertEquals("null", gson.toJson(Optional.empty()));
-    Assertions.assertEquals("{}", gson.toJson(new Dummy()));
-    Assertions.assertEquals("{\"value\":5}", gson.toJson(new Dummy(5)));
-    Assertions.assertEquals("{\"value\":\"Hello\"}", gson.toJson(new Dummy("Hello")));
-  }
-}
diff --git a/app/src/test/java/org/astraea/app/cost/ClusterInfoIntegratedTest.java b/app/src/test/java/org/astraea/app/cost/ClusterInfoIntegratedTest.java
new file mode 100644
index 00000000..4dbd3556
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/cost/ClusterInfoIntegratedTest.java
@@ -0,0 +1,58 @@
+/*
+ * 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.time.Duration;
+import java.util.Optional;
+import org.astraea.app.admin.Admin;
+import org.astraea.app.common.Utils;
+import org.astraea.app.service.RequireBrokerCluster;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class ClusterInfoIntegratedTest extends RequireBrokerCluster {
+
+  @Test
+  void testQuery() {
+    try (var admin = Admin.of(bootstrapServers())) {
+      admin.creator().topic(Utils.randomString()).numberOfPartitions(10).create();
+      Utils.sleep(Duration.ofSeconds(2));
+
+      var clusterInfo = admin.clusterInfo();
+
+      // search by replica
+      clusterInfo
+          .replicas()
+          .forEach(
+              r ->
+                  Assertions.assertNotEquals(
+                      Optional.empty(), clusterInfo.replica(r.topicPartitionReplica())));
+
+      // search by topic
+      clusterInfo
+          .topics()
+          .forEach(t -> Assertions.assertNotEquals(0, clusterInfo.replicas(t).size()));
+      clusterInfo
+          .topics()
+          .forEach(t -> Assertions.assertNotEquals(0, clusterInfo.availableReplicas(t).size()));
+      clusterInfo
+          .topics()
+          .forEach(
+              t -> Assertions.assertNotEquals(0, clusterInfo.availableReplicaLeaders(t).size()));
+    }
+  }
+}
diff --git a/app/src/test/java/org/astraea/app/metrics/client/consumer/ConsumerMetricsTest.java b/app/src/test/java/org/astraea/app/metrics/client/consumer/ConsumerMetricsTest.java
index e2936c36..0ffe87e1 100644
--- a/app/src/test/java/org/astraea/app/metrics/client/consumer/ConsumerMetricsTest.java
+++ b/app/src/test/java/org/astraea/app/metrics/client/consumer/ConsumerMetricsTest.java
@@ -38,7 +38,8 @@ public class ConsumerMetricsTest extends RequireBrokerCluster {
             Consumer.forTopics(Set.of(topic)).bootstrapServers(bootstrapServers()).build()) {
       admin.creator().topic(topic).numberOfPartitions(1).create();
       Utils.sleep(Duration.ofSeconds(3));
-      var owner = admin.replicas(Set.of(topic)).get(TopicPartition.of(topic, 0)).get(0).broker();
+      var owner =
+          admin.replicas(Set.of(topic)).get(TopicPartition.of(topic, 0)).get(0).nodeInfo().id();
       consumer.poll(Duration.ofSeconds(5));
       var metrics = ConsumerMetrics.node(MBeanClient.local(), owner);
       Assertions.assertEquals(1, metrics.size());
diff --git a/app/src/test/java/org/astraea/app/metrics/client/producer/ProducerMetricsTest.java b/app/src/test/java/org/astraea/app/metrics/client/producer/ProducerMetricsTest.java
index 17cbca27..504a7bdf 100644
--- a/app/src/test/java/org/astraea/app/metrics/client/producer/ProducerMetricsTest.java
+++ b/app/src/test/java/org/astraea/app/metrics/client/producer/ProducerMetricsTest.java
@@ -39,7 +39,8 @@ public class ProducerMetricsTest extends RequireBrokerCluster {
         var producer = Producer.of(bootstrapServers())) {
       admin.creator().topic(topic).numberOfPartitions(1).create();
       Utils.sleep(Duration.ofSeconds(3));
-      var owner = admin.replicas(Set.of(topic)).get(TopicPartition.of(topic, 0)).get(0).broker();
+      var owner =
+          admin.replicas(Set.of(topic)).get(TopicPartition.of(topic, 0)).get(0).nodeInfo().id();
       producer.sender().topic(topic).run().toCompletableFuture().get();
       var metrics = ProducerMetrics.node(MBeanClient.local(), owner);
       Assertions.assertEquals(1, metrics.size());
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 3ece0fa2..10c5f0d0 100644
--- a/app/src/test/java/org/astraea/app/web/BalancerHandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/BalancerHandlerTest.java
@@ -16,15 +16,15 @@
  */
 package org.astraea.app.web;
 
+import java.nio.charset.StandardCharsets;
 import java.time.Duration;
+import java.util.List;
 import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 import org.astraea.app.admin.Admin;
-import org.astraea.app.admin.ClusterBean;
-import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.common.Utils;
-import org.astraea.app.cost.ClusterCost;
-import org.astraea.app.cost.HasClusterCost;
+import org.astraea.app.producer.Producer;
 import org.astraea.app.service.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
@@ -33,11 +33,9 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
 
   @Test
   void testReport() {
-    var topicName = Utils.randomString(10);
+    var topicName = createAndProduceTopic(1).get(0);
     try (var admin = Admin.of(bootstrapServers())) {
-      admin.creator().topic(topicName).numberOfPartitions(10).numberOfReplicas((short) 3).create();
-      Utils.sleep(Duration.ofSeconds(3));
-      var handler = new BalancerHandler(admin, new MyCost());
+      var handler = new BalancerHandler(admin);
       var report =
           Assertions.assertInstanceOf(
               BalancerHandler.Report.class,
@@ -45,17 +43,94 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
       Assertions.assertEquals(30, report.limit);
       Assertions.assertNotEquals(0, report.changes.size());
       Assertions.assertTrue(report.cost >= report.newCost);
-      Assertions.assertEquals(MyCost.class.getSimpleName(), report.function);
+      Assertions.assertEquals(handler.costFunction.getClass().getSimpleName(), report.function);
+      // "before" should record size
+      report.changes.stream()
+          .flatMap(c -> c.before.stream())
+          .forEach(p -> Assertions.assertNotEquals(0, p.size));
+      // "after" should NOT record size
+      report.changes.stream()
+          .flatMap(c -> c.after.stream())
+          .forEach(p -> Assertions.assertNull(p.size));
+      Assertions.assertTrue(report.cost >= report.newCost);
     }
   }
 
-  private static class MyCost implements HasClusterCost {
-    private final AtomicInteger count = new AtomicInteger(0);
+  @Test
+  void testTopic() {
+    var topicNames = createAndProduceTopic(3);
+    try (var admin = Admin.of(bootstrapServers())) {
+      var handler = new BalancerHandler(admin);
+      var report =
+          Assertions.assertInstanceOf(
+              BalancerHandler.Report.class,
+              handler.get(
+                  Channel.ofQueries(
+                      Map.of(
+                          BalancerHandler.LIMIT_KEY,
+                          "30",
+                          BalancerHandler.TOPICS_KEY,
+                          topicNames.get(0)))));
+      var actual =
+          report.changes.stream().map(r -> r.topic).collect(Collectors.toUnmodifiableSet());
+      Assertions.assertEquals(1, actual.size());
+      Assertions.assertEquals(topicNames.get(0), actual.iterator().next());
+      Assertions.assertTrue(report.cost >= report.newCost);
+    }
+  }
 
-    @Override
-    public ClusterCost clusterCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
-      var cost = count.getAndIncrement() == 0 ? Double.MAX_VALUE : Math.random() * 100;
-      return () -> cost;
+  @Test
+  void testTopics() {
+    var topicNames = createAndProduceTopic(3);
+    try (var admin = Admin.of(bootstrapServers())) {
+      var handler = new BalancerHandler(admin);
+      var report =
+          Assertions.assertInstanceOf(
+              BalancerHandler.Report.class,
+              handler.get(
+                  Channel.ofQueries(
+                      Map.of(
+                          BalancerHandler.LIMIT_KEY,
+                          "30",
+                          BalancerHandler.TOPICS_KEY,
+                          topicNames.get(0) + "," + topicNames.get(1)))));
+      var actual =
+          report.changes.stream().map(r -> r.topic).collect(Collectors.toUnmodifiableSet());
+      Assertions.assertEquals(2, actual.size());
+      Assertions.assertTrue(actual.contains(topicNames.get(0)));
+      Assertions.assertTrue(actual.contains(topicNames.get(1)));
+      Assertions.assertTrue(report.cost >= report.newCost);
+    }
+  }
+
+  private static List<String> createAndProduceTopic(int topicCount) {
+    try (var admin = Admin.of(bootstrapServers())) {
+      var topics =
+          IntStream.range(0, topicCount)
+              .mapToObj(ignored -> Utils.randomString(10))
+              .collect(Collectors.toUnmodifiableList());
+      topics.forEach(
+          topic ->
+              admin
+                  .creator()
+                  .topic(topic)
+                  .numberOfPartitions(3)
+                  .numberOfReplicas((short) 1)
+                  .create());
+      Utils.sleep(Duration.ofSeconds(3));
+      try (var producer = Producer.of(bootstrapServers())) {
+        IntStream.range(0, 30)
+            .forEach(
+                index ->
+                    topics.forEach(
+                        topic ->
+                            producer
+                                .sender()
+                                .topic(topic)
+                                .key(String.valueOf(index).getBytes(StandardCharsets.UTF_8))
+                                .run()));
+      }
+      return topics;
     }
   }
 }
diff --git a/app/src/test/java/org/astraea/app/web/ReassignmentHandlerTest.java b/app/src/test/java/org/astraea/app/web/ReassignmentHandlerTest.java
index 181fee51..f2734096 100644
--- a/app/src/test/java/org/astraea/app/web/ReassignmentHandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/ReassignmentHandlerTest.java
@@ -36,7 +36,12 @@ public class ReassignmentHandlerTest extends RequireBrokerCluster {
       Utils.sleep(Duration.ofSeconds(3));
 
       var currentBroker =
-          admin.replicas(Set.of(topicName)).get(TopicPartition.of(topicName, 0)).get(0).broker();
+          admin
+              .replicas(Set.of(topicName))
+              .get(TopicPartition.of(topicName, 0))
+              .get(0)
+              .nodeInfo()
+              .id();
       var nextBroker = brokerIds().stream().filter(i -> i != currentBroker).findAny().get();
 
       var body =
@@ -60,7 +65,12 @@ public class ReassignmentHandlerTest extends RequireBrokerCluster {
 
       Assertions.assertEquals(
           nextBroker,
-          admin.replicas(Set.of(topicName)).get(TopicPartition.of(topicName, 0)).get(0).broker());
+          admin
+              .replicas(Set.of(topicName))
+              .get(TopicPartition.of(topicName, 0))
+              .get(0)
+              .nodeInfo()
+              .id());
     }
   }
 
@@ -74,8 +84,8 @@ public class ReassignmentHandlerTest extends RequireBrokerCluster {
 
       var currentReplica =
           admin.replicas(Set.of(topicName)).get(TopicPartition.of(topicName, 0)).get(0);
-      var currentBroker = currentReplica.broker();
-      var currentPath = currentReplica.path();
+      var currentBroker = currentReplica.nodeInfo().id();
+      var currentPath = currentReplica.dataFolder();
       var nextPath =
           logFolders().get(currentBroker).stream()
               .filter(p -> !p.equals(currentPath))
@@ -105,7 +115,11 @@ public class ReassignmentHandlerTest extends RequireBrokerCluster {
 
       Assertions.assertEquals(
           nextPath,
-          admin.replicas(Set.of(topicName)).get(TopicPartition.of(topicName, 0)).get(0).path());
+          admin
+              .replicas(Set.of(topicName))
+              .get(TopicPartition.of(topicName, 0))
+              .get(0)
+              .dataFolder());
     }
   }
 
diff --git a/app/src/test/java/org/astraea/app/web/ThrottleHandlerTest.java b/app/src/test/java/org/astraea/app/web/ThrottleHandlerTest.java
index da8e2225..05194bf8 100644
--- a/app/src/test/java/org/astraea/app/web/ThrottleHandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/ThrottleHandlerTest.java
@@ -16,11 +16,18 @@
  */
 package org.astraea.app.web;
 
+import static org.astraea.app.web.ThrottleHandler.LogIdentity.follower;
+import static org.astraea.app.web.ThrottleHandler.LogIdentity.leader;
+
 import com.google.gson.Gson;
+import com.google.gson.JsonElement;
 import com.google.gson.JsonObject;
 import com.google.gson.JsonPrimitive;
 import java.time.Duration;
 import java.util.Map;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.StreamSupport;
 import org.astraea.app.admin.Admin;
 import org.astraea.app.admin.TopicPartition;
 import org.astraea.app.admin.TopicPartitionReplica;
@@ -37,6 +44,7 @@ public class ThrottleHandlerTest extends RequireBrokerCluster {
     try (Admin admin = Admin.of(bootstrapServers())) {
       var handler = new ThrottleHandler(admin);
       var dataRate = DataRate.MiB.of(500).perSecond();
+      var longDataRate = (long) dataRate.byteRate();
       admin
           .replicationThrottler()
           .ingress(Map.of(0, dataRate, 2, dataRate))
@@ -47,27 +55,25 @@ public class ThrottleHandlerTest extends RequireBrokerCluster {
       var jsonString = handler.get(Channel.EMPTY).json();
       var json = new Gson().fromJson(jsonString, JsonObject.class);
 
+      Function<Integer, JsonObject> findByBrokerId =
+          (brokerId) ->
+              StreamSupport.stream(json.getAsJsonArray("brokers").spliterator(), false)
+                  .map(JsonElement::getAsJsonObject)
+                  .filter(item -> item.get("id").getAsInt() == brokerId)
+                  .findFirst()
+                  .orElseThrow();
+
       // broker 0
-      Assertions.assertEquals(
-          (long) dataRate.byteRate(),
-          json.getAsJsonObject("brokers").getAsJsonObject("0").get("ingress").getAsLong());
-      Assertions.assertFalse(
-          json.getAsJsonObject("brokers").getAsJsonObject("0").keySet().contains("egress"));
+      Assertions.assertEquals(longDataRate, findByBrokerId.apply(0).get("ingress").getAsLong());
+      Assertions.assertFalse(findByBrokerId.apply(0).keySet().contains("egress"));
 
       // broker 1
-      Assertions.assertFalse(
-          json.getAsJsonObject("brokers").getAsJsonObject("1").keySet().contains("ingress"));
-      Assertions.assertEquals(
-          (long) dataRate.byteRate(),
-          json.getAsJsonObject("brokers").getAsJsonObject("1").get("egress").getAsLong());
+      Assertions.assertFalse(findByBrokerId.apply(1).keySet().contains("ingress"));
+      Assertions.assertEquals(longDataRate, findByBrokerId.apply(1).get("egress").getAsLong());
 
       // broker 2
-      Assertions.assertEquals(
-          (long) dataRate.byteRate(),
-          json.getAsJsonObject("brokers").getAsJsonObject("2").get("ingress").getAsLong());
-      Assertions.assertEquals(
-          (long) dataRate.byteRate(),
-          json.getAsJsonObject("brokers").getAsJsonObject("2").get("egress").getAsLong());
+      Assertions.assertEquals(longDataRate, findByBrokerId.apply(2).get("ingress").getAsLong());
+      Assertions.assertEquals(longDataRate, findByBrokerId.apply(2).get("egress").getAsLong());
     }
   }
 
@@ -90,10 +96,10 @@ public class ThrottleHandlerTest extends RequireBrokerCluster {
           var theReplica = replica;
           var isLeader =
               currentReplicas.get(TopicPartition.of(topicName, partition)).stream()
-                  .filter(r -> r.broker() == theReplica)
+                  .filter(r -> r.nodeInfo().id() == theReplica)
                   .findFirst()
                   .orElseThrow()
-                  .leader();
+                  .isLeader();
           var expected = new JsonObject();
           expected.add("name", new JsonPrimitive(topicName));
           expected.add("partition", new JsonPrimitive(partition));
@@ -143,14 +149,12 @@ public class ThrottleHandlerTest extends RequireBrokerCluster {
 
   @Test
   void testThrottleTargetEqual() {
-    var target0 = new ThrottleHandler.ThrottleTarget("Topic", 0, 0);
-    var target1 = new ThrottleHandler.ThrottleTarget("Topic", 0, 0);
-    var target2 =
-        new ThrottleHandler.ThrottleTarget("Topic", 0, 0, ThrottleHandler.LogIdentity.leader);
-    var target3 =
-        new ThrottleHandler.ThrottleTarget("Topic", 0, 0, ThrottleHandler.LogIdentity.follower);
-    var target4 = new ThrottleHandler.ThrottleTarget("Topic", 1, 0);
-    var target5 = new ThrottleHandler.ThrottleTarget("Topic2", 0, 0);
+    var target0 = new ThrottleHandler.TopicThrottle("Topic", 0, 0, null);
+    var target1 = new ThrottleHandler.TopicThrottle("Topic", 0, 0, null);
+    var target2 = new ThrottleHandler.TopicThrottle("Topic", 0, 0, leader);
+    var target3 = new ThrottleHandler.TopicThrottle("Topic", 0, 0, follower);
+    var target4 = new ThrottleHandler.TopicThrottle("Topic", 1, 0, null);
+    var target5 = new ThrottleHandler.TopicThrottle("Topic2", 0, 0, null);
 
     Assertions.assertEquals(target0, target0);
     Assertions.assertEquals(target0, target1);
@@ -161,4 +165,51 @@ public class ThrottleHandlerTest extends RequireBrokerCluster {
     Assertions.assertNotEquals(target0, target5);
     Assertions.assertEquals(target0.hashCode(), target1.hashCode());
   }
+
+  @Test
+  void testSerializeDeserialize() {
+    var throttle0 = new ThrottleHandler.BrokerThrottle(1001, 1L, null);
+    var throttle1 = new ThrottleHandler.TopicThrottle("MyTopic", 0, 1001, null);
+    var throttle2 = new ThrottleHandler.TopicThrottle("MyTopic", 0, 1002, null);
+    var throttle3 = new ThrottleHandler.TopicThrottle("MyTopic", 0, 1003, null);
+    var set0 = Set.of(throttle0);
+    var set1 = Set.of(throttle1, throttle2, throttle3);
+    var setting = new ThrottleHandler.ThrottleSetting(set0, set1);
+
+    var serialized = setting.json();
+    var gson = new Gson();
+    var deserialized = gson.fromJson(serialized, ThrottleHandler.ThrottleSetting.class);
+
+    Assertions.assertEquals(set0, Set.copyOf(deserialized.brokers));
+    Assertions.assertEquals(set1, Set.copyOf(deserialized.topics));
+  }
+
+  @Test
+  void testDeserialize() {
+    final String rawJson =
+        "{\"brokers\":["
+            + "{\"id\": 1001, \"ingress\":1000,\"egress\":1000},"
+            + "{\"id\": 1002, \"ingress\":1000}],"
+            + "\"topics\":["
+            + "{\"name\":\"MyTopicA\"},"
+            + "{\"name\":\"MyTopicB\",\"partition\":2},"
+            + "{\"name\":\"MyTopicC\",\"partition\":3,\"broker\":1001},"
+            + "{\"name\":\"MyTopicD\",\"partition\":4,\"broker\":1001,\"type\":\"leader\"}]}";
+    var expectedBroker =
+        Set.of(
+            new ThrottleHandler.BrokerThrottle(1001, 1000L, 1000L),
+            new ThrottleHandler.BrokerThrottle(1002, 1000L, null));
+    var expectedTopic =
+        Set.of(
+            new ThrottleHandler.TopicThrottle("MyTopicA", null, null, null),
+            new ThrottleHandler.TopicThrottle("MyTopicB", 2, null, null),
+            new ThrottleHandler.TopicThrottle("MyTopicC", 3, 1001, null),
+            new ThrottleHandler.TopicThrottle("MyTopicD", 4, 1001, leader));
+
+    var gson = new Gson();
+    var deserialized = gson.fromJson(rawJson, ThrottleHandler.ThrottleSetting.class);
+
+    Assertions.assertEquals(expectedBroker, Set.copyOf(deserialized.brokers));
+    Assertions.assertEquals(expectedTopic, Set.copyOf(deserialized.topics));
+  }
 }
diff --git a/app/src/test/java/org/astraea/app/web/TopicHandlerTest.java b/app/src/test/java/org/astraea/app/web/TopicHandlerTest.java
index 86f47a2a..2ffcea67 100644
--- a/app/src/test/java/org/astraea/app/web/TopicHandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/TopicHandlerTest.java
@@ -16,6 +16,8 @@
  */
 package org.astraea.app.web;
 
+import static java.nio.charset.StandardCharsets.UTF_8;
+
 import java.time.Duration;
 import java.util.Map;
 import java.util.NoSuchElementException;
@@ -25,6 +27,8 @@ import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import org.astraea.app.admin.Admin;
 import org.astraea.app.common.Utils;
+import org.astraea.app.consumer.Consumer;
+import org.astraea.app.producer.Producer;
 import org.astraea.app.service.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
@@ -84,11 +88,20 @@ public class TopicHandlerTest extends RequireBrokerCluster {
       admin.creator().topic(topicName).create();
       Utils.sleep(Duration.ofSeconds(3));
       var handler = new TopicHandler(admin);
-      Assertions.assertEquals(Set.of(topicName), handler.topicNames(Optional.of(topicName)));
-      Assertions.assertThrows(
-          NoSuchElementException.class,
-          () -> handler.topicNames(Optional.of(Utils.randomString(10))));
-      Assertions.assertTrue(handler.topicNames(Optional.empty()).contains(topicName));
+
+      java.util.function.Consumer<Boolean> test =
+          (listInternal) -> {
+            Assertions.assertEquals(
+                Set.of(topicName), handler.topicNames(Optional.of(topicName), listInternal));
+            Assertions.assertThrows(
+                NoSuchElementException.class,
+                () -> handler.topicNames(Optional.of(Utils.randomString(10)), listInternal));
+            Assertions.assertTrue(
+                handler.topicNames(Optional.empty(), listInternal).contains(topicName));
+          };
+
+      test.accept(true);
+      test.accept(false);
     }
   }
 
@@ -141,6 +154,47 @@ public class TopicHandlerTest extends RequireBrokerCluster {
     }
   }
 
+  @Test
+  void testQueryWithListInternal() {
+    var bootstrapServers = bootstrapServers();
+    var topicName = Utils.randomString(10);
+    try (var admin = Admin.of(bootstrapServers);
+        var producer = Producer.of(bootstrapServers);
+        var consumer =
+            Consumer.forTopics(Set.of(topicName)).bootstrapServers(bootstrapServers).build()) {
+      // producer and consumer here are used to trigger kafka to create internal topic
+      // __consumer_offsets
+      producer
+          .sender()
+          .topic(topicName)
+          .key("foo".getBytes(UTF_8))
+          .run()
+          .toCompletableFuture()
+          .join();
+      consumer.poll(Duration.ofSeconds(1));
+
+      var handler = new TopicHandler(admin);
+
+      var withInternalTopics =
+          Assertions.assertInstanceOf(
+              TopicHandler.Topics.class,
+              handler.get(Channel.ofQueries(Map.of(TopicHandler.LIST_INTERNAL, "true"))));
+      Assertions.assertTrue(
+          withInternalTopics.topics.stream().anyMatch(t -> t.name.equals("__consumer_offsets")));
+      Assertions.assertEquals(
+          1, withInternalTopics.topics.stream().filter(t -> t.name.equals(topicName)).count());
+
+      var withoutInternalTopics =
+          Assertions.assertInstanceOf(
+              TopicHandler.Topics.class,
+              handler.get(Channel.ofQueries(Map.of(TopicHandler.LIST_INTERNAL, "false"))));
+      Assertions.assertFalse(
+          withoutInternalTopics.topics.stream().anyMatch(t -> t.name.equals("__consumer_offsets")));
+      Assertions.assertEquals(
+          1, withoutInternalTopics.topics.stream().filter(t -> t.name.equals(topicName)).count());
+    }
+  }
+
   @Test
   void testCreateTopicWithReplicas() {
     var topicName = Utils.randomString(10);