Repository: skiptests/astraea
Pull request #403 (2022-06-11 06:54:51)

Implementation of `BalancePlanExecutor`
這個 PR 實作搬移計劃的執行框架，順便提供一個最簡易的實作 `StraightPlanExecutor`。

這個 PR 相依這兩個 PR。

* #371 
* #400 

等上面兩個合併才能從 draft 轉正

garyparrot: https://github.com/skiptests/astraea/runs/6980662834?check_suite_focus=true#step:5:1830

Potential flaky test
garyparrot: 現在有一個測試錯了 https://github.com/skiptests/astraea/runs/6998761501?check_suite_focus=true#step:5:462

我看了一下，似乎是 `MigratorImpl` 變更 Placemnet 後，在經過一段時間之前，`Admin#repliacs` 回傳的結果會在 `變更前` 和 `變更後` 兩者狀態之間彈跳。下面這個程式碼可以反映錯誤

```java
  @RepeatedTest(value = 300)
  void testAllocationBouncing() throws InterruptedException {
    try (var admin = Admin.of(bootstrapServers())) {
      // Create topic
      var topicName = "Topic_" + Utils.randomString();
      admin.creator()
              .topic(topicName)
              .numberOfPartitions(1)
              .numberOfReplicas((short)3)
              .create();
      TimeUnit.MILLISECONDS.sleep(500);

      // Alter placement
      admin.migrator()
              .topic(topicName)
              .moveTo(List.of(2,1,0));
      // TimeUnit.MILLISECONDS.sleep(3000);

      // Wait Until the API result response
      Supplier<Boolean> alterPlacementWorks = () -> admin.replicas(Set.of(topicName))
                      .get(new TopicPartition(topicName, 0))
                      .stream()
                      .map(x -> x.broker())
                      .collect(Collectors.toUnmodifiableList()).equals(List.of(2,1,0));
      Utils.waitFor(alterPlacementWorks);

      // Let's see if the result will bounce :p
      final List<List<Integer>> repeatCalls = IntStream.range(0, 20)
              .mapToObj(i -> admin.replicas(Set.of(topicName)))
              .map(logs -> logs.get(new TopicPartition(topicName, 0)))
              .map(logs -> logs.stream().map(Replica::broker).collect(Collectors.toUnmodifiableList()))
              .peek(System.err::println)
              .collect(Collectors.toUnmodifiableList());

      Assertions.assertEquals(Collections.nCopies(20, List.of(2,1,0)), repeatCalls);
    }
  }

```

這個問題嚴重的時候可能會影響我們對 preferred leader 的判斷。我後面會讓 Leader Election 的 await 程式碼稍微做一些保護，可能在經過一定時間之前不能採用結果。
chia7712: > 這個問題嚴重的時候可能會影響我們對 preferred leader 的判斷。我後面會讓 Leader Election 的 await 程式碼稍微做一些保護，可能在經過一定時間之前不能採用結果。

這應該是kafka cluster的節點們還沒完成同步，`Admin#replicas`的結果是由多個請求來組成，因此在叢集尚未完成同步前就會撈到“中間”狀態的結果
garyparrot: 這個 PR 目前有包含 #440 的東西 (https://github.com/skiptests/astraea/pull/403/commits/97fcd00432a1744f17045052e75172b985482a40)

