Repository: skiptests/astraea
Pull request #385 (2022-05-31 07:03:47)

Add SmoothDispatch.md
#361 添加dispatch相關中文說明

chia7712: @wycccccc 有機會完成這個嗎？還是要交給別人
wycccccc: 他被遺忘在了角落，我會最近抽空完成它
wycccccc: 因爲年代久遠，我再結合實時修改了一下。麻煩再review。

Patch:
diff --git a/README.md b/README.md
index a06c1726..455df571 100644
--- a/README.md
+++ b/README.md
@@ -20,4 +20,5 @@
 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`叢集
\ No newline at end of file
+6. [Web Server](./docs/web_server/README.md): 可透過`Restful APIs`操作`Kafka`叢集 
+7. [Dispatcher](docs/dispatcher/README.md): 強大且高效率的 Kafka partitioner 實作
\ No newline at end of file
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 a57c9dfd..4ee9c335 100644
--- a/app/src/main/java/org/astraea/app/admin/Admin.java
+++ b/app/src/main/java/org/astraea/app/admin/Admin.java
@@ -290,6 +290,22 @@ public interface Admin extends Closeable {
    */
   void clearReplicationThrottle(TopicPartitionReplica log);
 
+  /**
+   * Clear the leader replication throttle related to the given topic/partition with specific broker
+   * id.
+   *
+   * @param log target to clear throttle.
+   */
+  void clearLeaderReplicationThrottle(TopicPartitionReplica log);
+
+  /**
+   * Clear the follower replication throttle related to the given topic/partition with specific
+   * broker id.
+   *
+   * @param log target to clear throttle.
+   */
+  void clearFollowerReplicationThrottle(TopicPartitionReplica log);
+
   /** Clear the ingress bandwidth of replication throttle for the specified brokers. */
   void clearIngressReplicationThrottle(Set<Integer> brokerIds);
 
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 1a2bbe11..a5abb519 100644
--- a/app/src/main/java/org/astraea/app/admin/Builder.java
+++ b/app/src/main/java/org/astraea/app/admin/Builder.java
@@ -463,8 +463,8 @@ public class Builder {
         }
 
         @Override
-        public List<Replica> replicas() {
-          return replicas;
+        public Stream<Replica> replicaStream() {
+          return replicas.stream();
         }
       };
     }
@@ -711,9 +711,35 @@ public class Builder {
         }
 
         @Override
-        public void apply() {
+        public AffectedResources apply() {
           applyBandwidth();
           applyThrottledReplicas();
+          return new AffectedResources() {
+            final Map<Integer, DataRate> ingressCopy = Map.copyOf(ingress);
+            final Map<Integer, DataRate> egressCopy = Map.copyOf(egress);
+            final Set<TopicPartitionReplica> leaderCopy = Set.copyOf(leaders);
+            final Set<TopicPartitionReplica> followerCopy = Set.copyOf(followers);
+
+            @Override
+            public Map<Integer, DataRate> ingress() {
+              return ingressCopy;
+            }
+
+            @Override
+            public Map<Integer, DataRate> egress() {
+              return egressCopy;
+            }
+
+            @Override
+            public Set<TopicPartitionReplica> leaders() {
+              return leaderCopy;
+            }
+
+            @Override
+            public Set<TopicPartitionReplica> followers() {
+              return followerCopy;
+            }
+          };
         }
 
         private void applyThrottledReplicas() {
@@ -874,16 +900,41 @@ public class Builder {
 
     @Override
     public void clearReplicationThrottle(TopicPartitionReplica log) {
+      // Attempt to submit two config alterations might encounter some bug.
+      // We have to submit all the changes in one API request.
+      // see https://github.com/skiptests/astraea/issues/649
       var configValue = log.partition() + ":" + log.brokerId();
       var configEntry0 = new ConfigEntry("leader.replication.throttled.replicas", configValue);
       var configEntry1 = new ConfigEntry("follower.replication.throttled.replicas", configValue);
-      var alterConfigOp0 = new AlterConfigOp(configEntry0, AlterConfigOp.OpType.SUBTRACT);
-      var alterConfigOp1 = new AlterConfigOp(configEntry1, AlterConfigOp.OpType.SUBTRACT);
       var configResource = new ConfigResource(ConfigResource.Type.TOPIC, log.topic());
       Utils.packException(
           () ->
               admin.incrementalAlterConfigs(
-                  Map.of(configResource, List.of(alterConfigOp0, alterConfigOp1))));
+                  Map.of(
+                      configResource,
+                      List.of(
+                          new AlterConfigOp(configEntry0, AlterConfigOp.OpType.SUBTRACT),
+                          new AlterConfigOp(configEntry1, AlterConfigOp.OpType.SUBTRACT)))));
+    }
+
+    @Override
+    public void clearLeaderReplicationThrottle(TopicPartitionReplica log) {
+      var configValue = log.partition() + ":" + log.brokerId();
+      var configEntry0 = new ConfigEntry("leader.replication.throttled.replicas", configValue);
+      var alterConfigOp0 = new AlterConfigOp(configEntry0, AlterConfigOp.OpType.SUBTRACT);
+      var configResource = new ConfigResource(ConfigResource.Type.TOPIC, log.topic());
+      Utils.packException(
+          () -> admin.incrementalAlterConfigs(Map.of(configResource, List.of(alterConfigOp0))));
+    }
+
+    @Override
+    public void clearFollowerReplicationThrottle(TopicPartitionReplica log) {
+      var configValue = log.partition() + ":" + log.brokerId();
+      var configEntry1 = new ConfigEntry("follower.replication.throttled.replicas", configValue);
+      var alterConfigOp1 = new AlterConfigOp(configEntry1, AlterConfigOp.OpType.SUBTRACT);
+      var configResource = new ConfigResource(ConfigResource.Type.TOPIC, log.topic());
+      Utils.packException(
+          () -> admin.incrementalAlterConfigs(Map.of(configResource, List.of(alterConfigOp1))));
     }
 
     @Override
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 3fb513ca..42f171f1 100644
--- a/app/src/main/java/org/astraea/app/admin/ClusterInfo.java
+++ b/app/src/main/java/org/astraea/app/admin/ClusterInfo.java
@@ -22,6 +22,7 @@ import java.util.NoSuchElementException;
 import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 public interface ClusterInfo<T extends ReplicaInfo> {
   ClusterInfo<ReplicaInfo> EMPTY =
@@ -33,11 +34,34 @@ public interface ClusterInfo<T extends ReplicaInfo> {
         }
 
         @Override
-        public List<ReplicaInfo> replicas() {
-          return List.of();
+        public Stream<ReplicaInfo> replicaStream() {
+          return Stream.of();
         }
       };
 
+  /**
+   * find the changed replicas between `before` and `after`. The diff is based on following
+   * conditions. 1) the replicas are existent only in the `before` cluster 2) find the changes based
+   * on either broker or data folder between `before` and `after`. Noted that the replicas existent
+   * only in the `after` cluster are NOT returned.
+   *
+   * @param before to be compared
+   * @param after to compare
+   * @return the diff replicas
+   */
+  static Set<Replica> diff(ClusterInfo<Replica> before, ClusterInfo<Replica> after) {
+    return before.replicas().stream()
+        .filter(
+            beforeReplica ->
+                after
+                    .replica(beforeReplica.topicPartitionReplica())
+                    // not equal so it is changed
+                    .map(newReplica -> !beforeReplica.dataFolder().equals(newReplica.dataFolder()))
+                    // no replica in the after cluster so it is changed
+                    .orElse(true))
+        .collect(Collectors.toSet());
+  }
+
   @SuppressWarnings("unchecked")
   static <T extends ReplicaInfo> ClusterInfo<T> empty() {
     return (ClusterInfo<T>) EMPTY;
@@ -59,6 +83,19 @@ public interface ClusterInfo<T extends ReplicaInfo> {
             .collect(Collectors.toUnmodifiableList()));
   }
 
+  /**
+   * build a cluster info based on replicas. Noted that the node info are collected by the replicas.
+   *
+   * @param replicas used to build cluster info
+   * @return cluster info
+   * @param <T> ReplicaInfo or Replica
+   */
+  static <T extends ReplicaInfo> ClusterInfo<T> of(List<T> replicas) {
+    return of(
+        replicas.stream().map(ReplicaInfo::nodeInfo).collect(Collectors.toUnmodifiableList()),
+        replicas);
+  }
+
   static <T extends ReplicaInfo> ClusterInfo<T> of(List<NodeInfo> nodes, List<T> replicas) {
     var topics = replicas.stream().map(ReplicaInfo::topic).collect(Collectors.toUnmodifiableSet());
     var replicasForTopic = replicas.stream().collect(Collectors.groupingBy(ReplicaInfo::topic));
@@ -89,12 +126,12 @@ public interface ClusterInfo<T extends ReplicaInfo> {
       }
 
       @Override
-      public List<T> availableReplicaLeaders(String topic) {
+      public List<T> replicaLeaders(String topic) {
         return availableReplicaLeadersForTopics.getOrDefault(topic, List.of());
       }
 
       @Override
-      public List<T> availableReplicaLeaders(int broker, String topic) {
+      public List<T> replicaLeaders(int broker, String topic) {
         return availableLeaderReplicasForBrokersTopics.getOrDefault(
             Map.entry(broker, topic), List.of());
       }
@@ -110,97 +147,169 @@ public interface ClusterInfo<T extends ReplicaInfo> {
       }
 
       @Override
-      public List<T> replicas() {
-        return replicas;
+      public Stream<T> replicaStream() {
+        return replicas.stream();
       }
     };
   }
 
+  // ---------------------[for leader]---------------------//
+
   /**
-   * find the node associated to node id.
+   * Get the list of replica leaders
    *
-   * @param id node id
-   * @return the node information. It throws NoSuchElementException if specify node id is not
-   *     associated to any node
+   * @return A list of {@link ReplicaInfo}.
    */
-  default NodeInfo node(int id) {
-    return nodes().stream()
-        .filter(n -> n.id() == id)
-        .findAny()
-        .orElseThrow(() -> new NoSuchElementException(id + " is nonexistent"));
+  default List<T> replicaLeaders() {
+    return replicaStream()
+        .filter(ReplicaInfo::isLeader)
+        .filter(ReplicaInfo::isOnline)
+        .collect(Collectors.toUnmodifiableList());
   }
 
   /**
-   * Get the list of replica leader information of each available partition for the given topic
+   * Get the list of replica leaders of given topic
    *
    * @param topic The Topic name
    * @return A list of {@link ReplicaInfo}.
    */
-  default List<T> availableReplicaLeaders(String topic) {
-    return replicas(topic).stream()
+  default List<T> replicaLeaders(String topic) {
+    return replicaStream()
+        .filter(r -> r.topic().equals(topic))
         .filter(ReplicaInfo::isLeader)
+        .filter(ReplicaInfo::isOnline)
         .collect(Collectors.toUnmodifiableList());
   }
 
   /**
-   * Get the list of replica leader information of each available partition for the given
-   * broker/topic
+   * Get the list of replica leaders of given topic on the given node
    *
    * @param broker the broker id
    * @param topic The Topic name
    * @return A list of {@link ReplicaInfo}.
    */
-  default List<T> availableReplicaLeaders(int broker, String topic) {
-    return availableReplicaLeaders(topic).stream()
+  default List<T> replicaLeaders(int broker, String topic) {
+    return replicaStream()
         .filter(r -> r.nodeInfo().id() == broker)
+        .filter(r -> r.topic().equals(topic))
+        .filter(ReplicaInfo::isLeader)
+        .filter(ReplicaInfo::isOnline)
         .collect(Collectors.toUnmodifiableList());
   }
 
   /**
-   * Get the list of replica information of each available partition/replica pair for the given
-   * topic
+   * Get the replica leader of given topic partition
+   *
+   * @param topicPartition topic and partition id
+   * @return {@link ReplicaInfo} or empty if there is no leader
+   */
+  default Optional<T> replicaLeader(TopicPartition topicPartition) {
+    return replicaStream()
+        .filter(r -> r.topicPartition().equals(topicPartition))
+        .filter(ReplicaInfo::isLeader)
+        .filter(ReplicaInfo::isOnline)
+        .findFirst();
+  }
+
+  // ---------------------[for available leader]---------------------//
+
+  /**
+   * Get the list of available replicas of given topic
    *
    * @param topic The topic name
    * @return A list of {@link ReplicaInfo}.
    */
   default List<T> availableReplicas(String topic) {
-    return replicas(topic).stream()
+    return replicaStream()
+        .filter(r -> r.topic().equals(topic))
         .filter(ReplicaInfo::isOnline)
         .collect(Collectors.toUnmodifiableList());
   }
 
-  /** @return The known set of nodes */
-  List<NodeInfo> nodes();
+  // ---------------------[for replicas]---------------------//
+
+  /** @return all replicas cached by this cluster info. */
+  default List<T> replicas() {
+    return replicaStream().collect(Collectors.toUnmodifiableList());
+  }
 
   /**
-   * All topic names
+   * Get the list of replica information of each partition/replica pair for the given topic
    *
-   * @return return a set of topic names
+   * @param topic The topic name
+   * @return A list of {@link ReplicaInfo}.
    */
-  default Set<String> topics() {
-    return replicas().stream().map(ReplicaInfo::topic).collect(Collectors.toUnmodifiableSet());
+  default List<T> replicas(String topic) {
+    return replicaStream()
+        .filter(r -> r.topic().equals(topic))
+        .collect(Collectors.toUnmodifiableList());
   }
 
   /**
-   * Get the list of replica information of each partition/replica pair for the given topic
+   * Get the list of replica information of each partition/replica pair for the given topic on given
+   * broker
    *
    * @param topic The topic name
    * @return A list of {@link ReplicaInfo}.
    */
-  default List<T> replicas(String topic) {
-    return replicas().stream()
+  default List<T> replicas(int broker, String topic) {
+    return replicaStream()
+        .filter(r -> r.nodeInfo().id() == broker)
         .filter(r -> r.topic().equals(topic))
         .collect(Collectors.toUnmodifiableList());
   }
 
+  /**
+   * Get the list of replica information of each partition/replica pair for the given topic on given
+   * broker
+   *
+   * @param topicPartition topic name and partition id
+   * @return A list of {@link ReplicaInfo}.
+   */
+  default List<T> replicas(TopicPartition topicPartition) {
+    return replicaStream()
+        .filter(r -> r.topicPartition().equals(topicPartition))
+        .collect(Collectors.toUnmodifiableList());
+  }
+
   /**
    * @param replica to search
    * @return the replica matched to input replica
    */
   default Optional<T> replica(TopicPartitionReplica replica) {
-    return replicas().stream().filter(r -> r.topicPartitionReplica().equals(replica)).findFirst();
+    return replicaStream().filter(r -> r.topicPartitionReplica().equals(replica)).findFirst();
   }
 
-  /** @return all replicas cached by this cluster info. */
-  List<T> replicas();
+  // ---------------------[others]---------------------//
+
+  /**
+   * All topic names
+   *
+   * @return return a set of topic names
+   */
+  default Set<String> topics() {
+    return replicaStream().map(ReplicaInfo::topic).collect(Collectors.toUnmodifiableSet());
+  }
+
+  /**
+   * find the node associated to node id.
+   *
+   * @param id node id
+   * @return the node information. It throws NoSuchElementException if specify node id is not
+   *     associated to any node
+   */
+  default NodeInfo node(int id) {
+    return nodes().stream()
+        .filter(n -> n.id() == id)
+        .findAny()
+        .orElseThrow(() -> new NoSuchElementException(id + " is nonexistent"));
+  }
+
+  // ---------------------[abstract methods]---------------------//
+
+  /** @return The known set of nodes */
+  List<NodeInfo> nodes();
+
+  /** @return replica stream to offer effective way to operate a bunch of replicas */
+  Stream<T> replicaStream();
 }
diff --git a/app/src/main/java/org/astraea/app/admin/Compression.java b/app/src/main/java/org/astraea/app/admin/Compression.java
index 274038c8..59cd61f2 100644
--- a/app/src/main/java/org/astraea/app/admin/Compression.java
+++ b/app/src/main/java/org/astraea/app/admin/Compression.java
@@ -16,21 +16,63 @@
  */
 package org.astraea.app.admin;
 
+import com.beust.jcommander.ParameterException;
+import java.util.Arrays;
 import java.util.Locale;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 public enum Compression {
-  NONE,
-  GZIP,
-  SNAPPY,
-  LZ4,
-  ZSTD;
-
-  public static Compression of(String name) {
-    return Compression.valueOf(name.toUpperCase(Locale.ROOT));
+  NONE("none"),
+  GZIP("gzip"),
+  SNAPPY("snappy"),
+  LZ4("lz4"),
+  ZSTD("zstd");
+
+  private final String alias;
+
+  Compression(String alias) {
+    this.alias = alias;
+  }
+
+  public String alias() {
+    return alias;
+  }
+
+  @Override
+  public String toString() {
+    return alias();
+  }
+
+  public static Compression ofAlias(String alias) {
+    return Arrays.stream(Compression.values())
+        .filter(a -> a.alias().equalsIgnoreCase(alias))
+        .findFirst()
+        .orElseThrow(
+            () ->
+                new IllegalArgumentException(
+                    "the "
+                        + alias
+                        + " is unsupported. The supported algorithms are "
+                        + Stream.of(Compression.values())
+                            .map(Compression::alias)
+                            .collect(Collectors.joining(","))));
   }
 
   /** @return the name parsed by kafka */
   public String nameOfKafka() {
     return name().toLowerCase(Locale.ROOT);
   }
+
+  public static class Field extends org.astraea.app.argument.Field<Compression> {
+    @Override
+    public Compression convert(String value) {
+      try {
+        // `CompressionType#forName` accept lower-case name only.
+        return Compression.ofAlias(value);
+      } catch (IllegalArgumentException e) {
+        throw new ParameterException(e);
+      }
+    }
+  }
 }
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 02536e5d..44bc703d 100644
--- a/app/src/main/java/org/astraea/app/admin/ReplicaInfo.java
+++ b/app/src/main/java/org/astraea/app/admin/ReplicaInfo.java
@@ -122,6 +122,15 @@ public interface ReplicaInfo {
     return TopicPartitionReplica.of(topic(), partition(), nodeInfo().id());
   }
 
+  /**
+   * a helper to build TopicPartition quickly
+   *
+   * @return TopicPartition
+   */
+  default TopicPartition topicPartition() {
+    return TopicPartition.of(topic(), partition());
+  }
+
   /** @return topic name */
   String topic();
 
diff --git a/app/src/main/java/org/astraea/app/admin/ReplicationThrottler.java b/app/src/main/java/org/astraea/app/admin/ReplicationThrottler.java
index d89317a0..0b64e728 100644
--- a/app/src/main/java/org/astraea/app/admin/ReplicationThrottler.java
+++ b/app/src/main/java/org/astraea/app/admin/ReplicationThrottler.java
@@ -17,6 +17,7 @@
 package org.astraea.app.admin;
 
 import java.util.Map;
+import java.util.Set;
 import org.astraea.app.common.DataRate;
 
 /**
@@ -120,6 +121,21 @@ public interface ReplicationThrottler {
    */
   ReplicationThrottler throttleFollower(TopicPartitionReplica replica);
 
-  /** Apply the throttle setting to the cluster. */
-  void apply();
+  /**
+   * Apply the throttle setting to the cluster.
+   *
+   * @return an {@link AffectedResources} object that describe the resource that will be affected
+   *     after applied this throttle.
+   */
+  AffectedResources apply();
+
+  interface AffectedResources {
+    Map<Integer, DataRate> ingress();
+
+    Map<Integer, DataRate> egress();
+
+    Set<TopicPartitionReplica> leaders();
+
+    Set<TopicPartitionReplica> followers();
+  }
 }
diff --git a/app/src/main/java/org/astraea/app/argument/CompressionField.java b/app/src/main/java/org/astraea/app/argument/CompressionField.java
deleted file mode 100644
index 4422c1e5..00000000
--- a/app/src/main/java/org/astraea/app/argument/CompressionField.java
+++ /dev/null
@@ -1,50 +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.argument;
-
-import com.beust.jcommander.ParameterException;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-import org.apache.kafka.common.record.CompressionType;
-import org.astraea.app.admin.Compression;
-
-public class CompressionField extends Field<Compression> {
-  /**
-   * @param value Name of compression type. Accept lower-case name only ("none", "gzip", "snappy",
-   *     "lz4", "zstd").
-   */
-  @Override
-  public Compression convert(String value) {
-    try {
-      // `CompressionType#forName` accept lower-case name only.
-      return Compression.of(value);
-    } catch (IllegalArgumentException e) {
-      throw new ParameterException(
-          "the "
-              + value
-              + " is unsupported. The supported algorithms are "
-              + Stream.of(CompressionType.values())
-                  .map(CompressionType::name)
-                  .collect(Collectors.joining(",")));
-    }
-  }
-
-  @Override
-  protected void check(String name, String value) throws ParameterException {
-    convert(value);
-  }
-}
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 4d7450f6..8a5d5349 100644
--- a/app/src/main/java/org/astraea/app/balancer/BalancerUtils.java
+++ b/app/src/main/java/org/astraea/app/balancer/BalancerUtils.java
@@ -26,6 +26,7 @@ import java.util.function.Function;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
+import java.util.stream.Stream;
 import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.admin.NodeInfo;
@@ -136,8 +137,8 @@ public class BalancerUtils {
       }
 
       @Override
-      public List<Replica> replicas() {
-        return replicas;
+      public Stream<Replica> replicaStream() {
+        return replicas.stream();
       }
     };
   }
diff --git a/app/src/main/java/org/astraea/app/common/DataRate.java b/app/src/main/java/org/astraea/app/common/DataRate.java
index 7f0b0d4b..cca1d5dc 100644
--- a/app/src/main/java/org/astraea/app/common/DataRate.java
+++ b/app/src/main/java/org/astraea/app/common/DataRate.java
@@ -24,6 +24,10 @@ import java.time.Duration;
 import java.time.temporal.ChronoUnit;
 import java.util.Comparator;
 import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 /** Data rate class */
 public class DataRate {
@@ -205,4 +209,78 @@ public class DataRate {
       return dataSize.dataRate(chronoUnit);
     }
   }
+
+  public static class Field extends org.astraea.app.argument.Field<DataRate> {
+
+    static final Duration DEFAULT_DURATION = Duration.ofSeconds(1);
+    private static final Pattern DATA_RATE_PATTERN =
+        Pattern.compile(
+            "(?<measurement>\\d+)\\s?(?<dataUnit>[a-zA-Z]+)(/(?<duration>[\\da-z-+.,A-Z]+))?");
+
+    /**
+     * Convert string to DataRate.
+     *
+     * <pre>{@code
+     * new DataRate.Field().convert("500KB/s");  // 500 KB  (500 * 1000 bytes)/ per second
+     * new DataRate.Field().convert("500KiB/H"); // 500 KiB (500 * 1024 bytes)/ per hour
+     * new DataRate.Field().convert("500Kb/minute");  // 500 Kb  (500 * 1000 bits)/ per minute
+     * new DataRate.Field().convert("500Kib/PT-20S"); // 500 Kib (500 * 1024 bits)/ per 20 second
+     * }</pre>
+     *
+     * @param argument number and the unit and duration e.g. "500MiB/s", "9876 KB/PT-20S"
+     * @return data size with data unit under duration.
+     */
+    @Override
+    public DataRate convert(String argument) {
+
+      Matcher matcher = DATA_RATE_PATTERN.matcher(argument);
+      if (matcher.matches()) {
+        var measurement = Long.parseLong(matcher.group("measurement"));
+        var dataUnit = DataUnit.valueOf(matcher.group("dataUnit"));
+        var duration = getDuration(matcher.group("duration"));
+        return new DataRate(dataUnit.of(measurement), duration);
+      } else {
+        throw new IllegalArgumentException("Unknown DataRate \"" + argument + "\"");
+      }
+    }
+
+    static Map<String, Duration> DURATION_MAP =
+        Map.ofEntries(
+            Map.entry("s", Duration.ofSeconds(1)),
+            Map.entry("second", Duration.ofSeconds(1)),
+            Map.entry("seconds", Duration.ofSeconds(1)),
+            Map.entry("m", Duration.ofMinutes(1)),
+            Map.entry("minute", Duration.ofMinutes(1)),
+            Map.entry("minutes", Duration.ofMinutes(1)),
+            Map.entry("h", Duration.ofHours(1)),
+            Map.entry("hour", Duration.ofHours(1)),
+            Map.entry("hours", Duration.ofHours(1)),
+            Map.entry("d", Duration.ofDays(1)),
+            Map.entry("day", Duration.ofDays(1)),
+            Map.entry("days", Duration.ofDays(1)));
+
+    private Duration getDuration(String duration) {
+      if (Objects.isNull(duration)) {
+        return DEFAULT_DURATION;
+      }
+      var lowerCaseDuration = duration.toLowerCase();
+      return DURATION_MAP.containsKey(lowerCaseDuration)
+          ? DURATION_MAP.get(lowerCaseDuration)
+          : Duration.parse(lowerCaseDuration);
+    }
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) return true;
+    if (o == null || getClass() != o.getClass()) return false;
+    DataRate dataRate = (DataRate) o;
+    return Objects.equals(totalBitTransmitted, dataRate.totalBitTransmitted)
+        && Objects.equals(durationInNanoSecond, dataRate.durationInNanoSecond);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(totalBitTransmitted, durationInNanoSecond);
+  }
 }
