Repository: skiptests/astraea
Pull request #356 (2022-05-23 14:17:39)

Simplify prometheus script
The following changes are included by this PR.
1. remove JMX address
2. remove refresh
3. mount configuration file directly
4. increase scrape timeout
5. remove volumes
6. refactor

garyparrot: It would be great if you could update the Prometheus section in `README.md`. Some of the content is stale.


Patch:
diff --git a/README.md b/README.md
index ad2a0403..94436051 100644
--- a/README.md
+++ b/README.md
@@ -97,7 +97,7 @@ There are 4 useful ENVs which can change JVM/container configuration.
 ### Run Node Exporter
 
 [Node Exporter](https://github.com/prometheus/node_exporter) is a famous utility for exporting machine metrics. It is 
-recommended using node exporter in conjunction with [Prometheus](#run-prometheus) to observe the test environment state.
+recommended using node exporter in conjunction with Prometheus to observe the test environment state.
 
 ```shell
 ./docker/start_node_exporter.sh
@@ -108,66 +108,10 @@ recommended using node exporter in conjunction with [Prometheus](#run-prometheus
 [INFO] node_exporter running at http://192.168.0.2:9100
 ```
 
-### Run Prometheus
-
-[Prometheus](https://github.com/prometheus/prometheus) is a famous application for monitor and gather time-series metrics. 
-
-#### Start Prometheus
-
-This project offers some scripts to set up Prometheus for your test environment quickly.
-
-If you have an exporter installed on your broker(all Kafka instances created by `start_broker.sh` will have JMX exporter installed),
-and [node exporter](#run-node-exporter) installed on your machine. You can follow the below instruction to observe the performance metrics
-of Kafka & your machine.
-
-For example. Assume you have two Kafka brokers, 
-
-* the first broker has an exporter running at `192.168.0.1:10558` 
-* the second broker has an exporter running at `192.168.0.2:10558`
-* the first machine has node exporter running at `192.168.0.1:9100`
-* the second machine has node exporter running at `192.168.0.2:9100`
-
-You can execute the following command to create a Prometheus instance that fetches data from the above 4 exporters.
-
-```shell
-./docker/start_prometheus.sh start
-```
-
-The console will show the http address of prometheus service, also some hints for you to set up Grafana. See next [section](#run-grafana) for
-further detail.
-
-```shell
-[INFO] Start existing prometheus instance
-prometheus-9090
-[INFO] =================================================
-[INFO] config file: /tmp/prometheus-9090.yml
-[INFO] prometheus address: http://192.168.0.2:9090
-[INFO] command to run grafana at this host: ./docker/start_grafana.sh start
-[INFO] command to add prometheus to grafana datasource: ./docker/start_grafana.sh add_prom_source <USERNAME>:<PASSWORD> Prometheus http://192.168.0.2:9090
-[INFO] =================================================
-```
-
-There are two ways to change the prometheus configuration.
-
-```shell
-# refresh prometheus config by command-line arguments
-./docker/start_prometheus.sh refresh host1:1111,host2:1111 host1:2222,host2:2222
-
-# refresh prometheus config by specific file
-./docker/start_prometheus.sh refresh ./path/to/config/file/prometheus.yml
-```
-
-#### Update Prometheus configuration
-
-To update your Prometheus configuration, there are two ways.
-
-1. Execute ``./docker/start_prometheus.sh refresh <kafka-exporter-addresses> <node-exporter-addresses>``.
-2. Go edit ``/tmp/prometheus-9090.yml``, and execute ``./docker/start_prometheus.sh refresh``.
-
 ### Run Grafana
 
 [Grafana](https://github.com/grafana/grafana) is a famous application for display system states. It is recommended to use Grafana
-in conjunction with [Prometheus](#run-prometheus) to observe the test environment state.
+in conjunction with Prometheus to observe the test environment state.
 
 #### Start Grafana
 
diff --git a/app/src/main/java/org/astraea/balancer/RebalancePlanProposal.java b/app/src/main/java/org/astraea/balancer/RebalancePlanProposal.java
new file mode 100644
index 00000000..a411a16a
--- /dev/null
+++ b/app/src/main/java/org/astraea/balancer/RebalancePlanProposal.java
@@ -0,0 +1,111 @@
+package org.astraea.balancer;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import org.astraea.balancer.log.ClusterLogAllocation;
+
+public interface RebalancePlanProposal {
+
+  Optional<ClusterLogAllocation> rebalancePlan();
+
+  List<String> info();
+
+  List<String> warnings();
+
+  static Build builder() {
+    return new Build();
+  }
+
+  class Build {
+    ClusterLogAllocation allocation = null;
+    List<String> info = Collections.synchronizedList(new ArrayList<>());
+    List<String> warnings = Collections.synchronizedList(new ArrayList<>());
+
+    // guard by this
+    private boolean built = false;
+
+    public synchronized Build noRebalancePlan() {
+      ensureNotBuiltYet();
+      this.allocation = null;
+      return this;
+    }
+
+    public synchronized Build withRebalancePlan(ClusterLogAllocation clusterLogAllocation) {
+      ensureNotBuiltYet();
+      this.allocation = Objects.requireNonNull(clusterLogAllocation);
+      return this;
+    }
+
+    public synchronized Build addWarning(String warning) {
+      ensureNotBuiltYet();
+      this.warnings.add(warning);
+      return this;
+    }
+
+    public synchronized Build addInfo(String info) {
+      ensureNotBuiltYet();
+      this.info.add(info);
+      return this;
+    }
+
+    private synchronized void ensureNotBuiltYet() {
+      if (built) throw new IllegalStateException("This builder already built.");
+    }
+
+    public synchronized RebalancePlanProposal build() {
+      final var allocationRef = allocation;
+      final var infoRef = info;
+      final var warningRef = warnings;
+
+      ensureNotBuiltYet();
+
+      built = true;
+      allocation = null;
+      info = null;
+      warnings = null;
+
+      return new RebalancePlanProposal() {
+
+        @Override
+        public Optional<ClusterLogAllocation> rebalancePlan() {
+          return Optional.ofNullable(allocationRef);
+        }
+
+        @Override
+        public List<String> info() {
+          // use Collections.unmodifiableList instead of List.copyOf to avoid excessive memory
+          // footprint
+          return Collections.unmodifiableList(infoRef);
+        }
+
+        @Override
+        public List<String> warnings() {
+          // use Collections.unmodifiableList instead of List.copyOf to avoid excessive memory
+          // footprint
+          return Collections.unmodifiableList(warningRef);
+        }
+
+        @Override
+        public String toString() {
+          StringBuilder sb = new StringBuilder();
+
+          sb.append("[RebalancePlanProposal]").append(System.lineSeparator());
+
+          sb.append("  Info:").append(System.lineSeparator());
+          if (info().isEmpty()) sb.append(String.format("    no information%n"));
+          else info().forEach(infoString -> sb.append(String.format("    * %s%n", infoString)));
+          if (!warnings().isEmpty()) {
+            sb.append("  Warning:").append(System.lineSeparator());
+            warnings()
+                .forEach(warningString -> sb.append(String.format("    * %s%n", warningString)));
+          }
+
+          return sb.toString();
+        }
+      };
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/balancer/generator/RebalancePlanGenerator.java b/app/src/main/java/org/astraea/balancer/generator/RebalancePlanGenerator.java
new file mode 100644
index 00000000..40dfa1e4
--- /dev/null
+++ b/app/src/main/java/org/astraea/balancer/generator/RebalancePlanGenerator.java
@@ -0,0 +1,37 @@
+package org.astraea.balancer.generator;
+
+import java.util.stream.Stream;
+import org.astraea.balancer.RebalancePlanProposal;
+import org.astraea.balancer.log.ClusterLogAllocation;
+import org.astraea.balancer.log.LayeredClusterLogAllocation;
+import org.astraea.cost.ClusterInfo;
+
+/** */
+public interface RebalancePlanGenerator {
+
+  /**
+   * Generate a rebalance proposal, noted that this function doesn't require proposing exactly the
+   * same plan for the same input argument. There can be some randomization that takes part in this
+   * process.
+   *
+   * @param clusterInfo the cluster state, implementation can take advantage of the data inside to
+   *     proposal the plan it feels confident to improve the cluster.
+   * @return a {@link Stream} generating rebalance plan regarding the given {@link ClusterInfo}
+   */
+  default Stream<RebalancePlanProposal> generate(ClusterInfo clusterInfo) {
+    return generate(clusterInfo, LayeredClusterLogAllocation.of(clusterInfo));
+  }
+
+  /**
+   * Generate a rebalance proposal, noted that this function doesn't require proposing exactly the
+   * same plan for the same input argument. There can be some randomization that takes part in this
+   * process.
+   *
+   * @param clusterInfo the cluster state, implementation can take advantage of the data inside to
+   *     proposal the plan it feels confident to improve the cluster.
+   * @param baseAllocation the cluster log allocation as the based of proposal generation.
+   * @return a {@link Stream} generating rebalance plan regarding the given {@link ClusterInfo}
+   */
+  Stream<RebalancePlanProposal> generate(
+      ClusterInfo clusterInfo, ClusterLogAllocation baseAllocation);
+}
diff --git a/app/src/main/java/org/astraea/balancer/generator/ShufflePlanGenerator.java b/app/src/main/java/org/astraea/balancer/generator/ShufflePlanGenerator.java
new file mode 100644
index 00000000..d1c55bb8
--- /dev/null
+++ b/app/src/main/java/org/astraea/balancer/generator/ShufflePlanGenerator.java
@@ -0,0 +1,227 @@
+package org.astraea.balancer.generator;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+import org.astraea.admin.TopicPartition;
+import org.astraea.balancer.RebalancePlanProposal;
+import org.astraea.balancer.log.ClusterLogAllocation;
+import org.astraea.balancer.log.LayeredClusterLogAllocation;
+import org.astraea.balancer.log.LogPlacement;
+import org.astraea.cost.ClusterInfo;
+import org.astraea.cost.NodeInfo;
+
+/**
+ * The {@link ShufflePlanGenerator} proposes a new log placement based on the current log placement,
+ * but with a few random placement changes. <br>
+ * <br>
+ * The following operations are considered as a valid shuffle action:
+ *
+ * <ol>
+ *   <li>Remove a replica from the replica set, then add another broker(must not be part of the
+ *       replica set before this action) into the replica set. Noted that this operation doesn't
+ *       specify which data directory the moving replica will eventually be on the destination
+ *       broker. It's likely that the replica will follow the default placement scheme, see the
+ *       LogManager#nextLogDirs method implementation in Apache Kafka server for more details.
+ *   <li>Change the leader/follower of a partition by a member of this replica set, the original
+ *       leader/follower becomes a follower/leader.
+ *   <li>Change the data directory of one replica from current data directory to another on the
+ *       current broker.
+ * </ol>
+ */
+public class ShufflePlanGenerator implements RebalancePlanGenerator {
+
+  private final Supplier<Integer> numberOfShuffle;
+
+  public ShufflePlanGenerator(int origin, int bound) {
+    this(() -> ThreadLocalRandom.current().nextInt(origin, bound));
+  }
+
+  public ShufflePlanGenerator(Supplier<Integer> numberOfShuffle) {
+    this.numberOfShuffle = numberOfShuffle;
+  }
+
+  private int sourceTopicPartitionSelector(List<TopicPartition> migrationCandidates) {
+    return ThreadLocalRandom.current().nextInt(0, migrationCandidates.size());
+  }
+
+  private int sourceLogPlacementSelector(List<LogPlacement> migrationCandidates) {
+    return ThreadLocalRandom.current().nextInt(0, migrationCandidates.size());
+  }
+
+  private int migrationSelector(List<Movement> movementCandidates) {
+    // There are two kinds of Movement, each operating on different kind of candidate.
+    // * The Movement#replicaSetMovement selecting target by the broker ID
+    // * The Movement#dataDirectoryMovement selecting target by the data dir on the specific broker
+    // If we mix all these movements and randomly picking one as the selected migration, doing so
+    // will cause fairness issue. Given a concrete example, there are 100 brokers, and each broker
+    // have 3 data directories. Now the leader log of a topic/partition with 3 replicas is being
+    // selected. How many possible migrations over here? The Answer is:
+    // 1. 99 from Movement#replicaSetMigration since the leader log can move to any other broker.
+    // 2. 2 from Movement#dataDirectoryMigration since the leader log can move to the other 2 dirs.
+    // So there are 101 possible migrations. But these "two ways" of migration have different
+    // probability to occur. We can expect a large number of replica set changes occur but only a
+    // few for the data directory migration. This will make the ShufflePlanGenerator hard to address
+    // specific kind of balance issue due to <strong>it rarely propose them</strong>.
+    final var migrationTypeList =
+        IntStream.range(0, movementCandidates.size())
+            .mapToObj(i -> Map.entry(movementCandidates.get(i).getClass(), i))
+            .collect(
+                Collectors.groupingBy(
+                    Map.Entry::getKey,
+                    Collectors.mapping(Map.Entry::getValue, Collectors.toUnmodifiableList())))
+            .values()
+            .stream()
+            .collect(Collectors.toUnmodifiableList());
+    final var selectedMigration =
+        migrationTypeList.get(ThreadLocalRandom.current().nextInt(0, migrationTypeList.size()));
+    return selectedMigration.get(ThreadLocalRandom.current().nextInt(0, selectedMigration.size()));
+  }
+
+  @Override
+  public Stream<RebalancePlanProposal> generate(
+      ClusterInfo clusterInfo, ClusterLogAllocation baseAllocation) {
+    return Stream.generate(
+        () -> {
+          final var rebalancePlanBuilder = RebalancePlanProposal.builder();
+          final var brokerIds =
+              clusterInfo.nodes().stream()
+                  .map(NodeInfo::id)
+                  .collect(Collectors.toUnmodifiableSet());
+
+          if (brokerIds.size() == 0)
+            return rebalancePlanBuilder
+                .addWarning("Why there is no broker?")
+                .noRebalancePlan()
+                .build();
+
+          if (brokerIds.size() == 1)
+            return rebalancePlanBuilder
+                .addWarning("Only one broker exists. There is no reason to rebalance.")
+                .noRebalancePlan()
+                .build();
+
+          if (clusterInfo.topics().size() == 0)
+            return rebalancePlanBuilder
+                .addWarning("No non-ignored topic to working on.")
+                .noRebalancePlan()
+                .build();
+
+          final var shuffleCount = numberOfShuffle.get();
+          final var newAllocation = LayeredClusterLogAllocation.of(baseAllocation);
+          final var pickingList =
+              newAllocation.topicPartitionStream().collect(Collectors.toUnmodifiableList());
+
+          rebalancePlanBuilder.addInfo(
+              "Make " + shuffleCount + (shuffleCount > 0 ? " shuffles." : " shuffle."));
+          for (int i = 0; i < shuffleCount; i++) {
+            final var sourceTopicPartitionIndex = sourceTopicPartitionSelector(pickingList);
+            final var sourceTopicPartition = pickingList.get(sourceTopicPartitionIndex);
+            final var sourceLogPlacements = newAllocation.logPlacements(sourceTopicPartition);
+            final var sourceLogPlacementIndex = sourceLogPlacementSelector(sourceLogPlacements);
+            final var sourceLogPlacement = sourceLogPlacements.get(sourceLogPlacementIndex);
+            final var sourceIsLeader = sourceLogPlacementIndex == 0;
+            final var sourceBroker = sourceLogPlacement.broker();
+
+            Consumer<Integer> replicaSetMigration =
+                (targetBroker) -> {
+                  newAllocation.migrateReplica(sourceTopicPartition, sourceBroker, targetBroker);
+                  rebalancePlanBuilder.addInfo(
+                      String.format(
+                          "Change replica set of topic %s partition %d, from %d to %d.",
+                          sourceTopicPartition.topic(),
+                          sourceTopicPartition.partition(),
+                          sourceLogPlacement.broker(),
+                          targetBroker));
+                };
+            Consumer<LogPlacement> leaderFollowerMigration =
+                (newLeaderReplicaCandidate) -> {
+                  newAllocation.letReplicaBecomeLeader(
+                      sourceTopicPartition, newLeaderReplicaCandidate.broker());
+                  rebalancePlanBuilder.addInfo(
+                      String.format(
+                          "Change the log identity of topic %s partition %d replica at broker %d, from %s to %s",
+                          sourceTopicPartition.topic(),
+                          sourceTopicPartition.partition(),
+                          sourceLogPlacement.broker(),
+                          sourceIsLeader ? "leader" : "follower",
+                          sourceIsLeader ? "follower" : "leader"));
+                };
+            Consumer<String> dataDirectoryMigration =
+                (dataDirectory) -> {
+                  newAllocation.changeDataDirectory(
+                      sourceTopicPartition, sourceBroker, dataDirectory);
+                  rebalancePlanBuilder.addInfo(
+                      String.format(
+                          "Change the data directory of topic %s partition %d replica at broker %d, from %s to %s",
+                          sourceTopicPartition.topic(),
+                          sourceTopicPartition.partition(),
+                          sourceLogPlacement.broker(),
+                          sourceLogPlacement.logDirectory().map(Object::toString).orElse("unknown"),
+                          dataDirectory));
+                };
+
+            // generate a set of valid migration broker for given placement.
+            final var validMigrationCandidates = new ArrayList<Movement>();
+            // [Valid movement 1] add all brokers and remove all broker in current replica set
+            brokerIds.stream()
+                .filter(
+                    broker -> sourceLogPlacements.stream().noneMatch(log -> log.broker() == broker))
+                .map(targetBroker -> (Runnable) () -> replicaSetMigration.accept(targetBroker))
+                .map(Movement::replicaSetMovement)
+                .forEach(validMigrationCandidates::add);
+            // [Valid movement 2] add all leader/follower change candidate
+            if (sourceLogPlacements.size() > 1) {
+              if (sourceIsLeader) {
+                // leader can migrate its identity to follower.
+                sourceLogPlacements.stream()
+                    .skip(1)
+                    .map(
+                        followerReplica ->
+                            (Runnable) () -> leaderFollowerMigration.accept(followerReplica))
+                    .map(Movement::replicaSetMovement)
+                    .forEach(validMigrationCandidates::add);
+              } else {
+                // follower can migrate its identity to leader.
+                validMigrationCandidates.add(
+                    Movement.replicaSetMovement(
+                        () -> leaderFollowerMigration.accept(sourceLogPlacement)));
+              }
+            }
+            // [Valid movement 3] change the data directory of selected replica
+            clusterInfo.dataDirectories(sourceLogPlacement.broker()).stream()
+                .filter(dir -> !dir.equals(sourceLogPlacement.logDirectory().orElse(null)))
+                .map(dir -> (Runnable) () -> dataDirectoryMigration.accept(dir))
+                .map(Movement::dataDirectoryMovement)
+                .forEach(validMigrationCandidates::add);
+
+            // pick a migration and execute
+            final var selectedMigrationIndex = migrationSelector(validMigrationCandidates);
+            validMigrationCandidates.get(selectedMigrationIndex).run();
+          }
+
+          return rebalancePlanBuilder.withRebalancePlan(newAllocation).build();
+        });
+  }
+
+  /** Action to trigger upon this migration */
+  interface Movement extends Runnable {
+    static ReplicaSetMovement replicaSetMovement(Runnable runnable) {
+      return runnable::run;
+    }
+
+    static DataDirectoryMovement dataDirectoryMovement(Runnable runnable) {
+      return runnable::run;
+    }
+  }
+
+  interface ReplicaSetMovement extends Movement {}
+
+  interface DataDirectoryMovement extends Movement {}
+}
diff --git a/app/src/main/java/org/astraea/balancer/log/ClusterLogAllocation.java b/app/src/main/java/org/astraea/balancer/log/ClusterLogAllocation.java
new file mode 100644
index 00000000..a0fe097b
--- /dev/null
+++ b/app/src/main/java/org/astraea/balancer/log/ClusterLogAllocation.java
@@ -0,0 +1,30 @@
+package org.astraea.balancer.log;
+
+import java.util.List;
+import java.util.stream.Stream;
+import org.astraea.admin.TopicPartition;
+
+/**
+ * Describe the log allocation state of a Kafka cluster. The implementation have to keep the cluster
+ * log allocation information, provide method for query the placement, and offer a set of log
+ * placement change operation.
+ */
+public interface ClusterLogAllocation {
+
+  /** let specific broker leave the replica set and let another broker join the replica set. */
+  void migrateReplica(TopicPartition topicPartition, int atBroker, int toBroker);
+
+  /** let specific follower log become the leader log of this topic/partition. */
+  void letReplicaBecomeLeader(TopicPartition topicPartition, int followerReplica);
+
+  /** change the data directory of specific log */
+  void changeDataDirectory(TopicPartition topicPartition, int atBroker, String newPath);
+
+  /** Retrieve the log placements of specific {@link TopicPartition}. */
+  List<LogPlacement> logPlacements(TopicPartition topicPartition);
+
+  /** Retrieve the stream of all topic/partition pairs in allocation. */
+  Stream<TopicPartition> topicPartitionStream();
+
+  // TODO: add a method to calculate the difference between two ClusterLogAllocation
+}
diff --git a/app/src/main/java/org/astraea/balancer/log/IllegalMigrationException.java b/app/src/main/java/org/astraea/balancer/log/IllegalMigrationException.java
new file mode 100644
index 00000000..bf3f26e8
--- /dev/null
+++ b/app/src/main/java/org/astraea/balancer/log/IllegalMigrationException.java
@@ -0,0 +1,7 @@
+package org.astraea.balancer.log;
+
+public class IllegalMigrationException extends RuntimeException {
+  public IllegalMigrationException(String s) {
+    super(s);
+  }
+}
diff --git a/app/src/main/java/org/astraea/balancer/log/LayeredClusterLogAllocation.java b/app/src/main/java/org/astraea/balancer/log/LayeredClusterLogAllocation.java
new file mode 100644
index 00000000..3588912c
--- /dev/null
+++ b/app/src/main/java/org/astraea/balancer/log/LayeredClusterLogAllocation.java
@@ -0,0 +1,263 @@
+package org.astraea.balancer.log;
+
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.OptionalInt;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+import org.astraea.admin.TopicPartition;
+import org.astraea.cost.ClusterInfo;
+import org.astraea.cost.ReplicaInfo;
+
+public class LayeredClusterLogAllocation implements ClusterLogAllocation {
+
+  // set operation is guard by this
+  private final AtomicBoolean isLocked = new AtomicBoolean(false);
+  private final LayeredClusterLogAllocation upperLayer;
+
+  // guard by this
+  private final Map<TopicPartition, List<LogPlacement>> allocation;
+
+  private LayeredClusterLogAllocation(
+      LayeredClusterLogAllocation upperLayer, Map<TopicPartition, List<LogPlacement>> allocation) {
+    allocation.keySet().stream()
+        .collect(Collectors.groupingBy(TopicPartition::topic))
+        .forEach(
+            (topic, tp) -> {
+              int maxPartitionId =
+                  tp.stream().mapToInt(TopicPartition::partition).max().orElseThrow();
+              if ((maxPartitionId + 1) != tp.size())
+                throw new IllegalArgumentException(
+                    "The partition size of " + topic + " is illegal");
+            });
+    allocation.forEach(
+        (tp, logs) -> {
+          long uniqueBrokers = logs.stream().map(LogPlacement::broker).distinct().count();
+          if (uniqueBrokers != logs.size() || logs.size() == 0)
+            throw new IllegalArgumentException(
+                "The topic "
+                    + tp.topic()
+                    + " partition "
+                    + tp.partition()
+                    + " has illegal replica set "
+                    + logs);
+        });
+    this.upperLayer = upperLayer;
+    this.allocation = allocation;
+  }
+
+  public static LayeredClusterLogAllocation of(ClusterInfo clusterInfo) {
+    final Map<TopicPartition, List<LogPlacement>> allocation =
+        clusterInfo.topics().stream()
+            .map(clusterInfo::partitions)
+            .flatMap(Collection::stream)
+            .collect(
+                Collectors.groupingBy(
+                    replica ->
+                        TopicPartition.of(replica.topic(), Integer.toString(replica.partition()))))
+            .entrySet()
+            .stream()
+            .map(
+                (entry) -> {
+                  // validate if the given log placements are valid
+                  if (entry.getValue().stream().filter(ReplicaInfo::isLeader).count() != 1)
+                    throw new IllegalArgumentException(
+                        "The " + entry.getKey() + " leader count mismatch 1.");
+
+                  final var topicPartition = entry.getKey();
+                  final var logPlacements =
+                      entry.getValue().stream()
+                          .sorted(Comparator.comparingInt(replica -> replica.isLeader() ? 0 : 1))
+                          .map(
+                              replica ->
+                                  LogPlacement.of(
+                                      replica.nodeInfo().id(), replica.dataFolder().orElse(null)))
+                          .collect(Collectors.toList());
+
+                  return Map.entry(topicPartition, logPlacements);
+                })
+            .collect(Collectors.toConcurrentMap(Map.Entry::getKey, Map.Entry::getValue));
+    return new LayeredClusterLogAllocation(null, allocation);
+  }
+
+  public static LayeredClusterLogAllocation of(
+      Map<TopicPartition, List<LogPlacement>> allocationMap) {
+    return new LayeredClusterLogAllocation(null, allocationMap);
+  }
+
+  /**
+   * Create a new {@link LayeredClusterLogAllocation}. If the given {@link ClusterLogAllocation} is
+   * an instance of {@link LayeredClusterLogAllocation}. Then the given {@link
+   * LayeredClusterLogAllocation} will be locked and become unmodifiable. The new log allocation
+   * will continue from the unmodifiable allocation. Doing so can improve the performance of new log
+   * allocation creation as we avoid the heavy data copy work.
+   */
+  public static LayeredClusterLogAllocation of(ClusterLogAllocation clusterLogAllocation) {
+    if (clusterLogAllocation instanceof LayeredClusterLogAllocation) {
+      return createNewLayer((LayeredClusterLogAllocation) clusterLogAllocation);
+    } else {
+      // no layer support, do the expensive copy work.
+      final var map = new ConcurrentHashMap<TopicPartition, List<LogPlacement>>();
+      clusterLogAllocation
+          .topicPartitionStream()
+          .forEach(
+              topicPartition ->
+                  map.put(topicPartition, clusterLogAllocation.logPlacements(topicPartition)));
+      return of(map);
+    }
+  }
+
+  /**
+   * Create a new {@link LayeredClusterLogAllocation} based on the given one. The given layered
+   * cluster log will become unmodifiable afterward.
+   */
+  private static LayeredClusterLogAllocation createNewLayer(
+      LayeredClusterLogAllocation baseLogAllocation) {
+    if (!baseLogAllocation.isLocked()) {
+      baseLogAllocation.lockLayer();
+    }
+    return new LayeredClusterLogAllocation(baseLogAllocation, new ConcurrentHashMap<>());
+  }
+
+  /** Make this {@link LayeredClusterLogAllocation} unmodifiable. */
+  private synchronized void lockLayer() {
+    isLocked.set(true);
+  }
+
+  /** Is this {@link LayeredClusterLogAllocation} unmodifiable. */
+  private boolean isLocked() {
+    return isLocked.get();
+  }
+
+  private synchronized void ensureNotLocked() {
+    if (isLocked())
+      throw new IllegalStateException("This layered cluster log allocation has been locked");
+  }
+
+  private static OptionalInt indexOfBroker(List<LogPlacement> logPlacements, int targetBroker) {
+    return IntStream.range(0, logPlacements.size())
+        .filter(index -> logPlacements.get(index).broker() == targetBroker)
+        .findFirst();
+  }
+
+  @Override
+  public synchronized void migrateReplica(
+      TopicPartition topicPartition, int broker, int destinationBroker) {
+    ensureNotLocked();
+
+    final List<LogPlacement> sourceLogPlacements = this.logPlacements(topicPartition);
+    if (sourceLogPlacements == null)
+      throw new IllegalMigrationException(
+          topicPartition.topic() + "-" + topicPartition.partition() + " no such topic/partition");
+
+    int sourceLogIndex = indexOfBroker(sourceLogPlacements, broker).orElse(-1);
+    if (sourceLogIndex == -1)
+      throw new IllegalMigrationException(
+          broker + " is not part of the replica set for " + topicPartition);
+
+    int destinationLogIndex = indexOfBroker(sourceLogPlacements, destinationBroker).orElse(-1);
+    if (destinationLogIndex != -1)
+      throw new IllegalArgumentException(
+          destinationBroker + " is already part of the replica set, no need to move");
+
+    this.allocation.put(
+        topicPartition,
+        IntStream.range(0, sourceLogPlacements.size())
+            .mapToObj(
+                index ->
+                    index == sourceLogIndex
+                        ? LogPlacement.of(destinationBroker)
+                        : sourceLogPlacements.get(index))
+            .collect(Collectors.toUnmodifiableList()));
+  }
+
+  @Override
+  public synchronized void letReplicaBecomeLeader(
+      TopicPartition topicPartition, int followerReplica) {
+    ensureNotLocked();
+
+    final List<LogPlacement> sourceLogPlacements = this.logPlacements(topicPartition);
+    if (sourceLogPlacements == null)
+      throw new IllegalMigrationException(
+          topicPartition.topic() + "-" + topicPartition.partition() + " no such topic/partition");
+
+    int leaderLogIndex = 0;
+    if (sourceLogPlacements.size() == 0)
+      throw new IllegalStateException("This partition has no log");
+
+    int followerLogIndex = indexOfBroker(sourceLogPlacements, followerReplica).orElse(-1);
+    if (followerLogIndex == -1)
+      throw new IllegalArgumentException(
+          followerReplica + " is not part of the replica set for " + topicPartition);
+
+    if (leaderLogIndex == followerLogIndex) return; // nothing to do
+
+    final var leaderLog = this.logPlacements(topicPartition).get(leaderLogIndex);
+    final var followerLog = this.logPlacements(topicPartition).get(followerLogIndex);
+
+    this.allocation.put(
+        topicPartition,
+        IntStream.range(0, sourceLogPlacements.size())
+            .mapToObj(
+                index ->
+                    index == leaderLogIndex
+                        ? followerLog
+                        : index == followerLogIndex ? leaderLog : sourceLogPlacements.get(index))
+            .collect(Collectors.toUnmodifiableList()));
+  }
+
+  @Override
+  public synchronized void changeDataDirectory(
+      TopicPartition topicPartition, int broker, String path) {
+    ensureNotLocked();
+
+    final List<LogPlacement> sourceLogPlacements = this.logPlacements(topicPartition);
+    if (sourceLogPlacements == null)
+      throw new IllegalMigrationException(
+          topicPartition.topic() + "-" + topicPartition.partition() + " no such topic/partition");
+
+    int sourceLogIndex = indexOfBroker(sourceLogPlacements, broker).orElse(-1);
+    if (sourceLogIndex == -1)
+      throw new IllegalMigrationException(
+          broker + " is not part of the replica set for " + topicPartition);
+
+    final var oldLog = this.logPlacements(topicPartition).get(sourceLogIndex);
+    final var newLog = LogPlacement.of(oldLog.broker(), path);
+    this.allocation.put(
+        topicPartition,
+        IntStream.range(0, sourceLogPlacements.size())
+            .mapToObj(index -> index == sourceLogIndex ? newLog : sourceLogPlacements.get(index))
+            .collect(Collectors.toUnmodifiableList()));
+  }
+
+  @Override
+  public List<LogPlacement> logPlacements(TopicPartition topicPartition) {
+    if (allocation.containsKey(topicPartition)) return allocation.get(topicPartition);
+    else if (upperLayer != null) return upperLayer.logPlacements(topicPartition);
+    else return null;
+  }
+
+  @Override
+  public Stream<TopicPartition> topicPartitionStream() {
+    if (upperLayer == null) return this.allocation.keySet().stream();
+    else {
+      LayeredClusterLogAllocation baseAllocationLookUp = this;
+      while (baseAllocationLookUp.upperLayer != null)
+        baseAllocationLookUp = baseAllocationLookUp.upperLayer;
+      // Ignore all the intermediate layers, just return the keys set of the base layer.
+      // THIS WILL WORK, AS LONG AS THE UPPER LAYERS DO NOT DO THE FOLLOWING.
+      // 1. delete topic (should balancer do this?)
+      // 2. create topic (should balancer do this?)
+      // 3. shrink partition size (impossible for the current Kafka)
+      // 4. expand partition size (should balancer do this?)
+      // Of course, we can consider every element in all the layers, but doing that hurt
+      // performance.
+      return baseAllocationLookUp.allocation.keySet().stream();
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/balancer/log/LogPlacement.java b/app/src/main/java/org/astraea/balancer/log/LogPlacement.java
new file mode 100644
index 00000000..2f75a23b
--- /dev/null
+++ b/app/src/main/java/org/astraea/balancer/log/LogPlacement.java
@@ -0,0 +1,43 @@
+package org.astraea.balancer.log;
+
+import java.util.Optional;
+
+/** This class describe the placement state of one kafka log. */
+public interface LogPlacement {
+
+  int broker();
+
+  Optional<String> logDirectory();
+
+  static LogPlacement of(int broker) {
+    return of(broker, null);
+  }
+
+  static LogPlacement of(int broker, String logDirectory) {
+    return new LogPlacement() {
+      @Override
+      public int broker() {
+        return broker;
+      }
+
+      @Override
+      public Optional<String> logDirectory() {
+        return Optional.ofNullable(logDirectory);
+      }
+
+      @Override
+      public boolean equals(Object obj) {
+        if (obj instanceof LogPlacement) {
+          final var that = (LogPlacement) obj;
+          return this.broker() == that.broker() && this.logDirectory().equals(that.logDirectory());
+        }
+        return false;
+      }
+
+      @Override
+      public String toString() {
+        return "LogPlacement{broker=" + broker() + " logDir=" + logDirectory() + "}";
+      }
+    };
+  }
+}
diff --git a/app/src/main/java/org/astraea/cost/ClusterInfo.java b/app/src/main/java/org/astraea/cost/ClusterInfo.java
index b147ff26..691ea3ab 100644
--- a/app/src/main/java/org/astraea/cost/ClusterInfo.java
+++ b/app/src/main/java/org/astraea/cost/ClusterInfo.java
@@ -20,6 +20,12 @@ public interface ClusterInfo {
         return cluster.nodes().stream().map(NodeInfo::of).collect(Collectors.toUnmodifiableList());
       }
 
+      @Override
+      public Set<String> dataDirectories(int brokerId) {
+        // org.apache.kafka.common.Cluster doesn't have such information.
+        throw new UnsupportedOperationException("This information is not available");
+      }
+
       public Set<String> topics() {
         return cluster.topics();
       }
@@ -82,6 +88,11 @@ public interface ClusterInfo {
         return cluster.nodes();
       }
 
+      @Override
+      public Set<String> dataDirectories(int brokerId) {
+        return cluster.dataDirectories(brokerId);
+      }
+
       @Override
       public List<ReplicaInfo> availablePartitionLeaders(String topic) {
         return cluster.availablePartitionLeaders(topic);
@@ -147,6 +158,10 @@ public interface ClusterInfo {
   /** @return The known set of nodes */
   List<NodeInfo> nodes();
 
+  /** @return return the data directories on specific broker */
+  Set<String> dataDirectories(int brokerId);
+  // TODO: provide a ClusterInfo implementation with this info
+
   /**
    * Get the list of replica leader information of each available partition for the given topic
    *
diff --git a/docker/start_prometheus.sh b/docker/start_prometheus.sh
index de0bacd4..843e2fd8 100755
--- a/docker/start_prometheus.sh
+++ b/docker/start_prometheus.sh
@@ -1,59 +1,23 @@
 #!/bin/bash
 
+declare -r DOCKER_FOLDER=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" &>/dev/null && pwd)
+source $DOCKER_FOLDER/docker_build_common.sh
+declare -r PROMETHEUS_PORT="${PROMETHEUS_PORT:-9090}"
+declare -r IMAGE_NAME=prom/prometheus:v2.32.1
+declare -r CONTAINER_NAME="prometheus-${PROMETHEUS_PORT}"
+declare -r CONFIGURATION_FILE="/tmp/prometheus-${PROMETHEUS_PORT}.yml"
+declare -r SCRAPE_INTERVAL="${SCRAPE_INTERVAL:-1s}"
+declare -r SCRAPE_TIMEOUT="${SCRAPE_TIMEOUT:-1s}"
+
 # =============================[functions]=============================
 
 function showHelp() {
   echo "Usage: start_prometheus.sh [ COMMAND ]"
   echo "COMMAND: "
-  echo "    start <kafka-broker-jmx-addresses> <node-exporter-addresses>    create/start the prometheus docker instance"
-  echo "    refresh <kafka-broker-jmx-addresses> <node-exporter-addresses>  refresh and apply the prometheus config"
-  echo "    refresh <config-file>                                           refresh and apply the prometheus config"
-  echo "    help                                                            show this dialog"
-}
-
-function gradle_project_directory {
-    local script_folder="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" &>/dev/null && pwd)"
-    local target_folder="$script_folder"
-
-    # attempt to find the nearest parent directory that contains the gradlew file
-    # we recognize such directory as a Gradle repo that contains the build directory
-    while [ "$target_folder" != "/" ]; do
-        target_folder="$(dirname $target_folder)"
-
-        if [ -f "$target_folder/gradlew" ]; then
-            break
-        fi
-    done
-
-    if [ "$target_folder" == "/" ]; then
-        echo "Cannot find the gradle directory from $script_folder."
-        echo "Is the directory exists?"
-        exit 5
-    fi
-
-    echo "$target_folder"
+  echo "    start <node-exporter-addresses>    create/start the prometheus docker instance"
+  echo "    help                               show this dialog"
 }
 
-# ===============================[checks]===============================
-
-if [[ "$(which docker)" == "" ]]; then
-  echo "you have to install docker"
-  exit 2
-fi
-
-if [[ "$(which ipconfig 2> /dev/null)" != "" ]]; then
-  address=$(ipconfig getifaddr en0)
-else
-  address=$(hostname -i)
-fi
-
-if [[ "$address" == "127.0.0.1" || "$address" == "127.0.1.1" ]]; then
-  echo "the address: Either 127.0.0.1 or 127.0.1.1 can't be used in this script. Please check /etc/hosts"
-  exit 2
-fi
-
-# =================================[main]=================================
-
 function wrap_address() {
   _value=""
   IFS=',' read -ra ADDR <<< "$1"
@@ -67,24 +31,10 @@ function wrap_address() {
   echo "$_value"
 }
 
-prometheus_port="${PROMETHEUS_PORT:-9090}"
-image_name=prom/prometheus:v2.32.1
-container_name="prometheus-${prometheus_port}"
-file_prefix="$(gradle_project_directory)/build/docker/prometheus"
-file="${file_prefix}/prometheus-${prometheus_port}.yml"
-temp_file="${file_prefix}/prometheus-${prometheus_port}-editing.yml"
-scrape_interval="${scrape_interval:-200ms}"
-scrape_timeout="${scrape_timeout:-200ms}"
-volume_name_1="prometheus-${prometheus_port}-etc"
-volume_name_2="prometheus-${prometheus_port}-prometheus"
-
-mkdir -p "$file_prefix"
-
 function write_config() {
-  kafka_jmx_addresses="$(wrap_address "$1")"
-  node_exporter_addresses="$(wrap_address "$2")"
+  node_exporter_addresses="$(wrap_address "$1")"
 
-  cat <<EOT > "$file"
+  cat <<EOT > "$CONFIGURATION_FILE"
 global:
   scrape_interval: 15s
   external_labels:
@@ -92,88 +42,44 @@ global:
 
 scrape_configs:
 
-  - job_name: 'kafka'
-    scrape_interval: $scrape_interval
-    scrape_timeout: $scrape_timeout
-    static_configs:
-      - targets: [$kafka_jmx_addresses]
-
   - job_name: 'node'
-    scrape_interval: $scrape_interval
-    scrape_timeout: $scrape_timeout
+    scrape_interval: $SCRAPE_INTERVAL
+    scrape_timeout: $SCRAPE_TIMEOUT
     static_configs:
       - targets: [$node_exporter_addresses]
 EOT
 }
 
-function is_prometheus_exists() {
-  docker ps -a | grep "$container_name" > /dev/null
-  if [ $? -eq 0 ]; then echo "yes"; else echo "no"; fi
-}
-
-function is_prometheus_running() {
-  docker ps | grep "$container_name" > /dev/null
-  if [ $? -eq 0 ]; then echo "yes"; else echo "no"; fi
-}
-
 function info() {
   # shellcheck disable=SC2068
   echo "[INFO]" $@
 }
 
-function refresh_config_from_file() {
-  info "Refresh prometheus configuration from config file $1"
-  docker cp "$1" "$container_name:/etc/prometheus/prometheus.yml"
-  docker kill --signal="SIGHUP" "$container_name"
-}
-
-function refresh_config() {
-  info "Refresh prometheus configuration"
-
-  write_config "$1" "$2"
-  docker cp "$file" "$container_name:/etc/prometheus/prometheus.yml"
-  docker kill --signal="SIGHUP" "$container_name"
-}
-
 function main() {
-  if [ "$(is_prometheus_running)" == "yes" ]; then
-    info "Prometheus is already running" at "http://$address:$prometheus_port"
-    showHelp
-    exit 0
-  fi
-
-  if [ "$(is_prometheus_exists)" == "yes" ]; then
-    info "Start existing prometheus instance"
-    docker start "$container_name"
-  else
-    info "Create a new prometheus docker instance"
-    docker run -d \
-        --name "$container_name" \
-        -p "$prometheus_port:9090" \
-        -v "${volume_name_1}:/etc/prometheus" \
-        -v "${volume_name_2}:/prometheus" \
-        "$image_name"
-  fi
-
-  refresh_config "$1" "$2"
+  docker run -d \
+      --name "$CONTAINER_NAME" \
+      -p "$PROMETHEUS_PORT:9090" \
+      -v "$CONFIGURATION_FILE:/etc/prometheus/prometheus.yml" \
+      "$IMAGE_NAME"
 
   info "================================================="
-  info "config file: $file"
-  info "prometheus address: http://${address}:$prometheus_port"
-  info "command to run grafana at this host: ./docker/start_grafana.sh start"
-  info "command to add prometheus to grafana datasource: ./docker/start_grafana.sh add_prom_source <USERNAME>:<PASSWORD> Prometheus http://$address:$prometheus_port"
+  info "config file: $CONFIGURATION_FILE"
+  info "prometheus address: http://${ADDRESS}:$PROMETHEUS_PORT"
+  info "command to run grafana at this host: $DOCKER_FOLDER/start_grafana.sh start"
+  info "command to add prometheus to grafana datasource: $DOCKER_FOLDER/start_grafana.sh add_prom_source <USERNAME>:<PASSWORD> Prometheus http://$ADDRESS:$PROMETHEUS_PORT"
   info "================================================="
 }
 
+# =================================[main]=================================
+
+checkDocker
+checkNetwork
+
 # shellcheck disable=SC2199
 if [[ "$1" == "start" ]]; then
   # shellcheck disable=SC2068
-  main "$2" "$3"
-elif [[ "$1" == "refresh" ]] && [[ "$#" -eq 3 ]]; then
-  # shellcheck disable=SC2068
-  refresh_config "$2" "$3"
-elif [[ "$1" == "refresh" ]] && [[ "$#" -eq 2 ]]; then
-  refresh_config_from_file "$2"
+  write_config "$2"
+  main
 elif [[ "$1" == "help" ]]; then
   showHelp
 else


Test patch:
diff --git a/app/src/test/java/org/astraea/balancer/RebalancePlanProposalTest.java b/app/src/test/java/org/astraea/balancer/RebalancePlanProposalTest.java
new file mode 100644
index 00000000..89c28eb7
--- /dev/null
+++ b/app/src/test/java/org/astraea/balancer/RebalancePlanProposalTest.java
@@ -0,0 +1,68 @@
+package org.astraea.balancer;
+
+import java.util.stream.Collectors;
+import org.astraea.balancer.log.LayeredClusterLogAllocation;
+import org.astraea.cost.ClusterInfoProvider;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class RebalancePlanProposalTest {
+
+  @Test
+  void testBuild() {
+    final var fakeClusterInfo = ClusterInfoProvider.fakeClusterInfo(10, 10, 10, 10);
+    final var thisAllocation = LayeredClusterLogAllocation.of(fakeClusterInfo);
+    final var build =
+        RebalancePlanProposal.builder()
+            .withRebalancePlan(thisAllocation)
+            .addInfo("Info0")
+            .addInfo("Info1")
+            .addInfo("Info2")
+            .addWarning("Warning0")
+            .addWarning("Warning1")
+            .addWarning("Warning2")
+            .build();
+
+    Assertions.assertTrue(build.rebalancePlan().isPresent());
+    final var thatAllocation = build.rebalancePlan().orElseThrow();
+    final var thisTps =
+        thisAllocation.topicPartitionStream().collect(Collectors.toUnmodifiableSet());
+    final var thatTps =
+        thatAllocation.topicPartitionStream().collect(Collectors.toUnmodifiableSet());
+    Assertions.assertEquals(thisTps, thatTps);
+    thisTps.forEach(
+        tp ->
+            Assertions.assertEquals(
+                thisAllocation.logPlacements(tp), thatAllocation.logPlacements(tp)));
+    Assertions.assertEquals("Info0", build.info().get(0));
+    Assertions.assertEquals("Info1", build.info().get(1));
+    Assertions.assertEquals("Info2", build.info().get(2));
+    Assertions.assertEquals("Warning0", build.warnings().get(0));
+    Assertions.assertEquals("Warning1", build.warnings().get(1));
+    Assertions.assertEquals("Warning2", build.warnings().get(2));
+  }
+
+  @Test
+  void testNoBuildTwice() {
+    // A builder should only build once. If a builder can build multiple times then it will have to
+    // do much copy work once a new build is requested. This will harm performance.
+    final var fakeClusterInfo = ClusterInfoProvider.fakeClusterInfo(10, 10, 10, 10);
+    final var logAllocation = LayeredClusterLogAllocation.of(fakeClusterInfo);
+    final var build = RebalancePlanProposal.builder().withRebalancePlan(logAllocation);
+
+    Assertions.assertDoesNotThrow(build::build);
+    Assertions.assertThrows(IllegalStateException.class, build::build);
+  }
+
+  @Test
+  void testNoModifyAfterBuild() {
+    final var fakeClusterInfo = ClusterInfoProvider.fakeClusterInfo(10, 10, 10, 10);
+    final var logAllocation = LayeredClusterLogAllocation.of(fakeClusterInfo);
+    final var build = RebalancePlanProposal.builder().withRebalancePlan(logAllocation);
+
+    RebalancePlanProposal proposal = build.build();
+    Assertions.assertThrows(
+        IllegalStateException.class, () -> build.addInfo("modify after built."));
+    Assertions.assertFalse(proposal.info().contains("modify after built."));
+  }
+}
diff --git a/app/src/test/java/org/astraea/balancer/generator/ShufflePlanGeneratorTest.java b/app/src/test/java/org/astraea/balancer/generator/ShufflePlanGeneratorTest.java
new file mode 100644
index 00000000..e6dac26e
--- /dev/null
+++ b/app/src/test/java/org/astraea/balancer/generator/ShufflePlanGeneratorTest.java
@@ -0,0 +1,173 @@
+package org.astraea.balancer.generator;
+
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.LongAdder;
+import java.util.stream.Collectors;
+import org.astraea.balancer.log.LayeredClusterLogAllocation;
+import org.astraea.cost.ClusterInfoProvider;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class ShufflePlanGeneratorTest {
+
+  @Test
+  void testRun() {
+    final var shufflePlanGenerator = new ShufflePlanGenerator(5, 10);
+    final var fakeCluster = ClusterInfoProvider.fakeClusterInfo(100, 10, 10, 3);
+    final var stream = shufflePlanGenerator.generate(fakeCluster);
+    final var iterator = stream.iterator();
+
+    Assertions.assertDoesNotThrow(() -> System.out.println(iterator.next()));
+    Assertions.assertDoesNotThrow(() -> System.out.println(iterator.next()));
+    Assertions.assertDoesNotThrow(() -> System.out.println(iterator.next()));
+  }
+
+  @ParameterizedTest
+  @ValueSource(ints = {3, 5, 7, 11, 13, 17, 19, 23, 29, 31, 301})
+  void testMovement(int shuffle) {
+    final var fakeClusterInfo = ClusterInfoProvider.fakeClusterInfo(30, 30, 20, 5);
+    final var allocation = LayeredClusterLogAllocation.of(fakeClusterInfo);
+    final var shufflePlanGenerator = new ShufflePlanGenerator(() -> shuffle);
+
+    shufflePlanGenerator
+        .generate(fakeClusterInfo)
+        .limit(100)
+        .forEach(
+            proposal -> {
+              final var that = proposal.rebalancePlan().orElseThrow();
+              final var thisTps =
+                  allocation.topicPartitionStream().collect(Collectors.toUnmodifiableSet());
+              final var thatTps =
+                  that.topicPartitionStream().collect(Collectors.toUnmodifiableSet());
+              final var thisMap =
+                  thisTps.stream()
+                      .collect(Collectors.toUnmodifiableMap(x -> x, allocation::logPlacements));
+              final var thatMap =
+                  thatTps.stream()
+                      .collect(Collectors.toUnmodifiableMap(x -> x, that::logPlacements));
+              Assertions.assertEquals(thisTps, thatTps);
+              Assertions.assertNotEquals(thisMap, thatMap);
+            });
+  }
+
+  @Test
+  void testNoNodes() {
+    final var fakeClusterInfo = ClusterInfoProvider.fakeClusterInfo(0, 0, 0, 0);
+    final var shufflePlanGenerator = new ShufflePlanGenerator(() -> 3);
+
+    final var proposal = shufflePlanGenerator.generate(fakeClusterInfo).iterator().next();
+
+    System.out.println(proposal);
+    Assertions.assertFalse(proposal.rebalancePlan().isPresent());
+    Assertions.assertTrue(proposal.warnings().size() >= 1);
+  }
+
+  @Test
+  void testOneNode() {
+    final var fakeClusterInfo = ClusterInfoProvider.fakeClusterInfo(1, 1, 1, 1);
+    final var shufflePlanGenerator = new ShufflePlanGenerator(() -> 3);
+
+    final var proposal = shufflePlanGenerator.generate(fakeClusterInfo).iterator().next();
+
+    System.out.println(proposal);
+    Assertions.assertFalse(proposal.rebalancePlan().isPresent());
+    Assertions.assertTrue(proposal.warnings().size() >= 1);
+  }
+
+  @Test
+  void testNoTopic() {
+    final var fakeClusterInfo = ClusterInfoProvider.fakeClusterInfo(3, 0, 0, 0);
+    final var shufflePlanGenerator = new ShufflePlanGenerator(() -> 3);
+
+    final var proposal = shufflePlanGenerator.generate(fakeClusterInfo).iterator().next();
+
+    System.out.println(proposal);
+    Assertions.assertFalse(proposal.rebalancePlan().isPresent());
+    Assertions.assertTrue(proposal.warnings().size() >= 1);
+  }
+
+  @ParameterizedTest(name = "[{0}] {1} nodes, {2} topics, {3} partitions, {4} replicas")
+  @CsvSource(
+      value = {
+        //      scenario, node, topic, partition, replica
+        "  small cluster,    3,    10,        30,       3",
+        " medium cluster,   30,    50,        50,       3",
+        "    big cluster,  100,   100,       100,       1",
+        "  many replicas, 1000,    30,       100,      30",
+      })
+  void performanceTest(
+      String scenario, int nodeCount, int topicCount, int partitionCount, int replicaCount) {
+    // This test is not intended for any performance guarantee.
+    // It only served the purpose of keeping track of the generator performance change in the CI
+    // log.
+    // Notice: Stream#limit() will hurt performance. the number here might not reflect the actual
+    // performance.
+    final var shufflePlanGenerator = new ShufflePlanGenerator(0, 10);
+    final var fakeClusterInfo =
+        ClusterInfoProvider.fakeClusterInfo(nodeCount, topicCount, partitionCount, replicaCount);
+    final var size = 1000;
+
+    final long s = System.nanoTime();
+    final var count = shufflePlanGenerator.generate(fakeClusterInfo).limit(size).count();
+    final long t = System.nanoTime();
+    Assertions.assertEquals(size, count);
+    System.out.printf("[%s]%n", scenario);
+    System.out.printf(
+        "%d nodes, %d topics, %d partitions, %d replicas.%n",
+        nodeCount, topicCount, partitionCount, replicaCount);
+    System.out.printf("Generate %.3f proposals per second.%n", count / (((double) (t - s) / 1e9)));
+    System.out.println();
+  }
+
+  @Test
+  void parallelStreamWorks() {
+    final var shufflePlanGenerator = new ShufflePlanGenerator(0, 10);
+    final var fakeClusterInfo = ClusterInfoProvider.fakeClusterInfo(10, 20, 10, 3);
+
+    // generator can do parallel without error.
+    Assertions.assertDoesNotThrow(
+        () -> shufflePlanGenerator.generate(fakeClusterInfo).parallel().limit(100).count());
+  }
+
+  @Test
+  @Disabled
+  void parallelPerformanceTests() throws InterruptedException {
+    final var shufflePlanGenerator = new ShufflePlanGenerator(0, 10);
+    final var fakeClusterInfo = ClusterInfoProvider.fakeClusterInfo(50, 500, 30, 2);
+    final var counter = new LongAdder();
+    final var forkJoinPool = new ForkJoinPool(ForkJoinPool.getCommonPoolParallelism());
+    final var startTime = System.nanoTime();
+
+    forkJoinPool.submit(
+        () ->
+            shufflePlanGenerator
+                .generate(fakeClusterInfo)
+                .parallel()
+                .forEach((ignore) -> counter.increment()));
+
+    // report progress
+    final var reportThread =
+        new Thread(
+            () -> {
+              while (!Thread.currentThread().isInterrupted()) {
+                try {
+                  TimeUnit.SECONDS.sleep(1);
+                  long now = System.nanoTime();
+                  System.out.printf(
+                      "%.3f proposal/s %n", counter.sum() / ((now - startTime) / 1e9));
+                } catch (InterruptedException e) {
+                  break;
+                }
+              }
+            });
+
+    reportThread.start();
+    reportThread.join();
+    forkJoinPool.shutdownNow();
+  }
+}
diff --git a/app/src/test/java/org/astraea/balancer/log/LayeredClusterLogAllocationTest.java b/app/src/test/java/org/astraea/balancer/log/LayeredClusterLogAllocationTest.java
new file mode 100644
index 00000000..d2c38609
--- /dev/null
+++ b/app/src/test/java/org/astraea/balancer/log/LayeredClusterLogAllocationTest.java
@@ -0,0 +1,166 @@
+package org.astraea.balancer.log;
+
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.astraea.admin.TopicPartition;
+import org.astraea.cost.ClusterInfoProvider;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class LayeredClusterLogAllocationTest {
+
+  @Test
+  void creation() {
+    // empty replica set
+    var badAllocation0 = Map.of(TopicPartition.of("topic", "0"), List.<LogPlacement>of());
+    Assertions.assertThrows(
+        IllegalArgumentException.class, () -> LayeredClusterLogAllocation.of(badAllocation0));
+
+    // partial topic/partition
+    var badAllocation1 = Map.of(TopicPartition.of("topic", "999"), List.of(LogPlacement.of(1001)));
+    Assertions.assertThrows(
+        IllegalArgumentException.class, () -> LayeredClusterLogAllocation.of(badAllocation1));
+
+    // duplicate replica
+    var badAllocation2 =
+        Map.of(
+            TopicPartition.of("topic", "0"),
+            List.of(LogPlacement.of(1001), LogPlacement.of(1001), LogPlacement.of(1001)));
+    Assertions.assertThrows(
+        IllegalArgumentException.class, () -> LayeredClusterLogAllocation.of(badAllocation2));
+  }
+
+  @Test
+  void migrateReplica() {
+    final var fakeClusterInfo =
+        ClusterInfoProvider.fakeClusterInfo(3, 1, 1, 1, (i) -> Set.of("topic"));
+    final var clusterLogAllocation = LayeredClusterLogAllocation.of(fakeClusterInfo);
+    final var sourceTopicPartition = TopicPartition.of("topic", "0");
+
+    clusterLogAllocation.migrateReplica(sourceTopicPartition, 0, 1);
+
+    Assertions.assertEquals(
+        1, clusterLogAllocation.logPlacements(sourceTopicPartition).get(0).broker());
+    Assertions.assertDoesNotThrow(() -> LayeredClusterLogAllocation.of(clusterLogAllocation));
+  }
+
+  @Test
+  void letReplicaBecomeLeader() {
+    final var fakeClusterInfo =
+        ClusterInfoProvider.fakeClusterInfo(3, 1, 1, 2, (i) -> Set.of("topic"));
+    final var clusterLogAllocation = LayeredClusterLogAllocation.of(fakeClusterInfo);
+    final var sourceTopicPartition = TopicPartition.of("topic", "0");
+
+    clusterLogAllocation.letReplicaBecomeLeader(sourceTopicPartition, 1);
+
+    Assertions.assertEquals(
+        1, clusterLogAllocation.logPlacements(sourceTopicPartition).get(0).broker());
+    Assertions.assertEquals(
+        0, clusterLogAllocation.logPlacements(sourceTopicPartition).get(1).broker());
+    Assertions.assertDoesNotThrow(() -> LayeredClusterLogAllocation.of(clusterLogAllocation));
+  }
+
+  @Test
+  void changeDataDirectory() {
+    final var fakeClusterInfo =
+        ClusterInfoProvider.fakeClusterInfo(3, 1, 1, 1, (i) -> Set.of("topic"));
+    final var clusterLogAllocation = LayeredClusterLogAllocation.of(fakeClusterInfo);
+    final var sourceTopicPartition = TopicPartition.of("topic", "0");
+
+    clusterLogAllocation.changeDataDirectory(sourceTopicPartition, 0, "/path/to/somewhere");
+
+    Assertions.assertEquals(
+        "/path/to/somewhere",
+        clusterLogAllocation
+            .logPlacements(sourceTopicPartition)
+            .get(0)
+            .logDirectory()
+            .orElseThrow());
+    Assertions.assertDoesNotThrow(() -> LayeredClusterLogAllocation.of(clusterLogAllocation));
+  }
+
+  @Test
+  void logPlacements() {
+    final var allocation =
+        LayeredClusterLogAllocation.of(
+            Map.of(TopicPartition.of("topic", "0"), List.of(LogPlacement.of(0, "/nowhere"))));
+
+    Assertions.assertEquals(1, allocation.logPlacements(TopicPartition.of("topic", "0")).size());
+    Assertions.assertEquals(
+        0, allocation.logPlacements(TopicPartition.of("topic", "0")).get(0).broker());
+    Assertions.assertEquals(
+        "/nowhere",
+        allocation
+            .logPlacements(TopicPartition.of("topic", "0"))
+            .get(0)
+            .logDirectory()
+            .orElseThrow());
+    Assertions.assertNull(allocation.logPlacements(TopicPartition.of("no", "0")));
+    allocation.logPlacements(TopicPartition.of("no", "0"));
+  }
+
+  @Test
+  void topicPartitionStream() {
+    final var fakeClusterInfo = ClusterInfoProvider.fakeClusterInfo(10, 10, 10, 3);
+    final var allocation0 = LayeredClusterLogAllocation.of(fakeClusterInfo);
+    final var allocation1 = LayeredClusterLogAllocation.of(allocation0);
+    fakeClusterInfo.topics().stream()
+        .flatMap(x -> fakeClusterInfo.partitions(x).stream())
+        .forEach(
+            replica ->
+                allocation1.changeDataDirectory(
+                    TopicPartition.of(replica.topic(), Integer.toString(replica.partition())),
+                    0,
+                    "/yeah"));
+
+    final var allTopicPartitions =
+        allocation1
+            .topicPartitionStream()
+            .sorted(
+                Comparator.comparing(TopicPartition::topic)
+                    .thenComparing(TopicPartition::partition))
+            .collect(Collectors.toUnmodifiableList());
+    Assertions.assertEquals(10 * 10, allTopicPartitions.size());
+    final var expectedTopicPartitions =
+        fakeClusterInfo.topics().stream()
+            .flatMap(x -> fakeClusterInfo.partitions(x).stream())
+            .map(x -> TopicPartition.of(x.topic(), Integer.toString(x.partition())))
+            .distinct()
+            .sorted(
+                Comparator.comparing(TopicPartition::topic)
+                    .thenComparing(TopicPartition::partition))
+            .collect(Collectors.toUnmodifiableList());
+    Assertions.assertEquals(expectedTopicPartitions, allTopicPartitions);
+  }
+
+  @Test
+  void lockWorks() {
+    final var fakeClusterInfo = ClusterInfoProvider.fakeClusterInfo(10, 10, 10, 3);
+    final var allocation = LayeredClusterLogAllocation.of(fakeClusterInfo);
+    final var toModify = allocation.topicPartitionStream().findFirst().orElseThrow();
+
+    // can modify before lock
+    Assertions.assertDoesNotThrow(() -> allocation.migrateReplica(toModify, 0, 9));
+    Assertions.assertDoesNotThrow(() -> allocation.letReplicaBecomeLeader(toModify, 1));
+    Assertions.assertDoesNotThrow(() -> allocation.changeDataDirectory(toModify, 2, "/nowhere"));
+
+    final var extended = LayeredClusterLogAllocation.of(allocation);
+
+    // cannot modify after some other layer rely on it
+    Assertions.assertThrows(
+        IllegalStateException.class, () -> allocation.migrateReplica(toModify, 9, 0));
+    Assertions.assertThrows(
+        IllegalStateException.class, () -> allocation.letReplicaBecomeLeader(toModify, 0));
+    Assertions.assertThrows(
+        IllegalStateException.class,
+        () -> allocation.changeDataDirectory(toModify, 2, "/anywhere"));
+
+    // the extended one can modify
+    Assertions.assertDoesNotThrow(() -> extended.migrateReplica(toModify, 9, 0));
+    Assertions.assertDoesNotThrow(() -> extended.letReplicaBecomeLeader(toModify, 0));
+    Assertions.assertDoesNotThrow(() -> extended.changeDataDirectory(toModify, 2, "/anywhere"));
+  }
+}
diff --git a/app/src/test/java/org/astraea/cost/ClusterInfoProvider.java b/app/src/test/java/org/astraea/cost/ClusterInfoProvider.java
new file mode 100644
index 00000000..509f5540
--- /dev/null
+++ b/app/src/test/java/org/astraea/cost/ClusterInfoProvider.java
@@ -0,0 +1,124 @@
+package org.astraea.cost;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.astraea.admin.TopicPartition;
+import org.astraea.metrics.HasBeanObject;
+
+public class ClusterInfoProvider {
+
+  public static ClusterInfo fakeClusterInfo(
+      int nodeCount, int topicCount, int partitionCount, int replicaCount) {
+    final var random = new Random();
+    random.setSeed(0);
+
+    return fakeClusterInfo(
+        nodeCount,
+        topicCount,
+        partitionCount,
+        replicaCount,
+        (partition) ->
+            IntStream.range(0, partition)
+                .mapToObj(
+                    (ignore) -> {
+                      final var suffix =
+                          IntStream.range(0, 5)
+                              .map(i -> random.nextInt(26))
+                              .map(i -> 'a' + i)
+                              .mapToObj(i -> String.valueOf((char) i))
+                              .collect(Collectors.joining());
+                      return "fake-topic-" + suffix;
+                    })
+                .collect(Collectors.toUnmodifiableSet()));
+  }
+
+  public static ClusterInfo fakeClusterInfo(
+      int nodeCount,
+      int topicCount,
+      int partitionCount,
+      int replicaCount,
+      Function<Integer, Set<String>> topicNameGenerator) {
+    final var nodes =
+        IntStream.range(0, nodeCount)
+            .mapToObj(nodeId -> NodeInfo.of(nodeId, "host" + nodeId, 9092))
+            .collect(Collectors.toUnmodifiableList());
+    final var dataDirCount = 3;
+    final var dataDirectories =
+        IntStream.range(0, dataDirCount)
+            .mapToObj(i -> "/tmp/data-directory-" + i)
+            .collect(Collectors.toUnmodifiableSet());
+    final var dataDirectoryList = List.copyOf(dataDirectories);
+    final var topics = topicNameGenerator.apply(topicCount);
+    final var replicas =
+        topics.stream()
+            .flatMap(
+                topic ->
+                    IntStream.range(0, partitionCount)
+                        .mapToObj(p -> TopicPartition.of(topic, Integer.toString(p))))
+            .flatMap(
+                tp ->
+                    IntStream.range(0, replicaCount)
+                        .mapToObj(
+                            r ->
+                                ReplicaInfo.of(
+                                    tp.topic(),
+                                    tp.partition(),
+                                    nodes.get(r),
+                                    r == 0,
+                                    true,
+                                    false,
+                                    dataDirectoryList.get(
+                                        tp.partition() % dataDirectories.size()))))
+            .collect(Collectors.groupingBy(ReplicaInfo::topic));
+
+    return new ClusterInfo() {
+      @Override
+      public List<NodeInfo> nodes() {
+        return nodes;
+      }
+
+      @Override
+      public Set<String> dataDirectories(int brokerId) {
+        return dataDirectories;
+      }
+
+      @Override
+      public List<ReplicaInfo> availablePartitionLeaders(String topic) {
+        return partitions(topic).stream()
+            .filter(ReplicaInfo::isLeader)
+            .collect(Collectors.toUnmodifiableList());
+      }
+
+      @Override
+      public List<ReplicaInfo> availablePartitions(String topic) {
+        return partitions(topic);
+      }
+
+      @Override
+      public Set<String> topics() {
+        return topics;
+      }
+
+      @Override
+      public List<ReplicaInfo> partitions(String topic) {
+        return replicas.get(topic);
+      }
+
+      @Override
+      public Collection<HasBeanObject> beans(int brokerId) {
+        throw new UnsupportedOperationException();
+      }
+
+      @Override
+      public Map<Integer, Collection<HasBeanObject>> allBeans() {
+        throw new UnsupportedOperationException();
+      }
+    };
+  }
+}
diff --git a/app/src/test/java/org/astraea/cost/FakeClusterInfo.java b/app/src/test/java/org/astraea/cost/FakeClusterInfo.java
index fd0f1d75..bae2062e 100644
--- a/app/src/test/java/org/astraea/cost/FakeClusterInfo.java
+++ b/app/src/test/java/org/astraea/cost/FakeClusterInfo.java
@@ -12,6 +12,11 @@ public class FakeClusterInfo implements ClusterInfo {
     return List.of();
   }
 
+  @Override
+  public Set<String> dataDirectories(int brokerId) {
+    return Set.of();
+  }
+
   @Override
   public List<ReplicaInfo> availablePartitionLeaders(String topic) {
     return List.of();