Patch:
diff --git a/app/src/main/java/org/astraea/app/balancer/executor/LeaderElectionTask.java b/app/src/main/java/org/astraea/app/balancer/executor/LeaderElectionTask.java
new file mode 100644
index 00000000..4816567f
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/balancer/executor/LeaderElectionTask.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.balancer.executor;
+
+import java.util.concurrent.CompletableFuture;
+import org.astraea.app.admin.TopicPartition;
+
+public class LeaderElectionTask {
+
+  private final TopicPartition topicPartition;
+  private final CompletableFuture<Boolean> completableFuture;
+
+  public LeaderElectionTask(RebalanceAdmin admin, TopicPartition topicPartition) {
+    this.topicPartition = topicPartition;
+    this.completableFuture = admin.waitPreferredLeaderSynced(topicPartition);
+  }
+
+  public TopicPartition topicPartition() {
+    return topicPartition;
+  }
+
+  public CompletableFuture<Boolean> completableFuture() {
+    return completableFuture;
+  }
+}
diff --git a/app/src/main/java/org/astraea/app/balancer/executor/RebalanceAdmin.java b/app/src/main/java/org/astraea/app/balancer/executor/RebalanceAdmin.java
new file mode 100644
index 00000000..c5cfdbab
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/balancer/executor/RebalanceAdmin.java
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.balancer.executor;
+
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Predicate;
+import org.astraea.app.admin.Admin;
+import org.astraea.app.admin.TopicPartition;
+import org.astraea.app.admin.TopicPartitionReplica;
+import org.astraea.app.balancer.log.LogPlacement;
+import org.astraea.app.cost.ClusterInfo;
+
+/**
+ * The wrapper of {@link Admin}. Offer only the essential functionalities & some utilities to
+ * perform rebalance operation.
+ */
+public interface RebalanceAdmin {
+
+  /**
+   * Construct an implementation of {@link RebalanceAdmin}
+   *
+   * @param topicFilter to determine which topics are permitted for balance operation
+   * @param admin the actual {@link Admin} implementation
+   */
+  static RebalanceAdmin of(Admin admin, Predicate<String> topicFilter) {
+    return new RebalanceAdminImpl(topicFilter, admin);
+  }
+
+  /**
+   * Attempt to migrate the target topic/partition to the given replica log placement state. This
+   * method will perform both replica list migration and data directory migration. This method will
+   * return after triggering the migration. It won't wait until the migration processes are
+   * fulfilled.
+   *
+   * @param topicPartition the topic/partition to perform migration
+   * @param expectedPlacement the expected placement after this request accomplished
+   * @return a list of task trackers regarding each log
+   */
+  List<ReplicaMigrationTask> alterReplicaPlacements(
+      TopicPartition topicPartition, List<LogPlacement> expectedPlacement);
+
+  /** @return a {@link CompletableFuture} that indicate the specific log has become synced. */
+  CompletableFuture<Boolean> waitLogSynced(TopicPartitionReplica log, Duration timeout);
+
+  /** @return a {@link CompletableFuture} that indicate the specific log has become synced. */
+  default CompletableFuture<Boolean> waitLogSynced(TopicPartitionReplica log) {
+    return waitLogSynced(log, ChronoUnit.FOREVER.getDuration());
+  }
+
+  /**
+   * @return a {@link CompletableFuture} that indicate the specific topic/partition has its
+   *     preferred leader becomes the actual leader.
+   */
+  CompletableFuture<Boolean> waitPreferredLeaderSynced(
+      TopicPartition topicPartition, Duration timeout);
+
+  /**
+   * @return a {@link CompletableFuture} that indicate the specific topic/partition has its
+   *     preferred leader becomes the actual leader.
+   */
+  default CompletableFuture<Boolean> waitPreferredLeaderSynced(TopicPartition topicPartition) {
+    return waitPreferredLeaderSynced(topicPartition, ChronoUnit.FOREVER.getDuration());
+  }
+
+  /**
+   * Perform preferred leader election for specific topic/partition.
+   *
+   * @param topicPartition the topic/partition to trigger preferred leader election
+   * @return a task tracker to track the election progress.
+   */
+  LeaderElectionTask leaderElection(TopicPartition topicPartition);
+
+  ClusterInfo clusterInfo();
+
+  /**
+   * @return a {@link Predicate<String>} indicate which topic name is allowed to operate by this
+   *     {@link RebalanceAdmin}.
+   */
+  Predicate<String> topicFilter();
+
+  // TODO: add method to apply reassignment bandwidth throttle.
+  // TODO: add method to fetch topic configuration
+  // TODO: add method to fetch broker configuration
+}
diff --git a/app/src/main/java/org/astraea/app/balancer/executor/RebalanceAdminImpl.java b/app/src/main/java/org/astraea/app/balancer/executor/RebalanceAdminImpl.java
new file mode 100644
index 00000000..891bb3e6
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/balancer/executor/RebalanceAdminImpl.java
@@ -0,0 +1,259 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.balancer.executor;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+import org.astraea.app.admin.Admin;
+import org.astraea.app.admin.Replica;
+import org.astraea.app.admin.TopicPartition;
+import org.astraea.app.admin.TopicPartitionReplica;
+import org.astraea.app.balancer.log.LogPlacement;
+import org.astraea.app.common.Utils;
+import org.astraea.app.cost.ClusterInfo;
+
+class RebalanceAdminImpl implements RebalanceAdmin {
+
+  private final Predicate<String> topicFilter;
+  private final Admin admin;
+
+  /**
+   * Construct an implementation of {@link RebalanceAdmin}
+   *
+   * @param topicFilter to determine which topics are permitted for balance operation
+   * @param admin the actual {@link Admin} implementation
+   */
+  public RebalanceAdminImpl(Predicate<String> topicFilter, Admin admin) {
+    this.topicFilter = topicFilter;
+    this.admin = admin;
+  }
+
+  private void ensureTopicPermitted(String topic) {
+    if (!topicFilter.test(topic))
+      throw new IllegalArgumentException("Operation to topic \"" + topic + "\" is not permitted");
+  }
+
+  private List<LogPlacement> fetchCurrentPlacement(TopicPartition topicPartition) {
+    ensureTopicPermitted(topicPartition.topic());
+    return admin.replicas(Set.of(topicPartition.topic())).get(topicPartition).stream()
+        .map(replica -> LogPlacement.of(replica.broker(), replica.path()))
+        .collect(Collectors.toUnmodifiableList());
+  }
+
+  /**
+   * Declare the preferred data directory at certain brokers.
+   *
+   * <p>By default, upon a new log creation with JBOD enabled broker. Kafka broker will pick up a
+   * data directory that has the fewest log maintained to be the data directory for the new log.
+   * This method declares the preferred data directory for a specific topic/partition on the certain
+   * broker. Upon the new log creation for the specific topic/partition on the certain broker. The
+   * preferred data directory will be used as the data directory for the new log, which replaces the
+   * default approach. This gives you the control to decide which data directory the replica log you
+   * are about to migrate will be.
+   *
+   * @param topicPartition the topic/partition to declare preferred data directory
+   * @param preferredPlacements the replica placements with their desired data directory at certain
+   *     brokers
+   */
+  private void declarePreferredDataDirectories(
+      TopicPartition topicPartition, List<LogPlacement> preferredPlacements) {
+    ensureTopicPermitted(topicPartition.topic());
+
+    final var currentPlacement = fetchCurrentPlacement(topicPartition);
+
+    final var currentBrokerAllocation =
+        currentPlacement.stream().map(LogPlacement::broker).collect(Collectors.toUnmodifiableSet());
+
+    // this operation is not supposed to trigger a log movement. But there might be a small window
+    // of time to actually trigger it (race condition).
+    final var declareMap =
+        preferredPlacements.stream()
+            .filter(futurePlacement -> !currentBrokerAllocation.contains(futurePlacement.broker()))
+            .filter(futurePlacement -> futurePlacement.logDirectory().isPresent())
+            .collect(
+                Collectors.toUnmodifiableMap(
+                    LogPlacement::broker, x -> x.logDirectory().orElseThrow()));
+
+    admin
+        .migrator()
+        .partition(topicPartition.topic(), topicPartition.partition())
+        .declarePreferredDir(declareMap);
+  }
+
+  @Override
+  public List<ReplicaMigrationTask> alterReplicaPlacements(
+      TopicPartition topicPartition, List<LogPlacement> expectedPlacement) {
+    ensureTopicPermitted(topicPartition.topic());
+
+    // ensure replica will be placed in the correct data directory at destination broker.
+    declarePreferredDataDirectories(topicPartition, expectedPlacement);
+
+    var currentReplicaBrokers =
+        fetchCurrentPlacement(topicPartition).stream()
+            .map(LogPlacement::broker)
+            .collect(Collectors.toUnmodifiableSet());
+
+    // do cross broker migration
+    admin
+        .migrator()
+        .partition(topicPartition.topic(), topicPartition.partition())
+        .moveTo(
+            expectedPlacement.stream()
+                .map(LogPlacement::broker)
+                .collect(Collectors.toUnmodifiableList()));
+    // do inter-data-directories migration
+    var forCrossDirMigration =
+        expectedPlacement.stream()
+            .filter(placement -> currentReplicaBrokers.contains(placement.broker()))
+            .filter(placement -> placement.logDirectory().isPresent())
+            .collect(
+                Collectors.toUnmodifiableMap(
+                    LogPlacement::broker, placement -> placement.logDirectory().orElseThrow()));
+    admin
+        .migrator()
+        .partition(topicPartition.topic(), topicPartition.partition())
+        .moveTo(forCrossDirMigration);
+
+    return expectedPlacement.stream()
+        .map(
+            log ->
+                new TopicPartitionReplica(
+                    topicPartition.topic(), topicPartition.partition(), log.broker()))
+        .map(log -> new ReplicaMigrationTask(this, log))
+        .collect(Collectors.toUnmodifiableList());
+  }
+
+  private long getEndTime(Duration timeout) {
+    try {
+      return Math.addExact(System.currentTimeMillis(), timeout.toMillis());
+    } catch (ArithmeticException e) {
+      return Long.MAX_VALUE;
+    }
+  }
+
+  @Override
+  public CompletableFuture<Boolean> waitLogSynced(TopicPartitionReplica log, Duration timeout) {
+    ensureTopicPermitted(log.topic());
+
+    return CompletableFuture.supplyAsync(
+        () -> {
+          // due to the state consistency issue in Kafka broker design. the cluster state returned
+          // from the API might bounce between the `old state` and the `new state` during the very
+          // beginning and accomplishment of the cluster state alteration API. to fix this we use
+          // debounce technique, to ensure the target condition is held over a few successive tries,
+          // which mean the cluster state alteration is considered stable.
+          var debounce = 2;
+          var endTime = getEndTime(timeout);
+          while (!Thread.currentThread().isInterrupted()) {
+            boolean synced =
+                admin.replicas(Set.of(log.topic())).entrySet().stream()
+                    .filter(x -> x.getKey().partition() == log.partition())
+                    .filter(x -> x.getKey().topic().equals(log.topic()))
+                    .flatMap(x -> x.getValue().stream())
+                    .filter(x -> x.broker() == log.brokerId())
+                    .findFirst()
+                    .map(x -> x.inSync() && !x.isFuture())
+                    .orElse(false);
+            // debounce & retrial interval
+            Utils.packException(() -> TimeUnit.MILLISECONDS.sleep(retrialTime.get().toMillis()));
+            debounce = synced ? (debounce - 1) : 2;
+            // synced
+            if (synced && debounce <= 0) return true;
+            // timeout
+            if (System.currentTimeMillis() > endTime) return false;
+          }
+          return false;
+        });
+  }
+
+  @Override
+  public CompletableFuture<Boolean> waitPreferredLeaderSynced(
+      TopicPartition topicPartition, Duration timeout) {
+    ensureTopicPermitted(topicPartition.topic());
+
+    return CompletableFuture.supplyAsync(
+        () -> {
+          // due to the state consistency issue in Kafka broker design. the cluster state returned
+          // from the API might bounce between the `old state` and the `new state` during the very
+          // beginning and accomplishment of the cluster state alteration API. to fix this we use
+          // debounce technique, to ensure the target condition is held over a few successive tries,
+          // which mean the cluster state alteration is considered stable.
+          var debounce = 2;
+          var endTime = getEndTime(timeout);
+          while (!Thread.currentThread().isInterrupted()) {
+            var synced =
+                admin.replicas(Set.of(topicPartition.topic())).entrySet().stream()
+                    .filter(x -> x.getKey().equals(topicPartition))
+                    .findFirst()
+                    .map(Map.Entry::getValue)
+                    .map(
+                        replicas -> {
+                          var preferred =
+                              replicas.stream()
+                                  .filter(Replica::isPreferredLeader)
+                                  .findFirst()
+                                  .orElseThrow();
+                          return preferred.leader();
+                        })
+                    .orElseThrow();
+            // debounce & retrial interval
+            Utils.sleep(retrialTime.get());
+            debounce = synced ? (debounce - 1) : 2;
+            // synced
+            if (synced && debounce <= 0) return true;
+            // timeout
+            if (System.currentTimeMillis() > endTime) return false;
+          }
+          return false;
+        });
+  }
+
+  @Override
+  public LeaderElectionTask leaderElection(TopicPartition topicPartition) {
+    ensureTopicPermitted(topicPartition.topic());
+
+    admin.preferredLeaderElection(topicPartition);
+
+    return new LeaderElectionTask(this, topicPartition);
+  }
+
+  @Override
+  public ClusterInfo clusterInfo() {
+    return admin.clusterInfo(
+        admin.topicNames().stream().filter(topicFilter).collect(Collectors.toUnmodifiableSet()));
+  }
+
+  @Override
+  public Predicate<String> topicFilter() {
+    return topicFilter;
+  }
+
+  private static final AtomicReference<Duration> retrialTime =
+      new AtomicReference<>(Duration.ofSeconds(1));
+
+  // visible for test
+  static void changeRetrialTime(Duration newDebounceTime) {
+    retrialTime.set(newDebounceTime);
+  }
+}
diff --git a/app/src/main/java/org/astraea/app/balancer/executor/RebalancePlanExecutor.java b/app/src/main/java/org/astraea/app/balancer/executor/RebalancePlanExecutor.java
new file mode 100644
index 00000000..ba064e58
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/balancer/executor/RebalancePlanExecutor.java
@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.balancer.executor;
+
+import org.astraea.app.balancer.log.ClusterLogAllocation;
+
+/** This class associate with the logic of fulfill given rebalance plan. */
+public interface RebalancePlanExecutor {
+
+  /** This method responsible for fulfill a rebalance plan. */
+  void run(RebalanceAdmin rebalanceAdmin, ClusterLogAllocation targetAllocation);
+}
diff --git a/app/src/main/java/org/astraea/app/balancer/executor/ReplicaMigrationTask.java b/app/src/main/java/org/astraea/app/balancer/executor/ReplicaMigrationTask.java
new file mode 100644
index 00000000..47488594
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/balancer/executor/ReplicaMigrationTask.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.balancer.executor;
+
+import java.util.concurrent.CompletableFuture;
+import org.astraea.app.admin.TopicPartitionReplica;
+
+public class ReplicaMigrationTask {
+
+  private final TopicPartitionReplica log;
+  private final CompletableFuture<Boolean> completableFuture;
+
+  public ReplicaMigrationTask(RebalanceAdmin admin, TopicPartitionReplica log) {
+    this.log = log;
+    this.completableFuture = admin.waitLogSynced(log);
+  }
+
+  public TopicPartitionReplica log() {
+    return log;
+  }
+
+  public CompletableFuture<Boolean> completableFuture() {
+    return completableFuture;
+  }
+}
diff --git a/app/src/main/java/org/astraea/app/balancer/executor/StraightPlanExecutor.java b/app/src/main/java/org/astraea/app/balancer/executor/StraightPlanExecutor.java
new file mode 100644
index 00000000..e11ff360
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/balancer/executor/StraightPlanExecutor.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.balancer.executor;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import org.astraea.app.admin.TopicPartition;
+import org.astraea.app.balancer.log.ClusterLogAllocation;
+import org.astraea.app.balancer.log.LayeredClusterLogAllocation;
+
+/** Execute every possible migration immediately. */
+public class StraightPlanExecutor implements RebalancePlanExecutor {
+
+  public StraightPlanExecutor() {}
+
+  @Override
+  public void run(RebalanceAdmin rebalanceAdmin, ClusterLogAllocation logAllocation) {
+    final var clusterInfo = rebalanceAdmin.clusterInfo();
+    final var currentLogAllocation = LayeredClusterLogAllocation.of(clusterInfo);
+    final var migrationTargets =
+        ClusterLogAllocation.findNonFulfilledAllocation(currentLogAllocation, logAllocation);
+
+    var executeReplicaMigration =
+        (Function<TopicPartition, List<ReplicaMigrationTask>>)
+            (topicPartition) ->
+                rebalanceAdmin.alterReplicaPlacements(
+                    topicPartition, logAllocation.logPlacements(topicPartition));
+
+    // do log migration
+    migrationTargets.stream()
+        .map(executeReplicaMigration)
+        .flatMap(Collection::stream)
+        .map(ReplicaMigrationTask::completableFuture)
+        .collect(Collectors.toUnmodifiableSet())
+        .forEach(CompletableFuture::join);
+
+    // do leader election
+    migrationTargets.stream()
+        .map(rebalanceAdmin::leaderElection)
+        .map(LeaderElectionTask::completableFuture)
+        .collect(Collectors.toUnmodifiableSet())
+        .forEach(CompletableFuture::join);
+  }
+}
diff --git a/app/src/main/java/org/astraea/app/balancer/log/ClusterLogAllocation.java b/app/src/main/java/org/astraea/app/balancer/log/ClusterLogAllocation.java
index 0daff1eb..850bc703 100644
--- a/app/src/main/java/org/astraea/app/balancer/log/ClusterLogAllocation.java
+++ b/app/src/main/java/org/astraea/app/balancer/log/ClusterLogAllocation.java
@@ -17,6 +17,8 @@
 package org.astraea.app.balancer.log;
 
 import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import org.astraea.app.admin.TopicPartition;
 