diff --git a/app/src/main/java/org/astraea/app/common/DataSize.java b/app/src/main/java/org/astraea/app/common/DataSize.java
index fff5274d..5ec6136d 100644
--- a/app/src/main/java/org/astraea/app/common/DataSize.java
+++ b/app/src/main/java/org/astraea/app/common/DataSize.java
@@ -311,10 +311,10 @@ public class DataSize implements Comparable<DataSize> {
      * Convert string to DataSize.
      *
      * <pre>{@code
-     * new DataSize.Converter().convert("500KB");  // 500 KB  (500 * 1000 bytes)
-     * new DataSize.Converter().convert("500KiB"); // 500 KiB (500 * 1024 bytes)
-     * new DataSize.Converter().convert("500Kb");  // 500 Kb  (500 * 1000 bits)
-     * new DataSize.Converter().convert("500Kib"); // 500 Kib (500 * 1024 bits)
+     * new DataSize.Field().convert("500KB");  // 500 KB  (500 * 1000 bytes)
+     * new DataSize.Field().convert("500KiB"); // 500 KiB (500 * 1024 bytes)
+     * new DataSize.Field().convert("500Kb");  // 500 Kb  (500 * 1000 bits)
+     * new DataSize.Field().convert("500Kib"); // 500 Kib (500 * 1024 bits)
      * }</pre>
      *
      * @param argument number and the unit. e.g. "500MiB", "9876 KB"
diff --git a/app/src/main/java/org/astraea/app/common/ExecutionRuntimeException.java b/app/src/main/java/org/astraea/app/common/ExecutionRuntimeException.java
index 2a535014..bd0c925e 100644
--- a/app/src/main/java/org/astraea/app/common/ExecutionRuntimeException.java
+++ b/app/src/main/java/org/astraea/app/common/ExecutionRuntimeException.java
@@ -30,14 +30,6 @@ public class ExecutionRuntimeException extends RuntimeException {
 
   /** @return the root cause of ExecutionException */
   public Throwable getRootCause() {
-    return unpack(executionException);
-  }
-
-  private static Throwable unpack(Throwable exception) {
-    Throwable current = exception;
-    while (current instanceof ExecutionException) {
-      current = current.getCause();
-    }
-    return current;
+    return executionException.getCause();
   }
 }
diff --git a/app/src/main/java/org/astraea/app/common/Utils.java b/app/src/main/java/org/astraea/app/common/Utils.java
index 2a2243ba..e73be0e8 100644
--- a/app/src/main/java/org/astraea/app/common/Utils.java
+++ b/app/src/main/java/org/astraea/app/common/Utils.java
@@ -138,7 +138,8 @@ public final class Utils {
   }
 
   /**
-   * loop the supplier until it returns non-null value
+   * loop the supplier until it returns non-null value. The exception arisen in the waiting get
+   * ignored if the supplier offers the non-null value in the end.
    *
    * @param supplier to loop
    * @param timeout to break the loop
@@ -271,5 +272,36 @@ public final class Utils {
     return (double) (value / duration.toNanos()) * 1000000000L;
   }
 
+  public static Object staticMember(Class<?> clz, String attribute) {
+    return reflectionAttribute(clz, null, attribute);
+  }
+
+  public static Object member(Object object, String attribute) {
+    return reflectionAttribute(object.getClass(), object, attribute);
+  }
+
+  /**
+   * reflection class attribute
+   *
+   * @param clz class type
+   * @param object object. Null means static member
+   * @param attribute attribute name
+   * @return attribute
+   */
+  private static Object reflectionAttribute(Class<?> clz, Object object, String attribute) {
+    do {
+      try {
+        var field = clz.getDeclaredField(attribute);
+        field.setAccessible(true);
+        return field.get(object);
+      } catch (NoSuchFieldException e) {
+        clz = clz.getSuperclass();
+      } catch (IllegalAccessException e) {
+        throw new RuntimeException(e);
+      }
+    } while (clz != null);
+    throw new RuntimeException(attribute + " is not existent in " + object.getClass().getName());
+  }
+
   private Utils() {}
 }
