Repository: skiptests/astraea
Pull request #306 (2022-04-20 16:51:15)

Four new cost functions
完成 #298 第一階段
它們大致會長這樣
等待 #303 合併後解決 costFunctions conflict

wycccccc: @chia7712 按照新的`HasBrokerCost`架構做了修改，現在他們彼此都很獨立。再麻煩學長review一下。

Patch:
diff --git a/README.md b/README.md
index b1913727..e71a5914 100644
--- a/README.md
+++ b/README.md
@@ -342,25 +342,25 @@ This tool offers an effective way to migrate specify replicas from specific brok
 ### Move all replicas of topic "abc" from broker_0 to broker_1
 
 ```shell
-./gradlew run --args="replica --bootstrap.servers 192.168.50.178:19993 --from 0,1 --to 2,3 --topics abc"
+./gradlew run --args="replica --bootstrap.servers 192.168.50.178:19993 --from 0 --to 1 --topics abc"
 ```
 
 ### Move all replicas of topic "abc" in broker_0 to other folders
 
 ```shell
-./gradlew run --args="replica --bootstrap.servers 192.168.50.178:19993 --from 0 --to 0 -topics abc"
+./gradlew run --args="replica --bootstrap.servers 192.168.50.178:19993 --from 0 --to 0 --topics abc"
 ```
 
 ### Move specify replicas of topic "abc" and "def" from broker_0 to broker_1
 
 ```shell
-./gradlew run --args="replica --bootstrap.servers 192.168.50.178:19993 --from 0 --to 1 --topic abc --partitions 0,1"
+./gradlew run --args="replica --bootstrap.servers 192.168.50.178:19993 --from 0 --to 1 --topic abc,def --partitions 0,1"
 ```
 
 ### Move specify replicas of topic "abc"  and "def" from broker_0 to broker_1 specify folder
 
 ```shell
-./gradlew run --args="replica --bootstrap.servers 192.168.50.178:19993 --from 0 --to 1 --topic abc --partitions 0,1 --path /tmp/log1"
+./gradlew run --args="replica --bootstrap.servers 192.168.50.178:19993 --from 0 --to 1 --topic abc,def --partitions 0,1 --path /tmp/log1"
 ```
 
 ### Replica Collie Configurations