@@ -62,5 +64,49 @@ public interface ClusterLogAllocation {
   /** Retrieve the stream of all topic/partition pairs in allocation. */
   Stream<TopicPartition> topicPartitionStream();
 
-  // TODO: add a method to calculate the difference between two ClusterLogAllocation
+  /**
+   * Find a subset of topic/partitions in the source allocation, that has any non-fulfilled log
+   * placement in the given target allocation. Note that the given two allocations must have the
+   * exactly same topic/partitions set. Otherwise, an {@link IllegalArgumentException} will be
+   * raised.
+   */
+  static Set<TopicPartition> findNonFulfilledAllocation(
+      ClusterLogAllocation source, ClusterLogAllocation target) {
+
+    final var sourceTopicPartition =
+        source.topicPartitionStream().collect(Collectors.toUnmodifiableSet());
+    final var targetTopicPartition =
+        target.topicPartitionStream().collect(Collectors.toUnmodifiableSet());
+
+    if (!sourceTopicPartition.equals(targetTopicPartition))
+      throw new IllegalArgumentException(
+          "source allocation and target allocation has different topic/partition set");
+
+    return sourceTopicPartition.stream()
+        .filter(tp -> !LogPlacement.isMatch(source.logPlacements(tp), target.logPlacements(tp)))
+        .collect(Collectors.toUnmodifiableSet());
+  }
+
+  static String toString(ClusterLogAllocation allocation) {
+    StringBuilder stringBuilder = new StringBuilder();
+
+    allocation
+        .topicPartitionStream()
+        .sorted()
+        .forEach(
+            tp -> {
+              stringBuilder.append("[").append(tp).append("] ");
+
+              allocation
+                  .logPlacements(tp)
+                  .forEach(
+                      log ->
+                          stringBuilder.append(
+                              String.format("%s(%s) ", log.broker(), log.logDirectory())));
+
+              stringBuilder.append(System.lineSeparator());
+            });
+
+    return stringBuilder.toString();
+  }
 }
diff --git a/app/src/main/java/org/astraea/app/balancer/log/LogPlacement.java b/app/src/main/java/org/astraea/app/balancer/log/LogPlacement.java
index a0a573c9..9425707f 100644
--- a/app/src/main/java/org/astraea/app/balancer/log/LogPlacement.java
+++ b/app/src/main/java/org/astraea/app/balancer/log/LogPlacement.java
@@ -16,7 +16,9 @@
  */
 package org.astraea.app.balancer.log;
 
+import java.util.List;
 import java.util.Optional;