diff --git a/app/src/main/java/org/astraea/app/consumer/Builder.java b/app/src/main/java/org/astraea/app/consumer/Builder.java
index 4ad6794f..737c09da 100644
--- a/app/src/main/java/org/astraea/app/consumer/Builder.java
+++ b/app/src/main/java/org/astraea/app/consumer/Builder.java
@@ -28,6 +28,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.BiConsumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.astraea.app.admin.TopicPartition;
+import org.astraea.app.common.Utils;
 
 public abstract class Builder<Key, Value> {
   protected final Map<String, Object> configs = new HashMap<>();
@@ -113,6 +114,11 @@ public abstract class Builder<Key, Value> {
     return this;
   }
 
+  public Builder<Key, Value> clientId(String clientId) {
+    this.configs.put(ConsumerConfig.CLIENT_ID_CONFIG, clientId);
+    return this;
+  }
+
   /** @return consumer instance. The different builders may return inherited consumer interface. */
   public abstract Consumer<Key, Value> build();
 
@@ -120,8 +126,12 @@ public abstract class Builder<Key, Value> {
     protected final org.apache.kafka.clients.consumer.Consumer<Key, Value> kafkaConsumer;
     private final AtomicBoolean subscribed = new AtomicBoolean(true);
 
+    private final String clientId;
+
     public BaseConsumer(org.apache.kafka.clients.consumer.Consumer<Key, Value> kafkaConsumer) {
       this.kafkaConsumer = kafkaConsumer;
+      // KafkaConsumer does not expose client-id
+      this.clientId = (String) Utils.member(kafkaConsumer, "clientId");
     }
 
     @Override
@@ -156,6 +166,11 @@ public abstract class Builder<Key, Value> {
       if (subscribed.compareAndSet(true, false)) kafkaConsumer.unsubscribe();
     }
 
+    @Override
+    public String clientId() {
+      return clientId;
+    }
+
     protected abstract void doResubscribe();
   }
 
diff --git a/app/src/main/java/org/astraea/app/consumer/Consumer.java b/app/src/main/java/org/astraea/app/consumer/Consumer.java
index a89bea8b..0d149a4a 100644
--- a/app/src/main/java/org/astraea/app/consumer/Consumer.java
+++ b/app/src/main/java/org/astraea/app/consumer/Consumer.java
@@ -58,6 +58,9 @@ public interface Consumer<Key, Value> extends AutoCloseable {
   /** @return current partitions assigned to this consumer */
   Set<TopicPartition> assignments();
 
+  /** @return client id of this consumer */
+  String clientId();
+
   /**
    * Create a consumer builder by setting specific topics
    *
diff --git a/app/src/main/java/org/astraea/app/consumer/PartitionsBuilder.java b/app/src/main/java/org/astraea/app/consumer/PartitionsBuilder.java
index 98821c07..8b5684b7 100644
--- a/app/src/main/java/org/astraea/app/consumer/PartitionsBuilder.java
+++ b/app/src/main/java/org/astraea/app/consumer/PartitionsBuilder.java
@@ -101,6 +101,12 @@ public class PartitionsBuilder<Key, Value> extends Builder<Key, Value> {
     return this;
   }
 
+  @Override
+  public PartitionsBuilder<Key, Value> clientId(String clientId) {
+    super.clientId(clientId);
+    return this;
+  }
+
   @SuppressWarnings("unchecked")
   @Override
   public AssignedConsumer<Key, Value> build() {
diff --git a/app/src/main/java/org/astraea/app/consumer/TopicsBuilder.java b/app/src/main/java/org/astraea/app/consumer/TopicsBuilder.java
index 4da24ad6..417891a5 100644
--- a/app/src/main/java/org/astraea/app/consumer/TopicsBuilder.java
+++ b/app/src/main/java/org/astraea/app/consumer/TopicsBuilder.java
@@ -124,6 +124,12 @@ public class TopicsBuilder<Key, Value> extends Builder<Key, Value> {
     return config(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
   }
 
+  @Override
+  public TopicsBuilder<Key, Value> clientId(String clientId) {
+    super.clientId(clientId);
+    return this;
+  }
+
   @SuppressWarnings("unchecked")
   @Override
   public SubscribedConsumer<Key, Value> build() {
diff --git a/app/src/main/java/org/astraea/app/cost/Dispersion.java b/app/src/main/java/org/astraea/app/cost/Dispersion.java
index effd87a1..9feffc89 100644
--- a/app/src/main/java/org/astraea/app/cost/Dispersion.java
+++ b/app/src/main/java/org/astraea/app/cost/Dispersion.java
@@ -26,7 +26,9 @@ public interface Dispersion {
    */
   static Dispersion correlationCoefficient() {
     return brokerCost -> {
+      if (brokerCost.size() == 0) return 0;
       var dataRateMean = brokerCost.stream().mapToDouble(x -> x).sum() / brokerCost.size();
+      if (dataRateMean == 0) return 0;
       var dataRateSD =
           Math.sqrt(
               brokerCost.stream().mapToDouble(score -> Math.pow((score - dataRateMean), 2)).sum()
diff --git a/app/src/main/java/org/astraea/app/cost/MoveCost.java b/app/src/main/java/org/astraea/app/cost/MoveCost.java
new file mode 100644
index 00000000..8ab9d426
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/cost/MoveCost.java
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.cost;
+
+import java.util.Map;
+
+/** Return type of cost function, `HasMoveCost`. It returns the score of migrate plan. */
+public interface MoveCost {
+  /** @return the function name of MoveCost */
+  String name();
+
+  /** @return cost of migrate plan */
+  long totalCost();
+
+  /** @return unit of cost */
+  String unit();
+
+  /** @return Changes per broker, negative if brokers moved out, positive if brokers moved in */
+  Map<Integer, Long> changes();
+}
diff --git a/app/src/main/java/org/astraea/app/cost/ReplicaLeaderCost.java b/app/src/main/java/org/astraea/app/cost/ReplicaLeaderCost.java
index 7ccbeea7..d8605181 100644
--- a/app/src/main/java/org/astraea/app/cost/ReplicaLeaderCost.java
+++ b/app/src/main/java/org/astraea/app/cost/ReplicaLeaderCost.java
@@ -72,8 +72,7 @@ public class ReplicaLeaderCost implements HasBrokerCost, HasClusterCost {
   }
 
   static Map<Integer, Integer> leaderCount(ClusterInfo<? extends ReplicaInfo> clusterInfo) {
-    return clusterInfo.topics().stream()
-        .flatMap(t -> clusterInfo.availableReplicaLeaders(t).stream())
+    return clusterInfo.replicaLeaders().stream()
         .collect(Collectors.groupingBy(r -> r.nodeInfo().id()))
         .entrySet()
         .stream()
diff --git a/app/src/main/java/org/astraea/app/cost/NodeTopicSizeCost.java b/app/src/main/java/org/astraea/app/cost/ReplicaSizeCost.java
similarity index 56%
rename from app/src/main/java/org/astraea/app/cost/NodeTopicSizeCost.java
rename to app/src/main/java/org/astraea/app/cost/ReplicaSizeCost.java
index 754a21fa..0fe7c240 100644
--- a/app/src/main/java/org/astraea/app/cost/NodeTopicSizeCost.java
+++ b/app/src/main/java/org/astraea/app/cost/ReplicaSizeCost.java
@@ -16,8 +16,11 @@
  */
 package org.astraea.app.cost;
 
+import java.util.Collection;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Collectors;
 import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
@@ -27,14 +30,46 @@ import org.astraea.app.admin.ReplicaInfo;
 import org.astraea.app.metrics.broker.LogMetrics;
 import org.astraea.app.metrics.collector.Fetcher;
 
-public class NodeTopicSizeCost implements HasBrokerCost, HasClusterCost {
+public class ReplicaSizeCost implements HasMoveCost, HasBrokerCost, HasClusterCost {
   private final Dispersion dispersion = Dispersion.correlationCoefficient();
 
+  /** @return the metrics getters. Those getters are used to fetch mbeans. */
   @Override
   public Optional<Fetcher> fetcher() {
     return Optional.of(LogMetrics.Log.SIZE::fetch);
   }
 
+  @Override
+  public MoveCost moveCost(
+      ClusterInfo<Replica> before, ClusterInfo<Replica> after, ClusterBean clusterBean) {
+    var removedReplicas = ClusterInfo.diff(before, after);
+    var addedReplicas = ClusterInfo.diff(after, before);
+    var migrateInfo = migrateInfo(removedReplicas, addedReplicas);
+    var sizeChanges = migrateInfo.sizeChange;
+    var totalMigrateSize = migrateInfo.totalMigrateSize;
+    return new MoveCost() {
+      @Override
+      public String name() {
+        return "size";
+      }
+
+      @Override
+      public long totalCost() {
+        return totalMigrateSize;
+      }
+
+      @Override
+      public String unit() {
+        return "byte";
+      }
+
+      @Override
+      public Map<Integer, Long> changes() {
+        return sizeChanges;
+      }
+    };
+  }
+
   /**
    * @param clusterInfo the clusterInfo that offers the metrics related to topic/partition size
    * @return a BrokerCost contains the used space for each broker
@@ -70,4 +105,36 @@ public class NodeTopicSizeCost implements HasBrokerCost, HasClusterCost {
         dispersion.calculate(
             brokerCost.values().stream().map(v -> (double) v).collect(Collectors.toList()));
   }
+
+  static class MigrateInfo {
+    long totalMigrateSize;
+    Map<Integer, Long> sizeChange;
+
+    MigrateInfo(long totalMigrateSize, Map<Integer, Long> sizeChange) {
+      this.totalMigrateSize = totalMigrateSize;
+      this.sizeChange = sizeChange;
+    }
+  }
+
+  static MigrateInfo migrateInfo(
+      Collection<Replica> removedReplicas, Collection<Replica> addedReplicas) {
+    var changes = new HashMap<Integer, Long>();
+    AtomicLong totalMigrateSize = new AtomicLong(0L);
+    removedReplicas.forEach(
+        replica ->
+            changes.compute(
+                replica.nodeInfo().id(),
+                (ignore, size) -> size == null ? -replica.size() : -replica.size() + size));
+
+    addedReplicas.forEach(
+        replica -> {
+          changes.compute(
+              replica.nodeInfo().id(),
+              (ignore, size) -> {
+                totalMigrateSize.set(totalMigrateSize.get() + replica.size());
+                return size == null ? replica.size() : replica.size() + size;
+              });
+        });
+    return new MigrateInfo(totalMigrateSize.get(), changes);
+  }
 }
diff --git a/app/src/main/java/org/astraea/app/metrics/client/HasNodeMetrics.java b/app/src/main/java/org/astraea/app/metrics/client/HasNodeMetrics.java
index 039a81ab..325aee1a 100644
--- a/app/src/main/java/org/astraea/app/metrics/client/HasNodeMetrics.java
+++ b/app/src/main/java/org/astraea/app/metrics/client/HasNodeMetrics.java
@@ -16,27 +16,17 @@
  */
 package org.astraea.app.metrics.client;
 
-import org.astraea.app.metrics.BeanObject;
+import java.util.function.Function;
 import org.astraea.app.metrics.HasBeanObject;
 
 public interface HasNodeMetrics extends HasBeanObject {
+  Function<String, Integer> BROKER_ID_FETCHER =
+      node -> Integer.parseInt(node.substring(node.indexOf("-") + 1));
 
-  static HasNodeMetrics of(BeanObject beanObject, int brokerId) {
-    return new HasNodeMetrics() {
-      @Override
-      public int brokerId() {
-        return brokerId;
-      }
-
-      @Override
-      public BeanObject beanObject() {
-        return beanObject;
-      }
-    };
+  default int brokerId() {
+    return BROKER_ID_FETCHER.apply(beanObject().properties().get("node-id"));
   }
 
-  int brokerId();
-
   default double incomingByteRate() {
     return (double) beanObject().attributes().get("incoming-byte-rate");
   }
diff --git a/app/src/main/java/org/astraea/app/metrics/client/consumer/ConsumerMetrics.java b/app/src/main/java/org/astraea/app/metrics/client/consumer/ConsumerMetrics.java
index ee6c7c38..c5a8dc65 100644
--- a/app/src/main/java/org/astraea/app/metrics/client/consumer/ConsumerMetrics.java
+++ b/app/src/main/java/org/astraea/app/metrics/client/consumer/ConsumerMetrics.java
@@ -17,57 +17,45 @@
 package org.astraea.app.metrics.client.consumer;
 
 import java.util.Collection;
-import java.util.Map;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 import org.astraea.app.metrics.BeanQuery;
 import org.astraea.app.metrics.MBeanClient;
 import org.astraea.app.metrics.client.HasNodeMetrics;
 
 public class ConsumerMetrics {
-  private static int brokerId(String node) {
-    return Integer.parseInt(node.substring(node.indexOf("-") + 1));
-  }
-
   /**
-   * node metrics traced by consumer
+   * collect HasNodeMetrics from all consumers.
    *
-   * @param mBeanClient to query beans
-   * @param brokerId broker ids
-   * @return key is client id used by consumer, and value is node metrics traced by each consumer
+   * @param mBeanClient to query metrics
+   * @return key is broker id, and value is associated to broker metrics recorded by all consumers
    */
-  public static Map<String, HasNodeMetrics> node(MBeanClient mBeanClient, int brokerId) {
+  public static Collection<HasNodeMetrics> nodes(MBeanClient mBeanClient) {
+    Function<String, Integer> brokerId =
+        node -> Integer.parseInt(node.substring(node.indexOf("-") + 1));
     return mBeanClient
         .queryBeans(
             BeanQuery.builder()
                 .domainName("kafka.consumer")
                 .property("type", "consumer-node-metrics")
-                .property("node-id", "node-" + brokerId)
+                .property("node-id", "*")
                 .property("client-id", "*")
                 .build())
         .stream()
-        .collect(
-            Collectors.toUnmodifiableMap(
-                b -> b.properties().get("client-id"),
-                b -> HasNodeMetrics.of(b, brokerId(b.properties().get("node-id")))));
+        .map(b -> (HasNodeMetrics) () -> b)
+        .collect(Collectors.toUnmodifiableList());
   }
 
-  /**
-   * collect HasNodeMetrics from all consumers.
-   *
-   * @param mBeanClient to query metrics
-   * @return key is broker id, and value is associated to broker metrics recorded by all consumers
-   */
-  public static Collection<HasNodeMetrics> nodes(MBeanClient mBeanClient) {
+  public static Collection<HasConsumerCoordinatorMetrics> coordinators(MBeanClient mBeanClient) {
     return mBeanClient
         .queryBeans(
             BeanQuery.builder()
                 .domainName("kafka.consumer")
-                .property("type", "consumer-node-metrics")
-                .property("node-id", "*")
+                .property("type", "consumer-coordinator-metrics")
                 .property("client-id", "*")
                 .build())
         .stream()
-        .map(b -> HasNodeMetrics.of(b, brokerId(b.properties().get("node-id"))))
+        .map(b -> (HasConsumerCoordinatorMetrics) () -> b)
         .collect(Collectors.toUnmodifiableList());
   }
 }
diff --git a/app/src/main/java/org/astraea/app/metrics/client/consumer/HasConsumerCoordinatorMetrics.java b/app/src/main/java/org/astraea/app/metrics/client/consumer/HasConsumerCoordinatorMetrics.java
new file mode 100644
index 00000000..33cf5bc2
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/metrics/client/consumer/HasConsumerCoordinatorMetrics.java
@@ -0,0 +1,142 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.metrics.client.consumer;
+
+import org.astraea.app.metrics.HasBeanObject;
+
+public interface HasConsumerCoordinatorMetrics extends HasBeanObject {
+
+  default String clientId() {
+    return beanObject().properties().get("client-id");
+  }
+
+  default double assignedPartitions() {
+    return (double) beanObject().attributes().get("assigned-partitions");
+  }
+
+  default double commitLatencyAvg() {
+    return (double) beanObject().attributes().get("commit-latency-avg");
+  }
+
+  default double commitLatencyMax() {
+    return (double) beanObject().attributes().get("commit-latency-max");
+  }
+
+  default double commitRate() {
+    return (double) beanObject().attributes().get("commit-rate");
+  }
+
+  default double commitTotal() {
+    return (double) beanObject().attributes().get("commit-total");
+  }
+
+  default double failedRebalanceRatePerHour() {
+    return (double) beanObject().attributes().get("failed-rebalance-rate-per-hour");
+  }
+
+  default double failedRebalanceTotal() {
+    return (double) beanObject().attributes().get("failed-rebalance-total");
+  }
+
+  default double heartbeatRate() {
+    return (double) beanObject().attributes().get("heartbeat-rate");
+  }
+
+  default double joinRate() {
+    return (double) beanObject().attributes().get("join-rate");
+  }
+
+  default double joinTimeAvg() {
+    return (double) beanObject().attributes().get("join-time-avg");
+  }
+
+  default double joinTimeMax() {
+    return (double) beanObject().attributes().get("join-time-max");
+  }
+
+  default double joinTotal() {
+    return (double) beanObject().attributes().get("join-total");
+  }
+
+  default double lastHeartbeatSecondsAgo() {
+    return (double) beanObject().attributes().get("last-heartbeat-seconds-ago");
+  }
+
+  default double lastRebalanceSecondsAgo() {
+    return (double) beanObject().attributes().get("last-rebalance-seconds-ago");
+  }
+
+  default double partitionAssignedLatencyAvg() {
+    return (double) beanObject().attributes().get("partition-assigned-latency-avg");
+  }
+
+  default double partitionAssignedLatencyMax() {
+    return (double) beanObject().attributes().get("partition-assigned-latency-max");
+  }
+
+  default double partitionLostLatencyAvg() {
+    return (double) beanObject().attributes().get("partition-lost-latency-avg");
+  }
+
+  default double partitionLostLatencyMax() {
+    return (double) beanObject().attributes().get("partition-lost-latency-max");
+  }
+
+  default double partitionRevokedLatencyAvg() {
+    return (double) beanObject().attributes().get("partition-revoked-latency-avg");
+  }
+
+  default double partitionRevokedLatencyMax() {
+    return (double) beanObject().attributes().get("partition-revoked-latency-max");
+  }
+
+  default double rebalanceLatencyAvg() {
+    return (double) beanObject().attributes().get("rebalance-latency-avg");
+  }
+
+  default double rebalanceLatencyMax() {
+    return (double) beanObject().attributes().get("rebalance-latency-max");
+  }
+
+  default double rebalanceLatencyTotal() {
+    return (double) beanObject().attributes().get("rebalance-latency-total");
+  }
+
+  default double rebalanceRatePerHour() {
+    return (double) beanObject().attributes().get("rebalance-rate-per-hour");
+  }
+
+  default double rebalanceTotal() {
+    return (double) beanObject().attributes().get("rebalance-total");
+  }
+
+  default double syncRate() {
+    return (double) beanObject().attributes().get("sync-rate");
+  }
+
+  default double syncTimeAvg() {
+    return (double) beanObject().attributes().get("sync-time-avg");
+  }
+
+  default double syncTimeMax() {
+    return (double) beanObject().attributes().get("sync-time-max");
+  }
+
+  default double syncTotal() {
+    return (double) beanObject().attributes().get("sync-total");
+  }
+}
diff --git a/app/src/main/java/org/astraea/app/metrics/client/producer/HasProducerTopicMetrics.java b/app/src/main/java/org/astraea/app/metrics/client/producer/HasProducerTopicMetrics.java
index 47a3b0d8..706e9326 100644
--- a/app/src/main/java/org/astraea/app/metrics/client/producer/HasProducerTopicMetrics.java
+++ b/app/src/main/java/org/astraea/app/metrics/client/producer/HasProducerTopicMetrics.java
@@ -20,6 +20,14 @@ import org.astraea.app.metrics.HasBeanObject;
 
 public interface HasProducerTopicMetrics extends HasBeanObject {
 
+  default String topic() {
+    return beanObject().properties().get("topic");
+  }
+
+  default String clientId() {
+    return beanObject().properties().get("client-id");
+  }
+
   default double byteRate() {
     return (double) beanObject().attributes().get("byte-rate");
   }
diff --git a/app/src/main/java/org/astraea/app/metrics/client/producer/ProducerMetrics.java b/app/src/main/java/org/astraea/app/metrics/client/producer/ProducerMetrics.java
index 52a34627..ba073f47 100644
--- a/app/src/main/java/org/astraea/app/metrics/client/producer/ProducerMetrics.java
+++ b/app/src/main/java/org/astraea/app/metrics/client/producer/ProducerMetrics.java
@@ -17,41 +17,13 @@
 package org.astraea.app.metrics.client.producer;
 
 import java.util.Collection;
-import java.util.Map;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 import org.astraea.app.metrics.BeanQuery;
 import org.astraea.app.metrics.MBeanClient;
 import org.astraea.app.metrics.client.HasNodeMetrics;
 
 public final class ProducerMetrics {
-
-  private static int brokerId(String node) {
-    return Integer.parseInt(node.substring(node.indexOf("-") + 1));
-  }
-
-  /**
-   * node metrics traced by producer
-   *
-   * @param mBeanClient to query beans
-   * @param brokerId broker ids
-   * @return key is client id used by producer, and value is node metrics traced by each producer
-   */
-  public static Map<String, HasNodeMetrics> node(MBeanClient mBeanClient, int brokerId) {
-    return mBeanClient
-        .queryBeans(
-            BeanQuery.builder()
-                .domainName("kafka.producer")
-                .property("type", "producer-node-metrics")
-                .property("node-id", "node-" + brokerId)
-                .property("client-id", "*")
-                .build())
-        .stream()
-        .collect(
-            Collectors.toUnmodifiableMap(
-                b -> b.properties().get("client-id"),
-                b -> HasNodeMetrics.of(b, brokerId(b.properties().get("node-id")))));
-  }
-
   /**
    * collect HasProducerNodeMetrics from all producers.
    *
@@ -59,6 +31,8 @@ public final class ProducerMetrics {
    * @return key is broker id, and value is associated to broker metrics recorded by all producers
    */
   public static Collection<HasNodeMetrics> nodes(MBeanClient mBeanClient) {
+    Function<String, Integer> brokerId =
+        node -> Integer.parseInt(node.substring(node.indexOf("-") + 1));
     return mBeanClient
         .queryBeans(
             BeanQuery.builder()
@@ -68,7 +42,7 @@ public final class ProducerMetrics {
                 .property("client-id", "*")
                 .build())
         .stream()
-        .map(b -> HasNodeMetrics.of(b, brokerId(b.properties().get("node-id"))))
+        .map(b -> (HasNodeMetrics) () -> b)
         .collect(Collectors.toUnmodifiableList());
   }
 
@@ -76,20 +50,20 @@ public final class ProducerMetrics {
    * topic metrics traced by producer
    *
    * @param mBeanClient to query beans
-   * @param topic topic name
    * @return key is client id used by producer, and value is topic metrics traced by each producer
    */
-  public static Map<String, HasProducerTopicMetrics> topic(MBeanClient mBeanClient, String topic) {
+  public static Collection<HasProducerTopicMetrics> topics(MBeanClient mBeanClient) {
     return mBeanClient
         .queryBeans(
             BeanQuery.builder()
                 .domainName("kafka.producer")
                 .property("type", "producer-topic-metrics")
                 .property("client-id", "*")
-                .property("topic", topic)
+                .property("topic", "*")
                 .build())
         .stream()
-        .collect(Collectors.toUnmodifiableMap(b -> b.properties().get("client-id"), b -> () -> b));
+        .map(b -> (HasProducerTopicMetrics) () -> b)
+        .collect(Collectors.toUnmodifiableList());
   }
 
   private ProducerMetrics() {}
diff --git a/app/src/main/java/org/astraea/app/partitioner/StrictCostDispatcher.java b/app/src/main/java/org/astraea/app/partitioner/StrictCostDispatcher.java
index eae00657..2071c6cd 100644
--- a/app/src/main/java/org/astraea/app/partitioner/StrictCostDispatcher.java
+++ b/app/src/main/java/org/astraea/app/partitioner/StrictCostDispatcher.java
@@ -105,7 +105,7 @@ public class StrictCostDispatcher implements Dispatcher {
   @Override
   public int partition(
       String topic, byte[] key, byte[] value, ClusterInfo<ReplicaInfo> clusterInfo) {
-    var partitionLeaders = clusterInfo.availableReplicaLeaders(topic);
+    var partitionLeaders = clusterInfo.replicaLeaders(topic);
     // just return first partition if there is no available partitions
     if (partitionLeaders.isEmpty()) return 0;
 
@@ -121,8 +121,7 @@ public class StrictCostDispatcher implements Dispatcher {
             next.getAndUpdate(previous -> previous >= roundRobin.length - 1 ? 0 : previous + 1)];
 
     // TODO: if the topic partitions are existent in fewer brokers, the target gets -1 in most cases
-    var candidate =
-        target < 0 ? partitionLeaders : clusterInfo.availableReplicaLeaders(target, topic);
+    var candidate = target < 0 ? partitionLeaders : clusterInfo.replicaLeaders(target, topic);
     candidate = candidate.isEmpty() ? partitionLeaders : candidate;
     return candidate.get((int) (Math.random() * candidate.size())).partition();
   }
diff --git a/app/src/main/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobin.java b/app/src/main/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobin.java
index 01b4fcba..f4a9ec2d 100644
--- a/app/src/main/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobin.java
+++ b/app/src/main/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobin.java
@@ -120,7 +120,7 @@ public final class SmoothWeightRoundRobin
         brokersIDofTopic.computeIfAbsent(
             topic,
             e ->
-                clusterInfo.availableReplicaLeaders(topic).stream()
+                clusterInfo.replicaLeaders(topic).stream()
                     .map(replicaInfo -> replicaInfo.nodeInfo().id())
                     .collect(Collectors.toList()));
     this.currentWeight =
diff --git a/app/src/main/java/org/astraea/app/performance/ConsumerThread.java b/app/src/main/java/org/astraea/app/performance/ConsumerThread.java
index f061b1b9..cdacbd38 100644
--- a/app/src/main/java/org/astraea/app/performance/ConsumerThread.java
+++ b/app/src/main/java/org/astraea/app/performance/ConsumerThread.java
@@ -18,7 +18,6 @@ package org.astraea.app.performance;
 
 import java.time.Duration;
 import java.util.List;
-import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executors;
@@ -28,7 +27,6 @@ import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import org.apache.kafka.common.errors.WakeupException;
-import org.astraea.app.admin.TopicPartition;
 import org.astraea.app.common.Utils;
 import org.astraea.app.consumer.ConsumerRebalanceListener;
 import org.astraea.app.consumer.SubscribedConsumer;
@@ -57,19 +55,19 @@ public interface ConsumerThread extends AbstractThread {
     return IntStream.range(0, consumers)
         .mapToObj(
             index -> {
-              var report = new Report();
-              var listener = new Listener(report);
+              @SuppressWarnings("resource")
+              var consumer = consumerSupplier.apply(ps -> {});
+              var report = new Report(consumer.clientId());
               var closeLatch = closeLatches.get(index);
               var closed = new AtomicBoolean(false);
               var subscribed = new AtomicBoolean(true);
               executors.execute(
                   () -> {
-                    try (var consumer = consumerSupplier.apply(listener)) {
+                    try {
                       while (!closed.get()) {
                         if (subscribed.get()) consumer.resubscribe();
                         else {
                           consumer.unsubscribe();
-                          report.assignments(Set.of());
                           Utils.sleep(Duration.ofSeconds(1));
                           continue;
                         }
@@ -85,11 +83,11 @@ public interface ConsumerThread extends AbstractThread {
                                         record.offset(),
                                         System.currentTimeMillis() - record.timestamp(),
                                         record.serializedKeySize() + record.serializedValueSize()));
-                        report.assignments(consumer.assignments());
                       }
                     } catch (WakeupException ignore) {
                       // Stop polling and being ready to clean up
                     } finally {
+                      Utils.swallowException(consumer::close);
                       closeLatch.countDown();
                     }
                   });
@@ -137,67 +135,16 @@ public interface ConsumerThread extends AbstractThread {
   /** @return report of this thread */
   Report report();
 
-  class Listener implements ConsumerRebalanceListener {
-    private final Report report;
-    private long previousCall = System.currentTimeMillis();
-    private long maxLatency = 0;
-    private long sumLatency = 0;
-    private long count = 0;
-
-    public Listener(Report report) {
-      this.report = report;
-    }
-
-    @Override
-    public void onPartitionAssigned(Set<TopicPartition> partitions) {
-      record();
-    }
-
-    @Override
-    public void onPartitionsRevoked(Set<TopicPartition> partitions) {
-      record();
-    }
-
-    private void record() {
-      count += 1;
-      var current = System.currentTimeMillis();
-      var diff = current - previousCall;
-      maxLatency = Math.max(maxLatency, diff);
-      sumLatency += diff;
-      previousCall = current;
-      report.maxSubscriptionLatency(maxLatency);
-      report.avgSubscriptionLatency((double) sumLatency / count);
-    }
-  }
-
   class Report extends org.astraea.app.performance.Report.Impl {
-    private volatile long maxSubscriptionLatency = 0;
-    private volatile double avgSubscriptionLatency = 0;
-
-    private volatile Set<TopicPartition> assignments;
 
-    public long maxSubscriptionLatency() {
-      return maxSubscriptionLatency;
-    }
-
-    public void maxSubscriptionLatency(long maxSubscriptionLatency) {
-      this.maxSubscriptionLatency = maxSubscriptionLatency;
-    }
-
-    public double avgSubscriptionLatency() {
-      return avgSubscriptionLatency;
-    }
-
-    public void avgSubscriptionLatency(double avgSubscriptionLatency) {
-      this.avgSubscriptionLatency = avgSubscriptionLatency;
-    }
+    private final String clientId;
 
-    public void assignments(Set<TopicPartition> assignments) {
-      this.assignments = assignments;
+    Report(String clientId) {
+      this.clientId = clientId;
     }
 
-    public Set<TopicPartition> assignments() {
-      return assignments;
+    public String clientId() {
+      return clientId;
     }
   }
 }
diff --git a/app/src/main/java/org/astraea/app/performance/DataSupplier.java b/app/src/main/java/org/astraea/app/performance/DataSupplier.java
index b3e62ab1..c27e2df1 100644
--- a/app/src/main/java/org/astraea/app/performance/DataSupplier.java
+++ b/app/src/main/java/org/astraea/app/performance/DataSupplier.java
@@ -21,8 +21,7 @@ import java.util.Random;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Supplier;
-import org.astraea.app.common.DataSize;
-import org.astraea.app.common.DataUnit;
+import org.astraea.app.common.DataRate;
 
 @FunctionalInterface
 interface DataSupplier extends Supplier<DataSupplier.Data> {
@@ -140,7 +139,7 @@ interface DataSupplier extends Supplier<DataSupplier.Data> {
    * @param keySizeDistribution supply the size of newly created key
    * @param valueDistribution supply abstract value which is represented by a 64-bit integer
    * @param valueSizeDistribution supply the size of newly created value
-   * @param throughput the limit on data produced per second
+   * @param throughput the limit on data produced
    * @return supply data with given distribution. It will map the 64-bit number supplied by
    *     key(/value) distribution to a byte array.
    */
@@ -150,7 +149,7 @@ interface DataSupplier extends Supplier<DataSupplier.Data> {
       Supplier<Long> keySizeDistribution,
       Supplier<Long> valueDistribution,
       Supplier<Long> valueSizeDistribution,
-      DataSize throughput) {
+      DataRate throughput) {
     return new DataSupplier() {
       private final Throttler throttler = new Throttler(throughput);
       private final long start = System.currentTimeMillis();
@@ -203,8 +202,8 @@ interface DataSupplier extends Supplier<DataSupplier.Data> {
     private final long throughput;
     private final AtomicLong totalBytes = new AtomicLong();
 
-    Throttler(DataSize max) {
-      throughput = max.measurement(DataUnit.Byte).longValue();
+    Throttler(DataRate max) {
+      throughput = Double.valueOf(max.byteRate()).longValue();
     }
 
     /**
diff --git a/app/src/main/java/org/astraea/app/performance/Performance.java b/app/src/main/java/org/astraea/app/performance/Performance.java
index bb52062f..c7cab370 100644
--- a/app/src/main/java/org/astraea/app/performance/Performance.java
+++ b/app/src/main/java/org/astraea/app/performance/Performance.java
@@ -34,7 +34,6 @@ import java.util.stream.Collectors;
 import org.astraea.app.admin.Admin;
 import org.astraea.app.admin.Compression;
 import org.astraea.app.admin.TopicPartition;
-import org.astraea.app.argument.CompressionField;
 import org.astraea.app.argument.DurationField;
 import org.astraea.app.argument.NonEmptyStringField;
 import org.astraea.app.argument.NonNegativeShortField;
@@ -44,39 +43,16 @@ import org.astraea.app.argument.PositiveLongField;
 import org.astraea.app.argument.PositiveShortField;
 import org.astraea.app.argument.PositiveShortListField;
 import org.astraea.app.argument.StringListField;
+import org.astraea.app.common.DataRate;
 import org.astraea.app.common.DataSize;
 import org.astraea.app.common.DataUnit;
 import org.astraea.app.common.Utils;
 import org.astraea.app.consumer.Consumer;
 import org.astraea.app.consumer.Isolation;
+import org.astraea.app.producer.Acks;
 import org.astraea.app.producer.Producer;
 
-/**
- * Performance benchmark which includes
- *
- * <ol>
- *   <li>publish latency: the time of completing producer data request
- *   <li>E2E latency: the time for a record to travel through Kafka
- *   <li>input rate: sum of consumer inputs in MByte per second
- *   <li>output rate: sum of producer outputs in MByte per second
- * </ol>
- *
- * With configurations:
- *
- * <ol>
- *   <li>--brokers: the server to connect to
- *   <li>--topic: the topic name. Create new topic when the given topic does not exist. Default:
- *       "testPerformance-" + System.currentTimeMillis()
- *   <li>--partitions: topic config when creating new topic. Default: 1
- *   <li>--replicationFactor: topic config when creating new topic. Default: 1
- *   <li>--producers: the number of producers (threads). Default: 1
- *   <li>--consumers: the number of consumers (threads). Default: 1
- *   <li>--records: the total number of records sent by the producers. Default: 1000
- *   <li>--recordSize: the record size in byte. Default: 1024
- * </ol>
- *
- * To avoid records being produced too fast, producer wait for one millisecond after each send.
- */
+/** see docs/performance_benchmark.md for man page */
 public class Performance {
   /** Used in Automation, to achieve the end of one Performance and then start another. */
   public static void main(String[] args)
@@ -207,14 +183,19 @@ public class Performance {
       try (var admin = Admin.of(configs())) {
         topicPattern.forEach(
             (topic, pr) -> {
-              for (var partitionReplica : pr.entrySet()) {
-                admin
-                    .creator()
-                    .topic(topic)
-                    .numberOfPartitions(partitionReplica.getKey())
-                    .numberOfReplicas(partitionReplica.getValue())
-                    .create();
-              }
+              pr.forEach(
+                  (p, r) ->
+                      Utils.waitFor(
+                          () -> {
+                            admin
+                                .creator()
+                                .topic(topic)
+                                .numberOfPartitions(p)
+                                .numberOfReplicas(r)
+                                .create();
+                            return true;
+                          },
+                          Duration.ofSeconds(30)));
             });
         Utils.waitFor(() -> admin.topicNames().containsAll(topics));
       }
@@ -242,16 +223,10 @@ public class Performance {
       try (var admin = Admin.of(configs())) {
         // the slow zk causes unknown error, so we have to wait it.
         return Utils.waitForNonNull(
-            () -> {
-              try {
-                return admin.offsets(new HashSet<>(topics)).entrySet().stream()
-                    .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().latest()));
-              } catch (Exception e) {
-                e.printStackTrace();
-                return null;
-              }
-            },
-            Duration.ofSeconds(5));
+            () ->
+                admin.offsets(new HashSet<>(topics)).entrySet().stream()
+                    .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().latest())),
+            Duration.ofSeconds(30));
       }
     }
 
@@ -306,7 +281,7 @@ public class Performance {
         names = {"--compression"},
         description =
             "String: the compression algorithm used by producer. Available algorithm are none, gzip, snappy, lz4, and zstd",
-        converter = CompressionField.class)
+        converter = Compression.Field.class)
     Compression compression = Compression.NONE;
 
     @Parameter(
@@ -327,12 +302,14 @@ public class Performance {
               .bootstrapServers(bootstrapServers())
               .compression(compression)
               .partitionClassName(partitioner)
+              .acks(acks)
               .buildTransactional()
           : Producer.builder()
               .configs(configs())
               .bootstrapServers(bootstrapServers())
               .compression(compression)
               .partitionClassName(partitioner)
+              .acks(acks)
               .build();
     }
 
@@ -363,10 +340,10 @@ public class Performance {
     DistributionType valueDistributionType = DistributionType.UNIFORM;
 
     @Parameter(
-        names = {"--specify.broker"},
+        names = {"--specify.brokers"},
         description =
-            "String: Used with SpecifyBrokerPartitioner to specify the brokers that partitioner can send.",
-        validateWith = NonEmptyStringField.class)
+            "String: The broker IDs to send to if the topic has partition on that broker.",
+        validateWith = PositiveIntegerListField.class)
     List<Integer> specifyBroker = List.of();
 
     Supplier<Integer> partitionSupplier() {
@@ -384,9 +361,10 @@ public class Performance {
     // replace DataSize by DataRate (see https://github.com/skiptests/astraea/issues/488)
     @Parameter(
         names = {"--throughput"},
-        description = "dataSize: size output per second. e.g. \"500KiB\"",
-        converter = DataSize.Field.class)
-    DataSize throughput = DataSize.GiB.of(500);
+        description =
+            "dataRate: size output/timeUnit. Default: second. e.g. \"500KiB/second\", \"100 MB/PT-10S\"",
+        converter = DataRate.Field.class)
+    DataRate throughput = DataRate.GiB.of(500).perSecond();
 
     @Parameter(
         names = {"--report.path"},
@@ -413,5 +391,11 @@ public class Performance {
         description = "Consumer group id",
         validateWith = NonEmptyStringField.class)
     String groupId = "groupId-" + System.currentTimeMillis();
+
+    @Parameter(
+        names = {"--acks"},
+        description = "How many replicas should be synced when producing records.",
+        converter = Acks.Field.class)
+    Acks acks = Acks.ISRS;
   }
 }
diff --git a/app/src/main/java/org/astraea/app/performance/TrackerThread.java b/app/src/main/java/org/astraea/app/performance/TrackerThread.java
index f1ac840f..d78d9dd0 100644
--- a/app/src/main/java/org/astraea/app/performance/TrackerThread.java
+++ b/app/src/main/java/org/astraea/app/performance/TrackerThread.java
@@ -27,8 +27,12 @@ import java.util.function.Supplier;
 import java.util.function.ToDoubleFunction;
 import org.astraea.app.common.DataSize;
 import org.astraea.app.common.Utils;
+import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.MBeanClient;
 import org.astraea.app.metrics.client.HasNodeMetrics;
+import org.astraea.app.metrics.client.consumer.ConsumerMetrics;
+import org.astraea.app.metrics.client.consumer.HasConsumerCoordinatorMetrics;
+import org.astraea.app.metrics.client.producer.HasProducerTopicMetrics;
 import org.astraea.app.metrics.client.producer.ProducerMetrics;
 
 /** Print out the given metrics. */
@@ -57,18 +61,21 @@ public interface TrackerThread extends AbstractThread {
           System.out.printf(
               "  current traffic: %s/second%n",
               DataSize.Byte.of(
-                  ((Double)
-                          sumOfAttribute(
-                              ProducerMetrics.nodes(mBeanClient), HasNodeMetrics::outgoingByteRate))
-                      .longValue()));
+                  (long)
+                      sumOfAttribute(
+                          ProducerMetrics.nodes(mBeanClient), HasNodeMetrics::outgoingByteRate)));
+          System.out.printf(
+              "  error: %.1f records/second%n",
+              sumOfAttribute(
+                  ProducerMetrics.topics(mBeanClient), HasProducerTopicMetrics::recordErrorRate));
           producerReports.stream()
               .mapToLong(Report::max)
               .max()
-              .ifPresent(i -> System.out.println("  publish max latency: " + i + " ms"));
+              .ifPresent(i -> System.out.printf("  publish max latency: %d ms%n", i));
           producerReports.stream()
-              .mapToLong(Report::min)
-              .min()
-              .ifPresent(i -> System.out.println("  publish mim latency: " + i + " ms"));
+              .mapToDouble(Report::avgLatency)
+              .average()
+              .ifPresent(i -> System.out.printf("  publish average latency: %.3f ms%n", i));
           for (int i = 0; i < producerReports.size(); ++i) {
             System.out.printf(
                 "  producer[%d] average throughput: %.3f MB%n",
@@ -104,29 +111,34 @@ public interface TrackerThread extends AbstractThread {
           System.out.printf(
               "  current traffic: %s/second%n",
               DataSize.Byte.of(
-                  ((Double)
-                          sumOfAttribute(
-                              ProducerMetrics.nodes(mBeanClient), HasNodeMetrics::incomingByteRate))
-                      .longValue()));
+                  (long)
+                      sumOfAttribute(
+                          ProducerMetrics.nodes(mBeanClient), HasNodeMetrics::incomingByteRate)));
           consumerReports.stream()
               .mapToLong(Report::max)
               .max()
-              .ifPresent(i -> System.out.println("  end-to-end max latency: " + i + " ms"));
-          consumerReports.stream()
-              .mapToLong(Report::min)
-              .min()
-              .ifPresent(i -> System.out.println("  end-to-end mim latency: " + i + " ms"));
+              .ifPresent(i -> System.out.printf("  end-to-end max latency: %d ms%n", i));
           consumerReports.stream()
-              .mapToLong(ConsumerThread.Report::maxSubscriptionLatency)
+              .mapToDouble(Report::avgLatency)
+              .average()
+              .ifPresent(i -> System.out.printf("  end-to-end average latency: %.3f ms%n", i));
+          var metrics = ConsumerMetrics.coordinators(mBeanClient);
+          metrics.stream()
+              .mapToDouble(HasConsumerCoordinatorMetrics::rebalanceLatencyMax)
               .max()
-              .ifPresent(i -> System.out.println("  subscription max latency: " + i + " ms"));
-          consumerReports.stream()
-              .mapToDouble(ConsumerThread.Report::avgSubscriptionLatency)
+              .ifPresent(i -> System.out.printf("  rebalance max latency: %.3f ms%n", i));
+          metrics.stream()
+              .mapToDouble(HasConsumerCoordinatorMetrics::rebalanceLatencyAvg)
               .average()
-              .ifPresent(i -> System.out.println("  subscription average latency: " + i + " ms"));
-          for (int i = 0; i < consumerReports.size(); ++i) {
+              .ifPresent(i -> System.out.printf("  rebalance average latency: %.3f ms%n", i));
+          for (var i = 0; i < consumerReports.size(); ++i) {
             var report = consumerReports.get(i);
-            System.out.printf("  consumer[%d] has %d partitions%n", i, report.assignments().size());
+            var ms =
+                metrics.stream().filter(m -> m.clientId().equals(report.clientId())).findFirst();
+            if (ms.isPresent()) {
+              System.out.printf(
+                  "  consumer[%d] has %d partitions%n", i, (int) ms.get().assignedPartitions());
+            }
             System.out.printf(
                 "  consumer[%d] average throughput: %.3f MB%n",
                 i, Utils.averageMB(duration, report.totalBytes()));
@@ -156,8 +168,13 @@ public interface TrackerThread extends AbstractThread {
                       + "min "
                       + duration.toSecondsPart()
                       + "sec");
-              if (!producerDone) producerDone = logProducers.apply(duration);
-              if (!consumerDone) consumerDone = logConsumers.apply(duration);
+              if (!producerDone) {
+                producerDone = logProducers.apply(duration);
+                consumerDone = logConsumers.apply(duration);
+              }
+              // if producers are not DONE, consumers should keep running as there are more data in
+              // the future.
+              if (!producerDone) consumerDone = false;
               if (producerDone && consumerDone) return;
 
               // Log after waiting for one second
@@ -199,8 +216,8 @@ public interface TrackerThread extends AbstractThread {
    * @param mbeans mBeans fetched by the receivers
    * @return sum of the latest given attribute of all beans which is instance of HasNodeMetrics.
    */
-  static double sumOfAttribute(
-      Collection<HasNodeMetrics> mbeans, ToDoubleFunction<HasNodeMetrics> targetAttribute) {
+  static <T extends HasBeanObject> double sumOfAttribute(
+      Collection<T> mbeans, ToDoubleFunction<T> targetAttribute) {
     return mbeans.stream().mapToDouble(targetAttribute).filter(d -> !Double.isNaN(d)).sum();
   }
 
diff --git a/app/src/main/java/org/astraea/app/producer/Acks.java b/app/src/main/java/org/astraea/app/producer/Acks.java
new file mode 100644
index 00000000..d6f0548f
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/producer/Acks.java
@@ -0,0 +1,84 @@
+/*
+ * 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.producer;
+
+import com.beust.jcommander.ParameterException;
+import java.util.Arrays;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public enum Acks {
+  /** wait for all isrs */
+  ISRS("isrs"),
+  /** wait for leader only */
+  ONLY_LEADER("leader"),
+  /** wait for nothing */
+  NONE("none");
+
+  private final String alias;
+
+  Acks(String alias) {
+    this.alias = alias;
+  }
+
+  public String alias() {
+    return alias;
+  }
+
+  @Override
+  public String toString() {
+    return alias();
+  }
+
+  public static Acks ofAlias(String alias) {
+    return Arrays.stream(Acks.values())
+        .filter(a -> a.alias().equalsIgnoreCase(alias))
+        .findFirst()
+        .orElseThrow(
+            () ->
+                new IllegalArgumentException(
+                    "the "
+                        + alias
+                        + " is unsupported. The supported algorithms are "
+                        + Stream.of(Acks.values())
+                            .map(Acks::alias)
+                            .collect(Collectors.joining(","))));
+  }
+
+  public String valueOfKafka() {
+    switch (this) {
+      case ISRS:
+        return "all";
+      case ONLY_LEADER:
+        return "1";
+      default:
+        return "0";
+    }
+  }
+
+  public static class Field extends org.astraea.app.argument.Field<Acks> {
+
+    @Override
+    public Acks convert(String value) {
+      try {
+        return Acks.ofAlias(value);
+      } catch (IllegalArgumentException e) {
+        throw new ParameterException(e);
+      }
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/app/producer/Builder.java b/app/src/main/java/org/astraea/app/producer/Builder.java
index da59a9d9..512ee1b4 100644
--- a/app/src/main/java/org/astraea/app/producer/Builder.java
+++ b/app/src/main/java/org/astraea/app/producer/Builder.java
@@ -80,6 +80,10 @@ public class Builder<Key, Value> {
     return config(ProducerConfig.COMPRESSION_TYPE_CONFIG, compression.nameOfKafka());
   }
 
+  public Builder<Key, Value> acks(Acks acks) {
+    return config(ProducerConfig.ACKS_CONFIG, acks.valueOfKafka());
+  }
+
   /**
    * set the transaction id. If you set the transaction id, the builder will always build
    * transactional producer.
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 7f2ff9c1..e4f5ea9e 100644
--- a/app/src/main/java/org/astraea/app/web/BalancerHandler.java
+++ b/app/src/main/java/org/astraea/app/web/BalancerHandler.java
@@ -27,6 +27,7 @@ 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.ClusterInfo;
 import org.astraea.app.admin.Replica;
 import org.astraea.app.admin.TopicPartitionReplica;
 import org.astraea.app.balancer.BalancerUtils;
@@ -34,8 +35,11 @@ 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.ClusterCost;
 import org.astraea.app.cost.HasClusterCost;
-import org.astraea.app.cost.NodeTopicSizeCost;
+import org.astraea.app.cost.HasMoveCost;
+import org.astraea.app.cost.MoveCost;
+import org.astraea.app.cost.ReplicaSizeCost;
 
 class BalancerHandler implements Handler {
 
@@ -47,14 +51,34 @@ class BalancerHandler implements Handler {
   private final Admin admin;
   private final RebalancePlanGenerator generator = RebalancePlanGenerator.random(30);
   final HasClusterCost costFunction;
+  final HasMoveCost moveCostFunction;
 
   BalancerHandler(Admin admin) {
-    this(admin, new NodeTopicSizeCost());
+    this(admin, new ReplicaSizeCost(), new ReplicaSizeCost());
   }
 
-  BalancerHandler(Admin admin, HasClusterCost costFunction) {
+  BalancerHandler(Admin admin, HasClusterCost costFunction, HasMoveCost moveCostFunction) {
     this.admin = admin;
     this.costFunction = costFunction;
+    this.moveCostFunction = moveCostFunction;
+  }
+
+  Map<ClusterLogAllocation, Map.Entry<ClusterCost, MoveCost>> planCosts(
+      ClusterInfo<Replica> clusterInfo, int limit, ClusterLogAllocation targetAllocations) {
+    return generator
+        .generate(admin.brokerFolders(), targetAllocations)
+        .limit(limit)
+        .collect(
+            Collectors.toMap(
+                RebalancePlanProposal::rebalancePlan,
+                rebalancePlanProposal ->
+                    Map.entry(
+                        costFunction.clusterCost(clusterInfo, ClusterBean.EMPTY),
+                        moveCostFunction.moveCost(
+                            clusterInfo,
+                            BalancerUtils.update(
+                                clusterInfo, rebalancePlanProposal.rebalancePlan()),
+                            ClusterBean.EMPTY))));
   }
 
   @Override
@@ -67,26 +91,19 @@ class BalancerHandler implements Handler {
     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(), targetAllocations)
-            .limit(limit)
-            .map(RebalancePlanProposal::rebalancePlan)
-            .map(
-                cla ->
-                    Map.entry(
-                        cla,
-                        costFunction
-                            .clusterCost(BalancerUtils.update(clusterInfo, cla), ClusterBean.EMPTY)
-                            .value()))
-            .filter(e -> e.getValue() <= cost)
-            .min(Comparator.comparingDouble(Map.Entry::getValue));
-
+        planCosts(clusterInfo, limit, targetAllocations).entrySet().stream()
+            .filter(e -> e.getValue().getKey().value() <= cost)
+            .min(Comparator.comparingDouble(x -> x.getValue().getKey().value()));
+    var migrationCosts =
+        planAndCost
+            .map(x -> x.getValue().getValue())
+            .map(value -> List.of(new MigrationCost(value)))
+            .orElseGet(List::of);
     return new Report(
         cost,
-        planAndCost.map(Map.Entry::getValue).orElse(cost),
+        planAndCost.map(x -> x.getValue().getKey().value()).orElse(cost),
         limit,
         costFunction.getClass().getSimpleName(),
         planAndCost
@@ -112,7 +129,8 @@ class BalancerHandler implements Handler {
                                                 .orElse(null)),
                                     placements(entry.getKey().logPlacements(tp), ignored -> null)))
                         .collect(Collectors.toUnmodifiableList()))
-            .orElse(List.of()));
+            .orElse(List.of()),
+        migrationCosts);
   }
 
   static List<Placement> placements(List<LogPlacement> lps, Function<LogPlacement, Long> size) {
@@ -149,21 +167,54 @@ class BalancerHandler implements Handler {
     }
   }
 
+  static class BrokerCost {
+    int brokerId;
+    long cost;
+
+    BrokerCost(int brokerId, long cost) {
+      this.brokerId = brokerId;
+      this.cost = cost;
+    }
+  }
+
+  static class MigrationCost {
+    final String function;
+    final long totalCost;
+    final List<BrokerCost> cost;
+    final String unit;
+
+    MigrationCost(MoveCost moveCost) {
+      this.function = moveCost.name();
+      this.totalCost = moveCost.totalCost();
+      this.cost =
+          moveCost.changes().entrySet().stream()
+              .map(x -> new BrokerCost(x.getKey(), x.getValue()))
+              .collect(Collectors.toList());
+      this.unit = moveCost.unit();
+    }
+  }
+
   static class Report implements Response {
     final double cost;
     final double newCost;
-
     final int limit;
-
     final String function;
     final List<Change> changes;
-
-    Report(double cost, double newCost, int limit, String function, List<Change> changes) {
+    final List<MigrationCost> migrationCosts;
+
+    Report(
+        double cost,
+        double newCost,
+        int limit,
+        String function,
+        List<Change> changes,
+        List<MigrationCost> migrationCosts) {
       this.cost = cost;
       this.newCost = newCost;
       this.limit = limit;
       this.function = function;
       this.changes = changes;
+      this.migrationCosts = migrationCosts;
     }
   }
 }
diff --git a/app/src/main/java/org/astraea/app/web/PostRequest.java b/app/src/main/java/org/astraea/app/web/PostRequest.java
index 2f0d972b..acf47052 100644
--- a/app/src/main/java/org/astraea/app/web/PostRequest.java
+++ b/app/src/main/java/org/astraea/app/web/PostRequest.java
@@ -19,6 +19,7 @@ package org.astraea.app.web;
 import com.google.gson.Gson;
 import com.google.gson.GsonBuilder;
 import com.google.gson.reflect.TypeToken;
+import java.lang.reflect.Type;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
@@ -74,6 +75,10 @@ public interface PostRequest {
     return Optional.ofNullable(raw().get(key)).map(v -> new Gson().fromJson(v, clz));
   }
 
+  default <T> Optional<T> get(String key, Type type) {
+    return Optional.ofNullable(raw().get(key)).map(v -> new Gson().fromJson(v, type));
+  }
+
   default <T> T value(String key, Class<T> clz) {
     return get(key, clz)
         .orElseThrow(() -> new NoSuchElementException("the value for " + key + " is nonexistent"));
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 96daac6a..3494fbee 100644
--- a/app/src/main/java/org/astraea/app/web/ThrottleHandler.java
+++ b/app/src/main/java/org/astraea/app/web/ThrottleHandler.java
@@ -16,14 +16,19 @@
  */
 package org.astraea.app.web;
 
+import com.google.gson.reflect.TypeToken;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.List;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import org.astraea.app.admin.Admin;
+import org.astraea.app.admin.TopicPartition;
 import org.astraea.app.admin.TopicPartitionReplica;
+import org.astraea.app.common.DataRate;
 
 public class ThrottleHandler implements Handler {
   private final Admin admin;
@@ -83,6 +88,128 @@ public class ThrottleHandler implements Handler {
     return new ThrottleSetting(brokers, simplify(leaderTargets, followerTargets));
   }
 
+  @Override
+  public Response post(Channel channel) {
+    var brokerToUpdate =
+        channel
+            .request()
+            .<Collection<BrokerThrottle>>get(
+                "brokers",
+                TypeToken.getParameterized(Collection.class, BrokerThrottle.class).getType())
+            .orElse(List.of());
+    var topics =
+        channel
+            .request()
+            .<Collection<TopicThrottle>>get(
+                "topics",
+                TypeToken.getParameterized(Collection.class, TopicThrottle.class).getType())
+            .orElse(List.of());
+
+    final var throttler = admin.replicationThrottler();
+    // ingress
+    throttler.ingress(
+        brokerToUpdate.stream()
+            .filter(broker -> broker.ingress != null)
+            .collect(
+                Collectors.toUnmodifiableMap(
+                    broker -> broker.id, broker -> DataRate.Byte.of(broker.ingress).perSecond())));
+    // egress
+    throttler.egress(
+        brokerToUpdate.stream()
+            .filter(broker -> broker.egress != null)
+            .collect(
+                Collectors.toUnmodifiableMap(
+                    broker -> broker.id, broker -> DataRate.Byte.of(broker.egress).perSecond())));
+
+    topics.forEach(
+        topic -> {
+          //noinspection ConstantConditions, the deserialization result must leave this value null
+          if (topic.name == null)
+            throw new IllegalArgumentException("The 'name' key of topic throttle must be given");
+          else if (topic.partition == null && topic.broker == null && topic.type == null) {
+            throttler.throttle(topic.name);
+          } else if (topic.partition != null && topic.broker == null && topic.type == null) {
+            throttler.throttle(TopicPartition.of(topic.name, topic.partition));
+          } else if (topic.partition != null && topic.broker != null && topic.type == null) {
+            throttler.throttle(TopicPartitionReplica.of(topic.name, topic.partition, topic.broker));
+          } else if (topic.partition != null && topic.broker != null && topic.type != null) {
+            var replica = TopicPartitionReplica.of(topic.name, topic.partition, topic.broker);
+            if (topic.type.equals("leader")) throttler.throttleLeader(replica);
+            else if (topic.type.equals("follower")) throttler.throttleFollower(replica);
+            else throw new IllegalArgumentException("Unknown throttle type: " + topic.type);
+          } else {
+            throw new IllegalArgumentException(
+                "The TopicThrottle argument is not supported: " + topic);
+          }
+        });
+
+    var affectedResources = throttler.apply();
+    var affectedBrokers =
+        Stream.concat(
+                affectedResources.ingress().keySet().stream(),
+                affectedResources.egress().keySet().stream())
+            .distinct()
+            .map(
+                broker ->
+                    BrokerThrottle.of(
+                        broker,
+                        affectedResources.ingress().get(broker),
+                        affectedResources.egress().get(broker)))
+            .collect(Collectors.toUnmodifiableList());
+    var affectedTopics = simplify(affectedResources.leaders(), affectedResources.followers());
+    return new ThrottleSetting(affectedBrokers, affectedTopics);
+  }
+
+  @Override
+  public Response delete(Channel channel) {
+    if (channel.queries().containsKey("topic")) {
+      var topic =
+          new TopicThrottle(
+              channel.queries().get("topic"),
+              Optional.ofNullable(channel.queries().get("partition"))
+                  .map(Integer::parseInt)
+                  .orElse(null),
+              Optional.ofNullable(channel.queries().get("replica"))
+                  .map(Integer::parseInt)
+                  .orElse(null),
+              channel.queries().get("type") == null
+                  ? null
+                  : Arrays.stream(LogIdentity.values())
+                      .filter(x -> x.name().equals(channel.queries().get("type")))
+                      .findFirst()
+                      .orElseThrow(IllegalArgumentException::new));
+
+      if (topic.partition == null && topic.broker == null && topic.type == null)
+        admin.clearReplicationThrottle(topic.name);
+      else if (topic.partition != null && topic.broker == null && topic.type == null)
+        admin.clearReplicationThrottle(TopicPartition.of(topic.name, topic.partition));
+      else if (topic.partition != null && topic.broker != null && topic.type == null)
+        admin.clearReplicationThrottle(
+            TopicPartitionReplica.of(topic.name, topic.partition, topic.broker));
+      else if (topic.partition != null && topic.broker != null && topic.type.equals("leader"))
+        admin.clearLeaderReplicationThrottle(
+            TopicPartitionReplica.of(topic.name, topic.partition, topic.broker));
+      else if (topic.partition != null && topic.broker != null && topic.type.equals("follower"))
+        admin.clearFollowerReplicationThrottle(
+            TopicPartitionReplica.of(topic.name, topic.partition, topic.broker));
+      else
+        throw new IllegalArgumentException("The argument is not supported: " + channel.queries());
+
+      return Response.ACCEPT;
+    } else if (channel.queries().containsKey("broker")) {
+      var broker = Integer.parseInt(channel.queries().get("broker"));
+      var bandwidth = channel.queries().get("type").split("\\+");
+      for (String target : bandwidth) {
+        if (target.equals("ingress")) admin.clearIngressReplicationThrottle(Set.of(broker));
+        else if (target.equals("egress")) admin.clearEgressReplicationThrottle(Set.of(broker));
+        else throw new IllegalArgumentException("Unknown clear target: " + target);
+      }
+      return Response.ACCEPT;
+    } else {
+      return Response.BAD_REQUEST;
+    }
+  }
+
   /**
    * break apart the {@code throttled.replica} string setting into a set of topic/partition/replicas
    */
@@ -110,37 +237,18 @@ public class ThrottleHandler implements Handler {
    */
   private Set<TopicThrottle> simplify(
       Set<TopicPartitionReplica> leaders, Set<TopicPartitionReplica> followers) {
-    var commonReplicas =
-        leaders.stream().filter(followers::contains).collect(Collectors.toUnmodifiableSet());
-
-    var simplifiedReplicas =
-        commonReplicas.stream()
-            .map(
-                replica ->
-                    new TopicThrottle(
-                        replica.topic(), replica.partition(), replica.brokerId(), null));
-    var leaderReplicas =
-        leaders.stream()
-            .filter(replica -> !commonReplicas.contains(replica))
-            .map(
-                replica ->
-                    new TopicThrottle(
-                        replica.topic(),
-                        replica.partition(),
-                        replica.brokerId(),
-                        LogIdentity.leader));
-    var followerReplicas =
-        followers.stream()
-            .filter(replica -> !commonReplicas.contains(replica))
-            .map(
-                replica ->
-                    new TopicThrottle(
-                        replica.topic(),
-                        replica.partition(),
-                        replica.brokerId(),
-                        LogIdentity.follower));
-
-    return Stream.concat(Stream.concat(simplifiedReplicas, leaderReplicas), followerReplicas)
+    return Stream.concat(leaders.stream(), followers.stream())
+        .distinct()
+        .map(
+            r -> {
+              if (leaders.contains(r) && followers.contains(r))
+                return new TopicThrottle(r.topic(), r.partition(), r.brokerId(), null);
+              if (leaders.contains(r))
+                return new TopicThrottle(
+                    r.topic(), r.partition(), r.brokerId(), LogIdentity.leader);
+              return new TopicThrottle(
+                  r.topic(), r.partition(), r.brokerId(), LogIdentity.follower);
+            })
         .collect(Collectors.toUnmodifiableSet());
   }
 
@@ -160,6 +268,13 @@ public class ThrottleHandler implements Handler {
     final Long ingress;
     final Long egress;
 
+    static BrokerThrottle of(int id, DataRate ingress, DataRate egress) {
+      return new BrokerThrottle(
+          id,
+          (ingress != null) ? ((long) ingress.byteRate()) : (null),
+          (egress != null) ? ((long) egress.byteRate()) : (null));
+    }
+
     BrokerThrottle(int id, Long ingress, Long egress) {
       this.id = id;
       this.ingress = ingress;
@@ -217,7 +332,7 @@ public class ThrottleHandler implements Handler {
     }
 
     TopicThrottle(String name, Integer partition, Integer broker, LogIdentity identity) {
-      this.name = name;
+      this.name = Objects.requireNonNull(name);
       this.partition = partition;
       this.broker = broker;
       this.type = (identity == null) ? null : identity.name();
diff --git a/config/partitionerConfig.properties b/config/smoothDispatchConfig.properties
similarity index 100%
rename from config/partitionerConfig.properties
rename to config/smoothDispatchConfig.properties
diff --git a/docker/build_deps_image.sh b/docker/build_deps_image.sh
index e05c9b67..2cf7c113 100755
--- a/docker/build_deps_image.sh
+++ b/docker/build_deps_image.sh
@@ -36,12 +36,17 @@ RUN apt-get update && apt-get install -y \
   libncurses5 \
   curl
 
+# download gradle 5 for previous kafka having no built-in gradlew
+WORKDIR /tmp
+RUN wget https://downloads.gradle-dn.com/distributions/gradle-5.6.4-bin.zip
+RUN unzip gradle-5.6.4-bin.zip
+
 # build code and download dependencies
 WORKDIR /astraea
 RUN git clone https://github.com/skiptests/astraea.git /astraea
 RUN ./gradlew clean build -x test
 # trigger download of database
-RUN ./gradlew cleanTest test --tests DatabaseTest
+RUN ./gradlew cleanTest app:test --tests DatabaseTest
 
 WORKDIR /root
 " >"$DOCKERFILE"
diff --git a/docker/start_broker.sh b/docker/start_broker.sh
index b3b87976..72ba11e3 100755
--- a/docker/start_broker.sh
+++ b/docker/start_broker.sh
@@ -18,6 +18,7 @@ declare -r DOCKER_FOLDER=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" &>/dev/null
 source $DOCKER_FOLDER/docker_build_common.sh
 
 # ===============================[global variables]===============================
+declare -r ACCOUNT=${ACCOUNT:-skiptests}
 declare -r VERSION=${REVISION:-${VERSION:-3.2.1}}
 declare -r DOCKERFILE=$DOCKER_FOLDER/broker.dockerfile
 declare -r CONFLUENT_BROKER=${CONFLUENT_BROKER:-false}
@@ -44,12 +45,10 @@ declare -r ZOOKEEPER_CONNECT=$(echo $1 | sed -n 's/^zookeeper.connect=\(.\+\)$/\
 declare -r HEAP_OPTS="${HEAP_OPTS:-"-Xmx2G -Xms2G"}"
 declare -r BROKER_PROPERTIES="/tmp/server-${BROKER_PORT}.properties"
 if [[ "$CONFLUENT_BROKER" = "true" ]]; then
-    declare -r REPO=${REPO:-ghcr.io/skiptests/astraea/confluent.broker}
-    declare -r IMAGE_NAME="$REPO:$CONFLUENT_VERSION"
+    declare -r IMAGE_NAME="ghcr.io/${ACCOUNT}/astraea/confluent.broker:$CONFLUENT_VERSION"
     declare -r SCRIPT_LOCATION_IN_CONTAINER="./bin/kafka-server-start"
 else
-    declare -r REPO=${REPO:-ghcr.io/skiptests/astraea/broker}
-    declare -r IMAGE_NAME="$REPO:$VERSION"
+    declare -r IMAGE_NAME="ghcr.io/${ACCOUNT}/astraea/broker:$VERSION"
     declare -r SCRIPT_LOCATION_IN_CONTAINER="./bin/kafka-server-start.sh"
 fi
 # cleanup the file if it is existent
@@ -65,7 +64,7 @@ function showHelp() {
   echo "    num.io.threads=10                        set broker I/O threads"
   echo "    num.network.threads=10                   set broker network threads"
   echo "ENV: "
-  echo "    REPO=astraea/broker                      set the docker repo"
+  echo "    ACCOUNT=skiptests                      set the github account"
   echo "    HEAP_OPTS=\"-Xmx2G -Xms2G\"                set broker JVM memory"
   echo "    REVISION=trunk                           set revision of kafka source code to build container"
   echo "    VERSION=3.2.1                            set version of kafka distribution"
@@ -113,14 +112,13 @@ WORKDIR /
 }
 
 function generateDockerfileBySource() {
-  echo "# this dockerfile is generated dynamically
-FROM ubuntu:22.04 AS build
-
-# install tools
-RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y openjdk-11-jdk wget git curl
+  local repo="https://github.com/apache/kafka"
+  if [[ "$ACCOUNT" != "skiptests" ]]; then
+    repo="https://github.com/${ACCOUNT}/kafka"
+  fi
 
-# add user
-RUN groupadd $USER && useradd -ms /bin/bash -g $USER $USER
+  echo "# this dockerfile is generated dynamically
+FROM ghcr.io/skiptests/astraea/deps AS build
 
 # download jmx exporter
 RUN mkdir /opt/jmx_exporter
@@ -129,12 +127,14 @@ RUN wget https://raw.githubusercontent.com/prometheus/jmx_exporter/master/exampl
 RUN wget https://REPO1.maven.org/maven2/io/prometheus/jmx/jmx_prometheus_javaagent/${EXPORTER_VERSION}/jmx_prometheus_javaagent-${EXPORTER_VERSION}.jar
 
 # build kafka from source code
-RUN git clone https://github.com/apache/kafka /tmp/kafka
+RUN git clone $repo /tmp/kafka
 WORKDIR /tmp/kafka
 RUN git checkout $VERSION
+# generate gradlew for previous
+RUN cp /tmp/kafka/gradlew /tmp/gradlew || /tmp/gradle-5.6.4/bin/gradle
 RUN ./gradlew clean releaseTarGz
 RUN mkdir /opt/kafka
-RUN tar -zxvf \$(find ./core/build/distributions/ -maxdepth 1 -type f -name kafka_*SNAPSHOT.tgz) -C /opt/kafka --strip-components=1
+RUN tar -zxvf \$(find ./core/build/distributions/ -maxdepth 1 -type f \( -iname \"kafka*tgz\" ! -iname \"*sit*\" \)) -C /opt/kafka --strip-components=1
 
 FROM ubuntu:22.04
 
diff --git a/docs/dispatcher/README.md b/docs/dispatcher/README.md
new file mode 100644
index 00000000..b628e5a6
--- /dev/null
+++ b/docs/dispatcher/README.md
@@ -0,0 +1,24 @@
+Astraea Dispatcher 中文文件
+===
+Astraea Dispatcher 是強大且高效率的 Kafka Partitioner 實作，提供豐富且彈性的叢集負載選項，從寫入端動態維持使用者定義後的負載平衡.
+
+### 通過gradle引入Astraea
+在build.gradle中添加以下內容
+```gradle
+repositories {
+    maven {
+        url = "https://maven.pkg.github.com/skiptests/astraea"
+        credentials {
+            username = System.getenv("GITHUB_ACTOR")
+            password = System.getenv("GITHUB_TOKEN")
+        }
+    }
+    
+    dependencies {
+        implementation 'org.astraea:app:0.0.1-SNAPSHOT'
+    }
+}
+```
+
+### Astraea Dispatcher實作
+1. [Smooth Dispatcher](smooth_dispatcher.md):  通過收集多metrics數據，結合熵權法與AHP進行節點狀況評估。再根據評估結果，使用smooth weight round-robin進行資料的調配。
\ No newline at end of file
diff --git a/docs/dispatcher/smooth_dispatcher.md b/docs/dispatcher/smooth_dispatcher.md
new file mode 100644
index 00000000..6cc535ba
--- /dev/null
+++ b/docs/dispatcher/smooth_dispatcher.md
@@ -0,0 +1,67 @@
+### SmoothDispatcher
+
+#### SmoothDispatcher爲了解決什麼問題
+分佈式流處理系統Kafka，因其容器在節點中分佈不均勻導致的叢集節點間負載不一致。負載不平衡的情況不僅會導致叢集中某些節點的硬體資源閒置，高負載節點處理資料的積壓會讓整個叢集處理資料的頻寬受制於它。我們提出SmoothDispatch ,這是一個基於叢集狀況監測，通過對收集的數據處理計算每臺節點的負載分數，以此來安排規劃每臺節點應承擔的資料負載量。當叢集中有節點達到負載上限時，Smooth Dispatch無須讓正在運行的業務下線，也能夠在負載不均勻的叢集中提升吞吐量效能，並且叢集負載越不均勻改善的程度越明顯。
+
+#### SmoothDispatcher的特點
+1. 綜合考量多維度Kafka metrics，得出能夠切實代表節點狀況的負載分數。
+2. 給予高負載節點低權重，將資料按節點權重均勻分配，緩解高負載節點負載的同時降低。
+3. 更平滑的資料發佈模式，高權重的節點也不會被連續選中變爲hotspot。
+
+#### SmoothDispatcher需要配置文檔信息
+在部署SmoothDispatcher，需要設置一些參數，這是因爲監測Kafka叢集狀況的數據來源於Kafka metrics。我們需要通過曝露的jmx.port來獲取這些信息。
+這些需要配置的信息被記錄在了config/partitionerConfig.properties中。
+
+#### SmoothDispatcher Configurations
+通過配置partitioner參數,指定Kafka運行SmoothDispatch。
+```java
+class demo{
+    void initConfig(){
+        Properties properties = new Properties();
+        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, org.astraea.partitioner.smooth.SmoothWeightRoundRobinDispatcher);
+    }
+}
+```
+
+#### 配置jmx port
+使用Dispatcher需要你配置kafka叢集各節點的jmx port。有兩種方法進行配置。
+
+1. 通過configs參數傳入dispatch配置文檔的地址。
+```bash
+--configs partitioner.config=~/astraea/config/smoothDispatchConfig.properties
+```
+
+如下是配置文檔內容，修改對應的參數即可。
+
+```bash
+#The properties are used to configure SmoothWeightPartitioner
+#
+# If broker.id.jmx.port is configured, then the jmx port of these brokers
+# will be configured first. If there are brokers that are not configured
+# to jmx port, then jmx.port will be used for configuration. And if any
+# broker is not configured, and jmx.port is not configured at the same
+# time, then an error will be reported.
+#
+###########################Smooth Partitioner Basics############################
+
+#If no jmx port is specified for the broker,jmx.port will be used.
+jmx.port=Default;
+
+# list of brokers used for jmx port about the rest of the cluster
+# format: broker.{brokerID}.jmx.port=jmx.port
+#broker.0.jmx.port=9999
+#broker.1.jmx.port=8888
+#......
+bootstrap.servers=localhost:9092
+broker.id.jmx.port=Default;
+```
+2. 通過configs傳入下列參數
+```bash
+--configs bootstrap.servers=localhost:9092, jmx.port=localhost:8000
+```
+
+|            參數名稱             | 說明                                                                      |
+|:---------------------------:|:------------------------------------------------------------------------|
+|      bootstrap.servers      | (必填) 欲連接的Kafka server address                                           |
+|          jmx.port           | (選填) 當你的jmx port擁有統一的port，那麼可以使用該參數進行一次性配置。你無需額外配置broker.id.jmx.port。   |
+|     broker.id.jmx.port      | (選填) 當每一臺節點的jmx port不一致，那麼可以根據broker id進行逐一配置。沒有被配置的broker會使用jmx.port值。 |
\ No newline at end of file
diff --git a/docs/performance_benchmark.md b/docs/performance_benchmark.md
index f2dac197..b7e464b7 100644
--- a/docs/performance_benchmark.md
+++ b/docs/performance_benchmark.md
@@ -16,29 +16,30 @@
 
 #### Performance Benchmark Configurations
 
-|      參數名稱      | 說明                                                         |              預設值              |
-| :----------------: | :----------------------------------------------------------- | :------------------------------: |
-| bootstrap.servers  | (必填) 欲連接的Kafka server address                          |                無                |
-|    compression     | (選填) Kafka Producer使用的壓縮演算法，可用的壓縮演算法為：`gzip`, `snappy`, `lz4`, `zstd` |               none               |
-|       topics       | (選填) 選擇topic names，可一次建立多個 topics<br />例如 : --topics test,test1,test2 | testPerformance-{Time in millis} |
-|     partitions     | (選填) 建立topic時，設定的partition數<br />目前有兩種建立模式，<br />1. 每個 topic 有各自的設置 例如 : --partitions 10,20,30<br />2. 每個 topic 有相同的設置 例如 : --partitions 5 |                1                 |
-|      replicas      | (選填) 建立topic時，設定的replica數<br />目前有兩種建立模式，<br />1. 每個 topic 有各自的設置 例如 : --replicas 3,2,1<br />2. 每個 topic 有相同的設置 例如 : --replicas 2 |                1                 |
-|     consumers      | (選填) 欲開啟的consumer thread(s)數量                        |                1                 |
-|     producers      | (選填) 欲開啟的producer thread(s)數量                        |                1                 |
+|        參數名稱        | 說明                                                                                                                                                                                                                                                                    |               預設值                |
+|:------------------:|:----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|:--------------------------------:|
+| bootstrap.servers  | (必填) 欲連接的Kafka server address                                                                                                                                                                                                                                         |                無                 |
+|    compression     | (選填) Kafka Producer使用的壓縮演算法，可用的壓縮演算法為：`gzip`, `snappy`, `lz4`, `zstd`                                                                                                                                                                                                 |               none               |
+|       topics       | (選填) 選擇topic names，可一次建立多個 topics<br />例如 : --topics test,test1,test2                                                                                                                                                                                                 | testPerformance-{Time in millis} |
+|     partitions     | (選填) 建立topic時，設定的partition數<br />目前有兩種建立模式，<br />1. 每個 topic 有各自的設置 例如 : --partitions 10,20,30<br />2. 每個 topic 有相同的設置 例如 : --partitions 5                                                                                                                            |                1                 |
+|      replicas      | (選填) 建立topic時，設定的replica數<br />目前有兩種建立模式，<br />1. 每個 topic 有各自的設置 例如 : --replicas 3,2,1<br />2. 每個 topic 有相同的設置 例如 : --replicas 2                                                                                                                                     |                1                 |
+|     consumers      | (選填) 欲開啟的consumer thread(s)數量                                                                                                                                                                                                                                         |                1                 |
+|     producers      | (選填) 欲開啟的producer thread(s)數量                                                                                                                                                                                                                                         |                1                 |
 |     run.until      | (選填) 可選擇兩種結束執行的模式，一種是發送records數量達到設定值，另一種則是執行時間達到設定值，格式為`數值`+`單位`<br />1. 選擇producers要送多少records，範例：發送89000 records 後結束，"--run.until 89000records"<br />2. 選擇producer在給定時間內發送資料，時間單位可以選擇`days`, `day`, `h`, `m`, `s`, `ms`, `us`, `ns`，範例：執行一分鐘後結束， "--run.until 1m"。 |           1000records            |
-|      key.size      | (選填) 每筆record key的大小上限                              |              4Byte               |
-|  key.distribution  | (選填) key的分佈，可用的分佈為：`uniform`, `zipfian`, `latest`, `fixed` |             uniform              |
-|     value.size     | (選填) 每筆record value的大小上限                            |               1KiB               |
-| value.distribution | (選填) value的分佈， 可用的分佈為: `uniform`, `zipfian`, `latest`, `fixed` |             uniform              |
-|     prop.file      | (選填) 配置property file的路徑                               |               none               |
-|    partitioner     | (選填) 配置producer使用的partitioner                         |               none               |
-|      configs       | (選填) 給partitioner的設置檔。 設置格式為 "\<key1\>=\<value1\>[,\<key2\>=\<value2\>]*"。 <br />例如: "--configs broker.1001.jmx.port=14338,org.astraea.cost.ThroughputCost=1" |               none               |
-|     throughput     | (選填) 用來限制輸出資料的速度, 範例： "--throughput 2MiB"    |           500 GiB/sec            |
-|   specify.broker   | (選填) 指定broker的ID，送資料到指定的broker                  |               none               |
-|    report.path     | (選填) report file的檔案路徑                                 |               none               |
-|   report.format    | (選填) 選擇輸出檔案格式, 可用的格式：`csv`, `json`           |               csv                |
-|  transaction.size  | (選填) 每個transaction的records數量。若設置1以上，會使用transaction，否則都是一般write |                1                 |
-|      group.id      | (選填) 設置 consumer group id                                |     groupId-{Time in millis}     |
+|      key.size      | (選填) 每筆record key的大小上限                                                                                                                                                                                                                                                |              4Byte               |
+|  key.distribution  | (選填) key的分佈，可用的分佈為：`uniform`, `zipfian`, `latest`, `fixed`                                                                                                                                                                                                            |             uniform              |
+|     value.size     | (選填) 每筆record value的大小上限                                                                                                                                                                                                                                              |               1KiB               |
+| value.distribution | (選填) value的分佈， 可用的分佈為: `uniform`, `zipfian`, `latest`, `fixed`                                                                                                                                                                                                        |             uniform              |
+|     prop.file      | (選填) 配置property file的路徑                                                                                                                                                                                                                                               |               none               |
+|    partitioner     | (選填) 配置producer使用的partitioner                                                                                                                                                                                                                                         |               none               |
+|      configs       | (選填) 給partitioner的設置檔。 設置格式為 "\<key1\>=\<value1\>[,\<key2\>=\<value2\>]*"。 <br />例如: "--configs broker.1001.jmx.port=14338,org.astraea.cost.ThroughputCost=1"                                                                                                         |               none               |
+|     throughput     | (選填) 用來限制輸出資料的速度, 範例： "--throughput 2MiB/m", "--throughput 2GB" 預設值是秒 <br/>大小單位: MB, MiB, Kb etc. <br />時間單位: second(s), minute(m), hour(h), day(d) or PT expression(PT30S)                                                                                           |          500 GiB/second          |
+|  specify.brokers   | (選填) 指定broker的ID，送資料到指定的broker，若 broker 上有 "目標 topic 的 partition"                                                                                                                                                                                                     |               none               |
+|    report.path     | (選填) report file的檔案路徑                                                                                                                                                                                                                                                 |               none               |
+|   report.format    | (選填) 選擇輸出檔案格式, 可用的格式：`csv`, `json`                                                                                                                                                                                                                                    |               csv                |
+|  transaction.size  | (選填) 每個transaction的records數量。若設置1以上，會使用transaction，否則都是一般write                                                                                                                                                                                                        |                1                 |
+|      group.id      | (選填) 設置 consumer group id                                                                                                                                                                                                                                             |     groupId-{Time in millis}     |
+|        acks        | (選填) 設置 producer acks，                                                                                                                                                                                                                                                |         all，代表要等所有isr同步          |
 
 #### 使用範例
 
diff --git a/docs/web_server/web_api_balancer_chinese.md b/docs/web_server/web_api_balancer_chinese.md
index f88124b3..2b9f95bd 100644
--- a/docs/web_server/web_api_balancer_chinese.md
+++ b/docs/web_server/web_api_balancer_chinese.md
@@ -33,6 +33,13 @@ JSON Response 範例
     - `directory`: replica 存在資料的路徑
     - `size`: replica 在硬碟上的資料大小
   - `after`: 比較好的配置
+- `migrations`: 計算搬移計畫的成本
+  * `function`: 用來評估成本的演算法
+  * `totalCost`: 各個broker的成本總和
+  * `cost`: 針對各個broker計算成本的改變
+    * `brokerId`: 有掌管 replica 的節點 id
+    * `cost`: 改變的量，負值表示移出，正值表示移入
+  * `unit`: 成本的單位
 ```json
 {
   "cost": 0.04948716593053935,
diff --git a/docs/web_server/web_api_throttles_chinese.md b/docs/web_server/web_api_throttles_chinese.md
index baf8e2bb..25044b7b 100644
--- a/docs/web_server/web_api_throttles_chinese.md
+++ b/docs/web_server/web_api_throttles_chinese.md
@@ -95,7 +95,9 @@ brokers 每個資料欄位
 |---------|---------------------------------------------------------------------------------------------------------------------------------------| ------ |
 | id      | (必填) 欲更新節點的 id                                                                                                                        | 無     |
 | ingress | (選填) 特定 Kafka 節點的 replication 流入流量的限制(單位 bytes/sec)，附註只有 `topics` 中有記錄到的對象，其 replication 流量會受限制，如果此欄位不存在則代表此節點的 replication 流入流量沒有被修改。 | 無     |
-| egress  | (選填) 特定 Kafka 節點的 replication 流出流量的限制(單位 bytes/sec)，附註只有 `topics` 中有記錄到的對象，其 replication 流量會受限制。                                      | 無     |
+| egress  | (選填) 特定 Kafka 節點的 replication 流出流量的限制(單位 bytes/sec)，附註只有 `topics` 中有記錄到的對象，其 replication 流量會受限制，如果此欄位不存在則代表此節點的 replication 流入流量沒有被修改。 | 無     |
+
+給定一個沒有指定 ingress/egress 的輸入不會使任何資訊被修改，如：`{ "id": 1002 }`。
 
 topics 每個資料欄位
 
@@ -106,6 +108,8 @@ topics 每個資料欄位
 | broker    | (選填) 要套用 replication throttle 的 replica (所在之 broker)，如果沒有指定，所有既有的 replicas 都會被套用 | 無     |
 | type      | (選填) 要套用 replication throttle 的 replica 身份，此值可以是 `leader` 或是 `follower`，如果沒有指定，則二者都會被套用 | 無     |
 
+topic 描述格式只支援 `name`, `name, partition`, `name, partition, broker`, `name, partition, broker, type` 這四種 key 的組合，目前此 API 不支援其他種類的組合，比如 `name, type`。當給與這類型的組合，API 會回傳錯誤。
+
 JSON Response 範例
 
 - `brokers`: 回傳被此 POST 影響的結果。
@@ -164,15 +168,15 @@ cURL 範例
 * ##### 移除整個 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"
+  # 移除與 TopicA/partition 3/replica at broker 4 相關的所有 replication throttle
+  curl -X DELETE "http://localhost:8001/throttles?topic=MyTopicA&partition=3&replica=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"
+  # 移除 TopicA/partition 3/replica at broker 4 的 leader replication throttle
+  curl -X DELETE "http://localhost:8001/throttles?topic=MyTopicA&partition=3&replica=4&type=leader"
   ```
 
 * ##### 移除特定節點的 replication 輸入限流


Test patch:
diff --git a/app/src/test/java/org/astraea/app/cost/NodeTopicSizeCostTest.java b/app/src/main/java/org/astraea/app/cost/HasMoveCost.java
similarity index 52%
rename from app/src/test/java/org/astraea/app/cost/NodeTopicSizeCostTest.java
rename to app/src/main/java/org/astraea/app/cost/HasMoveCost.java
index 15a296de..360e8b71 100644
--- a/app/src/test/java/org/astraea/app/cost/NodeTopicSizeCostTest.java
+++ b/app/src/main/java/org/astraea/app/cost/HasMoveCost.java
@@ -16,29 +16,21 @@
  */
 package org.astraea.app.cost;
 
-import static org.mockito.Mockito.mock;
-
-import java.util.List;
-import java.util.Map;
 import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
-import org.astraea.app.metrics.BeanObject;
-import org.astraea.app.metrics.broker.LogMetrics;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Test;
-
-class NodeTopicSizeCostTest {
-  private final BeanObject bean =
-      new BeanObject(
-          "domain", Map.of("topic", "t", "partition", "10", "name", "SIZE"), Map.of("Value", 777));
+import org.astraea.app.admin.Replica;
 
-  @Test
-  void testBrokerCost() {
-    var meter = new LogMetrics.Log.Gauge(bean);
-    var cost = new NodeTopicSizeCost();
-    var result =
-        cost.brokerCost(mock(ClusterInfo.class), ClusterBean.of(Map.of(1, List.of(meter))));
-    Assertions.assertEquals(1, result.value().size());
-    Assertions.assertEquals(777, result.value().entrySet().iterator().next().getValue());
-  }
+public interface HasMoveCost extends CostFunction {
+  /**
+   * score migrate cost from originClusterInfo to newClusterInfo .
+   *
+   * @param originClusterInfo the clusterInfo before migrate
+   * @param newClusterInfo the mocked clusterInfo generate from balancer
+   * @param clusterBean cluster metrics
+   * @return the score of migrate cost
+   */
+  MoveCost moveCost(
+      ClusterInfo<Replica> originClusterInfo,
+      ClusterInfo<Replica> newClusterInfo,
+      ClusterBean clusterBean);
 }
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 30a1f1ff..a4d41707 100644
--- a/app/src/test/java/org/astraea/app/admin/AdminTest.java
+++ b/app/src/test/java/org/astraea/app/admin/AdminTest.java
@@ -27,6 +27,7 @@ import java.util.Comparator;
 import java.util.List;
 import java.util.Map;
 import java.util.NoSuchElementException;
+import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
@@ -727,14 +728,30 @@ public class AdminTest extends RequireBrokerCluster {
       Assertions.assertEquals(
           partitionCount * replicaCount, clusterInfo.availableReplicas(topic2).size());
       // ClusterInfo#availableReplicaLeaders
-      Assertions.assertEquals(partitionCount, clusterInfo.availableReplicaLeaders(topic0).size());
-      Assertions.assertEquals(partitionCount, clusterInfo.availableReplicaLeaders(topic1).size());
-      Assertions.assertEquals(partitionCount, clusterInfo.availableReplicaLeaders(topic2).size());
+      Assertions.assertEquals(partitionCount, clusterInfo.replicaLeaders(topic0).size());
+      Assertions.assertEquals(partitionCount, clusterInfo.replicaLeaders(topic1).size());
+      Assertions.assertEquals(partitionCount, clusterInfo.replicaLeaders(topic2).size());
       // No resource match found will raise exception
       Assertions.assertEquals(List.of(), clusterInfo.replicas("Unknown Topic"));
       Assertions.assertEquals(List.of(), clusterInfo.availableReplicas("Unknown Topic"));
-      Assertions.assertEquals(List.of(), clusterInfo.availableReplicaLeaders("Unknown Topic"));
+      Assertions.assertEquals(List.of(), clusterInfo.replicaLeaders("Unknown Topic"));
       Assertions.assertThrows(NoSuchElementException.class, () -> clusterInfo.node(-1));
+
+      admin
+          .partitions(Set.of(topic0, topic1, topic2))
+          .forEach(
+              p ->
+                  Assertions.assertNotEquals(
+                      Optional.empty(), clusterInfo.replicaLeader(p), "partition: " + p));
+      Set.of(topic0, topic1, topic2)
+          .forEach(
+              t ->
+                  brokerIds()
+                      .forEach(
+                          id -> Assertions.assertNotEquals(0, clusterInfo.replicas(id, t).size())));
+      admin
+          .partitions(Set.of(topic0, topic1, topic2))
+          .forEach(p -> Assertions.assertNotEquals(0, clusterInfo.replicas(p).size()));
     }
   }
 
@@ -1559,12 +1576,12 @@ public class AdminTest extends RequireBrokerCluster {
     try (Admin admin = Admin.of(bootstrapServers())) {
       var topic = Utils.randomString();
       admin.creator().topic(topic).numberOfPartitions(10).numberOfReplicas((short) 3).create();
-      Utils.sleep(Duration.ofMillis(100));
+      Utils.sleep(Duration.ofSeconds(1));
       admin.replicationThrottler().throttle(topic).apply();
-      Utils.sleep(Duration.ofMillis(100));
+      Utils.sleep(Duration.ofSeconds(1));
 
       admin.clearReplicationThrottle(TopicPartitionReplica.of(topic, 0, 0));
-      Utils.sleep(Duration.ofMillis(500));
+      Utils.sleep(Duration.ofSeconds(1));
 
       Assertions.assertTrue(
           fetchLeaderThrottle(topic).stream()
@@ -1649,4 +1666,93 @@ public class AdminTest extends RequireBrokerCluster {
           1, admin.topicNames(false).stream().filter(t -> t.equals(topicName)).count());
     }
   }
+
+  @Test
+  void testClearLeaderReplicationThrottle() {
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var topic = Utils.randomString();
+      var log = TopicPartitionReplica.of(topic, 0, 0);
+      admin.creator().topic(topic).numberOfPartitions(1).numberOfReplicas((short) 3).create();
+      Utils.sleep(Duration.ofSeconds(1));
+      admin.replicationThrottler().throttleLeader(log).apply();
+      Utils.sleep(Duration.ofSeconds(1));
+      Assertions.assertEquals(
+          "0:0",
+          admin
+              .topics(Set.of(topic))
+              .get(topic)
+              .value("leader.replication.throttled.replicas")
+              .orElse(""));
+
+      admin.clearLeaderReplicationThrottle(log);
+      Utils.sleep(Duration.ofSeconds(1));
+      Assertions.assertEquals(
+          "",
+          admin
+              .topics(Set.of(topic))
+              .get(topic)
+              .value("leader.replication.throttled.replicas")
+              .orElse(""));
+    }
+  }
+
+  @Test
+  void testClearFollowerReplicationThrottle() {
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var topic = Utils.randomString();
+      var log = TopicPartitionReplica.of(topic, 0, 0);
+      admin.creator().topic(topic).numberOfPartitions(1).numberOfReplicas((short) 3).create();
+      Utils.sleep(Duration.ofSeconds(1));
+      admin.replicationThrottler().throttleFollower(log).apply();
+      Utils.sleep(Duration.ofSeconds(1));
+      Assertions.assertEquals(
+          "0:0",
+          admin
+              .topics(Set.of(topic))
+              .get(topic)
+              .value("follower.replication.throttled.replicas")
+              .orElse(""));
+
+      admin.clearFollowerReplicationThrottle(log);
+      Utils.sleep(Duration.ofSeconds(1));
+      Assertions.assertEquals(
+          "",
+          admin
+              .topics(Set.of(topic))
+              .get(topic)
+              .value("follower.replication.throttled.replicas")
+              .orElse(""));
+    }
+  }
+
+  @Test
+  void testReplicationThrottleAffectedTargets() {
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var topic = Utils.randomString();
+      var replica0 = TopicPartitionReplica.of(topic, 0, 0);
+      var replica1 = TopicPartitionReplica.of(topic, 1, 1);
+      var replica2 = TopicPartitionReplica.of(topic, 2, 2);
+      admin.creator().topic(topic).numberOfPartitions(3).numberOfReplicas((short) 3).create();
+      Utils.sleep(Duration.ofSeconds(1));
+
+      var affectedResources =
+          admin
+              .replicationThrottler()
+              .ingress(Map.of(0, DataRate.GiB.of(1).perSecond()))
+              .egress(Map.of(1, DataRate.GiB.of(1).perSecond()))
+              .throttle(replica0)
+              .throttleLeader(replica1)
+              .throttleFollower(replica2)
+              .apply();
+
+      Assertions.assertEquals(Set.of(replica0, replica1), affectedResources.leaders());
+      Assertions.assertEquals(Set.of(replica0, replica2), affectedResources.followers());
+      Assertions.assertEquals(
+          Map.of(0, DataRate.GiB.of(1).perSecond()).toString(),
+          affectedResources.ingress().toString());
+      Assertions.assertEquals(
+          Map.of(1, DataRate.GiB.of(1).perSecond()).toString(),
+          affectedResources.egress().toString());
+    }
+  }
 }
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 1842f3a9..aa329a2c 100644
--- a/app/src/test/java/org/astraea/app/admin/ClusterInfoWithOfflineNodeTest.java
+++ b/app/src/test/java/org/astraea/app/admin/ClusterInfoWithOfflineNodeTest.java
@@ -46,7 +46,7 @@ public class ClusterInfoWithOfflineNodeTest extends RequireBrokerCluster {
           before.replicas(topicName).stream().filter(x -> !x.isOffline()).count());
       Assertions.assertEquals(
           partitionCount * replicaCount, before.availableReplicas(topicName).size());
-      Assertions.assertEquals(partitionCount, before.availableReplicaLeaders(topicName).size());
+      Assertions.assertEquals(partitionCount, before.replicaLeaders(topicName).size());
 
       // act
       int brokerToClose = ThreadLocalRandom.current().nextInt(0, 3);
@@ -62,13 +62,13 @@ public class ClusterInfoWithOfflineNodeTest extends RequireBrokerCluster {
           partitionCount * (replicaCount - 1), after.availableReplicas(topicName).size());
       Assertions.assertEquals(
           partitionCount,
-          after.availableReplicaLeaders(topicName).size(),
+          after.replicaLeaders(topicName).size(),
           "One of the rest replicas should take over the leadership");
       Assertions.assertTrue(
           after.availableReplicas(topicName).stream()
               .allMatch(x -> x.nodeInfo().id() != brokerToClose));
       Assertions.assertTrue(
-          after.availableReplicaLeaders(topicName).stream()
+          after.replicaLeaders(topicName).stream()
               .allMatch(x -> x.nodeInfo().id() != brokerToClose));
       Assertions.assertTrue(
           after.replicas(topicName).stream()
diff --git a/app/src/test/java/org/astraea/app/admin/CompressionTest.java b/app/src/test/java/org/astraea/app/admin/CompressionTest.java
index e15485a8..22d7d56d 100644
--- a/app/src/test/java/org/astraea/app/admin/CompressionTest.java
+++ b/app/src/test/java/org/astraea/app/admin/CompressionTest.java
@@ -16,8 +16,12 @@
  */
 package org.astraea.app.admin;
 
+import com.beust.jcommander.Parameter;
+import com.beust.jcommander.ParameterException;
 import java.util.Arrays;
+import java.util.stream.Stream;
 import org.apache.kafka.common.record.CompressionType;
+import org.astraea.app.argument.Argument;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
@@ -32,4 +36,26 @@ public class CompressionTest {
                 Assertions.assertEquals(
                     c.nameOfKafka(), CompressionType.forName(c.nameOfKafka()).name));
   }
+
+  @Test
+  void testConversion() {
+    var arg = new Compression.Field();
+    Stream.of(Compression.values())
+        .forEach(type -> Assertions.assertEquals(type, arg.convert(type.name())));
+    Assertions.assertThrows(ParameterException.class, () -> arg.convert("aaa"));
+  }
+
+  @Test
+  void testParse() {
+    var arg = Argument.parse(new FakeParameter(), new String[] {"--field", "gzip"});
+    Assertions.assertEquals(Compression.GZIP, arg.value);
+  }
+
+  private static class FakeParameter {
+    @Parameter(
+        names = {"--field"},
+        converter = Compression.Field.class,
+        validateWith = Compression.Field.class)
+    public Compression value;
+  }
 }
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 7d9a3037..4e43b309 100644
--- a/app/src/test/java/org/astraea/app/balancer/FakeClusterInfo.java
+++ b/app/src/test/java/org/astraea/app/balancer/FakeClusterInfo.java
@@ -24,6 +24,7 @@ import java.util.Set;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
+import java.util.stream.Stream;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.admin.NodeInfo;
 import org.astraea.app.admin.Replica;
@@ -127,7 +128,7 @@ public class FakeClusterInfo implements ClusterInfo<Replica> {
   }
 
   @Override
-  public List<Replica> replicas() {
-    return replicas;
+  public Stream<Replica> replicaStream() {
+    return replicas.stream();
   }
 }
diff --git a/app/src/test/java/org/astraea/app/common/DataRateTest.java b/app/src/test/java/org/astraea/app/common/DataRateTest.java
index e9086bc9..a80e27a0 100644
--- a/app/src/test/java/org/astraea/app/common/DataRateTest.java
+++ b/app/src/test/java/org/astraea/app/common/DataRateTest.java
@@ -19,10 +19,12 @@ package org.astraea.app.common;
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 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 java.math.BigInteger;
 import java.time.Duration;
+import java.time.format.DateTimeParseException;
 import java.time.temporal.ChronoUnit;
 import java.util.function.BiConsumer;
 import org.junit.jupiter.api.Assertions;
@@ -197,4 +199,72 @@ class DataRateTest {
   void testZero() {
     Assertions.assertEquals(0, DataRate.ZERO.byteRate());
   }
+
+  @Test
+  void parseDataRate() {
+    var converter = new DataRate.Field();
+
+    // test Duration
+    assertEquals(DataRate.Bit.of(100).perSecond(), converter.convert("100Bit/S"));
+    assertEquals(DataRate.Bit.of(100).perSecond(), converter.convert("100Bit/Second"));
+    assertEquals(DataRate.Bit.of(100).perSecond(), converter.convert("100Bit/s"));
+    assertEquals(DataRate.Bit.of(100).perSecond(), converter.convert("100Bit/second"));
+    assertEquals(DataRate.Bit.of(100).perSecond(), converter.convert("100Bit/seconds"));
+    assertEquals(DataRate.Bit.of(100).perMinute(), converter.convert("100Bit/m"));
+    assertEquals(DataRate.Bit.of(100).perMinute(), converter.convert("100Bit/minute"));
+    assertEquals(DataRate.Bit.of(100).perMinute(), converter.convert("100Bit/minutes"));
+    assertEquals(DataRate.Bit.of(100).perHour(), converter.convert("100Bit/h"));
+    assertEquals(DataRate.Bit.of(100).perHour(), converter.convert("100Bit/hour"));
+    assertEquals(DataRate.Bit.of(100).perHour(), converter.convert("100Bit/hours"));
+    assertEquals(DataRate.Bit.of(100).perDay(), converter.convert("100Bit/d"));
+    assertEquals(DataRate.Bit.of(100).perDay(), converter.convert("100Bit/day"));
+    assertEquals(DataRate.Bit.of(100).perDay(), converter.convert("100Bit/days"));
+    assertEquals(
+        DataRate.Bit.of(100).over(Duration.of(20345, ChronoUnit.MILLIS)),
+        converter.convert("100Bit/PT20.345S"));
+    assertEquals(
+        DataRate.Bit.of(100).over(Duration.of(15, ChronoUnit.MINUTES)),
+        converter.convert("100Bit/PT15M"));
+
+    // test Space
+    assertEquals(DataRate.Bit.of(100).perHour(), converter.convert("100 Bit/hour"));
+    assertEquals(DataRate.MB.of(250).perDay(), converter.convert("250 MB/d"));
+    assertEquals(DataRate.GB.of(1).perSecond(), converter.convert("1 GB"));
+
+    // test Size and Unit
+    assertEquals(DataRate.Byte.of(100).perSecond(), converter.convert("100Byte"));
+    assertEquals(DataRate.KB.of(150).perSecond(), converter.convert("150 KB"));
+    assertEquals(DataRate.MB.of(200).perSecond(), converter.convert("200 MB"));
+    assertEquals(DataRate.GB.of(250).perSecond(), converter.convert("250 GB"));
+    assertEquals(DataRate.TB.of(300).perSecond(), converter.convert("300 TB"));
+    assertEquals(DataRate.PB.of(350).perSecond(), converter.convert("350 PB"));
+    assertEquals(DataRate.EB.of(400).perSecond(), converter.convert("400 EB"));
+    assertEquals(DataRate.ZB.of(450).perSecond(), converter.convert("450 ZB"));
+    assertEquals(DataRate.YB.of(500).perSecond(), converter.convert("500 YB"));
+
+    // some example
+    assertEquals(DataRate.GB.of(1).perSecond(), converter.convert("1GB"));
+    assertEquals(DataRate.KiB.of(175).perMinute(), converter.convert("175KiB/M"));
+    assertEquals(DataRate.MB.of(200).perDay(), converter.convert("200 MB/d"));
+    assertEquals(DataRate.GB.of(17).perHour(), converter.convert("17GB/Hour"));
+    assertEquals(
+        DataRate.KB.of(111).over(Duration.of(25, ChronoUnit.SECONDS)),
+        converter.convert("111 KB/PT25S"));
+    assertEquals(
+        DataRate.KB.of(1024).over(Duration.ofDays(2).plusHours(3)),
+        converter.convert("1024KB/P2DT3H"));
+  }
+
+  @Test
+  void parseDataRateWithException() {
+    var converter = new DataRate.Field();
+
+    assertThrows(IllegalArgumentException.class, () -> converter.convert("5000 MB/ second"));
+    assertThrows(IllegalArgumentException.class, () -> converter.convert("5000 MB per second"));
+    assertThrows(IllegalArgumentException.class, () -> converter.convert("xxx 5000 MB"));
+    assertThrows(IllegalArgumentException.class, () -> converter.convert("6.00 MB"));
+
+    // unsupported week
+    assertThrows(DateTimeParseException.class, () -> converter.convert("5000 MB/Week"));
+  }
 }
diff --git a/app/src/test/java/org/astraea/app/common/UtilsTest.java b/app/src/test/java/org/astraea/app/common/UtilsTest.java
index 0eb0f484..c88abb46 100644
--- a/app/src/test/java/org/astraea/app/common/UtilsTest.java
+++ b/app/src/test/java/org/astraea/app/common/UtilsTest.java
@@ -137,4 +137,30 @@ public class UtilsTest {
                   throw new IllegalArgumentException();
                 }));
   }
+
+  @Test
+  void testMember() {
+    var dumb = new Dumb(100);
+    var value = (int) Utils.member(dumb, "value");
+    Assertions.assertEquals(dumb.value, value);
+  }
+
+  @Test
+  void testStaticMember() {
+    Assertions.assertEquals(Dumb.CONSTANT, Utils.staticMember(Dumb.class, "CONSTANT"));
+  }
+
+  private static class Dumb {
+    private static final int CONSTANT = 1000;
+
+    private final int value;
+
+    Dumb(int v) {
+      this.value = v;
+    }
+
+    public int value() {
+      return value;
+    }
+  }
 }
diff --git a/app/src/test/java/org/astraea/app/consumer/ConsumerTest.java b/app/src/test/java/org/astraea/app/consumer/ConsumerTest.java
index 9b0f9721..6b3eaa20 100644
--- a/app/src/test/java/org/astraea/app/consumer/ConsumerTest.java
+++ b/app/src/test/java/org/astraea/app/consumer/ConsumerTest.java
@@ -447,4 +447,28 @@ public class ConsumerTest extends RequireBrokerCluster {
     closed.set(true);
     fs.get();
   }
+
+  @Test
+  void testClientId() {
+    var topic = Utils.randomString(10);
+    var clientId0 = Utils.randomString();
+    try (var consumer =
+        Consumer.forTopics(Set.of(topic))
+            .bootstrapServers(bootstrapServers())
+            .fromBeginning()
+            .clientId(clientId0)
+            .build()) {
+      Assertions.assertEquals(clientId0, consumer.clientId());
+    }
+
+    var clientId1 = Utils.randomString();
+    try (var consumer =
+        Consumer.forPartitions(Set.of(TopicPartition.of(topic, 0)))
+            .bootstrapServers(bootstrapServers())
+            .fromBeginning()
+            .clientId(clientId1)
+            .build()) {
+      Assertions.assertEquals(clientId1, consumer.clientId());
+    }
+  }
 }
diff --git a/app/src/test/java/org/astraea/app/cost/ClusterInfoIntegratedTest.java b/app/src/test/java/org/astraea/app/cost/ClusterInfoIntegratedTest.java
index 4dbd3556..2107ba61 100644
--- a/app/src/test/java/org/astraea/app/cost/ClusterInfoIntegratedTest.java
+++ b/app/src/test/java/org/astraea/app/cost/ClusterInfoIntegratedTest.java
@@ -51,8 +51,7 @@ public class ClusterInfoIntegratedTest extends RequireBrokerCluster {
           .forEach(t -> Assertions.assertNotEquals(0, clusterInfo.availableReplicas(t).size()));
       clusterInfo
           .topics()
-          .forEach(
-              t -> Assertions.assertNotEquals(0, clusterInfo.availableReplicaLeaders(t).size()));
+          .forEach(t -> Assertions.assertNotEquals(0, clusterInfo.replicaLeaders(t).size()));
     }
   }
 }
diff --git a/app/src/test/java/org/astraea/app/cost/ClusterInfoTest.java b/app/src/test/java/org/astraea/app/cost/ClusterInfoTest.java
index a80dad98..b8acdac8 100644
--- a/app/src/test/java/org/astraea/app/cost/ClusterInfoTest.java
+++ b/app/src/test/java/org/astraea/app/cost/ClusterInfoTest.java
@@ -22,12 +22,118 @@ import java.util.Set;
 import org.apache.kafka.common.Cluster;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.admin.NodeInfo;
+import org.astraea.app.admin.Replica;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
 
 public class ClusterInfoTest {
 
+  @Test
+  void testDiff() {
+    /*
+    test-1-0 : change only the data folder
+    test-1-1 : change the data folder and host
+    test-1-2 : no change
+     */
+    var beforeReplicas =
+        List.of(
+            Replica.of(
+                "test-1",
+                0,
+                NodeInfo.of(0, "", -1),
+                -1,
+                -1,
+                true,
+                true,
+                false,
+                false,
+                false,
+                "/data-folder-01"),
+            Replica.of(
+                "test-1",
+                1,
+                NodeInfo.of(1, "", -1),
+                -1,
+                -1,
+                false,
+                true,
+                false,
+                false,
+                false,
+                "/data-folder-02"),
+            Replica.of(
+                "test-1",
+                2,
+                NodeInfo.of(0, "", -1),
+                -1,
+                -1,
+                false,
+                true,
+                false,
+                false,
+                false,
+                "/data-folder-01"));
+    var afterReplicas =
+        List.of(
+            Replica.of(
+                "test-1",
+                0,
+                NodeInfo.of(0, "", -1),
+                -1,
+                -1,
+                true,
+                true,
+                false,
+                false,
+                false,
+                "/data-folder-02"),
+            Replica.of(
+                "test-1",
+                1,
+                NodeInfo.of(2, "", -1),
+                -1,
+                -1,
+                false,
+                true,
+                false,
+                false,
+                false,
+                "/data-folder-03"),
+            Replica.of(
+                "test-1",
+                2,
+                NodeInfo.of(0, "", -1),
+                -1,
+                -1,
+                false,
+                true,
+                false,
+                false,
+                false,
+                "/data-folder-01"));
+    var nodeInfos = List.of(NodeInfo.of(0, "", -1), NodeInfo.of(1, "", -1), NodeInfo.of(2, "", -1));
+    var before = ClusterInfo.of(nodeInfos, beforeReplicas);
+    var after = ClusterInfo.of(nodeInfos, afterReplicas);
+    var changes = ClusterInfo.diff(before, after);
+    Assertions.assertEquals(2, changes.size());
+    Assertions.assertEquals(
+        1,
+        changes.stream()
+            .filter(replica -> replica.topic().equals("test-1") && replica.partition() == 0)
+            .count());
+    Assertions.assertEquals(
+        1,
+        changes.stream()
+            .filter(replica -> replica.topic().equals("test-1") && replica.partition() == 1)
+            .count());
+    Assertions.assertEquals(
+        0,
+        changes.stream()
+            .filter(replica -> replica.topic().equals("test-1") && replica.partition() == 2)
+            .count());
+  }
+
   @Test
   void testNode() {
     var node = NodeInfoTest.node();
@@ -48,8 +154,7 @@ public class ClusterInfoTest {
     Assertions.assertEquals(
         NodeInfo.of(node), clusterInfo.availableReplicas(partition.topic()).get(0).nodeInfo());
     Assertions.assertEquals(
-        NodeInfo.of(node),
-        clusterInfo.availableReplicaLeaders(partition.topic()).get(0).nodeInfo());
+        NodeInfo.of(node), clusterInfo.replicaLeaders(partition.topic()).get(0).nodeInfo());
     Assertions.assertEquals(
         NodeInfo.of(node), clusterInfo.replicas(partition.topic()).get(0).nodeInfo());
   }
@@ -60,6 +165,6 @@ public class ClusterInfoTest {
     Assertions.assertEquals(0, clusterInfo.replicas("unknown").size());
     Assertions.assertThrows(NoSuchElementException.class, () -> clusterInfo.node(0));
     Assertions.assertEquals(0, clusterInfo.availableReplicas("unknown").size());
-    Assertions.assertEquals(0, clusterInfo.availableReplicaLeaders("unknown").size());
+    Assertions.assertEquals(0, clusterInfo.replicaLeaders("unknown").size());
   }
 }
diff --git a/app/src/test/java/org/astraea/app/cost/CpuCostTest.java b/app/src/test/java/org/astraea/app/cost/CpuCostTest.java
index 12569c99..172b8e4a 100644
--- a/app/src/test/java/org/astraea/app/cost/CpuCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/CpuCostTest.java
@@ -43,7 +43,7 @@ public class CpuCostTest {
                 4,
                 List.of()));
     var cpuCost = new CpuCost();
-    var result = cpuCost.brokerCost(Mockito.mock(ClusterInfo.class), clusterBean);
+    var result = cpuCost.brokerCost(ClusterInfo.empty(), clusterBean);
     Assertions.assertEquals(0.5, result.value().get(1));
     Assertions.assertEquals(0.6, result.value().get(2));
     Assertions.assertEquals(0.7, result.value().get(3));
diff --git a/app/src/test/java/org/astraea/app/cost/DispersionTest.java b/app/src/test/java/org/astraea/app/cost/DispersionTest.java
index 703baad7..a271d141 100644
--- a/app/src/test/java/org/astraea/app/cost/DispersionTest.java
+++ b/app/src/test/java/org/astraea/app/cost/DispersionTest.java
@@ -27,5 +27,10 @@ class DispersionTest {
     var dispersion = Dispersion.correlationCoefficient();
     var scores = List.of(0.2, 0.4, 0.7);
     Assertions.assertEquals(0.47418569253607507, dispersion.calculate(scores));
+
+    var zeroScores = List.of(0.0, 0.0, 0.0);
+    var score = dispersion.calculate(zeroScores);
+    Assertions.assertFalse(Double.isNaN(score));
+    Assertions.assertEquals(0.0, score);
   }
 }
diff --git a/app/src/test/java/org/astraea/app/cost/MemoryCostTest.java b/app/src/test/java/org/astraea/app/cost/MemoryCostTest.java
index ae553766..c3ffb1b9 100644
--- a/app/src/test/java/org/astraea/app/cost/MemoryCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/MemoryCostTest.java
@@ -43,7 +43,7 @@ public class MemoryCostTest {
                 3,
                 List.of()));
     var memoryCost = new MemoryCost();
-    var scores = memoryCost.brokerCost(Mockito.mock(ClusterInfo.class), clusterBean);
+    var scores = memoryCost.brokerCost(ClusterInfo.empty(), clusterBean);
     Assertions.assertEquals(0.4, scores.value().get(1));
     Assertions.assertEquals(0.5, scores.value().get(2));
     Assertions.assertEquals(0, scores.value().get(3));
diff --git a/app/src/test/java/org/astraea/app/cost/NeutralIntegratedCostTest.java b/app/src/test/java/org/astraea/app/cost/NeutralIntegratedCostTest.java
index f3bb4762..238b30b4 100644
--- a/app/src/test/java/org/astraea/app/cost/NeutralIntegratedCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/NeutralIntegratedCostTest.java
@@ -29,7 +29,6 @@ public class NeutralIntegratedCostTest {
   @Test
   void testSetBrokerMetrics() {
     var neutralIntegratedCost = new NeutralIntegratedCost();
-    var clusterInfo = Mockito.mock(ClusterInfo.class);
     neutralIntegratedCost.brokersMetric =
         new HashMap<>(
             Map.of(
@@ -50,14 +49,18 @@ public class NeutralIntegratedCostTest {
     var outputBrokerCost = getBrokerCost(new HashMap<>(Map.of(0, 500.0, 1, 1000.0, 2, 1500.0)));
     var cpuBrokerCost = getBrokerCost(new HashMap<>(Map.of(0, 0.05, 1, 0.1, 2, 0.15)));
     var memoryBrokerCost = getBrokerCost(new HashMap<>(Map.of(0, 0.1, 1, 0.3, 2, 0.5)));
-    Mockito.when(input.brokerCost(clusterInfo, ClusterBean.EMPTY)).thenReturn(inputBrokerCost);
-    Mockito.when(output.brokerCost(clusterInfo, ClusterBean.EMPTY)).thenReturn(outputBrokerCost);
-    Mockito.when(memory.brokerCost(clusterInfo, ClusterBean.EMPTY)).thenReturn(memoryBrokerCost);
-    Mockito.when(cpu.brokerCost(clusterInfo, ClusterBean.EMPTY)).thenReturn(cpuBrokerCost);
+    Mockito.when(input.brokerCost(ClusterInfo.empty(), ClusterBean.EMPTY))
+        .thenReturn(inputBrokerCost);
+    Mockito.when(output.brokerCost(ClusterInfo.empty(), ClusterBean.EMPTY))
+        .thenReturn(outputBrokerCost);
+    Mockito.when(memory.brokerCost(ClusterInfo.empty(), ClusterBean.EMPTY))
+        .thenReturn(memoryBrokerCost);
+    Mockito.when(cpu.brokerCost(ClusterInfo.empty(), ClusterBean.EMPTY)).thenReturn(cpuBrokerCost);
     List<HasBrokerCost> metricsCost = List.of(input, output, cpu, memory);
     metricsCost.forEach(
         hasBrokerCost ->
-            neutralIntegratedCost.setBrokerMetrics(hasBrokerCost, clusterInfo, ClusterBean.EMPTY));
+            neutralIntegratedCost.setBrokerMetrics(
+                hasBrokerCost, ClusterInfo.empty(), ClusterBean.EMPTY));
 
     Assertions.assertEquals(neutralIntegratedCost.brokersMetric.get(0).inputScore, 50.0);
     Assertions.assertEquals(neutralIntegratedCost.brokersMetric.get(1).inputScore, 100.0);
diff --git a/app/src/test/java/org/astraea/app/cost/NodeLatencyCostTest.java b/app/src/test/java/org/astraea/app/cost/NodeLatencyCostTest.java
index 5473f47d..e4b0ac0e 100644
--- a/app/src/test/java/org/astraea/app/cost/NodeLatencyCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/NodeLatencyCostTest.java
@@ -44,7 +44,7 @@ public class NodeLatencyCostTest extends RequireBrokerCluster {
     Mockito.when(bean.requestLatencyAvg()).thenReturn(Double.NaN);
     var clusterBean = ClusterBean.of(Map.of(-1, List.of(bean)));
     var function = new NodeLatencyCost();
-    var result = function.brokerCost(Mockito.mock(ClusterInfo.class), clusterBean);
+    var result = function.brokerCost(ClusterInfo.empty(), clusterBean);
     Assertions.assertEquals(0, result.value().size());
   }
 
@@ -55,7 +55,7 @@ public class NodeLatencyCostTest extends RequireBrokerCluster {
     Mockito.when(bean.requestLatencyAvg()).thenReturn(10D);
     var clusterBean = ClusterBean.of(Map.of(-1, List.of(bean)));
     var function = new NodeLatencyCost();
-    var result = function.brokerCost(Mockito.mock(ClusterInfo.class), clusterBean);
+    var result = function.brokerCost(ClusterInfo.empty(), clusterBean);
     Assertions.assertEquals(1, result.value().size());
     Assertions.assertEquals(10D, result.value().get(1));
   }
@@ -77,7 +77,7 @@ public class NodeLatencyCostTest extends RequireBrokerCluster {
           () ->
               function
                       .brokerCost(
-                          Mockito.mock(ClusterInfo.class),
+                          ClusterInfo.empty(),
                           ClusterBean.of(
                               Map.of(
                                   -1,
diff --git a/app/src/test/java/org/astraea/app/cost/NodeThroughputCostTest.java b/app/src/test/java/org/astraea/app/cost/NodeThroughputCostTest.java
index 22bdc46f..4c2b4772 100644
--- a/app/src/test/java/org/astraea/app/cost/NodeThroughputCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/NodeThroughputCostTest.java
@@ -37,7 +37,7 @@ public class NodeThroughputCostTest {
     Mockito.when(bean.outgoingByteRate()).thenReturn(Double.NaN);
     var clusterBean = ClusterBean.of(Map.of(-1, List.of(bean)));
     var function = new NodeThroughputCost();
-    var result = function.brokerCost(Mockito.mock(ClusterInfo.class), clusterBean);
+    var result = function.brokerCost(ClusterInfo.empty(), clusterBean);
     Assertions.assertEquals(0, result.value().size());
   }
 
@@ -49,7 +49,7 @@ public class NodeThroughputCostTest {
     Mockito.when(bean.outgoingByteRate()).thenReturn(10D);
     var clusterBean = ClusterBean.of(Map.of(-1, List.of(bean)));
     var function = new NodeThroughputCost();
-    var result = function.brokerCost(Mockito.mock(ClusterInfo.class), clusterBean);
+    var result = function.brokerCost(ClusterInfo.empty(), clusterBean);
     Assertions.assertEquals(1, result.value().size());
     Assertions.assertEquals(20D, result.value().get(1));
   }
@@ -66,8 +66,7 @@ public class NodeThroughputCostTest {
     Mockito.when(bean1.outgoingByteRate()).thenReturn(3D);
     Mockito.when(bean1.brokerId()).thenReturn(11);
     var clusterBean = ClusterBean.of(Map.of(0, List.of(bean0), 1, List.of(bean1)));
-    var clusterInfo = Mockito.mock(ClusterInfo.class);
-    var cost = throughputCost.brokerCost(clusterInfo, clusterBean);
+    var cost = throughputCost.brokerCost(ClusterInfo.empty(), clusterBean);
     Assertions.assertEquals(30D, cost.value().get(10));
     Assertions.assertEquals(5D, cost.value().get(11));
   }
diff --git a/app/src/test/java/org/astraea/app/cost/ReplicaDiskInCostTest.java b/app/src/test/java/org/astraea/app/cost/ReplicaDiskInCostTest.java
index 1365d508..e103e32c 100644
--- a/app/src/test/java/org/astraea/app/cost/ReplicaDiskInCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/ReplicaDiskInCostTest.java
@@ -19,7 +19,6 @@ package org.astraea.app.cost;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.admin.NodeInfo;
@@ -32,7 +31,6 @@ import org.astraea.app.partitioner.Configuration;
 import org.astraea.app.service.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
-import org.mockito.Mockito;
 
 class ReplicaDiskInCostTest extends RequireBrokerCluster {
   private static final HasGauge OLD_TP1_0 =
@@ -80,90 +78,81 @@ class ReplicaDiskInCostTest extends RequireBrokerCluster {
   }
 
   private ClusterInfo<Replica> clusterInfo() {
-    var clusterInfo = (ClusterInfo<Replica>) Mockito.mock(ClusterInfo.class);
-    Mockito.when(clusterInfo.nodes())
-        .thenReturn(
-            List.of(NodeInfo.of(1, "", -1), NodeInfo.of(2, "", -1), NodeInfo.of(3, "", -1)));
-    Mockito.when(clusterInfo.topics()).thenReturn(Set.of("test-1", "test-2"));
-    Mockito.when(clusterInfo.replicas(Mockito.anyString()))
-        .thenAnswer(
-            topic ->
-                topic.getArgument(0).equals("test-1")
-                    ? List.of(
-                        Replica.of(
-                            "test-1",
-                            0,
-                            NodeInfo.of(1, "", -1),
-                            0,
-                            100,
-                            true,
-                            true,
-                            false,
-                            false,
-                            true,
-                            "/tmp/aa"),
-                        Replica.of(
-                            "test-1",
-                            0,
-                            NodeInfo.of(2, "", -1),
-                            0,
-                            100,
-                            false,
-                            true,
-                            false,
-                            false,
-                            false,
-                            "/tmp/aa"),
-                        Replica.of(
-                            "test-1",
-                            1,
-                            NodeInfo.of(2, "", -1),
-                            0,
-                            100,
-                            false,
-                            true,
-                            false,
-                            false,
-                            false,
-                            "/tmp/aa"),
-                        Replica.of(
-                            "test-1",
-                            1,
-                            NodeInfo.of(3, "", -1),
-                            0,
-                            100,
-                            true,
-                            true,
-                            false,
-                            false,
-                            true,
-                            "/tmp/aa"))
-                    : List.of(
-                        Replica.of(
-                            "test-2",
-                            0,
-                            NodeInfo.of(1, "", -1),
-                            0,
-                            100,
-                            false,
-                            true,
-                            false,
-                            false,
-                            false,
-                            "/tmp/aa"),
-                        Replica.of(
-                            "test-2",
-                            0,
-                            NodeInfo.of(3, "", -1),
-                            0,
-                            100,
-                            true,
-                            true,
-                            false,
-                            false,
-                            true,
-                            "/tmp/aa")));
-    return clusterInfo;
+    return ClusterInfo.of(
+        List.of(NodeInfo.of(1, "", -1), NodeInfo.of(2, "", -1), NodeInfo.of(3, "", -1)),
+        List.of(
+            Replica.of(
+                "test-1",
+                0,
+                NodeInfo.of(1, "", -1),
+                0,
+                100,
+                true,
+                true,
+                false,
+                false,
+                true,
+                "/tmp/aa"),
+            Replica.of(
+                "test-1",
+                0,
+                NodeInfo.of(2, "", -1),
+                0,
+                100,
+                false,
+                true,
+                false,
+                false,
+                false,
+                "/tmp/aa"),
+            Replica.of(
+                "test-1",
+                1,
+                NodeInfo.of(2, "", -1),
+                0,
+                100,
+                false,
+                true,
+                false,
+                false,
+                false,
+                "/tmp/aa"),
+            Replica.of(
+                "test-1",
+                1,
+                NodeInfo.of(3, "", -1),
+                0,
+                100,
+                true,
+                true,
+                false,
+                false,
+                true,
+                "/tmp/aa"),
+            Replica.of(
+                "test-2",
+                0,
+                NodeInfo.of(1, "", -1),
+                0,
+                100,
+                false,
+                true,
+                false,
+                false,
+                false,
+                "/tmp/aa"),
+            Replica.of(
+                "test-2",
+                0,
+                NodeInfo.of(3, "", -1),
+                0,
+                100,
+                true,
+                true,
+                false,
+                false,
+                true,
+                "/tmp/aa")));
   }
 
   private static ClusterBean clusterBean() {
diff --git a/app/src/test/java/org/astraea/app/cost/ReplicaLeaderCostTest.java b/app/src/test/java/org/astraea/app/cost/ReplicaLeaderCostTest.java
index 9711f0b6..63b6fcb2 100644
--- a/app/src/test/java/org/astraea/app/cost/ReplicaLeaderCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/ReplicaLeaderCostTest.java
@@ -18,7 +18,6 @@ package org.astraea.app.cost;
 
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.admin.NodeInfo;
@@ -36,13 +35,10 @@ public class ReplicaLeaderCostTest {
   void testNoMetrics() {
     var replicas =
         List.of(
-            ReplicaInfo.of("topic", 0, NodeInfo.of(10, "broker0", 1111), true, true, true),
-            ReplicaInfo.of("topic", 0, NodeInfo.of(10, "broker0", 1111), true, true, true),
-            ReplicaInfo.of("topic", 0, NodeInfo.of(11, "broker1", 1111), true, true, true));
-    @SuppressWarnings("unchecked")
-    ClusterInfo<ReplicaInfo> clusterInfo = Mockito.mock(ClusterInfo.class);
-    Mockito.when(clusterInfo.topics()).thenReturn(Set.of("topic"));
-    Mockito.when(clusterInfo.availableReplicaLeaders(Mockito.anyString())).thenReturn(replicas);
+            ReplicaInfo.of("topic", 0, NodeInfo.of(10, "broker0", 1111), true, true, false),
+            ReplicaInfo.of("topic", 0, NodeInfo.of(10, "broker0", 1111), true, true, false),
+            ReplicaInfo.of("topic", 0, NodeInfo.of(11, "broker1", 1111), true, true, false));
+    var clusterInfo = ClusterInfo.of(replicas);
     var cost = ReplicaLeaderCost.leaderCount(clusterInfo);
     Assertions.assertTrue(cost.containsKey(10));
     Assertions.assertTrue(cost.containsKey(11));
diff --git a/app/src/test/java/org/astraea/app/cost/ReplicaSizeCostTest.java b/app/src/test/java/org/astraea/app/cost/ReplicaSizeCostTest.java
new file mode 100644
index 00000000..73f86b96
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/cost/ReplicaSizeCostTest.java
@@ -0,0 +1,241 @@
+/*
+ * 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 static org.mockito.Mockito.mock;
+
+import java.util.List;
+import java.util.Map;
+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.BeanObject;
+import org.astraea.app.metrics.broker.LogMetrics;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class ReplicaSizeCostTest {
+  private final BeanObject bean =
+      new BeanObject(
+          "domain", Map.of("topic", "t", "partition", "10", "name", "SIZE"), Map.of("Value", 777));
+
+  @Test
+  void testBrokerCost() {
+    var meter = new LogMetrics.Log.Gauge(bean);
+    var cost = new ReplicaSizeCost();
+    var result =
+        cost.brokerCost(mock(ClusterInfo.class), ClusterBean.of(Map.of(1, List.of(meter))));
+    Assertions.assertEquals(1, result.value().size());
+    Assertions.assertEquals(777, result.value().entrySet().iterator().next().getValue());
+  }
+
+  @Test
+  void testMoveCost() {
+    var cost = new ReplicaSizeCost();
+    var moveCost = cost.moveCost(originClusterInfo(), newClusterInfo(), ClusterBean.EMPTY);
+    var totalSize = moveCost.totalCost();
+    var changes = moveCost.changes();
+
+    // totalSize will also be calculated for the folder migrate in the same broker.
+    Assertions.assertEquals(6000000 + 700000 + 800000, totalSize);
+    Assertions.assertEquals(700000, changes.get(0));
+    Assertions.assertEquals(-6700000, changes.get(1));
+    Assertions.assertEquals(6000000, changes.get(2));
+  }
+
+  /*
+  origin replica distributed :
+    test-1-0 : 0,1
+    test-1-1 : 1,2
+    test-2-0 : 1,2
+
+  generated plan replica distributed :
+    test-1-0 : 0,2
+    test-1-1 : 0,2
+    test-2-0 : 1,2
+
+   */
+
+  static ClusterInfo<Replica> getClusterInfo(List<Replica> replicas) {
+    return ClusterInfo.of(
+        List.of(NodeInfo.of(1, "", -1), NodeInfo.of(2, "", -1), NodeInfo.of(3, "", -1)),
+        List.of(
+            replicas.get(0),
+            replicas.get(1),
+            replicas.get(2),
+            replicas.get(3),
+            replicas.get(4),
+            replicas.get(5)));
+  }
+
+  static ClusterInfo<Replica> originClusterInfo() {
+    var replicas =
+        List.of(
+            Replica.of(
+                "test-1",
+                0,
+                NodeInfo.of(0, "", -1),
+                -1,
+                6000000,
+                true,
+                true,
+                false,
+                false,
+                false,
+                ""),
+            Replica.of(
+                "test-1",
+                0,
+                NodeInfo.of(1, "", -1),
+                -1,
+                6000000,
+                false,
+                true,
+                false,
+                false,
+                false,
+                ""),
+            Replica.of(
+                "test-1",
+                1,
+                NodeInfo.of(1, "", -1),
+                -1,
+                700000,
+                true,
+                true,
+                false,
+                false,
+                false,
+                ""),
+            Replica.of(
+                "test-1",
+                1,
+                NodeInfo.of(2, "", -1),
+                -1,
+                700000,
+                false,
+                true,
+                false,
+                false,
+                false,
+                ""),
+            Replica.of(
+                "test-2",
+                0,
+                NodeInfo.of(1, "", -1),
+                -1,
+                800000,
+                true,
+                true,
+                false,
+                false,
+                false,
+                "/log-path-01"),
+            Replica.of(
+                "test-2",
+                0,
+                NodeInfo.of(2, "", -1),
+                -1,
+                800000,
+                false,
+                true,
+                false,
+                false,
+                false,
+                "/log-path-02"));
+    return getClusterInfo(replicas);
+  }
+
+  static ClusterInfo<Replica> newClusterInfo() {
+    var replicas =
+        List.of(
+            Replica.of(
+                "test-1",
+                0,
+                NodeInfo.of(0, "", -1),
+                -1,
+                6000000,
+                true,
+                true,
+                false,
+                false,
+                false,
+                ""),
+            Replica.of(
+                "test-1",
+                0,
+                NodeInfo.of(2, "", -1),
+                -1,
+                6000000,
+                false,
+                true,
+                false,
+                false,
+                false,
+                ""),
+            Replica.of(
+                "test-1",
+                1,
+                NodeInfo.of(0, "", -1),
+                -1,
+                700000,
+                true,
+                true,
+                false,
+                false,
+                false,
+                ""),
+            Replica.of(
+                "test-1",
+                1,
+                NodeInfo.of(2, "", -1),
+                -1,
+                700000,
+                false,
+                true,
+                false,
+                false,
+                false,
+                ""),
+            Replica.of(
+                "test-2",
+                0,
+                NodeInfo.of(1, "", -1),
+                -1,
+                800000,
+                true,
+                true,
+                false,
+                false,
+                false,
+                "/log-path-01"),
+            Replica.of(
+                "test-2",
+                0,
+                NodeInfo.of(2, "", -1),
+                -1,
+                800000,
+                false,
+                true,
+                false,
+                false,
+                false,
+                "/log-path-03"));
+    return getClusterInfo(replicas);
+  }
+}
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 0ffe87e1..f3b6be63 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
@@ -20,7 +20,6 @@ import java.time.Duration;
 import java.util.Set;
 import java.util.stream.Collectors;
 import org.astraea.app.admin.Admin;
-import org.astraea.app.admin.TopicPartition;
 import org.astraea.app.common.Utils;
 import org.astraea.app.consumer.Consumer;
 import org.astraea.app.metrics.MBeanClient;
@@ -30,23 +29,6 @@ import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 public class ConsumerMetricsTest extends RequireBrokerCluster {
-  @Test
-  void testSingleBroker() {
-    var topic = Utils.randomString(10);
-    try (var admin = Admin.of(bootstrapServers());
-        var consumer =
-            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).nodeInfo().id();
-      consumer.poll(Duration.ofSeconds(5));
-      var metrics = ConsumerMetrics.node(MBeanClient.local(), owner);
-      Assertions.assertEquals(1, metrics.size());
-      check(metrics.values().stream().findAny().get());
-    }
-  }
-
   @Test
   void testMultiBrokers() {
     var topic = Utils.randomString(10);
diff --git a/app/src/test/java/org/astraea/app/metrics/client/consumer/HasConsumerCoordinatorMetricsTest.java b/app/src/test/java/org/astraea/app/metrics/client/consumer/HasConsumerCoordinatorMetricsTest.java
new file mode 100644
index 00000000..4637be34
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/metrics/client/consumer/HasConsumerCoordinatorMetricsTest.java
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.metrics.client.consumer;
+
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.Set;
+import java.util.stream.IntStream;
+import org.astraea.app.common.Utils;
+import org.astraea.app.consumer.Consumer;
+import org.astraea.app.metrics.MBeanClient;
+import org.astraea.app.producer.Producer;
+import org.astraea.app.service.RequireSingleBrokerCluster;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class HasConsumerCoordinatorMetricsTest extends RequireSingleBrokerCluster {
+
+  @Test
+  void testClientId() {
+    var topic = Utils.randomString(10);
+
+    try (var producer = Producer.of(bootstrapServers())) {
+      IntStream.range(0, 10)
+          .forEach(
+              index ->
+                  producer
+                      .sender()
+                      .topic(topic)
+                      .key(String.valueOf(index).getBytes(StandardCharsets.UTF_8))
+                      .run());
+    }
+    try (var consumer =
+        Consumer.forTopics(Set.of(topic))
+            .bootstrapServers(bootstrapServers())
+            .fromBeginning()
+            .build()) {
+      Assertions.assertEquals(10, consumer.poll(10, Duration.ofSeconds(5)).size());
+      consumer.commitOffsets(Duration.ofSeconds(2));
+      var metrics = ConsumerMetrics.coordinators(MBeanClient.local());
+      Assertions.assertEquals(1, metrics.size());
+      var m = metrics.iterator().next();
+      Assertions.assertNotNull(m.clientId());
+      Assertions.assertEquals(1D, m.assignedPartitions());
+      Assertions.assertDoesNotThrow(m::commitLatencyAvg);
+      Assertions.assertDoesNotThrow(m::commitLatencyMax);
+      Assertions.assertDoesNotThrow(m::commitRate);
+      Assertions.assertDoesNotThrow(m::commitRate);
+      Assertions.assertDoesNotThrow(m::commitRate);
+      Assertions.assertDoesNotThrow(m::commitTotal);
+      Assertions.assertDoesNotThrow(m::failedRebalanceRatePerHour);
+      Assertions.assertDoesNotThrow(m::failedRebalanceTotal);
+      Assertions.assertDoesNotThrow(m::heartbeatRate);
+      Assertions.assertDoesNotThrow(m::joinRate);
+      Assertions.assertDoesNotThrow(m::joinTimeAvg);
+      Assertions.assertDoesNotThrow(m::joinTimeMax);
+      Assertions.assertDoesNotThrow(m::joinTotal);
+      Assertions.assertDoesNotThrow(m::lastHeartbeatSecondsAgo);
+      Assertions.assertDoesNotThrow(m::lastRebalanceSecondsAgo);
+      Assertions.assertDoesNotThrow(m::partitionAssignedLatencyAvg);
+      Assertions.assertDoesNotThrow(m::partitionAssignedLatencyMax);
+      // no partitions are revoked so no metrics
+      Assertions.assertTrue(Double.isNaN(m.partitionLostLatencyAvg()));
+      Assertions.assertTrue(Double.isNaN(m.partitionLostLatencyMax()));
+      Assertions.assertTrue(Double.isNaN(m.partitionRevokedLatencyAvg()));
+      Assertions.assertTrue(Double.isNaN(m.partitionRevokedLatencyMax()));
+      Assertions.assertDoesNotThrow(m::rebalanceLatencyAvg);
+      Assertions.assertDoesNotThrow(m::rebalanceLatencyMax);
+      Assertions.assertDoesNotThrow(m::rebalanceLatencyTotal);
+      Assertions.assertDoesNotThrow(m::rebalanceRatePerHour);
+      Assertions.assertDoesNotThrow(m::rebalanceTotal);
+      Assertions.assertDoesNotThrow(m::syncRate);
+      Assertions.assertDoesNotThrow(m::syncTimeAvg);
+      Assertions.assertDoesNotThrow(m::syncTimeMax);
+      Assertions.assertDoesNotThrow(m::syncTotal);
+    }
+  }
+}
diff --git a/app/src/test/java/org/astraea/app/metrics/client/producer/HasProducerTopicMetricsTest.java b/app/src/test/java/org/astraea/app/metrics/client/producer/HasProducerTopicMetricsTest.java
index ac14a2ed..766d5701 100644
--- a/app/src/test/java/org/astraea/app/metrics/client/producer/HasProducerTopicMetricsTest.java
+++ b/app/src/test/java/org/astraea/app/metrics/client/producer/HasProducerTopicMetricsTest.java
@@ -27,13 +27,14 @@ import org.junit.jupiter.api.Test;
 public class HasProducerTopicMetricsTest extends RequireSingleBrokerCluster {
 
   @Test
-  void testAttributes() throws ExecutionException, InterruptedException {
+  void testClientId() throws ExecutionException, InterruptedException {
     var topic = Utils.randomString(10);
     try (var producer = Producer.of(bootstrapServers())) {
       producer.sender().topic(topic).run().toCompletableFuture().get();
-      var metrics = ProducerMetrics.topic(MBeanClient.local(), topic);
-      Assertions.assertEquals(1, metrics.size());
-      var producerTopicMetrics = metrics.get("producer-1");
+      var metrics = ProducerMetrics.topics(MBeanClient.local());
+      Assertions.assertNotEquals(0, metrics.stream().filter(m -> m.topic().equals(topic)).count());
+      var producerTopicMetrics =
+          metrics.stream().filter(m -> m.clientId().equals("producer-1")).findFirst().get();
       Assertions.assertNotEquals(0D, producerTopicMetrics.byteRate());
       Assertions.assertNotEquals(0D, producerTopicMetrics.byteTotal());
       Assertions.assertEquals(1D, producerTopicMetrics.compressionRate());
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 504a7bdf..62187f97 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
@@ -17,11 +17,9 @@
 package org.astraea.app.metrics.client.producer;
 
 import java.time.Duration;
-import java.util.Set;
 import java.util.concurrent.ExecutionException;
 import java.util.stream.Collectors;
 import org.astraea.app.admin.Admin;
-import org.astraea.app.admin.TopicPartition;
 import org.astraea.app.common.Utils;
 import org.astraea.app.metrics.MBeanClient;
 import org.astraea.app.metrics.client.HasNodeMetrics;
@@ -32,22 +30,6 @@ import org.junit.jupiter.api.Test;
 
 public class ProducerMetricsTest extends RequireBrokerCluster {
 
-  @Test
-  void testSingleBroker() throws ExecutionException, InterruptedException {
-    var topic = Utils.randomString(10);
-    try (var admin = Admin.of(bootstrapServers());
-        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).nodeInfo().id();
-      producer.sender().topic(topic).run().toCompletableFuture().get();
-      var metrics = ProducerMetrics.node(MBeanClient.local(), owner);
-      Assertions.assertEquals(1, metrics.size());
-      check(metrics.get("producer-1"));
-    }
-  }
-
   @Test
   void testMultiBrokers() throws ExecutionException, InterruptedException {
     var topic = Utils.randomString(10);
diff --git a/app/src/test/java/org/astraea/app/partitioner/StrictCostDispatcherTest.java b/app/src/test/java/org/astraea/app/partitioner/StrictCostDispatcherTest.java
index b98392f2..7705ea3d 100644
--- a/app/src/test/java/org/astraea/app/partitioner/StrictCostDispatcherTest.java
+++ b/app/src/test/java/org/astraea/app/partitioner/StrictCostDispatcherTest.java
@@ -24,7 +24,6 @@ import java.util.Optional;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
-import java.util.stream.Stream;
 import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.admin.NodeInfo;
@@ -104,26 +103,23 @@ public class StrictCostDispatcherTest {
 
   @Test
   void testNoAvailableBrokers() {
-    var clusterInfo = Mockito.mock(ClusterInfo.class);
-    Mockito.when(clusterInfo.availableReplicaLeaders(Mockito.anyString())).thenReturn(List.of());
     try (var dispatcher = new StrictCostDispatcher()) {
       dispatcher.configure(Map.of(), Optional.empty(), Map.of(), Duration.ofSeconds(10));
       Assertions.assertEquals(
-          0, dispatcher.partition("topic", new byte[0], new byte[0], clusterInfo));
+          0, dispatcher.partition("topic", new byte[0], new byte[0], ClusterInfo.empty()));
     }
   }
 
   @Test
   void testSingleBroker() {
     var nodeInfo = NodeInfo.of(10, "host", 11111);
-    var replicaInfo = ReplicaInfo.of("topic", 10, nodeInfo, true, true, true);
-    var clusterInfo = Mockito.mock(ClusterInfo.class);
-    Mockito.when(clusterInfo.availableReplicaLeaders(Mockito.anyString()))
-        .thenReturn(List.of(replicaInfo));
+    var replicaInfo = ReplicaInfo.of("topic", 10, nodeInfo, true, true, false);
     try (var dispatcher = new StrictCostDispatcher()) {
       dispatcher.configure(Map.of(), Optional.empty(), Map.of(), Duration.ofSeconds(10));
       Assertions.assertEquals(
-          10, dispatcher.partition("topic", new byte[0], new byte[0], clusterInfo));
+          10,
+          dispatcher.partition(
+              "topic", new byte[0], new byte[0], ClusterInfo.of(List.of(replicaInfo))));
     }
   }
 
@@ -171,13 +167,11 @@ public class StrictCostDispatcherTest {
     var replicaInfo0 = ReplicaInfo.of("topic", 0, NodeInfo.of(10, "host", 11111), true, true, true);
     var replicaInfo1 =
         ReplicaInfo.of("topic", 1, NodeInfo.of(12, "host2", 11111), true, true, true);
-    var clusterInfo = Mockito.mock(ClusterInfo.class);
-    Mockito.when(clusterInfo.availableReplicaLeaders(Mockito.anyString()))
-        .thenReturn(List.of(replicaInfo0, replicaInfo1));
     try (var dispatcher = new StrictCostDispatcher()) {
       dispatcher.configure(
           Map.of(costFunction, 1D), Optional.empty(), Map.of(), Duration.ofSeconds(10));
-      dispatcher.partition("topic", new byte[0], new byte[0], clusterInfo);
+      dispatcher.partition(
+          "topic", new byte[0], new byte[0], ClusterInfo.of(List.of(replicaInfo0, replicaInfo1)));
       Assertions.assertEquals(0, dispatcher.receivers.size());
     }
   }
@@ -210,16 +204,13 @@ public class StrictCostDispatcherTest {
         };
     dispatcher.configure(
         Map.of(costFunction, 1D), Optional.of(jmxPort), Map.of(), Duration.ofSeconds(10));
-    var replicaInfo0 = ReplicaInfo.of("topic", 0, NodeInfo.of(10, "host", 11111), true, true, true);
-    var replicaInfo1 = ReplicaInfo.of("topic", 1, NodeInfo.of(10, "host", 11111), true, true, true);
+    var replicaInfo0 =
+        ReplicaInfo.of("topic", 0, NodeInfo.of(10, "host", 11111), true, true, false);
+    var replicaInfo1 =
+        ReplicaInfo.of("topic", 1, NodeInfo.of(10, "host", 11111), true, true, false);
     var replicaInfo2 =
         ReplicaInfo.of("topic", 1, NodeInfo.of(11, "host2", 11111), true, true, true);
-    var rs = List.of(replicaInfo0, replicaInfo1, replicaInfo2);
-    var clusterInfo = Mockito.mock(ClusterInfo.class);
-    Mockito.when(clusterInfo.availableReplicaLeaders(Mockito.anyString())).thenReturn(rs);
-    Mockito.when(clusterInfo.nodes())
-        .thenReturn(
-            rs.stream().map(ReplicaInfo::nodeInfo).collect(Collectors.toUnmodifiableList()));
+    var clusterInfo = ClusterInfo.of(List.of(replicaInfo0, replicaInfo1, replicaInfo2));
     // there is one local receiver by default
     Assertions.assertEquals(1, dispatcher.receivers.size());
     Assertions.assertEquals(-1, dispatcher.receivers.keySet().iterator().next());
@@ -274,21 +265,16 @@ public class StrictCostDispatcherTest {
 
       var replicaInfo0 =
           ReplicaInfo.of(
-              "topic", partitionId, NodeInfo.of(brokerId, "host", 11111), true, true, true);
+              "topic", partitionId, NodeInfo.of(brokerId, "host", 11111), true, true, false);
       var replicaInfo1 =
-          ReplicaInfo.of("topic", 1, NodeInfo.of(1111, "host2", 11111), true, true, true);
-      var clusterInfo = Mockito.mock(ClusterInfo.class);
-      Mockito.when(clusterInfo.availableReplicaLeaders(Mockito.anyString()))
-          .thenReturn(List.of(replicaInfo0, replicaInfo1));
-      Mockito.when(clusterInfo.nodes())
-          .thenReturn(
-              Stream.of(replicaInfo0, replicaInfo1)
-                  .map(ReplicaInfo::nodeInfo)
-                  .collect(Collectors.toUnmodifiableList()));
-      Mockito.when(clusterInfo.availableReplicaLeaders(Mockito.anyInt(), Mockito.anyString()))
-          .thenReturn(List.of(replicaInfo0));
+          ReplicaInfo.of("topic", 1, NodeInfo.of(1111, "host2", 11111), true, true, false);
       Assertions.assertEquals(
-          partitionId, dispatcher.partition("topic", new byte[0], new byte[0], clusterInfo));
+          partitionId,
+          dispatcher.partition(
+              "topic",
+              new byte[0],
+              new byte[0],
+              ClusterInfo.of(List.of(replicaInfo0, replicaInfo1))));
     }
   }
 
@@ -316,18 +302,17 @@ public class StrictCostDispatcherTest {
           Configuration.of(Map.of(StrictCostDispatcher.ROUND_ROBIN_LEASE_KEY, "2s")));
       Assertions.assertEquals(Duration.ofSeconds(2), dispatcher.roundRobinLease);
 
-      var clusterInfo = Mockito.mock(ClusterInfo.class);
-      dispatcher.tryToUpdateRoundRobin(clusterInfo);
+      dispatcher.tryToUpdateRoundRobin(ClusterInfo.empty());
       var t = dispatcher.timeToUpdateRoundRobin;
       var rr =
           Arrays.stream(dispatcher.roundRobin).boxed().collect(Collectors.toUnmodifiableList());
       Assertions.assertEquals(StrictCostDispatcher.ROUND_ROBIN_LENGTH, rr.size());
       // the rr is not updated yet
-      dispatcher.tryToUpdateRoundRobin(clusterInfo);
+      dispatcher.tryToUpdateRoundRobin(ClusterInfo.empty());
       IntStream.range(0, rr.size())
           .forEach(i -> Assertions.assertEquals(rr.get(i), dispatcher.roundRobin[i]));
       Utils.sleep(Duration.ofSeconds(3));
-      dispatcher.tryToUpdateRoundRobin(clusterInfo);
+      dispatcher.tryToUpdateRoundRobin(ClusterInfo.empty());
       // rr is updated already
       Assertions.assertNotEquals(t, dispatcher.timeToUpdateRoundRobin);
     }
@@ -345,8 +330,8 @@ public class StrictCostDispatcherTest {
           }
         }) {
       var nodeInfo = NodeInfo.of(10, "host", 2222);
-      var clusterInfo = Mockito.mock(ClusterInfo.class);
-      Mockito.when(clusterInfo.nodes()).thenReturn(List.of(nodeInfo));
+      var clusterInfo =
+          ClusterInfo.of(List.of(ReplicaInfo.of("topic", 0, nodeInfo, true, true, false)));
 
       Assertions.assertEquals(0, dispatcher.receivers.size());
       dispatcher.fetcher = Optional.of(Mockito.mock(Fetcher.class));
diff --git a/app/src/test/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobinDispatchTest.java b/app/src/test/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobinDispatchTest.java
index 85956798..e648427e 100644
--- a/app/src/test/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobinDispatchTest.java
+++ b/app/src/test/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobinDispatchTest.java
@@ -270,23 +270,18 @@ public class SmoothWeightRoundRobinDispatchTest extends RequireBrokerCluster {
   public void testGetAndChoose() {
     var topic = "test";
     var smoothWeight = new SmoothWeightRoundRobin(Map.of(1, 5.0, 2, 3.0, 3, 1.0));
-    var re1 = Mockito.mock(ReplicaInfo.class);
     var node1 = Mockito.mock(NodeInfo.class);
-    Mockito.when(re1.nodeInfo()).thenReturn(node1);
     Mockito.when(node1.id()).thenReturn(1);
+    var re1 = ReplicaInfo.of(topic, 0, node1, true, true, false);
 
-    var re2 = Mockito.mock(ReplicaInfo.class);
     var node2 = Mockito.mock(NodeInfo.class);
-    Mockito.when(re2.nodeInfo()).thenReturn(node2);
     Mockito.when(node2.id()).thenReturn(2);
+    var re2 = ReplicaInfo.of(topic, 1, node2, true, true, false);
 
-    var re3 = Mockito.mock(ReplicaInfo.class);
     var node3 = Mockito.mock(NodeInfo.class);
-    Mockito.when(re3.nodeInfo()).thenReturn(node3);
     Mockito.when(node3.id()).thenReturn(3);
-    var testCluster = Mockito.mock(ClusterInfo.class);
-    Mockito.when(testCluster.availableReplicaLeaders(Mockito.anyString()))
-        .thenReturn(List.of(re1, re2, re3));
+    var re3 = ReplicaInfo.of(topic, 2, node3, true, true, false);
+    var testCluster = ClusterInfo.of(List.of(re1, re2, re3));
     Assertions.assertEquals(1, smoothWeight.getAndChoose(topic, testCluster));
     Assertions.assertEquals(2, smoothWeight.getAndChoose(topic, testCluster));
     Assertions.assertEquals(3, smoothWeight.getAndChoose(topic, testCluster));
diff --git a/app/src/test/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobinTest.java b/app/src/test/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobinTest.java
index 6930bf2f..a5d67c8d 100644
--- a/app/src/test/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobinTest.java
+++ b/app/src/test/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobinTest.java
@@ -23,14 +23,13 @@ import org.astraea.app.admin.NodeInfo;
 import org.astraea.app.admin.ReplicaInfo;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
-import org.mockito.Mockito;
 
 public class SmoothWeightRoundRobinTest {
   @Test
   void testGetAndChoose() {
     var topic = "test";
     var smoothWeight = new SmoothWeightRoundRobin(Map.of(1, 5.0, 2, 3.0, 3, 1.0));
-    var testCluster = mockResult();
+    var testCluster = clusterInfo();
 
     Assertions.assertEquals(1, smoothWeight.getAndChoose(topic, testCluster));
     Assertions.assertEquals(2, smoothWeight.getAndChoose(topic, testCluster));
@@ -45,7 +44,7 @@ public class SmoothWeightRoundRobinTest {
   void testPartOfBrokerGetAndChoose() {
     var topic = "test";
     var smoothWeight = new SmoothWeightRoundRobin(Map.of(1, 5.0, 2, 3.0, 3, 1.0, 4, 1.0));
-    var testCluster = mockResult();
+    var testCluster = clusterInfo();
 
     Assertions.assertEquals(1, smoothWeight.getAndChoose(topic, testCluster));
     Assertions.assertEquals(2, smoothWeight.getAndChoose(topic, testCluster));
@@ -56,24 +55,11 @@ public class SmoothWeightRoundRobinTest {
     Assertions.assertEquals(1, smoothWeight.getAndChoose(topic, testCluster));
   }
 
-  ClusterInfo mockResult() {
-    var re1 = Mockito.mock(ReplicaInfo.class);
-    var node1 = Mockito.mock(NodeInfo.class);
-    Mockito.when(re1.nodeInfo()).thenReturn(node1);
-    Mockito.when(node1.id()).thenReturn(1);
-
-    var re2 = Mockito.mock(ReplicaInfo.class);
-    var node2 = Mockito.mock(NodeInfo.class);
-    Mockito.when(re2.nodeInfo()).thenReturn(node2);
-    Mockito.when(node2.id()).thenReturn(2);
-
-    var re3 = Mockito.mock(ReplicaInfo.class);
-    var node3 = Mockito.mock(NodeInfo.class);
-    Mockito.when(re3.nodeInfo()).thenReturn(node3);
-    Mockito.when(node3.id()).thenReturn(3);
-    var clusterInfo = Mockito.mock(ClusterInfo.class);
-    Mockito.when(clusterInfo.availableReplicaLeaders(Mockito.anyString()))
-        .thenReturn(List.of(re1, re2, re3));
-    return clusterInfo;
+  ClusterInfo<ReplicaInfo> clusterInfo() {
+    return ClusterInfo.of(
+        List.of(
+            ReplicaInfo.of("test", 1, NodeInfo.of(1, "host", 1111), true, true, false),
+            ReplicaInfo.of("test", 2, NodeInfo.of(2, "host", 1111), true, true, false),
+            ReplicaInfo.of("test", 3, NodeInfo.of(3, "host", 1111), true, true, false)));
   }
 }
diff --git a/app/src/test/java/org/astraea/app/performance/DataSupplierTest.java b/app/src/test/java/org/astraea/app/performance/DataSupplierTest.java
index 18774a3b..22768ce2 100644
--- a/app/src/test/java/org/astraea/app/performance/DataSupplierTest.java
+++ b/app/src/test/java/org/astraea/app/performance/DataSupplierTest.java
@@ -19,6 +19,7 @@ package org.astraea.app.performance;
 import java.time.Duration;
 import java.util.Arrays;
 import java.util.concurrent.atomic.AtomicLong;
+import org.astraea.app.common.DataRate;
 import org.astraea.app.common.DataSize;
 import org.astraea.app.common.DataUnit;
 import org.astraea.app.common.Utils;
@@ -31,7 +32,12 @@ public class DataSupplierTest {
   void testDuration() {
     var dataSupplier =
         DataSupplier.of(
-            ExeTime.of("2s"), () -> 1L, () -> 4L, () -> 1L, () -> 10L, DataSize.KiB.of(100));
+            ExeTime.of("2s"),
+            () -> 1L,
+            () -> 4L,
+            () -> 1L,
+            () -> 10L,
+            DataRate.KiB.of(100).perSecond());
     Assertions.assertTrue(dataSupplier.get().hasData());
     Utils.sleep(Duration.ofSeconds(3));
     Assertions.assertFalse(dataSupplier.get().hasData());
@@ -46,7 +52,7 @@ public class DataSupplierTest {
             () -> 10L,
             () -> DataSize.KiB.of(100).measurement(DataUnit.Byte).longValue(),
             () -> 10L,
-            DataSize.KiB.of(102));
+            DataRate.KiB.of(102).perSecond());
     Assertions.assertTrue(dataSupplier.get().hasData());
     Assertions.assertTrue(dataSupplier.get().hasData());
     Assertions.assertFalse(dataSupplier.get().hasData());
@@ -61,7 +67,7 @@ public class DataSupplierTest {
             () -> DataSize.Byte.of(20).measurement(DataUnit.Byte).longValue(),
             () -> 2L,
             () -> DataSize.KiB.of(100).measurement(DataUnit.Byte).longValue(),
-            DataSize.KiB.of(200));
+            DataRate.KiB.of(200).perSecond());
     var data1 = dataSupplier.get();
     Assertions.assertTrue(data1.hasData());
     var data2 = dataSupplier.get();
@@ -79,7 +85,7 @@ public class DataSupplierTest {
             () -> DataSize.Byte.of(20).measurement(DataUnit.Byte).longValue(),
             () -> 2L,
             () -> DataSize.KiB.of(100).measurement(DataUnit.Byte).longValue(),
-            DataSize.KiB.of(100));
+            DataRate.KiB.of(100).perSecond());
     var data = dataSupplier.get();
     Assertions.assertTrue(data.hasData());
     // initial value size is 100KB and the distributed is fixed to zero, so the final size is 102400
@@ -99,7 +105,7 @@ public class DataSupplierTest {
             () -> 100L + counter2.getAndIncrement(),
             () -> 10L,
             () -> 10L,
-            DataSize.KiB.of(200));
+            DataRate.KiB.of(200).perSecond());
     var data1 = dataSupplier.get();
     Assertions.assertTrue(data1.hasData());
     var data2 = dataSupplier.get();
@@ -121,7 +127,7 @@ public class DataSupplierTest {
             () -> 100L,
             () -> 10L,
             () -> 10L,
-            DataSize.KiB.of(200));
+            DataRate.KiB.of(200).perSecond());
     data1 = dataSupplier.get();
     Assertions.assertTrue(data1.hasData());
     data2 = dataSupplier.get();
@@ -145,7 +151,7 @@ public class DataSupplierTest {
             () -> 10L,
             () -> counter.getAndIncrement() % 2,
             () -> 100L + counter2.getAndIncrement(),
-            DataSize.KiB.of(100));
+            DataRate.KiB.of(100).perSecond());
     var data1 = dataSupplier.get();
     Assertions.assertTrue(data1.hasData());
     var data2 = dataSupplier.get();
@@ -167,7 +173,7 @@ public class DataSupplierTest {
             () -> 10L,
             () -> counter.getAndIncrement() % 2,
             () -> 100L,
-            DataSize.KiB.of(100));
+            DataRate.KiB.of(100).perSecond());
     data1 = dataSupplier.get();
     Assertions.assertTrue(data1.hasData());
     data2 = dataSupplier.get();
@@ -182,7 +188,7 @@ public class DataSupplierTest {
   void testThrottle() {
     var durationInSeconds = new AtomicLong(1);
     var throttler =
-        new DataSupplier.Throttler(DataSize.KiB.of(150)) {
+        new DataSupplier.Throttler(DataRate.KiB.of(150).perSecond()) {
           @Override
           long durationInSeconds() {
             return durationInSeconds.get();
@@ -200,7 +206,12 @@ public class DataSupplierTest {
   void testNoKey() {
     var dataSupplier =
         DataSupplier.of(
-            ExeTime.of("10s"), () -> 10L, () -> 0L, () -> 10L, () -> 10L, DataSize.KiB.of(200));
+            ExeTime.of("10s"),
+            () -> 10L,
+            () -> 0L,
+            () -> 10L,
+            () -> 10L,
+            DataRate.KiB.of(200).perSecond());
 
     var data = dataSupplier.get();
     Assertions.assertTrue(data.hasData());
@@ -211,7 +222,12 @@ public class DataSupplierTest {
   void testNoValue() {
     var dataSupplier =
         DataSupplier.of(
-            ExeTime.of("10s"), () -> 10L, () -> 10L, () -> 10L, () -> 0L, DataSize.KiB.of(200));
+            ExeTime.of("10s"),
+            () -> 10L,
+            () -> 10L,
+            () -> 10L,
+            () -> 0L,
+            DataRate.KiB.of(200).perSecond());
     var data = dataSupplier.get();
     Assertions.assertTrue(data.hasData());
     Assertions.assertNull(data.value());
diff --git a/app/src/test/java/org/astraea/app/performance/PerformanceTest.java b/app/src/test/java/org/astraea/app/performance/PerformanceTest.java
index 9146e699..34d00539 100644
--- a/app/src/test/java/org/astraea/app/performance/PerformanceTest.java
+++ b/app/src/test/java/org/astraea/app/performance/PerformanceTest.java
@@ -22,11 +22,14 @@ import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 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.TopicPartition;
 import org.astraea.app.argument.Argument;
+import org.astraea.app.common.DataRate;
 import org.astraea.app.common.Utils;
 import org.astraea.app.consumer.Isolation;
+import org.astraea.app.producer.Acks;
 import org.astraea.app.service.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
@@ -97,14 +100,17 @@ public class PerformanceTest extends RequireBrokerCluster {
       "4Byte",
       "--key.distribution",
       "zipfian",
-      "--specify.broker",
+      "--specify.brokers",
       "1",
+      "--throughput",
+      "100MB/m",
       "--configs",
       "key=value"
     };
 
     var arg = Argument.parse(new Performance.Argument(), arguments1);
     Assertions.assertEquals("value", arg.configs().get("key"));
+    Assertions.assertEquals(DataRate.MB.of(100).perMinute(), arg.throughput);
 
     String[] arguments2 = {"--bootstrap.servers", "localhost:9092", "--topics", ""};
     Assertions.assertThrows(
@@ -155,7 +161,7 @@ public class PerformanceTest extends RequireBrokerCluster {
     Assertions.assertThrows(
         ParameterException.class, () -> Argument.parse(new Performance.Argument(), arguments13));
 
-    String[] arguments14 = {"--bootstrap.servers", "localhost:9092", "--specify.broker", ""};
+    String[] arguments14 = {"--bootstrap.servers", "localhost:9092", "--specify.brokers", ""};
     Assertions.assertThrows(
         ParameterException.class, () -> Argument.parse(new Performance.Argument(), arguments14));
 
@@ -199,7 +205,7 @@ public class PerformanceTest extends RequireBrokerCluster {
                 bootstrapServers(),
                 "--topics",
                 topicName,
-                "--specify.broker",
+                "--specify.brokers",
                 "1"
               });
       var expectedPartitions =
@@ -412,4 +418,19 @@ public class PerformanceTest extends RequireBrokerCluster {
       Assertions.assertTrue(defaultPattern.get("test3").containsValue((short) 2));
     }
   }
+
+  @Test
+  void testAcks() {
+    Stream.of(Acks.values())
+        .forEach(
+            ack -> {
+              var arg =
+                  Argument.parse(
+                      new Performance.Argument(),
+                      new String[] {
+                        "--bootstrap.servers", bootstrapServers(), "--acks", ack.alias()
+                      });
+              Assertions.assertEquals(ack, arg.acks);
+            });
+  }
 }
diff --git a/app/src/test/java/org/astraea/app/performance/TrackerTest.java b/app/src/test/java/org/astraea/app/performance/TrackerTest.java
index 04030920..42b33883 100644
--- a/app/src/test/java/org/astraea/app/performance/TrackerTest.java
+++ b/app/src/test/java/org/astraea/app/performance/TrackerTest.java
@@ -50,7 +50,7 @@ public class TrackerTest {
   @Test
   void testExeTime() {
     var producerReport = new ProducerThread.Report();
-    var consumerReport = new ConsumerThread.Report();
+    var consumerReport = new ConsumerThread.Report("xxx");
     var tracker =
         TrackerThread.create(
             () -> List.of(producerReport), () -> List.of(consumerReport), ExeTime.of("2s"));
@@ -64,7 +64,7 @@ public class TrackerTest {
   @Test
   void testConsumerAndProducer() {
     var producerReport = new ProducerThread.Report();
-    var consumerReport = new ConsumerThread.Report();
+    var consumerReport = new ConsumerThread.Report("xxx");
     var tracker =
         TrackerThread.create(
             () -> List.of(producerReport), () -> List.of(consumerReport), ExeTime.of("1records"));
diff --git a/app/src/test/java/org/astraea/app/argument/CompressionFieldTest.java b/app/src/test/java/org/astraea/app/producer/AcksTest.java
similarity index 51%
rename from app/src/test/java/org/astraea/app/argument/CompressionFieldTest.java
rename to app/src/test/java/org/astraea/app/producer/AcksTest.java
index bf963c7c..8046a2a4 100644
--- a/app/src/test/java/org/astraea/app/argument/CompressionFieldTest.java
+++ b/app/src/test/java/org/astraea/app/producer/AcksTest.java
@@ -14,36 +14,37 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.astraea.app.argument;
+package org.astraea.app.producer;
 
-import com.beust.jcommander.Parameter;
-import com.beust.jcommander.ParameterException;
-import java.util.stream.Stream;
-import org.astraea.app.admin.Compression;
+import java.util.Arrays;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.config.ConfigDef;
+import org.astraea.app.common.Utils;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
-public class CompressionFieldTest {
+public class AcksTest {
 
-  private static class FakeParameter {
-    @Parameter(
-        names = {"--field"},
-        converter = CompressionField.class,
-        validateWith = CompressionField.class)
-    public Compression value;
+  @Test
+  void testKafkaConfig() {
+    var config =
+        ((ConfigDef) Utils.staticMember(ProducerConfig.class, "CONFIG"))
+            .configKeys()
+            .get(ProducerConfig.ACKS_CONFIG);
+    Arrays.stream(Acks.values())
+        .forEach(ack -> config.validator.ensureValid("acks", ack.valueOfKafka()));
   }
 
   @Test
-  void testConversion() {
-    var arg = new CompressionField();
-    Stream.of(Compression.values())
-        .forEach(type -> Assertions.assertEquals(type, arg.convert(type.name())));
-    Assertions.assertThrows(ParameterException.class, () -> arg.convert("aaa"));
+  void testAlias() {
+    Arrays.stream(Acks.values())
+        .forEach(acks -> Assertions.assertEquals(acks, Acks.ofAlias(acks.alias())));
   }
 
   @Test
-  void testParse() {
-    var arg = Argument.parse(new FakeParameter(), new String[] {"--field", "gzip"});
-    Assertions.assertEquals(Compression.GZIP, arg.value);
+  void testField() {
+    var field = new Acks.Field();
+    Arrays.stream(Acks.values())
+        .forEach(acks -> Assertions.assertEquals(acks, field.convert(acks.alias())));
   }
 }
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 10c5f0d0..cc28b2ef 100644
--- a/app/src/test/java/org/astraea/app/web/BalancerHandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/BalancerHandlerTest.java
@@ -53,6 +53,11 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
           .flatMap(c -> c.after.stream())
           .forEach(p -> Assertions.assertNull(p.size));
       Assertions.assertTrue(report.cost >= report.newCost);
+      var sizeMigration =
+          report.migrationCosts.stream().filter(x -> x.function.equals("size")).findFirst().get();
+      Assertions.assertTrue(sizeMigration.totalCost >= 0);
+      Assertions.assertTrue(sizeMigration.cost.size() > 0);
+      Assertions.assertEquals(0, sizeMigration.cost.stream().mapToLong(x -> x.cost).sum());
     }
   }
 
@@ -76,6 +81,11 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
       Assertions.assertEquals(1, actual.size());
       Assertions.assertEquals(topicNames.get(0), actual.iterator().next());
       Assertions.assertTrue(report.cost >= report.newCost);
+      var sizeMigration =
+          report.migrationCosts.stream().filter(x -> x.function.equals("size")).findFirst().get();
+      Assertions.assertTrue(sizeMigration.totalCost >= 0);
+      Assertions.assertTrue(sizeMigration.cost.size() > 0);
+      Assertions.assertEquals(0, sizeMigration.cost.stream().mapToLong(x -> x.cost).sum());
     }
   }
 
@@ -100,6 +110,11 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
       Assertions.assertTrue(actual.contains(topicNames.get(0)));
       Assertions.assertTrue(actual.contains(topicNames.get(1)));
       Assertions.assertTrue(report.cost >= report.newCost);
+      var sizeMigration =
+          report.migrationCosts.stream().filter(x -> x.function.equals("size")).findFirst().get();
+      Assertions.assertTrue(sizeMigration.totalCost >= 0);
+      Assertions.assertTrue(sizeMigration.cost.size() > 0);
+      Assertions.assertEquals(0, sizeMigration.cost.stream().mapToLong(x -> x.cost).sum());
     }
   }
 
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 05194bf8..db3360ab 100644
--- a/app/src/test/java/org/astraea/app/web/ThrottleHandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/ThrottleHandlerTest.java
@@ -24,9 +24,11 @@ import com.google.gson.JsonElement;
 import com.google.gson.JsonObject;
 import com.google.gson.JsonPrimitive;
 import java.time.Duration;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.function.Function;
+import java.util.function.Supplier;
 import java.util.stream.StreamSupport;
 import org.astraea.app.admin.Admin;
 import org.astraea.app.admin.TopicPartition;
@@ -45,6 +47,12 @@ public class ThrottleHandlerTest extends RequireBrokerCluster {
       var handler = new ThrottleHandler(admin);
       var dataRate = DataRate.MiB.of(500).perSecond();
       var longDataRate = (long) dataRate.byteRate();
+
+      // other tests might write this value too, ensure it is clean before we start
+      admin.clearIngressReplicationThrottle(brokerIds());
+      admin.clearEgressReplicationThrottle(brokerIds());
+      Utils.sleep(Duration.ofSeconds(1));
+
       admin
           .replicationThrottler()
           .ingress(Map.of(0, dataRate, 2, dataRate))
@@ -212,4 +220,387 @@ public class ThrottleHandlerTest extends RequireBrokerCluster {
     Assertions.assertEquals(expectedBroker, Set.copyOf(deserialized.brokers));
     Assertions.assertEquals(expectedTopic, Set.copyOf(deserialized.topics));
   }
+
+  @Test
+  void testPost() {
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var handler = new ThrottleHandler(admin);
+      var topicA = Utils.randomString();
+      var topicB = Utils.randomString();
+      var topicC = Utils.randomString();
+      var topicD = Utils.randomString();
+      admin.creator().topic(topicA).numberOfPartitions(2).numberOfReplicas((short) 3).create();
+      admin.creator().topic(topicB).numberOfPartitions(3).numberOfReplicas((short) 3).create();
+      admin.creator().topic(topicC).numberOfPartitions(4).numberOfReplicas((short) 3).create();
+      admin.creator().topic(topicD).numberOfPartitions(5).numberOfReplicas((short) 3).create();
+      Utils.sleep(Duration.ofSeconds(1));
+      admin.migrator().partition(topicA, 0).moveTo(List.of(0, 1, 2));
+      admin.migrator().partition(topicA, 1).moveTo(List.of(0, 1, 2));
+      admin.migrator().partition(topicB, 2).moveTo(List.of(0, 1, 2));
+      admin.migrator().partition(topicC, 3).moveTo(List.of(0, 1, 2));
+      admin.migrator().partition(topicD, 4).moveTo(List.of(0, 1, 2));
+      Utils.sleep(Duration.ofSeconds(1));
+      admin.preferredLeaderElection(TopicPartition.of(topicA, 0));
+      admin.preferredLeaderElection(TopicPartition.of(topicA, 1));
+      admin.preferredLeaderElection(TopicPartition.of(topicB, 2));
+      admin.preferredLeaderElection(TopicPartition.of(topicC, 3));
+      admin.preferredLeaderElection(TopicPartition.of(topicD, 4));
+      Utils.sleep(Duration.ofSeconds(1));
+      var rawJson =
+          "{\"brokers\":["
+              + "{\"id\":0,\"ingress\":1000,\"egress\":1000},"
+              + "{\"id\":1,\"ingress\":1000}],"
+              + "\"topics\":["
+              + "{\"name\":\""
+              + topicA
+              + "\"},"
+              + "{\"name\":\""
+              + topicB
+              + "\",\"partition\":2},"
+              + "{\"name\":\""
+              + topicC
+              + "\",\"partition\":3,\"broker\":0},"
+              + "{\"name\":\""
+              + topicD
+              + "\",\"partition\":4,\"broker\":1,\"type\":\"follower\"},"
+              + "{\"name\":\""
+              + topicD
+              + "\",\"partition\":4,\"broker\":0,\"type\":\"leader\"}]}";
+      var affectedBrokers =
+          Set.of(
+              new ThrottleHandler.BrokerThrottle(0, 1000L, 1000L),
+              new ThrottleHandler.BrokerThrottle(1, 1000L, null));
+      var affectedTopics =
+          Set.of(
+              new ThrottleHandler.TopicThrottle(topicA, 0, 0, leader),
+              new ThrottleHandler.TopicThrottle(topicA, 0, 1, follower),
+              new ThrottleHandler.TopicThrottle(topicA, 0, 2, follower),
+              new ThrottleHandler.TopicThrottle(topicA, 1, 0, leader),
+              new ThrottleHandler.TopicThrottle(topicA, 1, 1, follower),
+              new ThrottleHandler.TopicThrottle(topicA, 1, 2, follower),
+              new ThrottleHandler.TopicThrottle(topicB, 2, 0, leader),
+              new ThrottleHandler.TopicThrottle(topicB, 2, 1, follower),
+              new ThrottleHandler.TopicThrottle(topicB, 2, 2, follower),
+              new ThrottleHandler.TopicThrottle(topicC, 3, 0, null),
+              new ThrottleHandler.TopicThrottle(topicD, 4, 1, follower),
+              new ThrottleHandler.TopicThrottle(topicD, 4, 0, leader));
+      var affectedLeaders =
+          Set.of(
+              TopicPartitionReplica.of(topicA, 0, 0),
+              TopicPartitionReplica.of(topicA, 1, 0),
+              TopicPartitionReplica.of(topicB, 2, 0),
+              TopicPartitionReplica.of(topicC, 3, 0),
+              TopicPartitionReplica.of(topicD, 4, 0));
+      var affectedFollowers =
+          Set.of(
+              TopicPartitionReplica.of(topicA, 0, 1),
+              TopicPartitionReplica.of(topicA, 0, 2),
+              TopicPartitionReplica.of(topicA, 1, 1),
+              TopicPartitionReplica.of(topicA, 1, 2),
+              TopicPartitionReplica.of(topicB, 2, 1),
+              TopicPartitionReplica.of(topicB, 2, 2),
+              TopicPartitionReplica.of(topicC, 3, 0),
+              TopicPartitionReplica.of(topicD, 4, 1));
+
+      var post = handler.post(Channel.ofRequest(PostRequest.of(rawJson)));
+      var deserialized = new Gson().fromJson(post.json(), ThrottleHandler.ThrottleSetting.class);
+      Utils.sleep(Duration.ofSeconds(1));
+
+      // verify response content is correct
+      Assertions.assertEquals(200, post.code());
+      Assertions.assertEquals(affectedBrokers, Set.copyOf(deserialized.brokers));
+      Assertions.assertEquals(affectedTopics, Set.copyOf(deserialized.topics));
+
+      // verify topic/broker configs are correct
+      final var topicConfigs = admin.topics(Set.of(topicA, topicB, topicC, topicD));
+      Assertions.assertTrue(
+          affectedLeaders.stream()
+              .allMatch(
+                  log ->
+                      topicConfigs
+                          .get(log.topic())
+                          .value("leader.replication.throttled.replicas")
+                          .orElse("")
+                          .contains(log.partition() + ":" + log.brokerId())));
+      Assertions.assertTrue(
+          affectedFollowers.stream()
+              .allMatch(
+                  log ->
+                      topicConfigs
+                          .get(log.topic())
+                          .value("follower.replication.throttled.replicas")
+                          .orElse("")
+                          .contains(log.partition() + ":" + log.brokerId())));
+      final var brokerConfigs = admin.brokers();
+      Assertions.assertEquals(
+          1000L,
+          brokerConfigs
+              .get(0)
+              .value("leader.replication.throttled.rate")
+              .map(Long::parseLong)
+              .orElse(0L));
+      Assertions.assertEquals(
+          1000L,
+          brokerConfigs
+              .get(0)
+              .value("follower.replication.throttled.rate")
+              .map(Long::parseLong)
+              .orElse(0L));
+      Assertions.assertEquals(
+          1000L,
+          brokerConfigs
+              .get(1)
+              .value("follower.replication.throttled.rate")
+              .map(Long::parseLong)
+              .orElse(0L));
+    }
+  }
+
+  @Test
+  void testBadPost() {
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var handler = new ThrottleHandler(admin);
+
+      // empty
+      Assertions.assertThrows(
+          IllegalArgumentException.class,
+          () -> handler.post(Channel.ofRequest(PostRequest.of("{\"topics\":[{}]}"))));
+
+      // no key "name" specified
+      Assertions.assertThrows(
+          IllegalArgumentException.class,
+          () ->
+              handler.post(Channel.ofRequest(PostRequest.of("{\"topics\":[{\"partition\": 3}]}"))));
+
+      // this key combination is not supported
+      Assertions.assertThrows(
+          IllegalArgumentException.class,
+          () ->
+              handler.post(
+                  Channel.ofRequest(
+                      PostRequest.of("{\"topics\":[{\"name\": \"A\", \"broker\": 3}]}"))));
+
+      // illegal type value
+      Assertions.assertThrows(
+          IllegalArgumentException.class,
+          () ->
+              handler.post(
+                  Channel.ofRequest(
+                      PostRequest.of(
+                          "{\"topics\":[{\"name\": \"A\", \"partition\": 0, \"broker\": 3, \"type\": \"owo?\"}]}"))));
+    }
+  }
+
+  @Test
+  void testDelete() {
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var handler = new ThrottleHandler(admin);
+      var topic = Utils.randomString();
+      admin.creator().topic(topic).numberOfPartitions(3).numberOfReplicas((short) 3).create();
+      Utils.sleep(Duration.ofMillis(500));
+      admin.migrator().partition(topic, 0).moveTo(List.of(0, 1, 2));
+      admin.migrator().partition(topic, 1).moveTo(List.of(0, 1, 2));
+      admin.migrator().partition(topic, 2).moveTo(List.of(0, 1, 2));
+      Utils.sleep(Duration.ofMillis(500));
+      admin.preferredLeaderElection(TopicPartition.of(topic, 0));
+      admin.preferredLeaderElection(TopicPartition.of(topic, 1));
+      admin.preferredLeaderElection(TopicPartition.of(topic, 2));
+      Utils.sleep(Duration.ofMillis(500));
+
+      Supplier<String> leaderConfig =
+          () ->
+              admin
+                  .topics(Set.of(topic))
+                  .get(topic)
+                  .value("leader.replication.throttled.replicas")
+                  .orElse("");
+      Supplier<String> followerConfig =
+          () ->
+              admin
+                  .topics(Set.of(topic))
+                  .get(topic)
+                  .value("follower.replication.throttled.replicas")
+                  .orElse("");
+      Function<Integer, Long> egressRate =
+          (id) ->
+              admin
+                  .brokers()
+                  .get(id)
+                  .value("leader.replication.throttled.rate")
+                  .map(Long::parseLong)
+                  .orElse(-1L);
+      Function<Integer, Long> ingressRate =
+          (id) ->
+              admin
+                  .brokers()
+                  .get(id)
+                  .value("follower.replication.throttled.rate")
+                  .map(Long::parseLong)
+                  .orElse(-1L);
+      Runnable setThrottle =
+          () -> {
+            admin
+                .replicationThrottler()
+                .throttle(topic)
+                .ingress(DataRate.Byte.of(100).perSecond())
+                .egress(DataRate.Byte.of(100).perSecond())
+                .apply();
+            Utils.sleep(Duration.ofMillis(500));
+          };
+
+      // delete topic
+      setThrottle.run();
+      int code0 = handler.delete(Channel.ofQueries(Map.of("topic", topic))).code();
+      Utils.sleep(Duration.ofMillis(500));
+      Assertions.assertEquals(202, code0);
+      Assertions.assertEquals("", leaderConfig.get());
+      Assertions.assertEquals("", followerConfig.get());
+
+      // delete topic/partition
+      setThrottle.run();
+      int code1 =
+          handler.delete(Channel.ofQueries(Map.of("topic", topic, "partition", "0"))).code();
+      Utils.sleep(Duration.ofMillis(500));
+      Assertions.assertEquals(202, code1);
+      Assertions.assertFalse(leaderConfig.get().matches("0:[0-9]+"));
+      Assertions.assertFalse(followerConfig.get().matches("0:[0-9]+"));
+
+      // delete topic/partition/replica
+      setThrottle.run();
+      int code2 =
+          handler
+              .delete(
+                  Channel.ofQueries(
+                      Map.of(
+                          "topic", topic,
+                          "partition", "0",
+                          "replica", "0")))
+              .code();
+      Utils.sleep(Duration.ofMillis(500));
+      Assertions.assertEquals(202, code2);
+      Assertions.assertFalse(leaderConfig.get().matches("0:0"));
+      Assertions.assertFalse(followerConfig.get().matches("0:0"));
+
+      // delete topic/partition/replica/type
+      setThrottle.run();
+      int code3 =
+          handler
+              .delete(
+                  Channel.ofQueries(
+                      Map.of(
+                          "topic", topic,
+                          "partition", "0",
+                          "replica", "0",
+                          "type", "leader")))
+              .code();
+      int code4 =
+          handler
+              .delete(
+                  Channel.ofQueries(
+                      Map.of(
+                          "topic", topic,
+                          "partition", "0",
+                          "replica", "1",
+                          "type", "follower")))
+              .code();
+      Utils.sleep(Duration.ofMillis(500));
+      Assertions.assertEquals(202, code3);
+      Assertions.assertEquals(202, code4);
+      Assertions.assertFalse(leaderConfig.get().matches("0:0"));
+      Assertions.assertFalse(followerConfig.get().matches("0:1"));
+
+      // delete broker/type=ingress
+      setThrottle.run();
+      int code5 =
+          handler
+              .delete(
+                  Channel.ofQueries(
+                      Map.of(
+                          "broker", "0",
+                          "type", "ingress")))
+              .code();
+      Utils.sleep(Duration.ofMillis(500));
+      Assertions.assertEquals(202, code5);
+      Assertions.assertEquals(100L, egressRate.apply(0));
+      Assertions.assertEquals(-1L, ingressRate.apply(0));
+
+      // delete broker/type=egress
+      setThrottle.run();
+      int code6 =
+          handler
+              .delete(
+                  Channel.ofQueries(
+                      Map.of(
+                          "broker", "0",
+                          "type", "egress")))
+              .code();
+      Utils.sleep(Duration.ofMillis(500));
+      Assertions.assertEquals(202, code6);
+      Assertions.assertEquals(-1L, egressRate.apply(0));
+      Assertions.assertEquals(100L, ingressRate.apply(0));
+
+      // delete broker/type=ingress+egress
+      setThrottle.run();
+      int code7 =
+          handler
+              .delete(
+                  Channel.ofQueries(
+                      Map.of(
+                          "broker", "0",
+                          "type", "ingress+egress")))
+              .code();
+      Utils.sleep(Duration.ofMillis(500));
+      Assertions.assertEquals(202, code7);
+      Assertions.assertEquals(-1L, egressRate.apply(0));
+      Assertions.assertEquals(-1L, ingressRate.apply(0));
+    }
+  }
+
+  @Test
+  void testBadDelete() {
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var handler = new ThrottleHandler(admin);
+
+      // empty
+      Assertions.assertEquals(
+          Response.BAD_REQUEST.code(), handler.delete(Channel.ofQueries(Map.of())).code());
+
+      // no key "topic" specified
+      Assertions.assertEquals(
+          Response.BAD_REQUEST.code(),
+          handler.delete(Channel.ofQueries(Map.of("partition", "0"))).code());
+
+      // this key combination is not supported
+      Assertions.assertThrows(
+          IllegalArgumentException.class,
+          () ->
+              handler.delete(
+                  Channel.ofQueries(
+                      Map.of(
+                          "topic", "MyTopic",
+                          "replica", "0"))));
+
+      // illegal type value
+      Assertions.assertThrows(
+          IllegalArgumentException.class,
+          () ->
+              handler.delete(
+                  Channel.ofQueries(
+                      Map.of(
+                          "topic", "MyTopic",
+                          "partition", "0",
+                          "replica", "0",
+                          "type", "owo?"))));
+
+      // illegal clear target
+      Assertions.assertThrows(
+          IllegalArgumentException.class,
+          () ->
+              handler.delete(
+                  Channel.ofQueries(
+                      Map.of(
+                          "broker", "0",
+                          "type", "ingress+egress+everyTopic"))));
+    }
+  }
 }