diff --git a/app/src/main/java/org/astraea/cost/Periodic.java b/app/src/main/java/org/astraea/cost/Periodic.java
new file mode 100644
index 00000000..c3756c2a
--- /dev/null
+++ b/app/src/main/java/org/astraea/cost/Periodic.java
@@ -0,0 +1,28 @@
+package org.astraea.cost;
+
+import java.util.function.Supplier;
+import org.astraea.Utils;
+
+/** Enables methods to be updated periodically, not calculated every time they are called. */
+public abstract class Periodic<Value> {
+  private long lastUpdate = -1;
+  private Value value;
+
+  /**
+   * Updates the value each second.
+   *
+   * @param updater Methods that need to be updated regularly.
+   * @return an object of type Value created from the parameter value.
+   */
+  protected Value tryUpdate(Supplier<Value> updater) {
+    if (Utils.overSecond(lastUpdate, 1)) {
+      value = updater.get();
+      lastUpdate = currentTime();
+    }
+    return value;
+  }
+
+  protected long currentTime() {
+    return System.currentTimeMillis();
+  }
+}
diff --git a/app/src/main/java/org/astraea/cost/brokersMetrics/BrokerInputCost.java b/app/src/main/java/org/astraea/cost/brokersMetrics/BrokerInputCost.java
new file mode 100644
index 00000000..92740655
--- /dev/null
+++ b/app/src/main/java/org/astraea/cost/brokersMetrics/BrokerInputCost.java
@@ -0,0 +1,114 @@
+package org.astraea.cost.brokersMetrics;
+
+import static org.astraea.cost.brokersMetrics.CostUtils.TScore;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.astraea.cost.BrokerCost;
+import org.astraea.cost.ClusterInfo;
+import org.astraea.cost.HasBrokerCost;
+import org.astraea.cost.Periodic;
+import org.astraea.metrics.collector.Fetcher;
+import org.astraea.metrics.kafka.BrokerTopicMetricsResult;
+import org.astraea.metrics.kafka.KafkaMetrics;
+
+/**
+ * The result is computed by "BytesInPerSec.count". "BytesInPerSec.count" responds to the input
+ * throughput of brokers.
+ *
+ * <ol>
+ *   <li>We normalize the metric as score(by T-score).
+ *   <li>We record these data of each second.
+ *   <li>We only keep the last ten seconds of data.
+ *   <li>The final result is the average of the ten-second data.
+ * </ol>
+ */
+public class BrokerInputCost extends Periodic<Map<Integer, Double>> implements HasBrokerCost {
+  private final Map<Integer, BrokerMetric> brokersMetric = new HashMap<>();
+
+  @Override
+  public BrokerCost brokerCost(ClusterInfo clusterInfo) {
+    var brokerScore =
+        tryUpdate(
+            () -> {
+              var costMetrics =
+                  clusterInfo.allBeans().entrySet().stream()
+                      .collect(Collectors.toMap(Map.Entry::getKey, entry -> 0.0));
+              clusterInfo
+                  .allBeans()
+                  .forEach(
+                      (brokerID, value) -> {
+                        if (!brokersMetric.containsKey(brokerID)) {
+                          brokersMetric.put(brokerID, new BrokerMetric());
+                        }
+                        value.stream()
+                            .filter(
+                                hasBeanObject ->
+                                    hasBeanObject
+                                        .beanObject()
+                                        .getProperties()
+                                        .get("name")
+                                        .equals(
+                                            KafkaMetrics.BrokerTopic.BytesInPerSec.metricName()))
+                            .forEach(
+                                hasBeanObject -> {
+                                  var broker = brokersMetric.get(brokerID);
+                                  var inBean = (BrokerTopicMetricsResult) hasBeanObject;
+                                  costMetrics.put(
+                                      brokerID, (double) (inBean.count() - broker.accumulateCount));
+                                  broker.accumulateCount = inBean.count();
+                                });
+                      });
+              TScore(costMetrics).forEach((broker, v) -> brokersMetric.get(broker).updateLoad(v));
+
+              return computeLoad();
+            });
+    return () -> brokerScore;
+  }
+
+  Map<Integer, Double> computeLoad() {
+    return brokersMetric.entrySet().stream()
+        .collect(
+            Collectors.toMap(
+                Map.Entry::getKey,
+                e ->
+                    Math.round(
+                            e.getValue().load.stream()
+                                    .filter(aDouble -> !aDouble.equals(-1.0))
+                                    .mapToDouble(i -> i)
+                                    .average()
+                                    .orElse(0.5)
+                                * 100)
+                        / 100.0));
+  }
+
+  @Override
+  public Fetcher fetcher() {
+    return client -> List.of(KafkaMetrics.BrokerTopic.BytesInPerSec.fetch(client));
+  }
+
+  private static class BrokerMetric {
+    // mbean data.  BytesInPerSec.count
+    private long accumulateCount = 0L;
+
+    // Record the latest 10 numbers only.
+    private final List<Double> load =
+        IntStream.range(0, 10).mapToObj(i -> -1.0).collect(Collectors.toList());
+    private int loadIndex = 0;
+
+    private BrokerMetric() {}
+
+    /**
+     * This method record input data into a list. This list contains the latest 10 record. Each time
+     * it is called, the current index, "loadIndex", is increased by 1.
+     */
+    void updateLoad(Double load) {
+      var cLoad = load.isNaN() ? 0.5 : load;
+      this.load.set(loadIndex, cLoad);
+      loadIndex = (loadIndex + 1) % 10;
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/cost/brokersMetrics/BrokerOutputCost.java b/app/src/main/java/org/astraea/cost/brokersMetrics/BrokerOutputCost.java
new file mode 100644
index 00000000..012e54e7
--- /dev/null
+++ b/app/src/main/java/org/astraea/cost/brokersMetrics/BrokerOutputCost.java
@@ -0,0 +1,114 @@
+package org.astraea.cost.brokersMetrics;
+
+import static org.astraea.cost.brokersMetrics.CostUtils.TScore;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.astraea.cost.BrokerCost;
+import org.astraea.cost.ClusterInfo;
+import org.astraea.cost.HasBrokerCost;
+import org.astraea.cost.Periodic;
+import org.astraea.metrics.collector.Fetcher;
+import org.astraea.metrics.kafka.BrokerTopicMetricsResult;
+import org.astraea.metrics.kafka.KafkaMetrics;
+
+/**
+ * The result is computed by "BytesOutPerSec.count". "BytesOutPerSec.count" responds to the output
+ * throughput of brokers.
+ *
+ * <ol>
+ *   <li>We normalize the metric as score(by T-score).
+ *   <li>We record these data of each second.
+ *   <li>We only keep the last ten seconds of data.
+ *   <li>The final result is the average of the ten-second data.
+ * </ol>
+ */
+public class BrokerOutputCost extends Periodic<Map<Integer, Double>> implements HasBrokerCost {
+  private final Map<Integer, BrokerMetric> brokersMetric = new HashMap<>();
+
+  @Override
+  public BrokerCost brokerCost(ClusterInfo clusterInfo) {
+    var brokerScore =
+        tryUpdate(
+            () -> {
+              var costMetrics =
+                  clusterInfo.allBeans().entrySet().stream()
+                      .collect(Collectors.toMap(Map.Entry::getKey, entry -> 0.0));
+              clusterInfo
+                  .allBeans()
+                  .forEach(
+                      (brokerID, value) -> {
+                        if (!brokersMetric.containsKey(brokerID)) {
+                          brokersMetric.put(brokerID, new BrokerMetric());
+                        }
+                        value.stream()
+                            .filter(
+                                hasBeanObject ->
+                                    hasBeanObject
+                                        .beanObject()
+                                        .getProperties()
+                                        .get("name")
+                                        .equals(
+                                            KafkaMetrics.BrokerTopic.BytesOutPerSec.metricName()))
+                            .forEach(
+                                hasBeanObject -> {
+                                  var broker = brokersMetric.get(brokerID);
+                                  var outBean = (BrokerTopicMetricsResult) hasBeanObject;
+                                  costMetrics.put(
+                                      brokerID,
+                                      (double) (outBean.count() - broker.accumulateCount));
+                                  broker.accumulateCount = outBean.count();
+                                });
+                      });
+              TScore(costMetrics).forEach((broker, v) -> brokersMetric.get(broker).updateLoad(v));
+              return computeLoad();
+            });
+    return () -> brokerScore;
+  }
+
+  Map<Integer, Double> computeLoad() {
+    return brokersMetric.entrySet().stream()
+        .collect(
+            Collectors.toMap(
+                Map.Entry::getKey,
+                e ->
+                    Math.round(
+                            e.getValue().load.stream()
+                                    .filter(aDouble -> !aDouble.equals(-1.0))
+                                    .mapToDouble(i -> i)
+                                    .average()
+                                    .orElse(0.5)
+                                * 100)
+                        / 100.0));
+  }
+
+  @Override
+  public Fetcher fetcher() {
+    return client -> List.of(KafkaMetrics.BrokerTopic.BytesOutPerSec.fetch(client));
+  }
+
+  private static class BrokerMetric {
+    // mbean data. BytesOutPerSec.count
+    private long accumulateCount = 0L;
+
+    // Record the latest 10 numbers only.
+    private final List<Double> load =
+        IntStream.range(0, 10).mapToObj(i -> -1.0).collect(Collectors.toList());
+    private int loadIndex = 0;
+
+    private BrokerMetric() {}
+
+    /**
+     * This method record input data into a list. This list contains the latest 10 record. Each time
+     * it is called, the current index, "loadIndex", is increased by 1.
+     */
+    void updateLoad(Double load) {
+      var cLoad = load.isNaN() ? 0.5 : load;
+      this.load.set(loadIndex, cLoad);
+      loadIndex = (loadIndex + 1) % 10;
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/cost/brokersMetrics/CostUtils.java b/app/src/main/java/org/astraea/cost/brokersMetrics/CostUtils.java
new file mode 100644
index 00000000..7c788af9
--- /dev/null
+++ b/app/src/main/java/org/astraea/cost/brokersMetrics/CostUtils.java
@@ -0,0 +1,46 @@
+package org.astraea.cost.brokersMetrics;
+
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+
+public class CostUtils {
+  public static double standardDeviationImperative(
+      double avgMetrics, Map<Integer, Double> metrics) {
+    var variance = new AtomicReference<>(0.0);
+    metrics
+        .values()
+        .forEach(
+            metric ->
+                variance.updateAndGet(v -> v + (metric - avgMetrics) * (metric - avgMetrics)));
+    return Math.sqrt(variance.get() / metrics.size());
+  }
+
+  public static Map<Integer, Double> ZScore(Map<Integer, Double> metrics) {
+    var avg = metrics.values().stream().mapToDouble(d -> d).sum() / metrics.size();
+    var standardDeviation = standardDeviationImperative(avg, metrics);
+    return metrics.entrySet().stream()
+        .collect(
+            Collectors.toMap(Map.Entry::getKey, e -> (e.getValue() - avg) / standardDeviation));
+  }
+
+  public static Map<Integer, Double> TScore(Map<Integer, Double> metrics) {
+    var avg = metrics.values().stream().mapToDouble(d -> d).sum() / metrics.size();
+    var standardDeviation = standardDeviationImperative(avg, metrics);
+
+    return metrics.entrySet().stream()
+        .collect(
+            Collectors.toMap(
+                Map.Entry::getKey,
+                e -> {
+                  var score =
+                      Math.round((((e.getValue() - avg) / standardDeviation) * 10 + 50)) / 100.0;
+                  if (score > 1) {
+                    score = 1.0;
+                  } else if (score < 0) {
+                    score = 0.0;
+                  }
+                  return score;
+                }));
+  }
+}
diff --git a/app/src/main/java/org/astraea/cost/brokersMetrics/CpuCost.java b/app/src/main/java/org/astraea/cost/brokersMetrics/CpuCost.java
new file mode 100644
index 00000000..8222ba7e
--- /dev/null
+++ b/app/src/main/java/org/astraea/cost/brokersMetrics/CpuCost.java
@@ -0,0 +1,101 @@
+package org.astraea.cost.brokersMetrics;
+
+import static org.astraea.cost.brokersMetrics.CostUtils.TScore;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.astraea.cost.BrokerCost;
+import org.astraea.cost.ClusterInfo;
+import org.astraea.cost.HasBrokerCost;
+import org.astraea.cost.Periodic;
+import org.astraea.metrics.collector.Fetcher;
+import org.astraea.metrics.java.OperatingSystemInfo;
+import org.astraea.metrics.kafka.KafkaMetrics;
+
+/**
+ * The result is computed by "OperatingSystemInfo.systemCpuLoad".
+ * "OperatingSystemInfo.systemCpuLoad" responds to the cpu usage of brokers.
+ *
+ * <ol>
+ *   <li>We normalize the metric as score(by T-score).
+ *   <li>We record these data of each second.
+ *   <li>We only keep the last ten seconds of data.
+ *   <li>The final result is the average of the ten-second data.
+ * </ol>
+ */
+public class CpuCost extends Periodic<Map<Integer, Double>> implements HasBrokerCost {
+  private final Map<Integer, BrokerMetric> brokersMetric = new HashMap<>();
+
+  @Override
+  public BrokerCost brokerCost(ClusterInfo clusterInfo) {
+    var brokerScore =
+        tryUpdate(
+            () -> {
+              var costMetrics =
+                  clusterInfo.allBeans().entrySet().stream()
+                      .collect(Collectors.toMap(Map.Entry::getKey, entry -> 0.0));
+              clusterInfo
+                  .allBeans()
+                  .forEach(
+                      (brokerID, value) -> {
+                        if (!brokersMetric.containsKey(brokerID)) {
+                          brokersMetric.put(brokerID, new BrokerMetric());
+                        }
+                        value.stream()
+                            .filter(beanObject -> beanObject instanceof OperatingSystemInfo)
+                            .forEach(
+                                hasBeanObject -> {
+                                  var cpuBean = (OperatingSystemInfo) hasBeanObject;
+                                  costMetrics.put(brokerID, cpuBean.systemCpuLoad());
+                                });
+                      });
+              TScore(costMetrics).forEach((broker, v) -> brokersMetric.get(broker).updateLoad(v));
+              return computeLoad();
+            });
+    return () -> brokerScore;
+  }
+
+  Map<Integer, Double> computeLoad() {
+    return brokersMetric.entrySet().stream()
+        .collect(
+            Collectors.toMap(
+                Map.Entry::getKey,
+                e ->
+                    Math.round(
+                            e.getValue().load.stream()
+                                    .filter(aDouble -> !aDouble.equals(-1.0))
+                                    .mapToDouble(i -> i)
+                                    .average()
+                                    .orElse(0.5)
+                                * 100)
+                        / 100.0));
+  }
+
+  @Override
+  public Fetcher fetcher() {
+    return client -> List.of(KafkaMetrics.Host.jvmMemory(client));
+  }
+
+  private static class BrokerMetric {
+    // mbean data.CpuUsage
+    // Record the latest 10 numbers only.
+    private final List<Double> load =
+        IntStream.range(0, 10).mapToObj(i -> -1.0).collect(Collectors.toList());
+    private int loadIndex = 0;
+
+    private BrokerMetric() {}
+
+    /**
+     * This method record input data into a list. This list contains the latest 10 record. Each time
+     * it is called, the current index, "loadIndex", is increased by 1.
+     */
+    void updateLoad(Double load) {
+      var cLoad = load.isNaN() ? 0.5 : load;
+      this.load.set(loadIndex, cLoad);
+      loadIndex = (loadIndex + 1) % 10;
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/cost/brokersMetrics/MemoryCost.java b/app/src/main/java/org/astraea/cost/brokersMetrics/MemoryCost.java
new file mode 100644
index 00000000..dabf7855
--- /dev/null
+++ b/app/src/main/java/org/astraea/cost/brokersMetrics/MemoryCost.java
@@ -0,0 +1,104 @@
+package org.astraea.cost.brokersMetrics;
+
+import static org.astraea.cost.brokersMetrics.CostUtils.TScore;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.astraea.cost.BrokerCost;
+import org.astraea.cost.ClusterInfo;
+import org.astraea.cost.HasBrokerCost;
+import org.astraea.cost.Periodic;
+import org.astraea.metrics.collector.Fetcher;
+import org.astraea.metrics.java.HasJvmMemory;
+import org.astraea.metrics.kafka.KafkaMetrics;
+
+public class MemoryCost extends Periodic<Map<Integer, Double>> implements HasBrokerCost {
+  private final Map<Integer, BrokerMetric> brokersMetric = new HashMap<>();
+
+  /**
+   * The result is computed by "HasJvmMemory.getUsed/getMax". "HasJvmMemory.getUsed/getMax" responds
+   * to the memory usage of brokers.
+   *
+   * <ol>
+   *   <li>We normalize the metric as score(by T-score).
+   *   <li>We record these data of each second.
+   *   <li>We only keep the last ten seconds of data.
+   *   <li>The final result is the average of the ten-second data.
+   * </ol>
+   */
+  @Override
+  public BrokerCost brokerCost(ClusterInfo clusterInfo) {
+    var brokerScore =
+        tryUpdate(
+            () -> {
+              var costMetrics =
+                  clusterInfo.allBeans().entrySet().stream()
+                      .collect(Collectors.toMap(Map.Entry::getKey, entry -> 0.0));
+              clusterInfo
+                  .allBeans()
+                  .forEach(
+                      (brokerID, value) -> {
+                        if (!brokersMetric.containsKey(brokerID)) {
+                          brokersMetric.put(brokerID, new BrokerMetric());
+                        }
+                        value.stream()
+                            .filter(beanObject -> beanObject instanceof HasJvmMemory)
+                            .forEach(
+                                hasBeanObject -> {
+                                  var jvmBean = (HasJvmMemory) hasBeanObject;
+                                  costMetrics.put(
+                                      brokerID,
+                                      ((jvmBean.heapMemoryUsage().getUsed() + 0.0)
+                                          / (jvmBean.heapMemoryUsage().getMax() + 1)));
+                                });
+                      });
+              TScore(costMetrics).forEach((broker, v) -> brokersMetric.get(broker).updateLoad(v));
+              return computeLoad();
+            });
+    return () -> brokerScore;
+  }
+
+  Map<Integer, Double> computeLoad() {
+    return brokersMetric.entrySet().stream()
+        .collect(
+            Collectors.toMap(
+                Map.Entry::getKey,
+                e ->
+                    Math.round(
+                            e.getValue().load.stream()
+                                    .filter(aDouble -> !aDouble.equals(-1.0))
+                                    .mapToDouble(i -> i)
+                                    .average()
+                                    .orElse(0.5)
+                                * 100)
+                        / 100.0));
+  }
+
+  @Override
+  public Fetcher fetcher() {
+    return client -> List.of(KafkaMetrics.Host.jvmMemory(client));
+  }
+
+  private static class BrokerMetric {
+    // mbean data.JvmUsage
+    // Record the latest 10 numbers only.
+    private final List<Double> load =
+        IntStream.range(0, 10).mapToObj(i -> -1.0).collect(Collectors.toList());
+    private int loadIndex = 0;
+
+    private BrokerMetric() {}
+
+    /**
+     * This method record input data into a list. This list contains the latest 10 record. Each time
+     * it is called, the current index, "loadIndex", is increased by 1.
+     */
+    void updateLoad(Double load) {
+      var cLoad = load.isNaN() ? 0.5 : load;
+      this.load.set(loadIndex, cLoad);
+      loadIndex = (loadIndex + 1) % 10;
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/metrics/kafka/HasValue.java b/app/src/main/java/org/astraea/metrics/kafka/HasValue.java
new file mode 100644
index 00000000..6f0203d4
--- /dev/null
+++ b/app/src/main/java/org/astraea/metrics/kafka/HasValue.java
@@ -0,0 +1,15 @@
+package org.astraea.metrics.kafka;
+
+import org.astraea.metrics.HasBeanObject;
+import org.astraea.metrics.jmx.BeanObject;
+
+public interface HasValue extends HasBeanObject {
+  default long value() {
+    var value = beanObject().getAttributes().getOrDefault("Value", 0);
+    return ((Number) value).longValue();
+  }
+
+  static HasValue of(BeanObject beanObject) {
+    return () -> beanObject;
+  }
+}
diff --git a/app/src/main/java/org/astraea/metrics/kafka/KafkaMetrics.java b/app/src/main/java/org/astraea/metrics/kafka/KafkaMetrics.java
index 0b1e2a47..19587e5c 100644
--- a/app/src/main/java/org/astraea/metrics/kafka/KafkaMetrics.java
+++ b/app/src/main/java/org/astraea/metrics/kafka/KafkaMetrics.java
@@ -4,6 +4,7 @@ import java.util.Arrays;
 import java.util.Collection;
 import java.util.Map;
 import java.util.stream.Collectors;
+import org.astraea.metrics.HasBeanObject;
 import org.astraea.metrics.java.JvmMemory;
 import org.astraea.metrics.java.OperatingSystemInfo;
 import org.astraea.metrics.jmx.BeanObject;
@@ -120,13 +121,36 @@ public final class KafkaMetrics {
   }
 
   public enum Purgatory {
-    AlterAcls,
-    DeleteRecords,
-    ElectLeader,
-    Fetch,
-    Heartbeat,
-    Produce,
-    Rebalance;
+    AlterAcls("AlterAcls"),
+    DeleteRecords("DeleteRecords"),
+    ElectLeader("ElectLeader"),
+    Fetch("Fetch"),
+    Heartbeat("Heartbeat"),
+    Produce("Produce"),
+    Rebalance("Rebalance");
+
+    private final String metricName;
+
+    Purgatory(String name) {
+      this.metricName = name;
+    }
+
+    public String metricName() {
+      return metricName;
+    }
+
+    public Collection<HasBeanObject> fetch(MBeanClient mBeanClient) {
+      return mBeanClient
+          .queryBeans(
+              BeanQuery.builder("kafka.server")
+                  .property("type", "DelayedOperationPurgatory")
+                  .property("delayedOperation", metricName)
+                  .property("name", "PurgatorySize")
+                  .build())
+          .stream()
+          .map(HasValue::of)
+          .collect(Collectors.toUnmodifiableList());
+    }
 
     public int size(MBeanClient mBeanClient) {
       return (int)
@@ -158,9 +182,79 @@ public final class KafkaMetrics {
     }
   }
 
-  public static final class TopicPartition {
+  public enum ReplicaManager {
+    AtMinIsrPartitionCount("AtMinIsrPartitionCount"),
+    LeaderCount("LeaderCount"),
+    OfflineReplicaCount("OfflineReplicaCount"),
+    PartitionCount("PartitionCount"),
+    ReassigningPartitions("ReassigningPartitions"),
+    UnderMinIsrPartitionCount("UnderMinIsrPartitionCount"),
+    UnderReplicatedPartitions("UnderReplicatedPartition");
+    private final String metricName;
+
+    ReplicaManager(String name) {
+      this.metricName = name;
+    }
+
+    public String metricName() {
+      return metricName;
+    }
 
-    private TopicPartition() {}
+    public static ReplicaManager of(String metricName) {
+      return Arrays.stream(ReplicaManager.values())
+          .filter(metric -> metric.metricName().equalsIgnoreCase(metricName))
+          .findFirst()
+          .orElseThrow(() -> new IllegalArgumentException("No such metric: " + metricName));
+    }
+
+    public Collection<HasBeanObject> fetch(MBeanClient mBeanClient) {
+      return mBeanClient
+          .queryBeans(
+              BeanQuery.builder("kafka.server")
+                  .property("type", "ReplicaManager")
+                  .property("name", metricName)
+                  .build())
+          .stream()
+          .map(HasValue::of)
+          .collect(Collectors.toUnmodifiableList());
+    }
+  }
+
+  public enum TopicPartition {
+    LodEndOffset("LodEndOffset"),
+    LogStartOffset("LogStartOffset"),
+    NumLogSegments("NumLogSegments"),
+    Size("Size");
+    private final String metricName;
+
+    TopicPartition(String name) {
+      this.metricName = name;
+    }
+
+    public String metricName() {
+      return metricName;
+    }
+
+    public static TopicPartition of(String metricName) {
+      return Arrays.stream(TopicPartition.values())
+          .filter(metric -> metric.metricName().equalsIgnoreCase(metricName))
+          .findFirst()
+          .orElseThrow(() -> new IllegalArgumentException("No such metric: " + metricName));
+    }
+
+    public Collection<HasBeanObject> fetch(MBeanClient mBeanClient) {
+      return mBeanClient
+          .queryBeans(
+              BeanQuery.builder("kafka.log")
+                  .property("type", "Log")
+                  .property("topic", "*")
+                  .property("partition", "*")
+                  .property("name", metricName)
+                  .build())
+          .stream()
+          .map(HasValue::of)
+          .collect(Collectors.toUnmodifiableList());
+    }
 
     /**
      * Number of partitions across all topics in the cluster.
@@ -207,17 +301,15 @@ public final class KafkaMetrics {
      *     pair of partition id and its log size
      */
     public static Map<Integer, Long> size(MBeanClient client, String topicName) {
-      Collection<BeanObject> beanObjects =
-          client.queryBeans(
+      return client
+          .queryBeans(
               BeanQuery.builder("kafka.log")
                   .property("type", "Log")
                   .property("topic", topicName)
                   .property("partition", "*")
                   .property("name", "Size")
-                  .build());
-
-      // collect result as a map.
-      return beanObjects.stream()
+                  .build())
+          .stream()
           .collect(
               Collectors.toMap(
                   (BeanObject a) -> Integer.parseInt(a.getProperties().get("partition")),
diff --git a/app/src/main/java/org/astraea/topic/Builder.java b/app/src/main/java/org/astraea/topic/Builder.java
index 3d63ba4f..e89c7e29 100644
--- a/app/src/main/java/org/astraea/topic/Builder.java
+++ b/app/src/main/java/org/astraea/topic/Builder.java
@@ -12,6 +12,9 @@ import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.function.BiFunction;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -19,12 +22,14 @@ import org.apache.kafka.clients.admin.Admin;
 import org.apache.kafka.clients.admin.AdminClientConfig;
 import org.apache.kafka.clients.admin.ConfigEntry;
 import org.apache.kafka.clients.admin.ConsumerGroupListing;
+import org.apache.kafka.clients.admin.ElectLeadersResult;
 import org.apache.kafka.clients.admin.ListTopicsOptions;
 import org.apache.kafka.clients.admin.MemberDescription;
 import org.apache.kafka.clients.admin.NewPartitionReassignment;
 import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.clients.admin.OffsetSpec;
 import org.apache.kafka.clients.admin.ReplicaInfo;
+import org.apache.kafka.common.ElectionType;
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.TopicPartitionReplica;
@@ -73,7 +78,7 @@ public class Builder {
     }
 
     @Override
-    public Map<Integer, Set<String>> brokerFolders(Set<Integer> brokers) {
+    public Map<Integer, Set<String>> brokerFolders(Collection<Integer> brokers) {
       return Utils.handleException(
           () ->
               admin.describeLogDirs(brokers).allDescriptions().get().entrySet().stream()
@@ -231,6 +236,33 @@ public class Builder {
                   .collect(Collectors.toSet()));
     }
 
+    @Override
+    public Map<TopicPartition, Boolean> changeReplicaLeader(
+        Map<TopicPartition, Integer> partitions) {
+      Map<TopicPartition, Boolean> result = new HashMap<>();
+      partitions.forEach(
+          (tp, broker) -> {
+            var brokers =
+                replicas(Set.of(tp.topic())).get(tp).stream()
+                    .map(Replica::broker)
+                    .collect(Collectors.toList());
+            if (!brokers.contains(broker))
+              throw new IllegalArgumentException("replica " + tp + " is not in broker " + broker);
+            brokers.remove(broker);
+            brokers.add(0, broker);
+            migrator().partition(tp.topic(), tp.partition()).moveTo(brokers);
+            ElectLeadersResult electLeadersResult =
+                admin.electLeaders(ElectionType.PREFERRED, Set.of(tp));
+            try {
+              electLeadersResult.all().get(10L, TimeUnit.SECONDS);
+            } catch (InterruptedException | TimeoutException | ExecutionException e) {
+              throw new RuntimeException(e);
+            }
+            result.put(tp, electLeadersResult.partitions().isDone());
+          });
+      return result;
+    }
+
     @Override
     public List<TopicPartition> partitionsOfBrokers(Set<String> topics, Set<Integer> brokersID) {
       return replicas(topics).entrySet().stream()
@@ -467,7 +499,7 @@ public class Builder {
     }
 
     @Override
-    public void moveTo(Set<Integer> brokers) {
+    public void moveTo(List<Integer> brokers) {
       Utils.handleException(
           () ->
               admin
diff --git a/app/src/main/java/org/astraea/topic/Migrator.java b/app/src/main/java/org/astraea/topic/Migrator.java
index aeec5f19..9a9bfdbd 100644
--- a/app/src/main/java/org/astraea/topic/Migrator.java
+++ b/app/src/main/java/org/astraea/topic/Migrator.java
@@ -1,7 +1,7 @@
 package org.astraea.topic;
 
+import java.util.List;
 import java.util.Map;
-import java.util.Set;
 
 /** used to migrate partitions to another broker or broker folder. */
 public interface Migrator {
@@ -24,11 +24,11 @@ public interface Migrator {
 
   /**
    * move partitions to specify brokers. Noted that the number of brokers must be equal to the
-   * number of replicas
+   * number of replicas,and the first broker in the list will be set as the preferred leader
    *
    * @param brokers to host partitions
    */
-  void moveTo(Set<Integer> brokers);
+  void moveTo(List<Integer> brokers);
 
   /**
    * move partition to specify folder.
diff --git a/app/src/main/java/org/astraea/topic/ReplicaCollie.java b/app/src/main/java/org/astraea/topic/ReplicaCollie.java
index 8f702973..4b1960ba 100644
--- a/app/src/main/java/org/astraea/topic/ReplicaCollie.java
+++ b/app/src/main/java/org/astraea/topic/ReplicaCollie.java
@@ -2,9 +2,11 @@ package org.astraea.topic;
 
 import com.beust.jcommander.Parameter;
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.TreeMap;
@@ -18,8 +20,8 @@ public class ReplicaCollie {
   static final String UNKNOWN = "unknown";
 
   static class MigratorInfo {
-    Set<Integer> brokerSource;
-    Set<Integer> brokerSink;
+    List<Integer> brokerSource;
+    List<Integer> brokerSink;
     Set<String> pathSource;
     Set<String> pathSink;
   }
@@ -39,7 +41,7 @@ public class ReplicaCollie {
       argument.toBrokers =
           admin.brokerIds().stream()
               .filter(b -> !args.fromBrokers.contains(b))
-              .collect(Collectors.toSet());
+              .collect(Collectors.toList());
     } else {
       argument.toBrokers = args.toBrokers.isEmpty() ? args.fromBrokers : args.toBrokers;
     }
@@ -70,7 +72,6 @@ public class ReplicaCollie {
               + topics.iterator().next()
               + " does not exist partition: "
               + args.partitions.toString());
-
     if (args.fromBrokers.containsAll(admin.brokerIds()))
       throw new IllegalArgumentException(
           "No enough available brokers!" + " removed at least one: " + args.fromBrokers);
@@ -87,9 +88,9 @@ public class ReplicaCollie {
     }
   }
 
-  static TreeMap<TopicPartition, Map.Entry<Set<Integer>, Set<Integer>>> checkMigratorBroker(
+  static TreeMap<TopicPartition, Map.Entry<List<Integer>, List<Integer>>> checkMigratorBroker(
       TopicAdmin admin, Argument argument) {
-    TreeMap<TopicPartition, Map.Entry<Set<Integer>, Set<Integer>>> brokerMigrate =
+    TreeMap<TopicPartition, Map.Entry<List<Integer>, List<Integer>>> brokerMigrate =
         new TreeMap<>(
             Comparator.comparing(TopicPartition::topic).thenComparing(TopicPartition::partition));
     admin.replicas(argument.topics).entrySet().stream()
@@ -102,7 +103,7 @@ public class ReplicaCollie {
         .forEach(
             (tp) -> {
               var currentBrokers =
-                  tp.getValue().stream().map(Replica::broker).collect(Collectors.toSet());
+                  tp.getValue().stream().map(Replica::broker).collect(Collectors.toList());
               var keptBrokers =
                   currentBrokers.stream()
                       .filter(i -> !argument.fromBrokers.contains(i))
@@ -112,7 +113,7 @@ public class ReplicaCollie {
                 var availableBrokers =
                     argument.toBrokers.stream()
                         .filter(i -> !keptBrokers.contains(i) && !argument.fromBrokers.contains(i))
-                        .collect(Collectors.toSet());
+                        .collect(Collectors.toList());
                 if (availableBrokers.size() < numberOfMigratedReplicas)
                   throw new IllegalArgumentException(
                       "No enough available brokers! Available: "
@@ -121,7 +122,15 @@ public class ReplicaCollie {
                           + currentBrokers
                           + " removed: "
                           + argument.fromBrokers);
-                brokerMigrate.put(tp.getKey(), Map.entry(currentBrokers, argument.toBrokers));
+                var targetBrokers = new ArrayList<>(keptBrokers);
+                if (numberOfMigratedReplicas < argument.toBrokers.size())
+                  targetBrokers.addAll(
+                      new ArrayList<>(availableBrokers)
+                          .subList(0, argument.fromBrokers.size() - keptBrokers.size()));
+                else
+                  targetBrokers.addAll(
+                      new ArrayList<>(availableBrokers).subList(0, argument.toBrokers.size()));
+                brokerMigrate.put(tp.getKey(), Map.entry(argument.fromBrokers, targetBrokers));
               } else {
                 brokerMigrate.put(tp.getKey(), Map.entry(currentBrokers, currentBrokers));
               }
@@ -176,7 +185,7 @@ public class ReplicaCollie {
   }
 
   static void brokerMigrator(
-      TreeMap<TopicPartition, Map.Entry<Set<Integer>, Set<Integer>>> brokerMigrate,
+      TreeMap<TopicPartition, Map.Entry<List<Integer>, List<Integer>>> brokerMigrate,
       TopicAdmin admin) {
     brokerMigrate.forEach(
         (tp, assignments) -> {
@@ -201,7 +210,7 @@ public class ReplicaCollie {
   }
 
   static TreeMap<TopicPartition, MigratorInfo> getResult(
-      TreeMap<TopicPartition, Map.Entry<Set<Integer>, Set<Integer>>> brokerMigrate,
+      TreeMap<TopicPartition, Map.Entry<List<Integer>, List<Integer>>> brokerMigrate,
       TreeMap<TopicPartition, Map.Entry<Set<String>, Set<String>>> pathMigrate,
       Argument argument,
       TopicAdmin admin) {
@@ -223,8 +232,8 @@ public class ReplicaCollie {
         });
     pathMigrate.forEach(
         (tp, assignments) -> {
-          Set<Integer> fromBroker;
-          Set<Integer> toBroker;
+          List<Integer> fromBroker;
+          List<Integer> toBroker;
           if (argument.fromBrokers.containsAll(argument.toBrokers)) {
             fromBroker = argument.fromBrokers;
             toBroker = fromBroker;
@@ -241,15 +250,24 @@ public class ReplicaCollie {
             result.put(tp, migratorInfo);
           } else {
             if (assignments.getValue().contains(UNKNOWN)) {
-              Set<String> newPath = new HashSet<>();
-              newPath =
-                  !argument.verify
-                      ? admin.replicas(argument.topics).get(tp).stream()
-                          .map(Replica::path)
-                          .filter(path -> !assignments.getKey().contains(path))
-                          .collect(Collectors.toSet())
-                      : Set.of(UNKNOWN);
-              if (assignments.getKey().equals(newPath)) {
+              Set<String> newPath;
+              if (!argument.verify) {
+                var replicas = admin.replicas(argument.topics).get(tp);
+                newPath =
+                    replicas.stream()
+                            .map(Replica::path)
+                            .filter(p -> !assignments.getKey().contains(p))
+                            .collect(Collectors.toSet())
+                            .isEmpty()
+                        ? replicas.stream().map(Replica::path).collect(Collectors.toSet())
+                        : replicas.stream()
+                            .map(Replica::path)
+                            .filter(p -> !assignments.getKey().contains(p))
+                            .collect(Collectors.toSet());
+              } else {
+                newPath = Set.of(UNKNOWN);
+              }
+              if (assignments.getKey().equals(newPath) && fromBroker == toBroker) {
                 if (!argument.verify) result.remove(tp);
               } else {
                 MigratorInfo migratorInfo = new MigratorInfo();
@@ -279,6 +297,7 @@ public class ReplicaCollie {
 
   public static void main(String[] args) throws IOException {
     var argument = org.astraea.argument.Argument.parse(new Argument(), args);
+
     try (var admin = TopicAdmin.of(argument.props())) {
       execute(admin, argument)
           .forEach(
@@ -310,17 +329,13 @@ public class ReplicaCollie {
     @Parameter(
         names = {"--from"},
         description = "Those brokers won't hold any replicas of topics (defined by --topics)",
-        validateWith = IntegerSetField.class,
-        converter = IntegerSetField.class,
         required = true)
-    Set<Integer> fromBrokers;
+    List<Integer> fromBrokers;
 
     @Parameter(
         names = {"--to"},
-        description = "The replicas of topics (defined by --topic) will be moved to those brokers",
-        validateWith = IntegerSetField.class,
-        converter = IntegerSetField.class)
-    Set<Integer> toBrokers = Collections.emptySet();
+        description = "The replicas of topics (defined by --topic) will be moved to those brokers")
+    List<Integer> toBrokers = List.of();
 
     @Parameter(
         names = {"--partitions"},
diff --git a/app/src/main/java/org/astraea/topic/TopicAdmin.java b/app/src/main/java/org/astraea/topic/TopicAdmin.java
index 4f944611..c39c516b 100644
--- a/app/src/main/java/org/astraea/topic/TopicAdmin.java
+++ b/app/src/main/java/org/astraea/topic/TopicAdmin.java
@@ -1,6 +1,7 @@
 package org.astraea.topic;
 
 import java.io.Closeable;
+import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -59,6 +60,12 @@ public interface TopicAdmin extends Closeable {
   /** @return all brokers' ids */
   Set<Integer> brokerIds();
 
+  /**
+   * @parm partitions map of TopicPartition and target brokers
+   * @return true if the leader is change successful changed
+   */
+  Map<TopicPartition, Boolean> changeReplicaLeader(Map<TopicPartition, Integer> partitions);
+
   /**
    * @param topics topic names
    * @param brokersID brokers ID
@@ -70,7 +77,7 @@ public interface TopicAdmin extends Closeable {
    * @param brokers a Set containing broker's ID
    * @return all log directory
    */
-  Map<Integer, Set<String>> brokerFolders(Set<Integer> brokers);
+  Map<Integer, Set<String>> brokerFolders(Collection<Integer> brokers);
 
   /** @return a partition migrator used to move partitions to another broker or folder. */
   Migrator migrator();


Test patch:
diff --git a/app/src/test/java/org/astraea/cost/PeriodicTest.java b/app/src/test/java/org/astraea/cost/PeriodicTest.java
new file mode 100644
index 00000000..603fbe28
--- /dev/null
+++ b/app/src/test/java/org/astraea/cost/PeriodicTest.java
@@ -0,0 +1,37 @@
+package org.astraea.cost;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class PeriodicTest extends Periodic<Map<Integer, Double>> {
+  private double brokerValue = 0.0;
+
+  @Test
+  void testTryUpdate() {
+    var broker1 = tryUpdate(this::testMap);
+    Assertions.assertEquals(broker1.get(0), 0.0);
+    var broker2 = tryUpdate(this::testMap);
+    Assertions.assertEquals(broker2.get(0), 0.0);
+    sleep(1);
+    broker2 = tryUpdate(this::testMap);
+    Assertions.assertEquals(broker2.get(0), 1.0);
+  }
+
+  Map<Integer, Double> testMap() {
+    var broker = new HashMap<Integer, Double>();
+    broker.put(0, brokerValue);
+    brokerValue++;
+    return broker;
+  }
+
+  private static void sleep(int seconds) {
+    try {
+      TimeUnit.SECONDS.sleep(seconds);
+    } catch (InterruptedException e) {
+      throw new RuntimeException(e);
+    }
+  }
+}
diff --git a/app/src/test/java/org/astraea/cost/brokersMetrics/BrokerInputCostTest.java b/app/src/test/java/org/astraea/cost/brokersMetrics/BrokerInputCostTest.java
new file mode 100644
index 00000000..50228a95
--- /dev/null
+++ b/app/src/test/java/org/astraea/cost/brokersMetrics/BrokerInputCostTest.java
@@ -0,0 +1,60 @@
+package org.astraea.cost.brokersMetrics;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import org.astraea.cost.ClusterInfo;
+import org.astraea.cost.FakeClusterInfo;
+import org.astraea.metrics.HasBeanObject;
+import org.astraea.metrics.jmx.BeanObject;
+import org.astraea.metrics.kafka.BrokerTopicMetricsResult;
+import org.astraea.metrics.kafka.KafkaMetrics;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+public class BrokerInputCostTest {
+  @Test
+  void testCost() throws InterruptedException {
+    ClusterInfo clusterInfo = exampleClusterInfo(10000L, 20000L, 5000L);
+
+    var brokerInputCost = new BrokerInputCost();
+    var scores = brokerInputCost.brokerCost(clusterInfo).value();
+    Assertions.assertEquals(0.47, scores.get(1));
+    Assertions.assertEquals(0.63, scores.get(2));
+    Assertions.assertEquals(0.39, scores.get(3));
+    var scores5 = brokerInputCost.brokerCost(clusterInfo).value();
+    Thread.sleep(1000);
+
+    ClusterInfo clusterInfo2 = exampleClusterInfo(55555L, 25352L, 25000L);
+    scores = brokerInputCost.brokerCost(clusterInfo2).value();
+    Assertions.assertEquals(0.55, scores.get(1));
+    Assertions.assertEquals(0.51, scores.get(2));
+    Assertions.assertEquals(0.44, scores.get(3));
+  }
+
+  private ClusterInfo exampleClusterInfo(long in1, long in2, long in3) {
+    var BytesInPerSec1 = mockResult(KafkaMetrics.BrokerTopic.BytesInPerSec.metricName(), in1);
+    var BytesInPerSec2 = mockResult(KafkaMetrics.BrokerTopic.BytesInPerSec.metricName(), in2);
+    var BytesInPerSec3 = mockResult(KafkaMetrics.BrokerTopic.BytesInPerSec.metricName(), in3);
+
+    Collection<HasBeanObject> broker1 = List.of(BytesInPerSec1);
+    Collection<HasBeanObject> broker2 = List.of(BytesInPerSec2);
+    Collection<HasBeanObject> broker3 = List.of(BytesInPerSec3);
+    return new FakeClusterInfo() {
+      @Override
+      public Map<Integer, Collection<HasBeanObject>> allBeans() {
+        return Map.of(1, broker1, 2, broker2, 3, broker3);
+      }
+    };
+  }
+
+  private BrokerTopicMetricsResult mockResult(String name, long count) {
+    var result = Mockito.mock(BrokerTopicMetricsResult.class);
+    var bean = Mockito.mock(BeanObject.class);
+    Mockito.when(result.beanObject()).thenReturn(bean);
+    Mockito.when(bean.getProperties()).thenReturn(Map.of("name", name));
+    Mockito.when(result.count()).thenReturn(count);
+    return result;
+  }
+}
diff --git a/app/src/test/java/org/astraea/cost/brokersMetrics/BrokerOutPutCostTest.java b/app/src/test/java/org/astraea/cost/brokersMetrics/BrokerOutPutCostTest.java
new file mode 100644
index 00000000..1027c3fa
--- /dev/null
+++ b/app/src/test/java/org/astraea/cost/brokersMetrics/BrokerOutPutCostTest.java
@@ -0,0 +1,60 @@
+package org.astraea.cost.brokersMetrics;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import org.astraea.cost.ClusterInfo;
+import org.astraea.cost.FakeClusterInfo;
+import org.astraea.metrics.HasBeanObject;
+import org.astraea.metrics.jmx.BeanObject;
+import org.astraea.metrics.kafka.BrokerTopicMetricsResult;
+import org.astraea.metrics.kafka.KafkaMetrics;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+public class BrokerOutPutCostTest {
+  @Test
+  void testCost() throws InterruptedException {
+    ClusterInfo clusterInfo = exampleClusterInfo(10000L, 20000L, 5000L);
+
+    var brokerOutputCost = new BrokerOutputCost();
+    var scores = brokerOutputCost.brokerCost(clusterInfo).value();
+    Assertions.assertEquals(0.47, scores.get(1));
+    Assertions.assertEquals(0.63, scores.get(2));
+    Assertions.assertEquals(0.39, scores.get(3));
+
+    Thread.sleep(1000);
+
+    ClusterInfo clusterInfo2 = exampleClusterInfo(55555L, 25352L, 25000L);
+    scores = brokerOutputCost.brokerCost(clusterInfo2).value();
+    Assertions.assertEquals(0.55, scores.get(1));
+    Assertions.assertEquals(0.51, scores.get(2));
+    Assertions.assertEquals(0.44, scores.get(3));
+  }
+
+  private ClusterInfo exampleClusterInfo(long out1, long out2, long out3) {
+    var BytesInPerSec1 = mockResult(KafkaMetrics.BrokerTopic.BytesOutPerSec.metricName(), out1);
+    var BytesInPerSec2 = mockResult(KafkaMetrics.BrokerTopic.BytesOutPerSec.metricName(), out2);
+    var BytesInPerSec3 = mockResult(KafkaMetrics.BrokerTopic.BytesOutPerSec.metricName(), out3);
+
+    Collection<HasBeanObject> broker1 = List.of(BytesInPerSec1);
+    Collection<HasBeanObject> broker2 = List.of(BytesInPerSec2);
+    Collection<HasBeanObject> broker3 = List.of(BytesInPerSec3);
+    return new FakeClusterInfo() {
+      @Override
+      public Map<Integer, Collection<HasBeanObject>> allBeans() {
+        return Map.of(1, broker1, 2, broker2, 3, broker3);
+      }
+    };
+  }
+
+  private BrokerTopicMetricsResult mockResult(String name, long count) {
+    var result = Mockito.mock(BrokerTopicMetricsResult.class);
+    var bean = Mockito.mock(BeanObject.class);
+    Mockito.when(result.beanObject()).thenReturn(bean);
+    Mockito.when(bean.getProperties()).thenReturn(Map.of("name", name));
+    Mockito.when(result.count()).thenReturn(count);
+    return result;
+  }
+}
diff --git a/app/src/test/java/org/astraea/cost/brokersMetrics/CpuCostTest.java b/app/src/test/java/org/astraea/cost/brokersMetrics/CpuCostTest.java
new file mode 100644
index 00000000..67884c5a
--- /dev/null
+++ b/app/src/test/java/org/astraea/cost/brokersMetrics/CpuCostTest.java
@@ -0,0 +1,99 @@
+package org.astraea.cost.brokersMetrics;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.astraea.cost.ClusterInfo;
+import org.astraea.cost.FakeClusterInfo;
+import org.astraea.cost.NodeInfo;
+import org.astraea.cost.PartitionInfo;
+import org.astraea.metrics.HasBeanObject;
+import org.astraea.metrics.java.OperatingSystemInfo;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+public class CpuCostTest {
+  @Test
+  void testCost() throws InterruptedException {
+    var cpuUsage1 = mockResult(0.5);
+    var cpuUsage2 = mockResult(0.81);
+    var cpuUsage3 = mockResult(0.62);
+
+    Collection<HasBeanObject> broker1 = List.of(cpuUsage1);
+    Collection<HasBeanObject> broker2 = List.of(cpuUsage2);
+    Collection<HasBeanObject> broker3 = List.of(cpuUsage3);
+    ClusterInfo clusterInfo =
+        new FakeClusterInfo() {
+          @Override
+          public Map<Integer, Collection<HasBeanObject>> allBeans() {
+            return Map.of(1, broker1, 2, broker2, 3, broker3);
+          }
+
+          @Override
+          public Set<String> topics() {
+            return Set.of("t");
+          }
+
+          @Override
+          public List<PartitionInfo> availablePartitions(String topic) {
+            return List.of(
+                PartitionInfo.of(
+                    "t", 0, NodeInfo.of(1, "host1", 9092), List.of(), List.of(), List.of()),
+                PartitionInfo.of(
+                    "t", 0, NodeInfo.of(2, "host2", 9092), List.of(), List.of(), List.of()),
+                PartitionInfo.of(
+                    "t", 0, NodeInfo.of(3, "host3", 9092), List.of(), List.of(), List.of()));
+          }
+        };
+
+    var cpuCost = new CpuCost();
+    var scores = cpuCost.brokerCost(clusterInfo).value();
+    Assertions.assertEquals(0.39, scores.get(1));
+    Assertions.assertEquals(0.63, scores.get(2));
+    Assertions.assertEquals(0.48, scores.get(3));
+
+    Thread.sleep(1000);
+    cpuUsage1 = mockResult(0.8);
+    cpuUsage2 = mockResult(0.5);
+    cpuUsage3 = mockResult(0.3);
+
+    Collection<HasBeanObject> broker12 = List.of(cpuUsage1);
+    Collection<HasBeanObject> broker22 = List.of(cpuUsage2);
+    Collection<HasBeanObject> broker32 = List.of(cpuUsage3);
+    ClusterInfo clusterInfo2 =
+        new FakeClusterInfo() {
+          @Override
+          public Map<Integer, Collection<HasBeanObject>> allBeans() {
+            return Map.of(1, broker12, 2, broker22, 3, broker32);
+          }
+
+          @Override
+          public Set<String> topics() {
+            return Set.of("t");
+          }
+
+          @Override
+          public List<PartitionInfo> availablePartitions(String topic) {
+            return List.of(
+                PartitionInfo.of(
+                    "t", 0, NodeInfo.of(1, "host1", 9092), List.of(), List.of(), List.of()),
+                PartitionInfo.of(
+                    "t", 0, NodeInfo.of(2, "host2", 9092), List.of(), List.of(), List.of()),
+                PartitionInfo.of(
+                    "t", 0, NodeInfo.of(3, "host3", 9092), List.of(), List.of(), List.of()));
+          }
+        };
+    scores = cpuCost.brokerCost(clusterInfo2).value();
+    Assertions.assertEquals(0.51, scores.get(1));
+    Assertions.assertEquals(0.55, scores.get(2));
+    Assertions.assertEquals(0.44, scores.get(3));
+  }
+
+  private OperatingSystemInfo mockResult(double usage) {
+    var cpu = Mockito.mock(OperatingSystemInfo.class);
+    Mockito.when(cpu.systemCpuLoad()).thenReturn(usage);
+    return cpu;
+  }
+}
diff --git a/app/src/test/java/org/astraea/cost/brokersMetrics/MemoryCostTest.java b/app/src/test/java/org/astraea/cost/brokersMetrics/MemoryCostTest.java
new file mode 100644
index 00000000..393ff05e
--- /dev/null
+++ b/app/src/test/java/org/astraea/cost/brokersMetrics/MemoryCostTest.java
@@ -0,0 +1,103 @@
+package org.astraea.cost.brokersMetrics;
+
+import java.lang.management.MemoryUsage;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.astraea.cost.ClusterInfo;
+import org.astraea.cost.FakeClusterInfo;
+import org.astraea.cost.NodeInfo;
+import org.astraea.cost.PartitionInfo;
+import org.astraea.metrics.HasBeanObject;
+import org.astraea.metrics.java.HasJvmMemory;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+public class MemoryCostTest {
+  @Test
+  void testCost() throws InterruptedException {
+    var jvmMemory1 = mockResult(50L, 100L);
+    var jvmMemory2 = mockResult(81L, 100L);
+    var jvmMemory3 = mockResult(80L, 150L);
+
+    Collection<HasBeanObject> broker1 = List.of(jvmMemory1);
+    Collection<HasBeanObject> broker2 = List.of(jvmMemory2);
+    Collection<HasBeanObject> broker3 = List.of(jvmMemory3);
+    ClusterInfo clusterInfo =
+        new FakeClusterInfo() {
+          @Override
+          public Map<Integer, Collection<HasBeanObject>> allBeans() {
+            return Map.of(1, broker1, 2, broker2, 3, broker3);
+          }
+
+          @Override
+          public Set<String> topics() {
+            return Set.of("t");
+          }
+
+          @Override
+          public List<PartitionInfo> availablePartitions(String topic) {
+            return List.of(
+                PartitionInfo.of(
+                    "t", 0, NodeInfo.of(1, "host1", 9092), List.of(), List.of(), List.of()),
+                PartitionInfo.of(
+                    "t", 0, NodeInfo.of(2, "host2", 9092), List.of(), List.of(), List.of()),
+                PartitionInfo.of(
+                    "t", 0, NodeInfo.of(3, "host3", 9092), List.of(), List.of(), List.of()));
+          }
+        };
+
+    var memoryCost = new MemoryCost();
+    var scores = memoryCost.brokerCost(clusterInfo).value();
+    Assertions.assertEquals(0.42, scores.get(1));
+    Assertions.assertEquals(0.64, scores.get(2));
+    Assertions.assertEquals(0.44, scores.get(3));
+
+    Thread.sleep(1000);
+    jvmMemory1 = mockResult(50L, 50L);
+    jvmMemory2 = mockResult(30L, 100L);
+    jvmMemory3 = mockResult(80L, 150L);
+
+    Collection<HasBeanObject> broker12 = List.of(jvmMemory1);
+    Collection<HasBeanObject> broker22 = List.of(jvmMemory2);
+    Collection<HasBeanObject> broker32 = List.of(jvmMemory3);
+    ClusterInfo clusterInfo2 =
+        new FakeClusterInfo() {
+          @Override
+          public Map<Integer, Collection<HasBeanObject>> allBeans() {
+            return Map.of(1, broker12, 2, broker22, 3, broker32);
+          }
+
+          @Override
+          public Set<String> topics() {
+            return Set.of("t");
+          }
+
+          @Override
+          public List<PartitionInfo> availablePartitions(String topic) {
+            return List.of(
+                PartitionInfo.of(
+                    "t", 0, NodeInfo.of(1, "host1", 9092), List.of(), List.of(), List.of()),
+                PartitionInfo.of(
+                    "t", 0, NodeInfo.of(2, "host2", 9092), List.of(), List.of(), List.of()),
+                PartitionInfo.of(
+                    "t", 0, NodeInfo.of(3, "host3", 9092), List.of(), List.of(), List.of()));
+          }
+        };
+    scores = memoryCost.brokerCost(clusterInfo2).value();
+    Assertions.assertEquals(0.53, scores.get(1));
+    Assertions.assertEquals(0.52, scores.get(2));
+    Assertions.assertEquals(0.45, scores.get(3));
+  }
+
+  private HasJvmMemory mockResult(long used, long max) {
+    var jvmMemory = Mockito.mock(HasJvmMemory.class);
+    var memoryUsage = Mockito.mock(MemoryUsage.class);
+    Mockito.when(jvmMemory.heapMemoryUsage()).thenReturn(memoryUsage);
+    Mockito.when(memoryUsage.getUsed()).thenReturn(used);
+    Mockito.when(memoryUsage.getMax()).thenReturn(max);
+    return jvmMemory;
+  }
+}
diff --git a/app/src/test/java/org/astraea/topic/ReplicaCollieTest.java b/app/src/test/java/org/astraea/topic/ReplicaCollieTest.java
index 729d3cb0..8679633a 100644
--- a/app/src/test/java/org/astraea/topic/ReplicaCollieTest.java
+++ b/app/src/test/java/org/astraea/topic/ReplicaCollieTest.java
@@ -4,6 +4,7 @@ import static org.junit.jupiter.api.condition.OS.WINDOWS;
 
 import java.io.IOException;
 import java.util.Comparator;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.TreeMap;
@@ -48,14 +49,14 @@ public class ReplicaCollieTest extends RequireBrokerCluster {
       var brokerSource =
           partitionReplicas.get(new TopicPartition(topicName, 0)).stream()
               .map(Replica::broker)
-              .collect(Collectors.toSet());
+              .collect(Collectors.toList());
       var brokerSink =
           topicAdmin.brokerIds().stream().filter(b -> !brokerSource.contains(b)).iterator().next();
-      TreeMap<TopicPartition, Map.Entry<Set<Integer>, Set<Integer>>> brokerMigrate =
+      TreeMap<TopicPartition, Map.Entry<List<Integer>, List<Integer>>> brokerMigrate =
           new TreeMap<>(
               Comparator.comparing(TopicPartition::topic).thenComparing(TopicPartition::partition));
       brokerMigrate.put(
-          new TopicPartition(topicName, 0), Map.entry(brokerSource, Set.of(brokerSink)));
+          new TopicPartition(topicName, 0), Map.entry(brokerSource, List.of(brokerSink)));
       Assertions.assertEquals(
           topicAdmin.replicas(Set.of(topicName)).get(new TopicPartition(topicName, 0)).size(), 2);
       ReplicaCollie.brokerMigrator(brokerMigrate, topicAdmin);
@@ -135,14 +136,14 @@ public class ReplicaCollieTest extends RequireBrokerCluster {
               .filter(Replica::isCurrent)
               .collect(Collectors.toList());
       Assertions.assertEquals(1, replicas.size());
-      var badBroker = replicas.stream().map(Replica::broker).collect(Collectors.toSet());
+      var badBroker = replicas.stream().map(Replica::broker).collect(Collectors.toList());
       var targetBroker =
           topicAdmin.brokerIds().stream()
               .filter(b -> !badBroker.contains(b))
-              .collect(Collectors.toSet());
+              .collect(Collectors.toList());
       var argument = new ReplicaCollie.Argument();
       argument.fromBrokers = badBroker;
-      argument.toBrokers = targetBroker;
+      argument.toBrokers = targetBroker.subList(0, 1);
       argument.brokers = bootstrapServers();
       argument.topics = Set.of(topicName);
       argument.partitions = Set.of(0);
@@ -161,8 +162,7 @@ public class ReplicaCollieTest extends RequireBrokerCluster {
               for (var index = 0; index != rs.size(); ++index)
                 Assertions.assertEquals(rs.get(index), currentRs.get(index));
             });
-        Assertions.assertEquals(targetBroker, assignment.brokerSink);
-
+        Assertions.assertEquals(argument.toBrokers, assignment.brokerSink);
       } else {
         Utils.waitFor(
             () -> {
diff --git a/app/src/test/java/org/astraea/topic/ReplicaSyncingMonitorIntegrationTest.java b/app/src/test/java/org/astraea/topic/ReplicaSyncingMonitorIntegrationTest.java
index 8685ea17..a694460a 100644
--- a/app/src/test/java/org/astraea/topic/ReplicaSyncingMonitorIntegrationTest.java
+++ b/app/src/test/java/org/astraea/topic/ReplicaSyncingMonitorIntegrationTest.java
@@ -4,6 +4,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertSame;
 
 import java.io.IOException;
+import java.util.List;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
@@ -50,7 +51,7 @@ class ReplicaSyncingMonitorIntegrationTest extends RequireBrokerCluster {
       Thread executionThread =
           new Thread(
               () -> {
-                topicAdmin.migrator().partition(TOPIC_NAME, 0).moveTo(Set.of(moveToBroker));
+                topicAdmin.migrator().partition(TOPIC_NAME, 0).moveTo(List.of(moveToBroker));
                 ReplicaSyncingMonitor.execute(
                     topicAdmin,
                     org.astraea.argument.Argument.parse(
diff --git a/app/src/test/java/org/astraea/topic/TopicAdminTest.java b/app/src/test/java/org/astraea/topic/TopicAdminTest.java
index 463a8495..4f29ce77 100644
--- a/app/src/test/java/org/astraea/topic/TopicAdminTest.java
+++ b/app/src/test/java/org/astraea/topic/TopicAdminTest.java
@@ -4,6 +4,7 @@ import static org.junit.jupiter.api.condition.OS.WINDOWS;
 
 import java.time.Duration;
 import java.util.Collection;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
@@ -154,6 +155,8 @@ public class TopicAdminTest extends RequireBrokerCluster {
   }
 
   @Test
+  // There is a problem when migrating the log folder under Windows because the migrated source
+  // cannot be deleted, so disabled this test on Windows for now.
   @DisabledOnOs(WINDOWS)
   void testMigrateSinglePartition() throws InterruptedException {
     var topicName = "testMigrateSinglePartition";
@@ -162,7 +165,7 @@ public class TopicAdminTest extends RequireBrokerCluster {
       // wait for syncing topic creation
       TimeUnit.SECONDS.sleep(5);
       var broker = topicAdmin.brokerIds().iterator().next();
-      topicAdmin.migrator().partition(topicName, 0).moveTo(Set.of(broker));
+      topicAdmin.migrator().partition(topicName, 0).moveTo(List.of(broker));
       Utils.waitFor(
           () -> {
             var replicas = topicAdmin.replicas(Set.of(topicName));
@@ -205,6 +208,38 @@ public class TopicAdminTest extends RequireBrokerCluster {
     }
   }
 
+  @Test
+  @DisabledOnOs(WINDOWS)
+  void testChangeReplicaLeader() throws InterruptedException {
+    var topicName = "testChangeReplicaLeader";
+    var tp = new TopicPartition(topicName, 0);
+    try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
+      topicAdmin
+          .creator()
+          .topic(topicName)
+          .numberOfPartitions(1)
+          .numberOfReplicas((short) 2)
+          .create();
+      // wait for syncing topic creation
+      TimeUnit.SECONDS.sleep(5);
+      topicAdmin.replicas(Set.of(topicName)).get(tp);
+      var replicas = topicAdmin.replicas(Set.of(topicName)).get(tp);
+      var oldLeader =
+          replicas.stream().filter(Replica::leader).collect(Collectors.toList()).get(0).broker();
+      var newLeader =
+          replicas.stream().filter(b -> !b.leader()).collect(Collectors.toList()).get(0).broker();
+      topicAdmin.changeReplicaLeader(Map.of(tp, newLeader));
+      Assertions.assertNotEquals(oldLeader, newLeader);
+      Utils.waitFor(
+          () ->
+              topicAdmin.replicas(Set.of(topicName)).get(tp).stream()
+                  .filter(r -> r.broker() == newLeader)
+                  .iterator()
+                  .next()
+                  .leader());
+    }
+  }
+
   @Test
   @DisabledOnOs(WINDOWS)
   void testMigrateAllPartitions() throws InterruptedException {
@@ -214,7 +249,7 @@ public class TopicAdminTest extends RequireBrokerCluster {
       // wait for syncing topic creation
       TimeUnit.SECONDS.sleep(5);
       var broker = topicAdmin.brokerIds().iterator().next();
-      topicAdmin.migrator().topic(topicName).moveTo(Set.of(broker));
+      topicAdmin.migrator().topic(topicName).moveTo(List.of(broker));
       Utils.waitFor(
           () -> {
             var replicas = topicAdmin.replicas(Set.of(topicName));