+import java.util.stream.IntStream;
 
 /** This class describe the placement state of one kafka log. */
 public interface LogPlacement {
@@ -25,6 +27,44 @@ public interface LogPlacement {
 
   Optional<String> logDirectory();
 
+  static boolean isMatch(List<LogPlacement> sourcePlacements, List<LogPlacement> targetPlacements) {
+    if (sourcePlacements.size() != targetPlacements.size()) return false;
+    if (sourcePlacements.equals(targetPlacements)) return true;
+
+    final boolean brokerListMatch =
+        IntStream.range(0, sourcePlacements.size())
+            .allMatch(
+                index ->
+                    sourcePlacements.get(index).broker() == targetPlacements.get(index).broker());
+    if (!brokerListMatch) return false;
+
+    final boolean logDirectoryMatch =
+        IntStream.range(0, sourcePlacements.size())
+            .allMatch(
+                index ->
+                    meetLogDirectoryMigrationRequirement(
+                        sourcePlacements.get(index).logDirectory(),
+                        targetPlacements.get(index).logDirectory()));
+    //noinspection RedundantIfStatement
+    if (!logDirectoryMatch) return false;
+
+    return true;
+  }
+
+  @SuppressWarnings("OptionalUsedAsFieldOrParameterType")
+  private static boolean meetLogDirectoryMigrationRequirement(
+      Optional<String> sourceDir, Optional<String> targetDir) {
+    // don't care which log directory will eventually be in the destination.
+    if (targetDir.isEmpty()) return true;
+
+    // we care which data directory the target will eventually be, but we don't know.
+    if (sourceDir.isEmpty()) return false;
+
+    // both candidate is specified, if and only if two paths match, will consider as a requirement
+    // meet.
+    return sourceDir.get().equals(targetDir.get());
+  }
+
   static LogPlacement of(int broker) {
     return of(broker, null);
   }
diff --git a/app/src/main/java/org/astraea/app/cost/CostFunction.java b/app/src/main/java/org/astraea/app/cost/CostFunction.java
index 631f15af..91f22fa6 100644
--- a/app/src/main/java/org/astraea/app/cost/CostFunction.java
+++ b/app/src/main/java/org/astraea/app/cost/CostFunction.java
@@ -25,10 +25,6 @@ import org.astraea.app.metrics.collector.Fetcher;
  */
 public interface CostFunction {
 
-  static CostFunction throughput() {
-    return new ThroughputCost();
-  }
-
   /** @return the metrics getters. Those getters are used to fetch mbeans. */
   default Optional<Fetcher> fetcher() {
     return Optional.empty();
diff --git a/app/src/main/java/org/astraea/app/cost/NumberOfLeaderCost.java b/app/src/main/java/org/astraea/app/cost/NumberOfLeaderCost.java
deleted file mode 100644
index 7dd80c1f..00000000
--- a/app/src/main/java/org/astraea/app/cost/NumberOfLeaderCost.java
+++ /dev/null
@@ -1,67 +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.cost;
-
-import java.util.Comparator;
-import java.util.Map;
-import java.util.Optional;
-import java.util.stream.Collectors;
-import org.astraea.app.metrics.HasBeanObject;
-import org.astraea.app.metrics.collector.Fetcher;
-import org.astraea.app.metrics.kafka.HasValue;
-import org.astraea.app.metrics.kafka.KafkaMetrics;
-
-/**
- * The result is computed by "LeaderCount.Value". "LeaderCount.Value"" responds to the replica
- * leader number of brokers. The calculation method of the score is the total leader number divided
- * by the total leader number in broker
- */
-public class NumberOfLeaderCost implements HasBrokerCost {
-  @Override
-  public Optional<Fetcher> fetcher() {
-    return Optional.of(KafkaMetrics.ReplicaManager.LeaderCount::fetch);
-  }
-
-  /**
-   * @param clusterInfo cluster information
-   * @return a BrokerCost contain all ratio of leaders that exist on all brokers
-   */
-  @Override
-  public BrokerCost brokerCost(ClusterInfo clusterInfo) {
-    var leaderCount =
-        clusterInfo.clusterBean().all().entrySet().stream()
-            .flatMap(
-                e ->
-                    e.getValue().stream()
-                        .filter(x -> x instanceof HasValue)
-                        .filter(
-                            x -> x.beanObject().getProperties().get("name").equals("LeaderCount"))
-                        .filter(
-                            x ->
-                                x.beanObject().getProperties().get("type").equals("ReplicaManager"))
-                        .sorted(Comparator.comparing(HasBeanObject::createdTimestamp).reversed())
-                        .map(x -> (HasValue) x)
-                        .limit(1)
-                        .map(e2 -> Map.entry(e.getKey(), (int) e2.value())))
-            .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
-    var totalLeader = leaderCount.values().stream().mapToInt(Integer::intValue).sum();
-    var leaderCost =
-        leaderCount.entrySet().stream()
-            .collect(Collectors.toMap(Map.Entry::getKey, e -> (double) e.getValue() / totalLeader));
-    return () -> leaderCost;
-  }
-}
diff --git a/app/src/main/java/org/astraea/app/cost/ReplicaLeaderCost.java b/app/src/main/java/org/astraea/app/cost/ReplicaLeaderCost.java
new file mode 100644
index 00000000..9fe1a952
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/cost/ReplicaLeaderCost.java
@@ -0,0 +1,77 @@
+/*
+ * 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.Comparator;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import org.astraea.app.metrics.HasBeanObject;
+import org.astraea.app.metrics.collector.Fetcher;
+import org.astraea.app.metrics.kafka.HasValue;
+import org.astraea.app.metrics.kafka.KafkaMetrics;
+
+/** more replica leaders -> higher cost */
+public class ReplicaLeaderCost implements HasBrokerCost {
+
+  @Override
+  public BrokerCost brokerCost(ClusterInfo clusterInfo) {
+    var result =
+        leaderCount(clusterInfo).entrySet().stream()
+            .collect(Collectors.toMap(Map.Entry::getKey, e -> (double) e.getValue()));
+    return () -> result;
+  }
+
+  Map<Integer, Integer> leaderCount(ClusterInfo clusterInfo) {
+    return clusterInfo.clusterBean().all().entrySet().stream()
+        .flatMap(
+            e ->
+                e.getValue().stream()
+                    .filter(x -> x instanceof HasValue)
+                    .filter(x -> "LeaderCount".equals(x.beanObject().getProperties().get("name")))
+                    .filter(
+                        x -> "ReplicaManager".equals(x.beanObject().getProperties().get("type")))
+                    .sorted(Comparator.comparing(HasBeanObject::createdTimestamp).reversed())
+                    .map(x -> (HasValue) x)
+                    .limit(1)
+                    .map(e2 -> Map.entry(e.getKey(), (int) e2.value())))
+        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+  }
+
+  @Override
+  public Optional<Fetcher> fetcher() {
+    return Optional.of(KafkaMetrics.ReplicaManager.LeaderCount::fetch);
+  }
+
+  public static class NoMetrics extends ReplicaLeaderCost {
+
+    @Override
+    Map<Integer, Integer> leaderCount(ClusterInfo clusterInfo) {
+      return clusterInfo.topics().stream()
+          .flatMap(t -> clusterInfo.availableReplicaLeaders(t).stream())
+          .collect(Collectors.groupingBy(r -> r.nodeInfo().id()))
+          .entrySet()
+          .stream()
+          .collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, e -> e.getValue().size()));
+    }
+
+    @Override
+    public Optional<Fetcher> fetcher() {
+      return Optional.empty();
+    }
+  }
+}
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 878731c0..003f458d 100644
--- a/app/src/main/java/org/astraea/app/partitioner/StrictCostDispatcher.java
+++ b/app/src/main/java/org/astraea/app/partitioner/StrictCostDispatcher.java
@@ -31,14 +31,15 @@ import org.astraea.app.cost.CostFunction;
 import org.astraea.app.cost.HasBrokerCost;
 import org.astraea.app.cost.NodeInfo;
 import org.astraea.app.cost.ReplicaInfo;
+import org.astraea.app.cost.ReplicaLeaderCost;
 import org.astraea.app.metrics.collector.BeanCollector;
 import org.astraea.app.metrics.collector.Fetcher;
 import org.astraea.app.metrics.collector.Receiver;
 
 /**
  * this dispatcher scores the nodes by multiples cost functions. Each function evaluate the target
- * node by different metrics. The default cost function ranks nodes by throughput. It means the node
- * having lower throughput get higher score.
+ * node by different metrics. The default cost function ranks nodes by replica leader. It means the
+ * node having lower replica leaders get higher score.
  *
  * <p>The requisite config is JMX port. Most cost functions need the JMX metrics to score nodes.
  * Normally, all brokers use the same JMX port, so you can just define the `jmx.port=12345`. If one
@@ -57,7 +58,7 @@ public class StrictCostDispatcher implements Dispatcher {
       BeanCollector.builder().interval(Duration.ofSeconds(4)).build();
 
   // The cost-functions we consider and the weight of them. It is visible for test
-  Map<CostFunction, Double> functions = Map.of(CostFunction.throughput(), 1D);
+  Map<CostFunction, Double> functions = Map.of(new ReplicaLeaderCost.NoMetrics(), 1D);
 
   // all-in-one fetcher referenced to cost functions
   Optional<Fetcher> fetcher;
@@ -178,7 +179,7 @@ public class StrictCostDispatcher implements Dispatcher {
       Map<CostFunction, Double> functions,
       Optional<Integer> jmxPortDefault,
       Map<Integer, Integer> customJmxPort) {
-    this.functions = functions;
+    if (!functions.isEmpty()) this.functions = functions;
     this.fetcher = Fetcher.of(functions.keySet());
     this.jmxPortGetter =
         id ->
diff --git a/docker/start_broker.sh b/docker/start_broker.sh
index bbbde237..600032b1 100755
--- a/docker/start_broker.sh
+++ b/docker/start_broker.sh
@@ -337,6 +337,8 @@ setPropertyIfEmpty "num.partitions" "8"
 setPropertyIfEmpty "transaction.state.log.replication.factor" "1"
 setPropertyIfEmpty "offsets.topic.replication.factor" "1"
 setPropertyIfEmpty "transaction.state.log.min.isr" "1"
+setPropertyIfEmpty "min.insync.replicas" "1"
+
 if [[ "$CONFLUENT_BROKER" = "true" ]]; then
     rejectProperty "metric.reporters"
     setPropertyIfEmpty "confluent.metrics.reporter.zookeeper.connect" "$ZOOKEEPER_CONNECT"


Test patch:
diff --git a/app/src/test/java/org/astraea/app/balancer/executor/RebalanceAdminTest.java b/app/src/test/java/org/astraea/app/balancer/executor/RebalanceAdminTest.java
new file mode 100644
index 00000000..95ed015d
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/balancer/executor/RebalanceAdminTest.java
@@ -0,0 +1,326 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.balancer.executor;
+
+import java.time.Duration;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.function.Predicate;
+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.Admin;
+import org.astraea.app.admin.Replica;
+import org.astraea.app.admin.TopicPartition;
+import org.astraea.app.admin.TopicPartitionReplica;
+import org.astraea.app.balancer.log.LogPlacement;
+import org.astraea.app.common.DataSize;
+import org.astraea.app.common.DataUnit;
+import org.astraea.app.common.Utils;
+import org.astraea.app.producer.Producer;
+import org.astraea.app.service.RequireBrokerCluster;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.RepeatedTest;
+import org.junit.jupiter.api.Test;
+
+class RebalanceAdminTest extends RequireBrokerCluster {
+
+  @Test
+  void alterReplicaPlacementByList() throws InterruptedException {
+    // arrange
+    try (var admin = Admin.of(bootstrapServers())) {
+      var topic = prepareTopic(admin, 1, (short) 1);
+      var rebalanceAdmin = prepareRebalanceAdmin(admin);
+      RebalanceAdminImpl.changeRetrialTime(Duration.ofMillis(50));
+
+      // scale the replica size from 1 to 3, to the following data dir
+      var logFolder0 = randomElement(logFolders().get(0));
+      var logFolder1 = randomElement(logFolders().get(1));
+      var logFolder2 = randomElement(logFolders().get(2));
+
+      // act
+      var tasks =
+          rebalanceAdmin.alterReplicaPlacements(
+              new TopicPartition(topic, 0),
+              List.of(
+                  LogPlacement.of(0, logFolder0),
+                  LogPlacement.of(1, logFolder1),
+                  LogPlacement.of(2, logFolder2)));
+      tasks.forEach(
+          task -> Utils.packException(() -> task.completableFuture().get(5, TimeUnit.SECONDS)));
+
+      // assert
+      var topicPartition = new TopicPartition(topic, 0);
+      var replicas = admin.replicas(Set.of(topic)).get(topicPartition);
+
+      Assertions.assertEquals(
+          List.of(0, 1, 2), replicas.stream().map(Replica::broker).collect(Collectors.toList()));
+      Assertions.assertEquals(logFolder0, replicas.get(0).path());
+      Assertions.assertEquals(logFolder1, replicas.get(1).path());
+      Assertions.assertEquals(logFolder2, replicas.get(2).path());
+    }
+  }
+
+  // repeat the test so it has higher chance to fail
+  @RepeatedTest(value = 3)
+  void alterReplicaPlacementByDirectory() throws InterruptedException {
+    // arrange
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var topic = prepareTopic(admin, 1, (short) 1);
+      var topicPartition = new TopicPartition(topic, 0);
+      var rebalanceAdmin = prepareRebalanceAdmin(admin);
+      // decrease the debouncing time so the test has higher chance to fail
+      RebalanceAdminImpl.changeRetrialTime(Duration.ofMillis(150));
+      prepareData(topic, 0, DataUnit.MiB.of(256));
+      Supplier<Replica> replicaNow = () -> admin.replicas(Set.of(topic)).get(topicPartition).get(0);
+      var originalReplica = replicaNow.get();
+      var nextDir =
+          logFolders().get(originalReplica.broker()).stream()
+              .filter(name -> !name.equals(originalReplica.path()))
+              .findAny()
+              .orElseThrow();
+      var expectedPlacement = LogPlacement.of(originalReplica.broker(), nextDir);
+
+      // act, change the dir of the only replica
+      var task =
+          rebalanceAdmin.alterReplicaPlacements(topicPartition, List.of(expectedPlacement)).get(0);
+
+      // assert
+      task.completableFuture().join();
+      var finalReplica = replicaNow.get();
+      Assertions.assertTrue(finalReplica.inSync());
+      Assertions.assertFalse(finalReplica.isFuture());
+      Assertions.assertEquals(originalReplica.broker(), finalReplica.broker());
+    }
+  }
+
+  @Test
+  void clusterInfo() throws InterruptedException {
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      // test if all topics are covered
+      final var rebalanceAdmin = RebalanceAdmin.of(admin, (ignore) -> true);
+      final var clusterInfo = rebalanceAdmin.clusterInfo();
+      Assertions.assertEquals(admin.topicNames(), clusterInfo.topics());
+
+      // create topic
+      final var topic = prepareTopic(admin, 3, (short) 1);
+
+      // test if topic filter works
+      final var rebalanceAdmin1 = RebalanceAdmin.of(admin, topic::equals);
+      final var clusterInfo1 = rebalanceAdmin1.clusterInfo();
+      Assertions.assertEquals(Set.of(topic), clusterInfo1.topics());
+    }
+  }
+
+  @Test
+  void checkLogSynced() throws InterruptedException, ExecutionException {
+    // arrange
+    try (var admin = Admin.of(bootstrapServers())) {
+      var topic = prepareTopic(admin, 1, (short) 1);
+      var topicPartition = new TopicPartition(topic, 0);
+      var rebalanceAdmin = prepareRebalanceAdmin(admin);
+      var beginReplica = admin.replicas().get(topicPartition).get(0);
+      var otherDataDir =
+          admin.brokerFolders().get(beginReplica.broker()).stream()
+              .filter(dir -> !dir.equals(beginReplica.path()))
+              .findAny()
+              .orElseThrow();
+      RebalanceAdminImpl.changeRetrialTime(Duration.ofMillis(150));
+      prepareData(topic, 0, DataUnit.MiB.of(32));
+      // let two brokers join the replica list
+      admin.migrator().partition(topic, 0).moveTo(List.of(0, 1, 2));
+      // let the existing replica change its directory
+      admin.migrator().partition(topic, 0).moveTo(Map.of(beginReplica.broker(), otherDataDir));
+
+      // act
+      long time0 = System.currentTimeMillis();
+      rebalanceAdmin.waitLogSynced(new TopicPartitionReplica(topic, 0, 0)).get();
+      rebalanceAdmin.waitLogSynced(new TopicPartitionReplica(topic, 0, 1)).get();
+      rebalanceAdmin.waitLogSynced(new TopicPartitionReplica(topic, 0, 2)).get();
+      long time1 = System.currentTimeMillis();
+
+      // assert all replica synced
+      Assertions.assertTrue(
+          admin.replicas(Set.of(topic)).entrySet().stream()
+              .flatMap(x -> x.getValue().stream())
+              .allMatch(Replica::inSync));
+      // assert all data directory migration synced
+      Assertions.assertTrue(
+          admin.replicas(Set.of(topic)).entrySet().stream()
+              .flatMap(x -> x.getValue().stream())
+              .noneMatch(Replica::isFuture));
+      Assertions.assertTrue((time1 - time0) > 100, "This should takes awhile");
+    }
+  }
+
+  @Test
+  void checkPreferredLeaderSynced()
+      throws InterruptedException, ExecutionException, TimeoutException {
+    try (var admin = Admin.of(bootstrapServers())) {
+      var topic = prepareTopic(admin, 1, (short) 3);
+      var rebalanceAdmin = prepareRebalanceAdmin(admin);
+      var topicPartition = new TopicPartition(topic, 0);
+
+      var leaderNow =
+          (Supplier<Integer>)
+              () ->
+                  admin.replicas(Set.of(topic)).entrySet().stream()
+                      .filter(x -> x.getKey().topic().equals(topic))
+                      .filter(x -> x.getKey().partition() == 0)
+                      .flatMap(x -> x.getValue().stream())
+                      .filter(Replica::leader)
+                      .findFirst()
+                      .orElseThrow()
+                      .broker();
+
+      int oldLeader = leaderNow.get();
+
+      // change the preferred leader
+      int newPreferredLeader = (oldLeader + 2) % 3;
+      admin
+          .migrator()
+          .partition(topic, 0)
+          .moveTo(List.of(newPreferredLeader, (oldLeader + 1) % 3, oldLeader));
+
+      // assert not leader yet
+      Assertions.assertNotEquals(newPreferredLeader, leaderNow.get());
+
+      // do election
+      rebalanceAdmin.leaderElection(topicPartition).completableFuture().get();
+
+      // wait for this
+      rebalanceAdmin.waitPreferredLeaderSynced(topicPartition).get(5, TimeUnit.SECONDS);
+
+      // assert it is the leader
+      Assertions.assertEquals(newPreferredLeader, leaderNow.get());
+    }
+  }
+
+  @Test
+  void leaderElection() throws InterruptedException, ExecutionException, TimeoutException {
+    try (var admin = Admin.of(bootstrapServers())) {
+      var topic = prepareTopic(admin, 1, (short) 3);
+      var rebalanceAdmin = prepareRebalanceAdmin(admin);
+      var topicPartition = new TopicPartition(topic, 0);
+
+      var leaderNow =
+          (Supplier<Integer>)
+              () ->
+                  admin.replicas(Set.of(topic)).entrySet().stream()
+                      .filter(x -> x.getKey().topic().equals(topic))
+                      .filter(x -> x.getKey().partition() == 0)
+                      .flatMap(x -> x.getValue().stream())
+                      .filter(Replica::leader)
+                      .findFirst()
+                      .orElseThrow()
+                      .broker();
+
+      int oldLeader = leaderNow.get();
+
+      // change the preferred leader
+      int newPreferredLeader = (oldLeader + 2) % 3;
+      admin
+          .migrator()
+          .partition(topic, 0)
+          .moveTo(List.of(newPreferredLeader, (oldLeader + 1) % 3, oldLeader));
+
+      // assert not leader yet
+      Assertions.assertNotEquals(newPreferredLeader, leaderNow.get());
+
+      // do election
+      var task = rebalanceAdmin.leaderElection(topicPartition);
+
+      // wait for this
+      rebalanceAdmin.waitPreferredLeaderSynced(topicPartition).get(5, TimeUnit.SECONDS);
+
+      // assert it is the leader now
+      Assertions.assertEquals(newPreferredLeader, leaderNow.get());
+    }
+  }
+
+  @Test
+  void testTopicFilter() {
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var topic1 = Utils.randomString();
+      var topic2 = Utils.randomString();
+      var topic3 = Utils.randomString();
+      var topicPartition1 = new TopicPartition(topic1, 0);
+      var topicPartition2 = new TopicPartition(topic2, 0);
+      var topicPartition3 = new TopicPartition(topic3, 0);
+      Stream.of(topic1, topic2, topic3)
+          .forEach(i -> admin.creator().topic(i).numberOfPartitions(1).create());
+      var allowed = List.of(topic1, topic2);
+      Predicate<String> filter = allowed::contains;
+      var rebalanceAdmin = RebalanceAdmin.of(admin, filter);
+
+      Assertions.assertTrue(rebalanceAdmin.topicFilter().test(topic1));
+      Assertions.assertTrue(rebalanceAdmin.topicFilter().test(topic2));
+      Assertions.assertFalse(rebalanceAdmin.topicFilter().test(topic3));
+      Assertions.assertFalse(rebalanceAdmin.topicFilter().test("Something"));
+      Assertions.assertDoesNotThrow(() -> rebalanceAdmin.leaderElection(topicPartition1));
+      Assertions.assertDoesNotThrow(() -> rebalanceAdmin.leaderElection(topicPartition2));
+      Assertions.assertThrows(
+          IllegalArgumentException.class, () -> rebalanceAdmin.leaderElection(topicPartition3));
+    }
+  }
+
+  String prepareTopic(Admin admin, int partition, short replica) throws InterruptedException {
+    var topicName = Utils.randomString();
+
+    admin
+        .creator()
+        .topic(topicName)
+        .numberOfPartitions(partition)
+        .numberOfReplicas(replica)
+        .create();
+    TimeUnit.SECONDS.sleep(1);
+
+    return topicName;
+  }
+
+  RebalanceAdmin prepareRebalanceAdmin(Admin admin) {
+    return RebalanceAdmin.of(admin, (ignore) -> true);
+  }
+
+  void prepareData(String topic, int partition, DataSize dataSize) {
+    var dummy = new byte[1024];
+    int sends = dataSize.measurement(DataUnit.KiB).intValue();
+    try (var producer = Producer.of(bootstrapServers())) {
+      var sender = producer.sender().topic(topic).partition(partition).value(dummy);
+      IntStream.range(0, sends)
+          .parallel()
+          .mapToObj(i -> sender.run().toCompletableFuture())
+          .collect(Collectors.toSet())
+          .forEach(CompletableFuture::join);
+    }
+  }
+
+  <T> T randomElement(Collection<T> collection) {
+    return collection.stream()
+        .skip(ThreadLocalRandom.current().nextInt(0, collection.size()))
+        .findFirst()
+        .orElseThrow();
+  }
+}
diff --git a/app/src/test/java/org/astraea/app/balancer/executor/StraightPlanExecutorTest.java b/app/src/test/java/org/astraea/app/balancer/executor/StraightPlanExecutorTest.java
new file mode 100644
index 00000000..59873bfa
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/balancer/executor/StraightPlanExecutorTest.java
@@ -0,0 +1,134 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.balancer.executor;
+
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.astraea.app.admin.Admin;
+import org.astraea.app.admin.TopicPartition;
+import org.astraea.app.balancer.log.ClusterLogAllocation;
+import org.astraea.app.balancer.log.LayeredClusterLogAllocation;
+import org.astraea.app.balancer.log.LogPlacement;
+import org.astraea.app.common.Utils;
+import org.astraea.app.service.RequireBrokerCluster;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class StraightPlanExecutorTest extends RequireBrokerCluster {
+
+  @Test
+  void testRun() throws InterruptedException {
+    // arrange
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      final var topicName = "StraightPlanExecutorTest_" + Utils.randomString(8);
+      admin.creator().topic(topicName).numberOfPartitions(10).numberOfReplicas((short) 2).create();
+      TimeUnit.SECONDS.sleep(2);
+      final var originalAllocation =
+          LayeredClusterLogAllocation.of(admin.clusterInfo(Set.of(topicName)));
+      TimeUnit.SECONDS.sleep(3);
+      final var broker0 = 0;
+      final var broker1 = 1;
+      final var logFolder0 = logFolders().get(broker0).stream().findAny().orElseThrow();
+      final var logFolder1 = logFolders().get(broker1).stream().findAny().orElseThrow();
+      final var onlyPlacement =
+          List.of(LogPlacement.of(broker0, logFolder0), LogPlacement.of(broker1, logFolder1));
+      final var allocationMap =
+          IntStream.range(0, 10)
+              .mapToObj(i -> new TopicPartition(topicName, i))
+              .collect(Collectors.toUnmodifiableMap(tp -> tp, tp -> onlyPlacement));
+      final var expectedAllocation = LayeredClusterLogAllocation.of(allocationMap);
+      final var expectedTopicPartition =
+          expectedAllocation.topicPartitionStream().collect(Collectors.toUnmodifiableSet());
+      final var rebalanceAdmin = RebalanceAdmin.of(admin, (s) -> s.equals(topicName));
+
+      // act
+      new StraightPlanExecutor().run(rebalanceAdmin, expectedAllocation);
+
+      // assert
+      final var currentAllocation =
+          LayeredClusterLogAllocation.of(admin.clusterInfo(Set.of(topicName)));
+      final var currentTopicPartition =
+          currentAllocation.topicPartitionStream().collect(Collectors.toUnmodifiableSet());
+      Assertions.assertEquals(expectedTopicPartition, currentTopicPartition);
+      expectedTopicPartition.forEach(
+          topicPartition ->
+              Assertions.assertEquals(
+                  expectedAllocation.logPlacements(topicPartition),
+                  currentAllocation.logPlacements(topicPartition),
+                  "Testing for " + topicPartition));
+
+      System.out.println("Expected:");
+      System.out.println(ClusterLogAllocation.toString(expectedAllocation));
+      System.out.println("Current:");
+      System.out.println(ClusterLogAllocation.toString(currentAllocation));
+      System.out.println("Original:");
+      System.out.println(ClusterLogAllocation.toString(originalAllocation));
+    }
+  }
+
+  @Test
+  void testRunNoDirSpecified() throws InterruptedException {
+    // arrange
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      final var topicName = "StraightPlanExecutorTest_" + Utils.randomString(8);
+      admin.creator().topic(topicName).numberOfPartitions(10).numberOfReplicas((short) 3).create();
+      TimeUnit.SECONDS.sleep(2);
+      final var originalAllocation =
+          LayeredClusterLogAllocation.of(admin.clusterInfo(Set.of(topicName)));
+      TimeUnit.SECONDS.sleep(3);
+      final var broker0 = 0;
+      final var broker1 = 1;
+      final var broker2 = 2;
+      final var onlyPlacement =
+          List.of(LogPlacement.of(broker0), LogPlacement.of(broker1), LogPlacement.of(broker2));
+      final var allocationMap =
+          IntStream.range(0, 10)
+              .mapToObj(i -> new TopicPartition(topicName, i))
+              .collect(Collectors.toUnmodifiableMap(tp -> tp, tp -> onlyPlacement));
+      final var expectedAllocation = LayeredClusterLogAllocation.of(allocationMap);
+      final var expectedTopicPartition =
+          expectedAllocation.topicPartitionStream().collect(Collectors.toUnmodifiableSet());
+      final var rebalanceAdmin = RebalanceAdmin.of(admin, (s) -> s.equals(topicName));
+
+      // act
+      new StraightPlanExecutor().run(rebalanceAdmin, expectedAllocation);
+
+      // assert
+      final var currentAllocation =
+          LayeredClusterLogAllocation.of(admin.clusterInfo(Set.of(topicName)));
+      final var currentTopicPartition =
+          currentAllocation.topicPartitionStream().collect(Collectors.toUnmodifiableSet());
+      Assertions.assertEquals(expectedTopicPartition, currentTopicPartition);
+      expectedTopicPartition.forEach(
+          topicPartition ->
+              Assertions.assertTrue(
+                  LogPlacement.isMatch(
+                      currentAllocation.logPlacements(topicPartition),
+                      expectedAllocation.logPlacements(topicPartition))));
+
+      System.out.println("Expected:");
+      System.out.println(ClusterLogAllocation.toString(expectedAllocation));
+      System.out.println("Current:");
+      System.out.println(ClusterLogAllocation.toString(currentAllocation));
+      System.out.println("Original:");
+      System.out.println(ClusterLogAllocation.toString(originalAllocation));
+    }
+  }
+}
diff --git a/app/src/test/java/org/astraea/app/balancer/log/LayeredClusterLogAllocationTest.java b/app/src/test/java/org/astraea/app/balancer/log/LayeredClusterLogAllocationTest.java
index 233f671c..42556139 100644
--- a/app/src/test/java/org/astraea/app/balancer/log/LayeredClusterLogAllocationTest.java
+++ b/app/src/test/java/org/astraea/app/balancer/log/LayeredClusterLogAllocationTest.java
@@ -202,4 +202,49 @@ class LayeredClusterLogAllocationTest {
     Assertions.assertDoesNotThrow(() -> extended.migrateReplica(toModify, 8, 1, "somewhere"));
     Assertions.assertDoesNotThrow(() -> extended.letReplicaBecomeLeader(toModify, 0));
   }
+
+  @Test
+  void findNonFulfilledAllocation() {
+    final var clusterInfo = ClusterInfoProvider.fakeClusterInfo(3, 10, 10, 2);
+    final var a = LayeredClusterLogAllocation.of(clusterInfo);
+    final var b = LayeredClusterLogAllocation.of(clusterInfo);
+    Assertions.assertEquals(Set.of(), ClusterLogAllocation.findNonFulfilledAllocation(a, b));
+
+    final var source = LayeredClusterLogAllocation.of(clusterInfo);
+    final var oneTopicPartition = source.topicPartitionStream().findFirst().orElseThrow();
+    final var twoTopicPartition = source.topicPartitionStream().skip(1).findFirst().orElseThrow();
+
+    final var target0 = LayeredClusterLogAllocation.of(a);
+    target0.migrateReplica(oneTopicPartition, 0, 0, "/somewhere");
+    Assertions.assertEquals(
+        Set.of(oneTopicPartition),
+        ClusterLogAllocation.findNonFulfilledAllocation(source, target0));
+
+    final var target1 = LayeredClusterLogAllocation.of(a);
+    target1.migrateReplica(oneTopicPartition, 0, 2);
+    Assertions.assertEquals(
+        Set.of(oneTopicPartition),
+        ClusterLogAllocation.findNonFulfilledAllocation(source, target1));
+
+    final var target2 = LayeredClusterLogAllocation.of(a);
+    target2.letReplicaBecomeLeader(oneTopicPartition, 1);
+    Assertions.assertEquals(
+        Set.of(oneTopicPartition),
+        ClusterLogAllocation.findNonFulfilledAllocation(source, target2));
+
+    final var target3 = LayeredClusterLogAllocation.of(a);
+    target3.migrateReplica(oneTopicPartition, 0, 2);
+    target3.migrateReplica(oneTopicPartition, 2, 2, "/somewhere");
+    target3.letReplicaBecomeLeader(twoTopicPartition, 1);
+    Assertions.assertEquals(
+        Set.of(oneTopicPartition, twoTopicPartition),
+        ClusterLogAllocation.findNonFulfilledAllocation(source, target3));
+
+    final var map4 = a.topicPartitionStream().collect(Collectors.toMap(x -> x, a::logPlacements));
+    map4.put(new TopicPartition("NewTopic", 0), List.of(LogPlacement.of(0, "?")));
+    final var target4 = LayeredClusterLogAllocation.of(map4);
+    Assertions.assertThrows(
+        IllegalArgumentException.class,
+        () -> ClusterLogAllocation.findNonFulfilledAllocation(source, target4));
+  }
 }
diff --git a/app/src/test/java/org/astraea/app/balancer/log/LogPlacementTest.java b/app/src/test/java/org/astraea/app/balancer/log/LogPlacementTest.java
new file mode 100644
index 00000000..6592b5e6
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/balancer/log/LogPlacementTest.java
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.balancer.log;
+
+import java.util.List;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+class LogPlacementTest {
+
+  @Test
+  @DisplayName("Equals")
+  void isMatch0() {
+    final var placement0 =
+        List.of(LogPlacement.of(0, "/A"), LogPlacement.of(1, "/B"), LogPlacement.of(2, "/C"));
+    final var placement1 =
+        List.of(LogPlacement.of(0, "/A"), LogPlacement.of(1, "/B"), LogPlacement.of(2, "/C"));
+    Assertions.assertTrue(LogPlacement.isMatch(placement0, placement1));
+  }
+
+  @Test
+  @DisplayName("Wrong order")
+  void noMatch0() {
+    final var placement0 =
+        List.of(LogPlacement.of(0, "/A"), LogPlacement.of(1, "/B"), LogPlacement.of(2, "/C"));
+    final var placement1 =
+        List.of(LogPlacement.of(1, "/B"), LogPlacement.of(0, "/A"), LogPlacement.of(2, "/C"));
+    Assertions.assertFalse(LogPlacement.isMatch(placement0, placement1));
+  }
+
+  @Test
+  @DisplayName("No log dir name match")
+  void noMatch1() {
+    final var placement0 =
+        List.of(LogPlacement.of(0, "/A"), LogPlacement.of(1, "/B"), LogPlacement.of(2, "/C"));
+    final var placement1 =
+        List.of(LogPlacement.of(0, "/Aaa"), LogPlacement.of(1, "/B"), LogPlacement.of(2, "/C"));
+    Assertions.assertFalse(LogPlacement.isMatch(placement0, placement1));
+  }
+
+  @Test
+  @DisplayName("Optional log dir")
+  void noMatch2() {
+    final var sourcePlacement =
+        List.of(LogPlacement.of(0, "/A"), LogPlacement.of(1, "/B"), LogPlacement.of(2, "/C"));
+    final var targetPlacement =
+        List.of(LogPlacement.of(0), LogPlacement.of(1, "/B"), LogPlacement.of(2, "/C"));
+    // don't care which log dir placement[0] will eventually be.
+    Assertions.assertTrue(LogPlacement.isMatch(sourcePlacement, targetPlacement));
+  }
+
+  @Test
+  @DisplayName("Optional log dir")
+  void noMatch3() {
+    final var sourcePlacement =
+        List.of(LogPlacement.of(0), LogPlacement.of(1, "/B"), LogPlacement.of(2, "/C"));
+    final var targetPlacement =
+        List.of(LogPlacement.of(0, "/A"), LogPlacement.of(1, "/B"), LogPlacement.of(2, "/C"));
+    // do care which log dir placement[0] will eventually be.
+    Assertions.assertFalse(LogPlacement.isMatch(sourcePlacement, targetPlacement));
+  }
+}
diff --git a/app/src/test/java/org/astraea/app/cost/NumberOfLeaderCostTest.java b/app/src/test/java/org/astraea/app/cost/NumberOfLeaderCostTest.java
deleted file mode 100644
index 74ccd364..00000000
--- a/app/src/test/java/org/astraea/app/cost/NumberOfLeaderCostTest.java
+++ /dev/null
@@ -1,122 +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.cost;
-
-import java.time.Duration;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-import org.astraea.app.admin.Admin;
-import org.astraea.app.admin.ClusterBean;
-import org.astraea.app.metrics.HasBeanObject;
-import org.astraea.app.metrics.collector.BeanCollector;
-import org.astraea.app.metrics.jmx.BeanObject;
-import org.astraea.app.metrics.kafka.HasValue;
-import org.astraea.app.metrics.kafka.KafkaMetrics;
-import org.astraea.app.service.RequireBrokerCluster;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Test;
-import org.mockito.Mockito;
-
-class NumberOfLeaderCostTest extends RequireBrokerCluster {
-
-  @Test
-  void testBrokerCost() {
-    var topicName = List.of("testLeaderCost-1", "testLeaderCost-2", "testLeaderCost-3");
-    try (var admin = Admin.of(bootstrapServers())) {
-      var host = "localhost";
-      for (var topic : topicName)
-        admin.creator().topic(topic).numberOfPartitions(4).numberOfReplicas((short) 1).create();
-      // wait for topic creation
-      TimeUnit.SECONDS.sleep(5);
-      admin.migrator().partition(topicName.get(0), 0).moveTo(List.of(0));
-      admin.migrator().partition(topicName.get(0), 1).moveTo(List.of(0));
-      admin.migrator().partition(topicName.get(0), 2).moveTo(List.of(0));
-      admin.migrator().partition(topicName.get(0), 3).moveTo(List.of(1));
-      // wait for replica migrate
-      TimeUnit.SECONDS.sleep(1);
-      admin.migrator().topic(topicName.get(1)).moveTo(List.of(1));
-      // wait for replica migrate
-      TimeUnit.SECONDS.sleep(1);
-      admin.migrator().topic(topicName.get(2)).moveTo(List.of(2));
-
-      var allBeans = new HashMap<Integer, Collection<HasBeanObject>>();
-      var jmxAddress = Map.of(1001, jmxServiceURL().getPort());
-
-      NumberOfLeaderCost costFunction = new NumberOfLeaderCost();
-      jmxAddress.forEach(
-          (b, port) -> {
-            var firstBeanObjects =
-                BeanCollector.builder()
-                    .interval(Duration.ofSeconds(4))
-                    .build()
-                    .register()
-                    .host(host)
-                    .port(port)
-                    .fetcher(costFunction.fetcher().get())
-                    .build()
-                    .current();
-            allBeans.put(
-                b,
-                allBeans.containsKey(b)
-                    ? Stream.concat(allBeans.get(b).stream(), firstBeanObjects.stream())
-                        .collect(Collectors.toList())
-                    : firstBeanObjects);
-          });
-      var loadCostFunction = new NumberOfLeaderCost();
-      var load = loadCostFunction.brokerCost(exampleClusterInfo());
-
-      Assertions.assertEquals(3.0 / (3 + 4 + 5), load.value().get(1));
-      Assertions.assertEquals(4.0 / (3 + 4 + 5), load.value().get(2));
-      Assertions.assertEquals(5.0 / (3 + 4 + 5), load.value().get(3));
-    } catch (InterruptedException e) {
-      throw new RuntimeException(e);
-    }
-  }
-
-  private ClusterInfo exampleClusterInfo() {
-    var LeaderCount1 =
-        mockResult("ReplicaManager", KafkaMetrics.ReplicaManager.LeaderCount.metricName(), 3);
-    var LeaderCount2 =
-        mockResult("ReplicaManager", KafkaMetrics.ReplicaManager.LeaderCount.metricName(), 4);
-    var LeaderCount3 =
-        mockResult("ReplicaManager", KafkaMetrics.ReplicaManager.LeaderCount.metricName(), 5);
-
-    Collection<HasBeanObject> broker1 = List.of(LeaderCount1);
-    Collection<HasBeanObject> broker2 = List.of(LeaderCount2);
-    Collection<HasBeanObject> broker3 = List.of(LeaderCount3);
-    return new FakeClusterInfo() {
-      @Override
-      public ClusterBean clusterBean() {
-        return ClusterBean.of(Map.of(1, broker1, 2, broker2, 3, broker3));
-      }
-    };
-  }
-
-  private HasValue mockResult(String type, String name, long count) {
-    var result = Mockito.mock(HasValue.class);
-    var bean = Mockito.mock(BeanObject.class);
-    Mockito.when(result.beanObject()).thenReturn(bean);
-    Mockito.when(bean.getProperties()).thenReturn(Map.of("name", name, "type", type));
-    Mockito.when(result.value()).thenReturn(count);
-    return result;
-  }
-}
diff --git a/app/src/test/java/org/astraea/app/cost/ReplicaLeaderCostTest.java b/app/src/test/java/org/astraea/app/cost/ReplicaLeaderCostTest.java
new file mode 100644
index 00000000..3ce35a60
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/cost/ReplicaLeaderCostTest.java
@@ -0,0 +1,82 @@
+/*
+ * 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.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.astraea.app.admin.ClusterBean;
+import org.astraea.app.metrics.HasBeanObject;
+import org.astraea.app.metrics.jmx.BeanObject;
+import org.astraea.app.metrics.kafka.HasValue;
+import org.astraea.app.metrics.kafka.KafkaMetrics;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+public class ReplicaLeaderCostTest {
+
+  @Test
+  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));
+    var function = new ReplicaLeaderCost.NoMetrics();
+    var clusterInfo = Mockito.mock(ClusterInfo.class);
+    Mockito.when(clusterInfo.topics()).thenReturn(Set.of("topic"));
+    Mockito.when(clusterInfo.availableReplicaLeaders(Mockito.anyString())).thenReturn(replicas);
+    var cost = function.leaderCount(clusterInfo);
+    Assertions.assertTrue(cost.containsKey(10));
+    Assertions.assertTrue(cost.containsKey(11));
+    Assertions.assertEquals(2, cost.size());
+    Assertions.assertTrue(cost.get(10) > cost.get(11));
+  }
+
+  @Test
+  void testWithMetrics() {
+    var topicName = List.of("testLeaderCost-1", "testLeaderCost-2", "testLeaderCost-3");
+    var costFunction = new ReplicaLeaderCost();
+    var LeaderCount1 = mockResult(KafkaMetrics.ReplicaManager.LeaderCount.metricName(), 3);
+    var LeaderCount2 = mockResult(KafkaMetrics.ReplicaManager.LeaderCount.metricName(), 4);
+    var LeaderCount3 = mockResult(KafkaMetrics.ReplicaManager.LeaderCount.metricName(), 5);
+
+    Collection<HasBeanObject> broker1 = List.of(LeaderCount1);
+    Collection<HasBeanObject> broker2 = List.of(LeaderCount2);
+    Collection<HasBeanObject> broker3 = List.of(LeaderCount3);
+    var clusterBean = ClusterBean.of(Map.of(1, broker1, 2, broker2, 3, broker3));
+    var clusterInfo = Mockito.mock(ClusterInfo.class);
+    Mockito.when(clusterInfo.clusterBean()).thenReturn(clusterBean);
+    var load = costFunction.brokerCost(clusterInfo);
+
+    Assertions.assertEquals(3, load.value().size());
+    Assertions.assertEquals(3.0, load.value().get(1));
+    Assertions.assertEquals(4.0, load.value().get(2));
+    Assertions.assertEquals(5.0, load.value().get(3));
+  }
+
+  private HasValue mockResult(String name, long count) {
+    var result = Mockito.mock(HasValue.class);
+    var bean = Mockito.mock(BeanObject.class);
+    Mockito.when(result.beanObject()).thenReturn(bean);
+    Mockito.when(bean.getProperties()).thenReturn(Map.of("name", name, "type", "ReplicaManager"));
+    Mockito.when(result.value()).thenReturn(count);
+    return result;
+  }
+}
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 6b1bd454..fcfbc2c3 100644
--- a/app/src/test/java/org/astraea/app/partitioner/StrictCostDispatcherTest.java
+++ b/app/src/test/java/org/astraea/app/partitioner/StrictCostDispatcherTest.java
@@ -29,6 +29,7 @@ import org.astraea.app.cost.CostFunction;
 import org.astraea.app.cost.HasBrokerCost;
 import org.astraea.app.cost.NodeInfo;
 import org.astraea.app.cost.ReplicaInfo;
+import org.astraea.app.cost.ReplicaLeaderCost;
 import org.astraea.app.cost.ThroughputCost;
 import org.astraea.app.metrics.collector.Fetcher;
 import org.astraea.app.metrics.collector.Receiver;
@@ -65,13 +66,14 @@ public class StrictCostDispatcherTest {
       Assertions.assertThrows(
           IllegalArgumentException.class,
           () ->
-              dispatcher.configure(Configuration.of(Map.of(ThroughputCost.class.getName(), "-1"))));
+              dispatcher.configure(
+                  Configuration.of(Map.of(ReplicaLeaderCost.NoMetrics.class.getName(), "-1"))));
 
       // Test for cost functions configuring
       dispatcher.configure(
           Configuration.of(
               Map.of(
-                  ThroughputCost.class.getName(),
+                  ReplicaLeaderCost.NoMetrics.class.getName(),
                   "0.1",
                   BrokerInputCost.class.getName(),
                   "2",
@@ -87,7 +89,7 @@ public class StrictCostDispatcherTest {
       dispatcher.configure(
           Configuration.of(
               Map.of(
-                  ThroughputCost.class.getName(),
+                  ReplicaLeaderCost.NoMetrics.class.getName(),
                   "0.1",
                   BrokerInputCost.class.getName(),
                   "2",
@@ -226,15 +228,13 @@ public class StrictCostDispatcherTest {
     var dispatcher = new StrictCostDispatcher();
     Assertions.assertThrows(
         IllegalArgumentException.class,
-        () ->
-            dispatcher.configure(
-                Map.of(CostFunction.throughput(), 1D), Optional.empty(), Map.of()));
+        () -> dispatcher.configure(Map.of(new ThroughputCost(), 1D), Optional.empty(), Map.of()));
 
     // pass due to default port
-    dispatcher.configure(Map.of(CostFunction.throughput(), 1D), Optional.of(111), Map.of());
+    dispatcher.configure(Map.of(new ThroughputCost(), 1D), Optional.of(111), Map.of());
 
     // pass due to specify port
-    dispatcher.configure(Map.of(CostFunction.throughput(), 1D), Optional.empty(), Map.of(222, 111));
+    dispatcher.configure(Map.of(new ThroughputCost(), 1D), Optional.empty(), Map.of(222, 111));
   }
 
   @Test
@@ -264,4 +264,11 @@ public class StrictCostDispatcherTest {
     Assertions.assertEquals(
         partitionId, dispatcher.partition("topic", new byte[0], new byte[0], clusterInfo));
   }
+
+  @Test
+  void testDefaultFunction() {
+    var dispatcher = new StrictCostDispatcher();
+    dispatcher.configure(Map.of(), Optional.empty(), Map.of());
+    Assertions.assertEquals(1, dispatcher.functions.size());
+  }
 }