Repository: skiptests/astraea
Pull request #357 (2022-05-24 09:00:30)

Decrease number of records buffered
#323 

解決consumer throughput 時常為0 的怪象。

發生原因於 #323 留言中解釋。


Patch:
diff --git a/.gitignore b/.gitignore
index 810e79ed..3f50cc0e 100644
--- a/.gitignore
+++ b/.gitignore
@@ -15,5 +15,6 @@ zookeeper.dockerfile
 broker.dockerfile
 kafka_tool.dockerfile
 spark.dockerfile
+deps.dockerfile
 # we don't put binary file to git repo
 gradle-wrapper.jar
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/build.gradle b/app/build.gradle
index b4ec5650..55cca03a 100644
--- a/app/build.gradle
+++ b/app/build.gradle
@@ -17,6 +17,8 @@ def versions = [
     slf4j: project.properties['slf4j.version'] ?: "1.7.32",
     zookeeper: project.properties['zookeeper.version'] ?: "3.6.3",
     gson: project.properties['gson.version'] ?: "2.9.0",
+    wix: project.properties['wix.version'] ?: "4.6.2",
+    mysql: project.properties['mysql.version'] ?: "8.0.29",
 ]
 
 dependencies {
@@ -25,6 +27,8 @@ dependencies {
     testImplementation "org.mockito:mockito-inline:${versions["mockito"]}"
     testImplementation "org.apache.zookeeper:zookeeper:${versions["zookeeper"]}"
     testImplementation "org.apache.kafka:kafka_2.12:${versions["kafka"]}"
+    testImplementation "com.wix:wix-embedded-mysql:${versions["wix"]}"
+    testImplementation "mysql:mysql-connector-java:${versions["mysql"]}"
 
 
     implementation "org.apache.kafka:kafka-clients:${versions["kafka"]}"
diff --git a/app/src/main/java/org/astraea/admin/Builder.java b/app/src/main/java/org/astraea/admin/Builder.java
index 8906b422..cce17579 100644
--- a/app/src/main/java/org/astraea/admin/Builder.java
+++ b/app/src/main/java/org/astraea/admin/Builder.java
@@ -32,7 +32,7 @@ import org.apache.kafka.common.quota.ClientQuotaAlteration;
 import org.apache.kafka.common.quota.ClientQuotaEntity;
 import org.apache.kafka.common.quota.ClientQuotaFilter;
 import org.apache.kafka.common.quota.ClientQuotaFilterComponent;
-import org.astraea.Utils;
+import org.astraea.common.Utils;
 
 public class Builder {
 
@@ -69,7 +69,7 @@ public class Builder {
 
     @Override
     public Set<Integer> brokerIds() {
-      return Utils.handleException(
+      return Utils.packException(
           () ->
               admin.describeCluster().nodes().get().stream()
                   .map(Node::id)
@@ -78,7 +78,7 @@ public class Builder {
 
     @Override
     public Map<Integer, Set<String>> brokerFolders(Set<Integer> brokers) {
-      return Utils.handleException(
+      return Utils.packException(
           () ->
               admin.describeLogDirs(brokers).allDescriptions().get().entrySet().stream()
                   .collect(Collectors.toMap(Map.Entry::getKey, map -> map.getValue().keySet())));
@@ -92,7 +92,7 @@ public class Builder {
     @Override
     public Map<TopicPartition, Collection<ProducerState>> producerStates(
         Set<TopicPartition> partitions) {
-      return Utils.handleException(
+      return Utils.packException(
           () ->
               admin
                   .describeProducers(
@@ -115,14 +115,14 @@ public class Builder {
 
     @Override
     public Set<String> consumerGroupIds() {
-      return Utils.handleException(() -> admin.listConsumerGroups().all().get()).stream()
+      return Utils.packException(() -> admin.listConsumerGroups().all().get()).stream()
           .map(ConsumerGroupListing::groupId)
           .collect(Collectors.toUnmodifiableSet());
     }
 
     @Override
     public Map<String, ConsumerGroup> consumerGroups(Set<String> consumerGroupNames) {
-      return Utils.handleException(
+      return Utils.packException(
           () -> {
             var consumerGroupDescriptions =
                 admin.describeConsumerGroups(consumerGroupNames).all().get();
@@ -134,7 +134,7 @@ public class Builder {
                         x ->
                             Map.entry(
                                 x.getKey(),
-                                Utils.handleException(
+                                Utils.packException(
                                     () -> x.getValue().partitionsToOffsetAndMetadata().get())))
                     .collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, Map.Entry::getValue));
 
@@ -175,7 +175,7 @@ public class Builder {
 
     private Map<TopicPartition, Long> earliestOffset(Set<TopicPartition> partitions) {
 
-      return Utils.handleException(
+      return Utils.packException(
           () ->
               admin
                   .listOffsets(
@@ -193,7 +193,7 @@ public class Builder {
     }
 
     private Map<TopicPartition, Long> latestOffset(Set<TopicPartition> partitions) {
-      return Utils.handleException(
+      return Utils.packException(
           () ->
               admin
                   .listOffsets(
@@ -212,7 +212,7 @@ public class Builder {
 
     @Override
     public Map<String, Config> topics(Set<String> topicNames) {
-      return Utils.handleException(
+      return Utils.packException(
               () ->
                   admin
                       .describeConfigs(
@@ -228,13 +228,13 @@ public class Builder {
 
     @Override
     public Set<String> topicNames() {
-      return Utils.handleException(
+      return Utils.packException(
           () -> admin.listTopics(new ListTopicsOptions().listInternal(true)).names().get());
     }
 
     @Override
     public Map<Integer, Config> brokers(Set<Integer> brokerIds) {
-      return Utils.handleException(
+      return Utils.packException(
               () ->
                   admin
                       .describeConfigs(
@@ -267,7 +267,7 @@ public class Builder {
 
     @Override
     public Set<TopicPartition> partitions(Set<String> topics) {
-      return Utils.handleException(
+      return Utils.packException(
           () ->
               admin.describeTopics(topics).all().get().entrySet().stream()
                   .flatMap(
@@ -296,7 +296,7 @@ public class Builder {
     @Override
     public Map<TopicPartition, List<Replica>> replicas(Set<String> topics) {
       var replicaInfos =
-          Utils.handleException(() -> admin.describeLogDirs(brokerIds()).allDescriptions().get());
+          Utils.packException(() -> admin.describeLogDirs(brokerIds()).allDescriptions().get());
 
       BiFunction<Integer, TopicPartition, List<Map.Entry<String, ReplicaInfo>>> findReplicas =
           (id, partition) ->
@@ -311,7 +311,7 @@ public class Builder {
                       })
                   .collect(Collectors.toList());
 
-      return Utils.handleException(
+      return Utils.packException(
           () ->
               admin.describeTopics(topics).all().get().entrySet().stream()
                   .flatMap(
@@ -381,7 +381,7 @@ public class Builder {
 
     private Collection<Quota> quotas(ClientQuotaFilter filter) {
       return Quota.of(
-          Utils.handleException(() -> admin.describeClientQuotas(filter).entities().get()));
+          Utils.packException(() -> admin.describeClientQuotas(filter).entities().get()));
     }
   }
 
@@ -470,7 +470,7 @@ public class Builder {
 
     @Override
     public void create() {
-      if (Utils.handleException(() -> admin.listTopics().names().get()).contains(topic)) {
+      if (Utils.packException(() -> admin.listTopics().names().get()).contains(topic)) {
         var partitionReplicas = replicasGetter.apply(topic);
         partitionReplicas.forEach(
             (tp, replicas) -> {
@@ -483,7 +483,7 @@ public class Builder {
                         + numberOfReplicas);
             });
         var result =
-            Utils.handleException(() -> admin.describeTopics(Set.of(topic)).all().get().get(topic));
+            Utils.packException(() -> admin.describeTopics(Set.of(topic)).all().get().get(topic));
         if (result.partitions().size() != numberOfPartitions)
           throw new IllegalArgumentException(
               topic
@@ -512,7 +512,7 @@ public class Builder {
         return;
       }
 
-      Utils.handleException(
+      Utils.packException(
           () ->
               admin
                   .createTopics(
@@ -551,7 +551,7 @@ public class Builder {
 
     @Override
     public void moveTo(Map<Integer, String> brokerFolders) {
-      Utils.handleException(
+      Utils.packException(
           () ->
               admin.alterReplicaLogDirs(
                   brokerFolders.entrySet().stream()
@@ -567,7 +567,7 @@ public class Builder {
 
     @Override
     public void moveTo(List<Integer> brokers) {
-      Utils.handleException(
+      Utils.packException(
           () ->
               admin
                   .alterPartitionReassignments(
@@ -587,7 +587,7 @@ public class Builder {
       moveTo(all);
       // kafka produces error if re-election happens in single node
       if (!followers.isEmpty())
-        Utils.handleException(
+        Utils.packException(
             () ->
                 admin
                     .electLeaders(
@@ -619,7 +619,7 @@ public class Builder {
         @Override
         public void create() {
           if (connectionRate == Integer.MAX_VALUE) return;
-          Utils.handleException(
+          Utils.packException(
               () ->
                   admin
                       .alterClientQuotas(
@@ -666,7 +666,7 @@ public class Builder {
                 new ClientQuotaAlteration.Op(
                     Quota.Limit.CONSUMER_BYTE_RATE.nameOfKafka(), (double) consumeRate));
           if (!q.isEmpty())
-            Utils.handleException(
+            Utils.packException(
                 () ->
                     admin
                         .alterClientQuotas(
diff --git a/app/src/main/java/org/astraea/admin/ReplicaSyncingMonitor.java b/app/src/main/java/org/astraea/admin/ReplicaSyncingMonitor.java
index 91eb18c0..816a861f 100644
--- a/app/src/main/java/org/astraea/admin/ReplicaSyncingMonitor.java
+++ b/app/src/main/java/org/astraea/admin/ReplicaSyncingMonitor.java
@@ -16,12 +16,12 @@ import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import org.apache.kafka.common.TopicPartitionReplica;
-import org.astraea.Utils;
 import org.astraea.argument.DurationField;
 import org.astraea.argument.NonEmptyStringField;
-import org.astraea.utils.DataRate;
-import org.astraea.utils.DataSize;
-import org.astraea.utils.DataUnit;
+import org.astraea.common.DataRate;
+import org.astraea.common.DataSize;
+import org.astraea.common.DataUnit;
+import org.astraea.common.Utils;
 
 public class ReplicaSyncingMonitor {
 
@@ -168,7 +168,7 @@ public class ReplicaSyncingMonitor {
             .forEach(tpr -> previousCheckedSize.remove(tpr.getKey()));
       }
 
-      Utils.handleException(
+      Utils.packException(
           () -> {
             long expectedWaitNs = argument.interval.toNanos();
             long elapsedNs = (System.nanoTime() - startTime);
diff --git a/app/src/main/java/org/astraea/admin/TopicExplorer.java b/app/src/main/java/org/astraea/admin/TopicExplorer.java
index 6020e884..4c3cfb21 100644
--- a/app/src/main/java/org/astraea/admin/TopicExplorer.java
+++ b/app/src/main/java/org/astraea/admin/TopicExplorer.java
@@ -19,8 +19,8 @@ import java.util.stream.IntStream;
 import java.util.stream.Stream;
 import org.astraea.argument.NonEmptyStringField;
 import org.astraea.argument.StringSetField;
-import org.astraea.utils.DataSize;
-import org.astraea.utils.DataUnit;
+import org.astraea.common.DataSize;
+import org.astraea.common.DataUnit;
 
 public class TopicExplorer {
 
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..56f838da
--- /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::replicas)
+            .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/utils/DataRate.java b/app/src/main/java/org/astraea/common/DataRate.java
similarity index 99%
rename from app/src/main/java/org/astraea/utils/DataRate.java
rename to app/src/main/java/org/astraea/common/DataRate.java
index 94bd315a..40f1e686 100644
--- a/app/src/main/java/org/astraea/utils/DataRate.java
+++ b/app/src/main/java/org/astraea/common/DataRate.java
@@ -1,4 +1,4 @@
-package org.astraea.utils;
+package org.astraea.common;
 
 import java.math.BigDecimal;
 import java.math.BigInteger;
diff --git a/app/src/main/java/org/astraea/utils/DataSize.java b/app/src/main/java/org/astraea/common/DataSize.java
similarity index 99%
rename from app/src/main/java/org/astraea/utils/DataSize.java
rename to app/src/main/java/org/astraea/common/DataSize.java
index 98e9f774..f0165df9 100644
--- a/app/src/main/java/org/astraea/utils/DataSize.java
+++ b/app/src/main/java/org/astraea/common/DataSize.java
@@ -1,4 +1,4 @@
-package org.astraea.utils;
+package org.astraea.common;
 
 import java.math.BigDecimal;
 import java.math.BigInteger;
diff --git a/app/src/main/java/org/astraea/utils/DataUnit.java b/app/src/main/java/org/astraea/common/DataUnit.java
similarity index 99%
rename from app/src/main/java/org/astraea/utils/DataUnit.java
rename to app/src/main/java/org/astraea/common/DataUnit.java
index 69e2500b..a197d0ae 100644
--- a/app/src/main/java/org/astraea/utils/DataUnit.java
+++ b/app/src/main/java/org/astraea/common/DataUnit.java
@@ -1,4 +1,4 @@
-package org.astraea.utils;
+package org.astraea.common;
 
 import java.math.BigInteger;
 import java.util.Arrays;
diff --git a/app/src/main/java/org/astraea/Utils.java b/app/src/main/java/org/astraea/common/Utils.java
similarity index 59%
rename from app/src/main/java/org/astraea/Utils.java
rename to app/src/main/java/org/astraea/common/Utils.java
index 4fc5f55d..af9ee34f 100644
--- a/app/src/main/java/org/astraea/Utils.java
+++ b/app/src/main/java/org/astraea/common/Utils.java
@@ -1,16 +1,10 @@
-package org.astraea;
+package org.astraea.common;
 
 import java.io.File;
-import java.io.IOException;
-import java.io.UncheckedIOException;
 import java.net.InetAddress;
-import java.net.UnknownHostException;
+import java.net.ServerSocket;
 import java.nio.file.Files;
 import java.time.Duration;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
@@ -26,7 +20,15 @@ public final class Utils {
     return current;
   }
 
-  public static <R> R handleException(Getter<R> getter) {
+  /**
+   * Convert the exception thrown by getter to RuntimeException. This method can eliminate the
+   * exception from Java signature.
+   *
+   * @param getter to execute
+   * @param <R> type of returned object
+   * @return the object created by getter
+   */
+  public static <R> R packException(Getter<R> getter) {
     try {
       return getter.get();
     } catch (Throwable exception) {
@@ -37,49 +39,63 @@ public final class Utils {
     }
   }
 
+  /**
+   * Convert the exception thrown by getter to RuntimeException. This method can eliminate the
+   * exception from Java signature.
+   *
+   * @param runner to execute
+   */
+  public static void packException(Runner runner) {
+    Utils.packException(
+        () -> {
+          runner.run();
+          return null;
+        });
+  }
+
+  /**
+   * Swallow any exception caused by runner.
+   *
+   * @param runner to execute
+   */
+  public static void swallowException(Runner runner) {
+    Utils.packException(
+        () -> {
+          runner.run();
+          return null;
+        });
+  }
+
   public interface Getter<R> {
     R get() throws Exception;
   }
 
+  public interface Runner {
+    void run() throws Exception;
+  }
+
   /**
    * Delete the file or folder
    *
    * @param file path to file or folder
    */
   public static void delete(File file) {
-    try {
-      if (file.isDirectory()) {
-        var fs = file.listFiles();
-        if (fs != null) Stream.of(fs).forEach(Utils::delete);
-      }
-      Files.deleteIfExists(file.toPath());
-    } catch (IOException e) {
-      throw new IllegalStateException(e);
-    }
+    Utils.packException(
+        () -> {
+          if (file.isDirectory()) {
+            var fs = file.listFiles();
+            if (fs != null) Stream.of(fs).forEach(Utils::delete);
+          }
+          Files.deleteIfExists(file.toPath());
+        });
   }
 
   public static String hostname() {
-    try {
-      return InetAddress.getLocalHost().getHostName();
-    } catch (UnknownHostException e) {
-      throw new IllegalArgumentException(e);
-    }
+    return Utils.packException(() -> InetAddress.getLocalHost().getHostName());
   }
 
   public static File createTempDirectory(String prefix) {
-    try {
-      return Files.createTempDirectory(prefix).toFile();
-    } catch (IOException e) {
-      throw new UncheckedIOException(e);
-    }
-  }
-
-  public static void close(AutoCloseable closeable) {
-    try {
-      if (closeable != null) closeable.close();
-    } catch (Exception e) {
-      e.printStackTrace();
-    }
+    return Utils.packException(() -> Files.createTempDirectory(prefix).toFile());
   }
 
   /**
@@ -110,39 +126,6 @@ public final class Utils {
     throw new RuntimeException("Timeout to wait procedure");
   }
 
-  /**
-   * Get the field of the object.
-   *
-   * @param object reflected object.
-   * @param fieldName reflected field name.
-   * @return Required field.
-   */
-  public static Object requireField(Object object, String fieldName) {
-    try {
-      var field = object.getClass().getDeclaredField(fieldName);
-      field.setAccessible(true);
-      return field.get(object);
-    } catch (NoSuchFieldException | IllegalAccessException e) {
-      throw new IllegalArgumentException(e);
-    }
-  }
-
-  /**
-   * Separate host and port
-   *
-   * @param address like 0.0.0.0:9092
-   * @return (0.0.0.0,9092)
-   */
-  public static Map<String, Integer> requireHostPort(List<String> address) {
-    var mapAddress = new HashMap<String, Integer>();
-    address.forEach(
-        str ->
-            mapAddress.put(
-                Arrays.asList(str.split(":")).get(0),
-                Integer.parseInt(Arrays.asList(str.split(":")).get(1))));
-    return mapAddress;
-  }
-
   public static int requirePositive(int value) {
     if (value <= 0)
       throw new IllegalArgumentException("the value: " + value + " must be bigger than zero");
@@ -161,10 +144,7 @@ public final class Utils {
   }
 
   public static void sleep(Duration duration) {
-    try {
-      TimeUnit.MILLISECONDS.sleep(duration.toMillis());
-    } catch (InterruptedException ignored) {
-    }
+    Utils.swallowException(() -> TimeUnit.MILLISECONDS.sleep(duration.toMillis()));
   }
 
   public static String randomString(int len) {
@@ -184,5 +164,22 @@ public final class Utils {
     return java.util.UUID.randomUUID().toString().replaceAll("-", "");
   }
 
+  public static int availablePort() {
+    return Utils.packException(
+        () -> {
+          try (ServerSocket socket = new ServerSocket(0)) {
+            socket.setReuseAddress(true);
+            var port = socket.getLocalPort();
+            // the port smaller than 1024 may be protected by OS.
+            return port > 1024 ? port : port + 1024;
+          }
+        });
+  }
+
+  public static int resolvePort(int port) {
+    if (port <= 0) return availablePort();
+    return port;
+  }
+
   private Utils() {}
 }
diff --git a/app/src/main/java/org/astraea/consumer/Consumer.java b/app/src/main/java/org/astraea/consumer/Consumer.java
index e2234213..1848ed22 100644
--- a/app/src/main/java/org/astraea/consumer/Consumer.java
+++ b/app/src/main/java/org/astraea/consumer/Consumer.java
@@ -8,7 +8,7 @@ import java.util.Map;
 public interface Consumer<Key, Value> extends AutoCloseable {
 
   default Collection<Record<Key, Value>> poll(Duration timeout) {
-    return poll(Integer.MAX_VALUE, timeout);
+    return poll(1, timeout);
   }
 
   /**
diff --git a/app/src/main/java/org/astraea/cost/ClusterInfo.java b/app/src/main/java/org/astraea/cost/ClusterInfo.java
index b147ff26..408c459b 100644
--- a/app/src/main/java/org/astraea/cost/ClusterInfo.java
+++ b/app/src/main/java/org/astraea/cost/ClusterInfo.java
@@ -20,12 +20,18 @@ 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();
       }
 
       @Override
-      public List<ReplicaInfo> availablePartitionLeaders(String topic) {
+      public List<ReplicaInfo> availableReplicaLeaders(String topic) {
         return cluster.availablePartitionsForTopic(topic).stream()
             .map(ReplicaInfo::of)
             .map(
@@ -35,7 +41,7 @@ public interface ClusterInfo {
       }
 
       @Override
-      public List<ReplicaInfo> availablePartitions(String topic) {
+      public List<ReplicaInfo> availableReplicas(String topic) {
         return cluster.availablePartitionsForTopic(topic).stream()
             .map(ReplicaInfo::of)
             .flatMap(Collection::stream)
@@ -43,7 +49,7 @@ public interface ClusterInfo {
       }
 
       @Override
-      public List<ReplicaInfo> partitions(String topic) {
+      public List<ReplicaInfo> replicas(String topic) {
         return cluster.partitionsForTopic(topic).stream()
             .map(ReplicaInfo::of)
             .flatMap(Collection::stream)
@@ -83,13 +89,18 @@ public interface ClusterInfo {
       }
 
       @Override
-      public List<ReplicaInfo> availablePartitionLeaders(String topic) {
-        return cluster.availablePartitionLeaders(topic);
+      public Set<String> dataDirectories(int brokerId) {
+        return cluster.dataDirectories(brokerId);
       }
 
       @Override
-      public List<ReplicaInfo> availablePartitions(String topic) {
-        return cluster.availablePartitions(topic);
+      public List<ReplicaInfo> availableReplicaLeaders(String topic) {
+        return cluster.availableReplicaLeaders(topic);
+      }
+
+      @Override
+      public List<ReplicaInfo> availableReplicas(String topic) {
+        return cluster.availableReplicas(topic);
       }
 
       @Override
@@ -98,8 +109,8 @@ public interface ClusterInfo {
       }
 
       @Override
-      public List<ReplicaInfo> partitions(String topic) {
-        return cluster.partitions(topic);
+      public List<ReplicaInfo> replicas(String topic) {
+        return cluster.replicas(topic);
       }
 
       @Override
@@ -147,13 +158,17 @@ 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
    *
    * @param topic The Topic name
    * @return A list of {@link ReplicaInfo}
    */
-  List<ReplicaInfo> availablePartitionLeaders(String topic);
+  List<ReplicaInfo> availableReplicaLeaders(String topic);
 
   /**
    * Get the list of replica information of each available partition/replica pair for the given
@@ -162,7 +177,7 @@ public interface ClusterInfo {
    * @param topic The topic name
    * @return A list of {@link ReplicaInfo}
    */
-  List<ReplicaInfo> availablePartitions(String topic);
+  List<ReplicaInfo> availableReplicas(String topic);
 
   /**
    * All topic names
@@ -177,7 +192,7 @@ public interface ClusterInfo {
    * @param topic The topic name
    * @return A list of {@link ReplicaInfo}
    */
-  List<ReplicaInfo> partitions(String topic);
+  List<ReplicaInfo> replicas(String topic);
 
   /**
    * @param brokerId broker id
diff --git a/app/src/main/java/org/astraea/cost/Periodic.java b/app/src/main/java/org/astraea/cost/Periodic.java
index 03814ff8..3ddf331b 100644
--- a/app/src/main/java/org/astraea/cost/Periodic.java
+++ b/app/src/main/java/org/astraea/cost/Periodic.java
@@ -2,7 +2,7 @@ package org.astraea.cost;
 
 import java.time.Duration;
 import java.util.function.Supplier;
-import org.astraea.Utils;
+import org.astraea.common.Utils;
 
 /** Enables methods to be updated periodically, not calculated every time they are called. */
 public abstract class Periodic<Value> {
diff --git a/app/src/main/java/org/astraea/database/Builder.java b/app/src/main/java/org/astraea/database/Builder.java
new file mode 100644
index 00000000..b7ead224
--- /dev/null
+++ b/app/src/main/java/org/astraea/database/Builder.java
@@ -0,0 +1,231 @@
+package org.astraea.database;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.astraea.common.Utils;
+
+public class Builder {
+  private String url;
+  private String user;
+  private String password;
+
+  Builder() {}
+
+  public Builder url(String url) {
+    this.url = url;
+    return this;
+  }
+
+  public Builder user(String user) {
+    this.user = user;
+    return this;
+  }
+
+  public Builder password(String password) {
+    this.password = password;
+    return this;
+  }
+
+  public DatabaseClient build() {
+    return new DatabaseClient() {
+      private final Connection connection =
+          Utils.packException(
+              () ->
+                  DriverManager.getConnection(
+                      Objects.requireNonNull(url, "url must be defined"),
+                      Objects.requireNonNull(user, "user must be defined"),
+                      Objects.requireNonNull(password, "password must be defined")));
+
+      @Override
+      public TableQuery query() {
+        return new TableQuery() {
+          private String catalog;
+          private String schema;
+          private String tableName;
+
+          @Override
+          public TableQuery catalog(String catalog) {
+            this.catalog = catalog;
+            return this;
+          }
+
+          @Override
+          public TableQuery schema(String schema) {
+            this.schema = schema;
+            return this;
+          }
+
+          @Override
+          public TableQuery tableName(String tableName) {
+            this.tableName = tableName;
+            return this;
+          }
+
+          @Override
+          public Collection<TableInfo> run() {
+            return Utils.packException(
+                () -> {
+                  var md = connection.getMetaData();
+                  return catalogAndNames(md.getTables(catalog, schema, tableName, null)).stream()
+                      .map(
+                          e -> {
+                            var catalog = e.getKey();
+                            var tableName = e.getValue();
+                            var pks =
+                                primaryKeys(
+                                    Utils.packException(
+                                        () -> md.getPrimaryKeys(catalog, null, tableName)));
+                            var columnNameAndTypes =
+                                columnNameAndTypes(
+                                    Utils.packException(
+                                        () -> md.getColumns(catalog, null, tableName, null)));
+                            return new TableInfo(
+                                tableName,
+                                columnNameAndTypes.entrySet().stream()
+                                    .map(
+                                        c ->
+                                            new ColumnInfo(
+                                                c.getKey(), c.getValue(), pks.contains(c.getKey())))
+                                    .collect(Collectors.toUnmodifiableList()));
+                          })
+                      .collect(Collectors.toUnmodifiableList());
+                });
+          }
+        };
+      }
+
+      @Override
+      public TableCreator tableCreator() {
+        return new TableCreator() {
+          private String tableName;
+          private final Map<String, String> nameAndType = new LinkedHashMap<>();
+          private final Set<String> pks = new LinkedHashSet<>();
+
+          @Override
+          public TableCreator name(String name) {
+            this.tableName = name;
+            return this;
+          }
+
+          @Override
+          public TableCreator column(String name, String type) {
+            nameAndType.put(name, type);
+            return this;
+          }
+
+          @Override
+          public TableCreator primaryKey(String name, String type) {
+            nameAndType.put(name, type);
+            pks.add(name);
+            return this;
+          }
+
+          @Override
+          public void run() {
+            if (pks.isEmpty()) throw new IllegalArgumentException("primary keys must be defined");
+            var sql =
+                "CREATE TABLE \""
+                    + Objects.requireNonNull(tableName, "table name must be defined")
+                    + "\" ( "
+                    + nameAndType.entrySet().stream()
+                        .map(e -> "\"" + e.getKey() + "\" " + e.getValue())
+                        .collect(Collectors.joining(","))
+                    + ", PRIMARY KEY ( "
+                    + pks.stream().map(c -> "\"" + c + "\"").collect(Collectors.joining(","))
+                    + " ))";
+            execute(sql);
+          }
+        };
+      }
+
+      @Override
+      public void deleteTable(String name) {
+        execute("DROP TABLE \"" + name + "\"");
+      }
+
+      @Override
+      public void close() {
+        Utils.packException(connection::close);
+      }
+
+      private void execute(String sql) {
+        try (var statement = connection.createStatement()) {
+          statement.execute(sql);
+        } catch (SQLException e) {
+          throw new RuntimeException(e);
+        }
+      }
+    };
+  }
+
+  private static Map<String, String> columnNameAndTypes(ResultSet rs) {
+    try (rs) {
+      var result = new HashMap<String, String>();
+      while (rs.next()) result.put(columnName(rs), columnType(rs));
+      return Collections.unmodifiableMap(result);
+    } catch (SQLException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  private static Set<String> primaryKeys(ResultSet rs) {
+    try (rs) {
+      var result = new HashSet<String>();
+      while (rs.next()) result.add(columnName(rs));
+      return Collections.unmodifiableSet(result);
+    } catch (SQLException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  private static Collection<Map.Entry<String, String>> catalogAndNames(ResultSet rs) {
+    try (rs) {
+      var result = new ArrayList<Map.Entry<String, String>>();
+      while (rs.next()) {
+        if (isNormalTable(rs)) result.add(Map.entry(catalog(rs), tableName(rs)));
+      }
+      return Collections.unmodifiableList(result);
+    } catch (SQLException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  private static String catalog(ResultSet rs) {
+    return Utils.packException(() -> rs.getString("TABLE_CAT"));
+  }
+
+  private static String tableName(ResultSet rs) {
+    return Utils.packException(() -> rs.getString("TABLE_NAME"));
+  }
+
+  private static String columnName(ResultSet rs) {
+    return Utils.packException(() -> rs.getString("COLUMN_NAME"));
+  }
+
+  private static String columnType(ResultSet rs) {
+    return Utils.packException(() -> rs.getString("TYPE_NAME"));
+  }
+
+  private static boolean isNormalTable(ResultSet rs) {
+    return Utils.packException(
+        () -> {
+          var r = rs.getString("TABLE_TYPE");
+          if (r == null) return true;
+          return !Arrays.asList(r.split(" ")).contains("SYSTEM");
+        });
+  }
+}
diff --git a/app/src/main/java/org/astraea/database/ColumnInfo.java b/app/src/main/java/org/astraea/database/ColumnInfo.java
new file mode 100644
index 00000000..efd5ad9c
--- /dev/null
+++ b/app/src/main/java/org/astraea/database/ColumnInfo.java
@@ -0,0 +1,46 @@
+package org.astraea.database;
+
+import java.util.Objects;
+
+public class ColumnInfo {
+
+  private final String name;
+  private final String type;
+  private final boolean pk;
+
+  public ColumnInfo(String name, String type, boolean pk) {
+    this.name = name;
+    this.type = type;
+    this.pk = pk;
+  }
+
+  public String name() {
+    return name;
+  }
+
+  public String type() {
+    return type;
+  }
+
+  public boolean pk() {
+    return pk;
+  }
+
+  @Override
+  public String toString() {
+    return "ColumnInfo{" + "name='" + name + '\'' + ", type='" + type + '\'' + ", pk=" + pk + '}';
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) return true;
+    if (o == null || getClass() != o.getClass()) return false;
+    ColumnInfo that = (ColumnInfo) o;
+    return pk == that.pk && Objects.equals(name, that.name) && Objects.equals(type, that.type);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(name, type, pk);
+  }
+}
diff --git a/app/src/main/java/org/astraea/database/DatabaseClient.java b/app/src/main/java/org/astraea/database/DatabaseClient.java
new file mode 100644
index 00000000..2db85383
--- /dev/null
+++ b/app/src/main/java/org/astraea/database/DatabaseClient.java
@@ -0,0 +1,28 @@
+package org.astraea.database;
+
+public interface DatabaseClient extends AutoCloseable {
+
+  static Builder builder() {
+    return new Builder();
+  }
+
+  /**
+   * start to query tables
+   *
+   * @return Table Query
+   */
+  TableQuery query();
+
+  /**
+   * start to create new table
+   *
+   * @return Table Creator
+   */
+  TableCreator tableCreator();
+
+  /** @param name to delete */
+  void deleteTable(String name);
+
+  @Override
+  void close();
+}
diff --git a/app/src/main/java/org/astraea/database/TableCreator.java b/app/src/main/java/org/astraea/database/TableCreator.java
new file mode 100644
index 00000000..7758ef47
--- /dev/null
+++ b/app/src/main/java/org/astraea/database/TableCreator.java
@@ -0,0 +1,30 @@
+package org.astraea.database;
+
+public interface TableCreator {
+
+  /**
+   * @param name table name
+   * @return this creator
+   */
+  TableCreator name(String name);
+
+  /**
+   * create a normal column
+   *
+   * @param name column name
+   * @param type column type
+   * @return this creator
+   */
+  TableCreator column(String name, String type);
+
+  /**
+   * create a primary key
+   *
+   * @param name primary key name
+   * @param type primary key type
+   * @return this creator
+   */
+  TableCreator primaryKey(String name, String type);
+
+  void run();
+}
diff --git a/app/src/main/java/org/astraea/database/TableInfo.java b/app/src/main/java/org/astraea/database/TableInfo.java
new file mode 100644
index 00000000..7c382c48
--- /dev/null
+++ b/app/src/main/java/org/astraea/database/TableInfo.java
@@ -0,0 +1,40 @@
+package org.astraea.database;
+
+import java.util.Collection;
+import java.util.Objects;
+
+public class TableInfo {
+  private final String name;
+  private final Collection<ColumnInfo> columns;
+
+  public TableInfo(String name, Collection<ColumnInfo> columns) {
+    this.name = name;
+    this.columns = columns;
+  }
+
+  public String name() {
+    return name;
+  }
+
+  public Collection<ColumnInfo> columns() {
+    return columns;
+  }
+
+  @Override
+  public String toString() {
+    return "TableInfo{" + "name='" + name + '\'' + ", columns=" + columns + '}';
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) return true;
+    if (o == null || getClass() != o.getClass()) return false;
+    TableInfo tableInfo = (TableInfo) o;
+    return Objects.equals(name, tableInfo.name) && Objects.equals(columns, tableInfo.columns);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(name, columns);
+  }
+}
diff --git a/app/src/main/java/org/astraea/database/TableQuery.java b/app/src/main/java/org/astraea/database/TableQuery.java
new file mode 100644
index 00000000..9d88cb73
--- /dev/null
+++ b/app/src/main/java/org/astraea/database/TableQuery.java
@@ -0,0 +1,31 @@
+package org.astraea.database;
+
+import java.util.Collection;
+
+public interface TableQuery {
+
+  /**
+   * Normally, catalog in database is server instance or database instance.
+   *
+   * @param catalog to search
+   * @return this query
+   */
+  TableQuery catalog(String catalog);
+
+  /**
+   * Normally, schema in database is namespace.
+   *
+   * @param schema to search
+   * @return this query
+   */
+  TableQuery schema(String schema);
+
+  /**
+   * @param tableName to search
+   * @return this query
+   */
+  TableQuery tableName(String tableName);
+
+  /** @return the tables matched to this query. */
+  Collection<TableInfo> run();
+}
diff --git a/app/src/main/java/org/astraea/metrics/collector/BeanCollector.java b/app/src/main/java/org/astraea/metrics/collector/BeanCollector.java
index 5a52b83f..1976a96f 100644
--- a/app/src/main/java/org/astraea/metrics/collector/BeanCollector.java
+++ b/app/src/main/java/org/astraea/metrics/collector/BeanCollector.java
@@ -1,6 +1,8 @@
 package org.astraea.metrics.collector;
 
 import java.time.Duration;
+import java.util.Collection;
+import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
@@ -11,7 +13,7 @@ import java.util.concurrent.ConcurrentSkipListMap;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.BiFunction;
-import org.astraea.Utils;
+import org.astraea.common.Utils;
 import org.astraea.metrics.HasBeanObject;
 import org.astraea.metrics.jmx.MBeanClient;
 import org.astraea.metrics.kafka.KafkaMetrics;
@@ -108,9 +110,9 @@ public class BeanCollector {
               }
 
               @Override
-              public List<HasBeanObject> current() {
+              public Collection<HasBeanObject> current() {
                 tryUpdate();
-                return List.copyOf(objects.values());
+                return Collections.unmodifiableCollection(objects.values());
               }
 
               @Override
@@ -118,7 +120,8 @@ public class BeanCollector {
                 node.lock.lock();
                 try {
                   node.receivers.remove(this);
-                  if (node.receivers.isEmpty()) Utils.close(node.mBeanClient);
+                  if (node.receivers.isEmpty() && node.mBeanClient != null)
+                    Utils.swallowException(node.mBeanClient::close);
                   node.mBeanClient = null;
                 } finally {
                   node.lock.unlock();
diff --git a/app/src/main/java/org/astraea/metrics/jmx/MBeanClientImpl.java b/app/src/main/java/org/astraea/metrics/jmx/MBeanClientImpl.java
index 3cbd3253..cf38e66e 100644
--- a/app/src/main/java/org/astraea/metrics/jmx/MBeanClientImpl.java
+++ b/app/src/main/java/org/astraea/metrics/jmx/MBeanClientImpl.java
@@ -22,7 +22,7 @@ import javax.management.ReflectionException;
 import javax.management.remote.JMXConnector;
 import javax.management.remote.JMXConnectorFactory;
 import javax.management.remote.JMXServiceURL;
-import org.astraea.Utils;
+import org.astraea.common.Utils;
 
 class MBeanClientImpl implements MBeanClient {
 
@@ -178,7 +178,7 @@ class MBeanClientImpl implements MBeanClient {
   public void close() {
     if (!isClosed) {
       isClosed = true;
-      Utils.close(jmxConnector);
+      Utils.packException(jmxConnector::close);
     }
   }
 }
diff --git a/app/src/main/java/org/astraea/metrics/kafka/KafkaMetrics.java b/app/src/main/java/org/astraea/metrics/kafka/KafkaMetrics.java
index 19587e5c..74b36379 100644
--- a/app/src/main/java/org/astraea/metrics/kafka/KafkaMetrics.java
+++ b/app/src/main/java/org/astraea/metrics/kafka/KafkaMetrics.java
@@ -221,7 +221,7 @@ public final class KafkaMetrics {
   }
 
   public enum TopicPartition {
-    LodEndOffset("LodEndOffset"),
+    LogEndOffset("LogEndOffset"),
     LogStartOffset("LogStartOffset"),
     NumLogSegments("NumLogSegments"),
     Size("Size");
diff --git a/app/src/main/java/org/astraea/partitioner/StrictCostDispatcher.java b/app/src/main/java/org/astraea/partitioner/StrictCostDispatcher.java
index 58a11193..9b6b6bc6 100644
--- a/app/src/main/java/org/astraea/partitioner/StrictCostDispatcher.java
+++ b/app/src/main/java/org/astraea/partitioner/StrictCostDispatcher.java
@@ -9,7 +9,7 @@ import java.util.Objects;
 import java.util.Optional;
 import java.util.TreeMap;
 import java.util.stream.Collectors;
-import org.astraea.Utils;
+import org.astraea.common.Utils;
 import org.astraea.cost.ClusterInfo;
 import org.astraea.cost.CostFunction;
 import org.astraea.cost.HasBrokerCost;
@@ -58,7 +58,7 @@ public class StrictCostDispatcher implements Dispatcher {
 
   @Override
   public int partition(String topic, byte[] key, byte[] value, ClusterInfo clusterInfo) {
-    var partitionLeaders = clusterInfo.availablePartitionLeaders(topic);
+    var partitionLeaders = clusterInfo.availableReplicaLeaders(topic);
     // just return first partition if there is no available partitions
     if (partitionLeaders.isEmpty()) return 0;
 
@@ -200,7 +200,7 @@ public class StrictCostDispatcher implements Dispatcher {
 
   @Override
   public void close() {
-    receivers.values().forEach(Utils::close);
+    receivers.values().forEach(r -> Utils.swallowException(r::close));
     receivers.clear();
   }
 }
diff --git a/app/src/main/java/org/astraea/partitioner/smooth/SmoothWeightRoundRobin.java b/app/src/main/java/org/astraea/partitioner/smooth/SmoothWeightRoundRobin.java
index 5608f424..32887eeb 100644
--- a/app/src/main/java/org/astraea/partitioner/smooth/SmoothWeightRoundRobin.java
+++ b/app/src/main/java/org/astraea/partitioner/smooth/SmoothWeightRoundRobin.java
@@ -2,8 +2,11 @@ package org.astraea.partitioner.smooth;
 
 import java.time.Duration;
 import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
+import org.astraea.cost.ClusterInfo;
 import org.astraea.cost.Periodic;
 import org.astraea.cost.broker.CostUtils;
 
@@ -34,7 +37,9 @@ import org.astraea.cost.broker.CostUtils;
 public final class SmoothWeightRoundRobin
     extends Periodic<SmoothWeightRoundRobin.EffectiveWeightResult> {
   private EffectiveWeightResult effectiveWeightResult;
-  public Map<Integer, Double> currentWeight;
+  private Map<Integer, Double> currentWeight;
+
+  private final Map<String, List<Integer>> brokersIDofTopic = new HashMap<>();
 
   public SmoothWeightRoundRobin(Map<Integer, Double> effectiveWeight) {
     effectiveWeightResult =
@@ -75,15 +80,27 @@ public final class SmoothWeightRoundRobin
    *
    * @return the preferred ID
    */
-  public synchronized int getAndChoose() {
+  public synchronized int getAndChoose(String topic, ClusterInfo clusterInfo) {
+    // TODO Update brokerID with ClusterInfo frequency.
+    var brokerID =
+        brokersIDofTopic.computeIfAbsent(
+            topic,
+            e ->
+                clusterInfo.availableReplicaLeaders(topic).stream()
+                    .map(replicaInfo -> replicaInfo.nodeInfo().id())
+                    .collect(Collectors.toList()));
     this.currentWeight =
         this.currentWeight.entrySet().stream()
             .collect(
                 Collectors.toMap(
                     Map.Entry::getKey,
-                    e -> e.getValue() + effectiveWeightResult.effectiveWeight.get(e.getKey())));
+                    e ->
+                        brokerID.contains(e.getKey())
+                            ? e.getValue() + effectiveWeightResult.effectiveWeight.get(e.getKey())
+                            : e.getValue()));
     var maxID =
         this.currentWeight.entrySet().stream()
+            .filter(entry -> brokerID.contains(entry.getKey()))
             .max(Comparator.comparingDouble(Map.Entry::getValue))
             .map(Map.Entry::getKey)
             .orElse(0);
@@ -94,18 +111,23 @@ public final class SmoothWeightRoundRobin
                     Map.Entry::getKey,
                     e ->
                         e.getKey().equals(maxID)
-                            ? e.getValue() - effectiveWeightResult.effectiveWeightSum
+                            ? e.getValue() - effectiveWeightResult.effectiveWeightSum(brokerID)
                             : e.getValue()));
     return maxID;
   }
 
   public static class EffectiveWeightResult {
     private final Map<Integer, Double> effectiveWeight;
-    private final double effectiveWeightSum;
 
     EffectiveWeightResult(Map<Integer, Double> effectiveWeight) {
       this.effectiveWeight = effectiveWeight;
-      this.effectiveWeightSum = effectiveWeight.values().stream().mapToDouble(i -> i).sum();
+    }
+
+    double effectiveWeightSum(List<Integer> brokerID) {
+      return effectiveWeight.entrySet().stream()
+          .filter(entry -> brokerID.contains(entry.getKey()))
+          .mapToDouble(Map.Entry::getValue)
+          .sum();
     }
   }
 }
diff --git a/app/src/main/java/org/astraea/partitioner/smooth/SmoothWeightRoundRobinDispatcher.java b/app/src/main/java/org/astraea/partitioner/smooth/SmoothWeightRoundRobinDispatcher.java
index 3dff8429..318f9535 100644
--- a/app/src/main/java/org/astraea/partitioner/smooth/SmoothWeightRoundRobinDispatcher.java
+++ b/app/src/main/java/org/astraea/partitioner/smooth/SmoothWeightRoundRobinDispatcher.java
@@ -16,7 +16,7 @@ import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 import org.apache.kafka.common.Cluster;
-import org.astraea.Utils;
+import org.astraea.common.Utils;
 import org.astraea.cost.ClusterInfo;
 import org.astraea.cost.NodeInfo;
 import org.astraea.cost.Periodic;
@@ -82,7 +82,7 @@ public class SmoothWeightRoundRobinDispatcher extends Periodic<Map<Integer, Doub
     if (partitions.size() == 1) return partitions.iterator().next().partition();
 
     if (targetPartition == null) {
-      var targetBroker = smoothWeightRoundRobinCal.getAndChoose();
+      var targetBroker = smoothWeightRoundRobinCal.getAndChoose(topic, clusterInfo);
       targetPartition =
           hasPartitions
               .get(targetBroker)
@@ -96,7 +96,7 @@ public class SmoothWeightRoundRobinDispatcher extends Periodic<Map<Integer, Doub
 
   @Override
   public void close() {
-    receivers.values().forEach(Utils::close);
+    receivers.values().forEach(r -> Utils.swallowException(r::close));
     receivers.clear();
   }
 
@@ -148,7 +148,7 @@ public class SmoothWeightRoundRobinDispatcher extends Periodic<Map<Integer, Doub
   }
 
   private void refreshPartitionMetaData(ClusterInfo clusterInfo, String topic) {
-    partitions = clusterInfo.availablePartitions(topic);
+    partitions = clusterInfo.availableReplicas(topic);
     partitions.forEach(
         p ->
             hasPartitions
diff --git a/app/src/main/java/org/astraea/performance/DataSupplier.java b/app/src/main/java/org/astraea/performance/DataSupplier.java
index 657949a1..9c6b5538 100644
--- a/app/src/main/java/org/astraea/performance/DataSupplier.java
+++ b/app/src/main/java/org/astraea/performance/DataSupplier.java
@@ -4,8 +4,8 @@ import java.util.Arrays;
 import java.util.Random;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Supplier;
-import org.astraea.utils.DataSize;
-import org.astraea.utils.DataUnit;
+import org.astraea.common.DataSize;
+import org.astraea.common.DataUnit;
 
 @FunctionalInterface
 interface DataSupplier extends Supplier<DataSupplier.Data> {
diff --git a/app/src/main/java/org/astraea/performance/Manager.java b/app/src/main/java/org/astraea/performance/Manager.java
index 25743440..fac125cb 100644
--- a/app/src/main/java/org/astraea/performance/Manager.java
+++ b/app/src/main/java/org/astraea/performance/Manager.java
@@ -2,7 +2,7 @@ package org.astraea.performance;
 
 import java.util.List;
 import java.util.function.Supplier;
-import org.astraea.utils.DataUnit;
+import org.astraea.common.DataUnit;
 
 /**
  * Thread safe This class is used for managing the start/end of the producer/consumer threads.
diff --git a/app/src/main/java/org/astraea/performance/Performance.java b/app/src/main/java/org/astraea/performance/Performance.java
index a3c16afd..adb10730 100644
--- a/app/src/main/java/org/astraea/performance/Performance.java
+++ b/app/src/main/java/org/astraea/performance/Performance.java
@@ -17,7 +17,6 @@ import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import org.apache.kafka.clients.producer.internals.DefaultPartitioner;
 import org.apache.kafka.common.errors.WakeupException;
-import org.astraea.Utils;
 import org.astraea.admin.Admin;
 import org.astraea.admin.Compression;
 import org.astraea.admin.TopicPartition;
@@ -27,14 +26,15 @@ import org.astraea.argument.NonNegativeShortField;
 import org.astraea.argument.PathField;
 import org.astraea.argument.PositiveLongField;
 import org.astraea.argument.PositiveShortField;
+import org.astraea.common.DataSize;
+import org.astraea.common.DataUnit;
+import org.astraea.common.Utils;
 import org.astraea.concurrent.Executor;
 import org.astraea.concurrent.State;
 import org.astraea.concurrent.ThreadPool;
 import org.astraea.consumer.Consumer;
 import org.astraea.consumer.Isolation;
 import org.astraea.producer.Producer;
-import org.astraea.utils.DataSize;
-import org.astraea.utils.DataUnit;
 
 /**
  * Performance benchmark which includes
diff --git a/app/src/main/java/org/astraea/performance/ProducerExecutor.java b/app/src/main/java/org/astraea/performance/ProducerExecutor.java
index 12c8f42c..a248c22d 100644
--- a/app/src/main/java/org/astraea/performance/ProducerExecutor.java
+++ b/app/src/main/java/org/astraea/performance/ProducerExecutor.java
@@ -7,7 +7,7 @@ import java.util.function.BiConsumer;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
-import org.astraea.Utils;
+import org.astraea.common.Utils;
 import org.astraea.concurrent.Executor;
 import org.astraea.concurrent.State;
 import org.astraea.producer.Producer;
diff --git a/app/src/main/java/org/astraea/performance/ReportFormat.java b/app/src/main/java/org/astraea/performance/ReportFormat.java
index 98c452b4..7a0b6200 100644
--- a/app/src/main/java/org/astraea/performance/ReportFormat.java
+++ b/app/src/main/java/org/astraea/performance/ReportFormat.java
@@ -12,10 +12,10 @@ import java.time.Duration;
 import java.util.Date;
 import java.util.function.Supplier;
 import java.util.stream.IntStream;
-import org.astraea.Utils;
+import org.astraea.common.DataUnit;
+import org.astraea.common.Utils;
 import org.astraea.concurrent.Executor;
 import org.astraea.concurrent.State;
-import org.astraea.utils.DataUnit;
 
 public enum ReportFormat {
   CSV("csv"),
@@ -70,7 +70,7 @@ public enum ReportFormat {
 
           @Override
           public void close() {
-            Utils.close(writer);
+            Utils.packException(writer::close);
           }
         };
       case JSON:
@@ -85,11 +85,8 @@ public enum ReportFormat {
 
           @Override
           public void close() {
-            try {
-              writer.write("}");
-              Utils.close(writer);
-            } catch (IOException ignore) {
-            }
+            Utils.swallowException(() -> writer.write("}"));
+            Utils.packException(writer::close);
           }
         };
       default:
diff --git a/app/src/main/java/org/astraea/performance/Tracker.java b/app/src/main/java/org/astraea/performance/Tracker.java
index 229df870..1c968a1d 100644
--- a/app/src/main/java/org/astraea/performance/Tracker.java
+++ b/app/src/main/java/org/astraea/performance/Tracker.java
@@ -5,9 +5,9 @@ import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.function.Supplier;
+import org.astraea.common.DataUnit;
 import org.astraea.concurrent.Executor;
 import org.astraea.concurrent.State;
-import org.astraea.utils.DataUnit;
 
 /** Print out the given metrics. */
 public class Tracker implements Executor {
diff --git a/app/src/main/java/org/astraea/web/PipelineHandler.java b/app/src/main/java/org/astraea/web/PipelineHandler.java
new file mode 100644
index 00000000..851f15ce
--- /dev/null
+++ b/app/src/main/java/org/astraea/web/PipelineHandler.java
@@ -0,0 +1,121 @@
+package org.astraea.web;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.Function;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+import org.astraea.admin.Admin;
+import org.astraea.admin.Member;
+import org.astraea.admin.ProducerState;
+
+class PipelineHandler implements Handler {
+
+  static final String ACTIVE_KEY = "active";
+
+  private final Admin admin;
+
+  PipelineHandler(Admin admin) {
+    this.admin = admin;
+  }
+
+  @Override
+  public JsonObject get(Optional<String> target, Map<String, String> queries) {
+    var tps =
+        topicPartitions(admin).stream()
+            .filter(filter(queries))
+            .collect(Collectors.toUnmodifiableList());
+    return new TopicPartitions(tps);
+  }
+
+  static Predicate<TopicPartition> filter(Map<String, String> queries) {
+    var flag = queries.get(ACTIVE_KEY);
+    // remove the topic-partitions having no producers/consumers
+    if (flag != null && flag.equalsIgnoreCase("true"))
+      return tp -> !tp.to.isEmpty() || !tp.from.isEmpty();
+    // remove the topic-partitions having producers/consumers
+    if (flag != null && flag.equalsIgnoreCase("false"))
+      return tp -> tp.to.isEmpty() && tp.from.isEmpty();
+    return tp -> true;
+  }
+
+  static Collection<TopicPartition> topicPartitions(Admin admin) {
+    var result =
+        admin.partitions().stream()
+            .collect(
+                Collectors.toMap(
+                    Function.identity(), tp -> new TopicPartition(tp.topic(), tp.partition())));
+    admin
+        .consumerGroups()
+        .values()
+        .forEach(
+            cg ->
+                cg.assignment()
+                    .forEach(
+                        (m, tps) ->
+                            tps.stream()
+                                // it could return new topic partition, so we have to remove them.
+                                .filter(result::containsKey)
+                                .forEach(tp -> result.get(tp).to.add(new Consumer(m)))));
+    admin
+        .producerStates(result.keySet())
+        .forEach((tp, p) -> p.forEach(s -> result.get(tp).from.add(new Producer(s))));
+    return result.values().stream()
+        .sorted(
+            Comparator.comparing((TopicPartition tp) -> tp.topic).thenComparing(tp -> tp.partition))
+        .collect(Collectors.toUnmodifiableList());
+  }
+
+  static class Producer implements JsonObject {
+    final long producerId;
+    final int producerEpoch;
+    final int lastSequence;
+    final long lastTimestamp;
+
+    Producer(ProducerState state) {
+      this.producerId = state.producerId();
+      this.producerEpoch = state.producerEpoch();
+      this.lastSequence = state.lastSequence();
+      this.lastTimestamp = state.lastTimestamp();
+    }
+  }
+
+  static class Consumer implements JsonObject {
+    final String groupId;
+    final String memberId;
+    final String groupInstanceId;
+    final String clientId;
+    final String host;
+
+    Consumer(Member member) {
+      this.groupId = member.groupId();
+      this.memberId = member.memberId();
+      this.groupInstanceId = member.groupInstanceId().orElse(null);
+      this.clientId = member.clientId();
+      this.host = member.host();
+    }
+  }
+
+  static class TopicPartition implements JsonObject {
+    final String topic;
+    final int partition;
+    final Collection<Producer> from = new ArrayList<>();
+    final Collection<Consumer> to = new ArrayList<>();
+
+    TopicPartition(String topic, int partition) {
+      this.topic = topic;
+      this.partition = partition;
+    }
+  }
+
+  static class TopicPartitions implements JsonObject {
+    final Collection<TopicPartition> topicPartitions;
+
+    TopicPartitions(Collection<TopicPartition> topicPartitions) {
+      this.topicPartitions = topicPartitions;
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/web/WebService.java b/app/src/main/java/org/astraea/web/WebService.java
index d0821feb..c47cf379 100644
--- a/app/src/main/java/org/astraea/web/WebService.java
+++ b/app/src/main/java/org/astraea/web/WebService.java
@@ -20,6 +20,7 @@ public class WebService {
     server.createContext("/brokers", new BrokerHandler(Admin.of(arg.configs())));
     server.createContext("/producers", new ProducerHandler(Admin.of(arg.configs())));
     server.createContext("/quotas", new QuotaHandler(Admin.of(arg.configs())));
+    server.createContext("/pipelines", new PipelineHandler(Admin.of(arg.configs())));
     server.start();
   }
 
diff --git a/docker/build_deps_image.sh b/docker/build_deps_image.sh
new file mode 100755
index 00000000..313e7193
--- /dev/null
+++ b/docker/build_deps_image.sh
@@ -0,0 +1,46 @@
+#!/bin/bash
+
+declare -r DOCKER_FOLDER=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" &>/dev/null && pwd)
+source $DOCKER_FOLDER/docker_build_common.sh
+declare -r REPO=${REPO:-ghcr.io/skiptests/astraea/deps}
+declare -r VERSION=${VERSION:-latest}
+declare -r IMAGE_NAME="$REPO:$VERSION"
+declare -r DOCKERFILE=$DOCKER_FOLDER/deps.dockerfile
+# ===================================[functions]===================================
+function generateDockerfile() {
+  echo "# this dockerfile is generated dynamically
+FROM ubuntu:22.04
+
+# install tools
+RUN apt-get update && apt-get install -y \
+  git \
+  openjdk-11-jdk \
+  wget \
+  unzip \
+  libaio1 \
+  numactl \
+  libncurses5 \
+  curl
+
+# build code and download dependencies
+WORKDIR /astraea
+RUN git clone https://github.com/skiptests/astraea.git /astraea
+RUN ./gradlew clean build -x test
+# trigger download of database
+RUN ./gradlew cleanTest test --tests DatabaseTest
+
+WORKDIR /root
+" >"$DOCKERFILE"
+}
+
+function buildBaseImageIfNeed() {
+  if [[ "$(docker images -q $IMAGE_NAME 2>/dev/null)" == "" ]]; then
+    docker build --no-cache -t "$IMAGE_NAME" -f "$DOCKERFILE" "$DOCKER_FOLDER"
+    docker push $IMAGE_NAME
+  else
+    echo "$IMAGE_NAME is existent in local"
+  fi
+}
+# ===================================[main]===================================
+generateDockerfile
+buildBaseImageIfNeed
\ No newline at end of file
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/admin/AdminTest.java b/app/src/test/java/org/astraea/admin/AdminTest.java
index cfef8718..6fc39611 100644
--- a/app/src/test/java/org/astraea/admin/AdminTest.java
+++ b/app/src/test/java/org/astraea/admin/AdminTest.java
@@ -13,7 +13,7 @@ import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import java.util.stream.StreamSupport;
 import org.apache.kafka.common.config.TopicConfig;
-import org.astraea.Utils;
+import org.astraea.common.Utils;
 import org.astraea.consumer.Consumer;
 import org.astraea.consumer.Deserializer;
 import org.astraea.producer.Producer;
diff --git a/app/src/test/java/org/astraea/admin/ReplicaCollieTest.java b/app/src/test/java/org/astraea/admin/ReplicaCollieTest.java
index 28aadf35..ac946b17 100644
--- a/app/src/test/java/org/astraea/admin/ReplicaCollieTest.java
+++ b/app/src/test/java/org/astraea/admin/ReplicaCollieTest.java
@@ -8,8 +8,8 @@ import java.util.Set;
 import java.util.TreeMap;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
-import org.astraea.Utils;
 import org.astraea.argument.Argument;
+import org.astraea.common.Utils;
 import org.astraea.service.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
diff --git a/app/src/test/java/org/astraea/admin/ReplicaSyncingMonitorTest.java b/app/src/test/java/org/astraea/admin/ReplicaSyncingMonitorTest.java
index e23f0b7e..4c78d46b 100644
--- a/app/src/test/java/org/astraea/admin/ReplicaSyncingMonitorTest.java
+++ b/app/src/test/java/org/astraea/admin/ReplicaSyncingMonitorTest.java
@@ -28,7 +28,7 @@ import java.util.function.BiFunction;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
-import org.astraea.utils.DataUnit;
+import org.astraea.common.DataUnit;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
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..32427f7d
--- /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.replicas(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.replicas(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/utils/DataRateTest.java b/app/src/test/java/org/astraea/common/DataRateTest.java
similarity index 99%
rename from app/src/test/java/org/astraea/utils/DataRateTest.java
rename to app/src/test/java/org/astraea/common/DataRateTest.java
index 7f674a37..050dd571 100644
--- a/app/src/test/java/org/astraea/utils/DataRateTest.java
+++ b/app/src/test/java/org/astraea/common/DataRateTest.java
@@ -1,4 +1,4 @@
-package org.astraea.utils;
+package org.astraea.common;
 
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
diff --git a/app/src/test/java/org/astraea/utils/DataSizeTest.java b/app/src/test/java/org/astraea/common/DataSizeTest.java
similarity index 99%
rename from app/src/test/java/org/astraea/utils/DataSizeTest.java
rename to app/src/test/java/org/astraea/common/DataSizeTest.java
index 7ee141c2..4e18039f 100644
--- a/app/src/test/java/org/astraea/utils/DataSizeTest.java
+++ b/app/src/test/java/org/astraea/common/DataSizeTest.java
@@ -1,4 +1,4 @@
-package org.astraea.utils;
+package org.astraea.common;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
diff --git a/app/src/test/java/org/astraea/UtilsTest.java b/app/src/test/java/org/astraea/common/UtilsTest.java
similarity index 86%
rename from app/src/test/java/org/astraea/UtilsTest.java
rename to app/src/test/java/org/astraea/common/UtilsTest.java
index 57ef6415..5d71837b 100644
--- a/app/src/test/java/org/astraea/UtilsTest.java
+++ b/app/src/test/java/org/astraea/common/UtilsTest.java
@@ -1,4 +1,4 @@
-package org.astraea;
+package org.astraea.common;
 
 import java.io.IOException;
 import java.util.concurrent.ExecutionException;
@@ -12,7 +12,7 @@ public class UtilsTest {
     Assertions.assertThrows(
         IllegalArgumentException.class,
         () ->
-            Utils.handleException(
+            Utils.packException(
                 () -> {
                   throw new ExecutionException(new IllegalArgumentException());
                 }));
@@ -20,7 +20,7 @@ public class UtilsTest {
     Assertions.assertThrows(
         IllegalArgumentException.class,
         () ->
-            Utils.handleException(
+            Utils.packException(
                 () -> {
                   throw new IllegalArgumentException();
                 }));
@@ -28,7 +28,7 @@ public class UtilsTest {
     Assertions.assertThrows(
         RuntimeException.class,
         () ->
-            Utils.handleException(
+            Utils.packException(
                 () -> {
                   throw new IOException();
                 }));
diff --git a/app/src/test/java/org/astraea/consumer/ConsumerTest.java b/app/src/test/java/org/astraea/consumer/ConsumerTest.java
index 7f99c882..1848e480 100644
--- a/app/src/test/java/org/astraea/consumer/ConsumerTest.java
+++ b/app/src/test/java/org/astraea/consumer/ConsumerTest.java
@@ -7,7 +7,7 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.IntStream;
 import org.apache.kafka.common.errors.WakeupException;
-import org.astraea.Utils;
+import org.astraea.common.Utils;
 import org.astraea.producer.Producer;
 import org.astraea.service.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
@@ -152,4 +152,22 @@ public class ConsumerTest extends RequireBrokerCluster {
       Assertions.assertEquals(10, consumer.poll(11, Duration.ofSeconds(5)).size());
     }
   }
+
+  @Test
+  void testRecordsPollingTime() {
+    var count = 1;
+    var topic = "testPollingTime";
+    try (var consumer =
+        Consumer.builder()
+            .bootstrapServers(bootstrapServers())
+            .topics(Set.of(topic))
+            .fromBeginning()
+            .build()) {
+
+      // poll() returns immediately, if there is(/are) record(s) to poll.
+      produceData(topic, count);
+      Assertions.assertTimeout(
+          Duration.ofSeconds(10), () -> consumer.poll(Duration.ofSeconds(Integer.MAX_VALUE)));
+    }
+  }
 }
diff --git a/app/src/test/java/org/astraea/consumer/RebalanceListenerTest.java b/app/src/test/java/org/astraea/consumer/RebalanceListenerTest.java
index 8cf6ac55..49d92866 100644
--- a/app/src/test/java/org/astraea/consumer/RebalanceListenerTest.java
+++ b/app/src/test/java/org/astraea/consumer/RebalanceListenerTest.java
@@ -3,7 +3,7 @@ package org.astraea.consumer;
 import java.time.Duration;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
-import org.astraea.Utils;
+import org.astraea.common.Utils;
 import org.astraea.concurrent.State;
 import org.astraea.concurrent.ThreadPool;
 import org.astraea.service.RequireBrokerCluster;
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..f69926b9
--- /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> availableReplicaLeaders(String topic) {
+        return replicas(topic).stream()
+            .filter(ReplicaInfo::isLeader)
+            .collect(Collectors.toUnmodifiableList());
+      }
+
+      @Override
+      public List<ReplicaInfo> availableReplicas(String topic) {
+        return replicas(topic);
+      }
+
+      @Override
+      public Set<String> topics() {
+        return topics;
+      }
+
+      @Override
+      public List<ReplicaInfo> replicas(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/ClusterInfoTest.java b/app/src/test/java/org/astraea/cost/ClusterInfoTest.java
index 32d4d2f8..26fabe9d 100644
--- a/app/src/test/java/org/astraea/cost/ClusterInfoTest.java
+++ b/app/src/test/java/org/astraea/cost/ClusterInfoTest.java
@@ -26,15 +26,15 @@ public class ClusterInfoTest {
     Assertions.assertEquals(1, clusterInfo.nodes().size());
     Assertions.assertEquals(NodeInfo.of(node), clusterInfo.nodes().get(0));
     Assertions.assertEquals(clusterInfo.nodes().get(0), clusterInfo.node(node.host(), node.port()));
-    Assertions.assertEquals(1, clusterInfo.availablePartitions(partition.topic()).size());
-    Assertions.assertEquals(1, clusterInfo.partitions(partition.topic()).size());
+    Assertions.assertEquals(1, clusterInfo.availableReplicas(partition.topic()).size());
+    Assertions.assertEquals(1, clusterInfo.replicas(partition.topic()).size());
     Assertions.assertEquals(
-        NodeInfo.of(node), clusterInfo.availablePartitions(partition.topic()).get(0).nodeInfo());
+        NodeInfo.of(node), clusterInfo.availableReplicas(partition.topic()).get(0).nodeInfo());
     Assertions.assertEquals(
         NodeInfo.of(node),
-        clusterInfo.availablePartitionLeaders(partition.topic()).get(0).nodeInfo());
+        clusterInfo.availableReplicaLeaders(partition.topic()).get(0).nodeInfo());
     Assertions.assertEquals(
-        NodeInfo.of(node), clusterInfo.partitions(partition.topic()).get(0).nodeInfo());
+        NodeInfo.of(node), clusterInfo.replicas(partition.topic()).get(0).nodeInfo());
   }
 
   @Test
diff --git a/app/src/test/java/org/astraea/cost/FakeClusterInfo.java b/app/src/test/java/org/astraea/cost/FakeClusterInfo.java
index fd0f1d75..0a2e71e1 100644
--- a/app/src/test/java/org/astraea/cost/FakeClusterInfo.java
+++ b/app/src/test/java/org/astraea/cost/FakeClusterInfo.java
@@ -13,12 +13,17 @@ public class FakeClusterInfo implements ClusterInfo {
   }
 
   @Override
-  public List<ReplicaInfo> availablePartitionLeaders(String topic) {
+  public Set<String> dataDirectories(int brokerId) {
+    return Set.of();
+  }
+
+  @Override
+  public List<ReplicaInfo> availableReplicaLeaders(String topic) {
     return List.of();
   }
 
   @Override
-  public List<ReplicaInfo> availablePartitions(String topic) {
+  public List<ReplicaInfo> availableReplicas(String topic) {
     return List.of();
   }
 
@@ -28,7 +33,7 @@ public class FakeClusterInfo implements ClusterInfo {
   }
 
   @Override
-  public List<ReplicaInfo> partitions(String topic) {
+  public List<ReplicaInfo> replicas(String topic) {
     return List.of();
   }
 
diff --git a/app/src/test/java/org/astraea/cost/ThroughputCostTest.java b/app/src/test/java/org/astraea/cost/ThroughputCostTest.java
index 8b5ae285..580ae0f1 100644
--- a/app/src/test/java/org/astraea/cost/ThroughputCostTest.java
+++ b/app/src/test/java/org/astraea/cost/ThroughputCostTest.java
@@ -35,7 +35,7 @@ public class ThroughputCostTest {
     Mockito.when(cluster.nodes()).thenReturn(List.of(node));
     Mockito.when(cluster.allBeans()).thenReturn(Map.of());
     Mockito.when(cluster.topics()).thenReturn(Set.of("t"));
-    Mockito.when(cluster.availablePartitions("t"))
+    Mockito.when(cluster.availableReplicas("t"))
         .thenReturn(List.of(ReplicaInfo.of("t", 0, node, true, true, false)));
 
     var cost =
diff --git a/app/src/test/java/org/astraea/cost/broker/CpuCostTest.java b/app/src/test/java/org/astraea/cost/broker/CpuCostTest.java
index f8042f59..516501d3 100644
--- a/app/src/test/java/org/astraea/cost/broker/CpuCostTest.java
+++ b/app/src/test/java/org/astraea/cost/broker/CpuCostTest.java
@@ -37,7 +37,7 @@ public class CpuCostTest {
           }
 
           @Override
-          public List<ReplicaInfo> availablePartitions(String topic) {
+          public List<ReplicaInfo> availableReplicas(String topic) {
             return List.of(
                 ReplicaInfo.of("t", 0, NodeInfo.of(1, "host1", 9092), true, true, false),
                 ReplicaInfo.of("t", 0, NodeInfo.of(2, "host2", 9092), false, true, false),
@@ -72,7 +72,7 @@ public class CpuCostTest {
           }
 
           @Override
-          public List<ReplicaInfo> availablePartitions(String topic) {
+          public List<ReplicaInfo> availableReplicas(String topic) {
             return List.of(
                 ReplicaInfo.of("t", 0, NodeInfo.of(1, "host1", 9092), true, true, false),
                 ReplicaInfo.of("t", 0, NodeInfo.of(2, "host2", 9092), false, true, false),
diff --git a/app/src/test/java/org/astraea/cost/broker/MemoryCostTest.java b/app/src/test/java/org/astraea/cost/broker/MemoryCostTest.java
index 1415a87c..434d15be 100644
--- a/app/src/test/java/org/astraea/cost/broker/MemoryCostTest.java
+++ b/app/src/test/java/org/astraea/cost/broker/MemoryCostTest.java
@@ -38,7 +38,7 @@ public class MemoryCostTest {
           }
 
           @Override
-          public List<ReplicaInfo> availablePartitions(String topic) {
+          public List<ReplicaInfo> availableReplicas(String topic) {
             return List.of(
                 ReplicaInfo.of("t", 0, NodeInfo.of(1, "host1", 9092), true, false, false),
                 ReplicaInfo.of("t", 0, NodeInfo.of(2, "host2", 9092), false, true, false),
@@ -73,7 +73,7 @@ public class MemoryCostTest {
           }
 
           @Override
-          public List<ReplicaInfo> availablePartitions(String topic) {
+          public List<ReplicaInfo> availableReplicas(String topic) {
             return List.of(
                 ReplicaInfo.of("t", 0, NodeInfo.of(1, "host1", 9092), true, true, false),
                 ReplicaInfo.of("t", 0, NodeInfo.of(2, "host2", 9092), false, true, false),
diff --git a/app/src/test/java/org/astraea/database/DatabaseClientTest.java b/app/src/test/java/org/astraea/database/DatabaseClientTest.java
new file mode 100644
index 00000000..8c61e96e
--- /dev/null
+++ b/app/src/test/java/org/astraea/database/DatabaseClientTest.java
@@ -0,0 +1,75 @@
+package org.astraea.database;
+
+import org.astraea.common.Utils;
+import org.astraea.service.Database;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class DatabaseClientTest {
+  @Test
+  void testCreate() {
+    try (var database = Database.builder().build();
+        var client =
+            DatabaseClient.builder()
+                .user(database.user())
+                .password(database.password())
+                .url(database.url())
+                .build()) {
+      var tableCount = client.query().run().size();
+      var tableName = Utils.randomString(10);
+      var c0 = new ColumnInfo(Utils.randomString(5), "INT", true);
+      var c1 = new ColumnInfo(Utils.randomString(5), "INT", false);
+      client
+          .tableCreator()
+          .name(tableName)
+          .primaryKey(c0.name(), c0.type())
+          .column(c1.name(), c1.type())
+          .run();
+
+      Assertions.assertEquals(tableCount + 1, client.query().run().size());
+      Assertions.assertEquals(1, client.query().tableName(tableName).run().size());
+      var table = client.query().tableName(tableName).run().iterator().next();
+      Assertions.assertEquals(2, table.columns().size());
+      Assertions.assertEquals(
+          c0, table.columns().stream().filter(c -> c.name().equals(c0.name())).findFirst().get());
+      Assertions.assertEquals(
+          c1, table.columns().stream().filter(c -> c.name().equals(c1.name())).findFirst().get());
+    }
+  }
+
+  @Test
+  void testDelete() {
+    try (var database = Database.builder().build();
+        var client =
+            DatabaseClient.builder()
+                .user(database.user())
+                .password(database.password())
+                .url(database.url())
+                .build()) {
+      var tableName = Utils.randomString(10);
+      var tableCount = client.query().run().size();
+      client.tableCreator().name(tableName).primaryKey(Utils.randomString(5), "INT").run();
+      Assertions.assertEquals(tableCount + 1, client.query().run().size());
+      client.deleteTable(tableName);
+      Assertions.assertEquals(tableCount, client.query().run().size());
+    }
+  }
+
+  @Test
+  void testNoPrimaryKeyOrTableName() {
+    try (var database = Database.builder().build();
+        var client =
+            DatabaseClient.builder()
+                .user(database.user())
+                .password(database.password())
+                .url(database.url())
+                .build()) {
+      Assertions.assertThrows(
+          NullPointerException.class,
+          () -> client.tableCreator().primaryKey(Utils.randomString(5), "INT").run());
+      Assertions.assertThrows(
+          IllegalArgumentException.class,
+          () -> client.tableCreator().name("a").column(Utils.randomString(5), "INT").run());
+    }
+  }
+}
diff --git a/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricsTest.java b/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricsTest.java
index 85495d94..3306a3b2 100644
--- a/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricsTest.java
+++ b/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricsTest.java
@@ -2,6 +2,7 @@ package org.astraea.metrics.kafka;
 
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.condition.OS.LINUX;
 import static org.junit.jupiter.api.condition.OS.WINDOWS;
@@ -10,12 +11,14 @@ import java.io.IOException;
 import java.lang.management.ManagementFactory;
 import java.util.Arrays;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import javax.management.MBeanServer;
 import javax.management.remote.JMXConnectorServer;
 import javax.management.remote.JMXConnectorServerFactory;
 import javax.management.remote.JMXServiceURL;
 import org.astraea.admin.Admin;
+import org.astraea.common.Utils;
 import org.astraea.metrics.java.JvmMemory;
 import org.astraea.metrics.java.OperatingSystemInfo;
 import org.astraea.metrics.jmx.MBeanClient;
@@ -109,6 +112,21 @@ class KafkaMetricsTest extends RequireBrokerCluster {
     assertDoesNotThrow(totalTimeMs::stdDev);
   }
 
+  @ParameterizedTest()
+  @EnumSource(value = KafkaMetrics.TopicPartition.class)
+  void testTopicPartitionMetrics(KafkaMetrics.TopicPartition request) throws InterruptedException {
+    try (var admin = Admin.of(bootstrapServers())) {
+      // there are only 3 brokers, so 10 partitions can make each broker has some partitions
+      admin.creator().topic(Utils.randomString(5)).numberOfPartitions(10).create();
+    }
+
+    // wait for topic creation
+    TimeUnit.SECONDS.sleep(2);
+
+    var beans = request.fetch(mBeanClient);
+    assertNotEquals(0, beans.size());
+  }
+
   @Test
   void testGlobalPartitionCount() {
     // act
@@ -121,7 +139,7 @@ class KafkaMetricsTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testSize() throws IOException {
+  void testSize() {
     // arrange
     try (Admin admin = Admin.of(bootstrapServers())) {
       String topicName = getClass().getName();
diff --git a/app/src/test/java/org/astraea/partitioner/StrictCostDispatcherTest.java b/app/src/test/java/org/astraea/partitioner/StrictCostDispatcherTest.java
index 7df0b8ba..ca541842 100644
--- a/app/src/test/java/org/astraea/partitioner/StrictCostDispatcherTest.java
+++ b/app/src/test/java/org/astraea/partitioner/StrictCostDispatcherTest.java
@@ -137,16 +137,16 @@ public class StrictCostDispatcherTest {
       var clusterInfo = Mockito.mock(ClusterInfo.class);
 
       // there is no available partition
-      Mockito.when(clusterInfo.availablePartitionLeaders("aa")).thenReturn(List.of());
+      Mockito.when(clusterInfo.availableReplicaLeaders("aa")).thenReturn(List.of());
       Mockito.when(clusterInfo.topics()).thenReturn(Set.of("aa"));
       Assertions.assertEquals(0, dispatcher.partition("aa", new byte[0], new byte[0], clusterInfo));
 
       // there is only one available partition
-      Mockito.when(clusterInfo.availablePartitionLeaders("aa")).thenReturn(List.of(p0));
+      Mockito.when(clusterInfo.availableReplicaLeaders("aa")).thenReturn(List.of(p0));
       Assertions.assertEquals(1, dispatcher.partition("aa", new byte[0], new byte[0], clusterInfo));
 
       // there is no beans, so it just returns the partition.
-      Mockito.when(clusterInfo.availablePartitionLeaders("aa")).thenReturn(List.of(p0, p1));
+      Mockito.when(clusterInfo.availableReplicaLeaders("aa")).thenReturn(List.of(p0, p1));
       Assertions.assertEquals(2, dispatcher.partition("aa", new byte[0], new byte[0], clusterInfo));
 
       // cost functions with weight
diff --git a/app/src/test/java/org/astraea/partitioner/smooth/SmoothWeightRoundRobinDispatchTest.java b/app/src/test/java/org/astraea/partitioner/smooth/SmoothWeightRoundRobinDispatchTest.java
index c0e84377..06ee2aa6 100644
--- a/app/src/test/java/org/astraea/partitioner/smooth/SmoothWeightRoundRobinDispatchTest.java
+++ b/app/src/test/java/org/astraea/partitioner/smooth/SmoothWeightRoundRobinDispatchTest.java
@@ -23,11 +23,15 @@ import org.astraea.concurrent.ThreadPool;
 import org.astraea.consumer.Consumer;
 import org.astraea.consumer.Deserializer;
 import org.astraea.consumer.Header;
+import org.astraea.cost.FakeClusterInfo;
+import org.astraea.cost.NodeInfo;
+import org.astraea.cost.ReplicaInfo;
 import org.astraea.producer.Producer;
 import org.astraea.producer.Serializer;
 import org.astraea.service.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
 
 public class SmoothWeightRoundRobinDispatchTest extends RequireBrokerCluster {
   private final String brokerList = bootstrapServers();
@@ -273,14 +277,35 @@ public class SmoothWeightRoundRobinDispatchTest extends RequireBrokerCluster {
 
   @Test
   public void testGetAndChoose() {
+    var topic = "test";
     var smoothWeight = new SmoothWeightRoundRobin(Map.of(1, 5.0, 2, 3.0, 3, 1.0));
+    var re1 = Mockito.mock(ReplicaInfo.class);
+    var node1 = Mockito.mock(NodeInfo.class);
+    Mockito.when(re1.nodeInfo()).thenReturn(node1);
+    Mockito.when(node1.id()).thenReturn(1);
 
-    Assertions.assertEquals(1, smoothWeight.getAndChoose());
-    Assertions.assertEquals(2, smoothWeight.getAndChoose());
-    Assertions.assertEquals(3, smoothWeight.getAndChoose());
-    Assertions.assertEquals(1, smoothWeight.getAndChoose());
-    Assertions.assertEquals(2, smoothWeight.getAndChoose());
-    Assertions.assertEquals(3, smoothWeight.getAndChoose());
-    Assertions.assertEquals(1, smoothWeight.getAndChoose());
+    var re2 = Mockito.mock(ReplicaInfo.class);
+    var node2 = Mockito.mock(NodeInfo.class);
+    Mockito.when(re2.nodeInfo()).thenReturn(node2);
+    Mockito.when(node2.id()).thenReturn(2);
+
+    var re3 = Mockito.mock(ReplicaInfo.class);
+    var node3 = Mockito.mock(NodeInfo.class);
+    Mockito.when(re3.nodeInfo()).thenReturn(node3);
+    Mockito.when(node3.id()).thenReturn(3);
+    var testCluster =
+        new FakeClusterInfo() {
+          @Override
+          public List<ReplicaInfo> availableReplicaLeaders(String topic) {
+            return List.of(re1, re2, re3);
+          }
+        };
+    Assertions.assertEquals(1, smoothWeight.getAndChoose(topic, testCluster));
+    Assertions.assertEquals(2, smoothWeight.getAndChoose(topic, testCluster));
+    Assertions.assertEquals(3, smoothWeight.getAndChoose(topic, testCluster));
+    Assertions.assertEquals(1, smoothWeight.getAndChoose(topic, testCluster));
+    Assertions.assertEquals(2, smoothWeight.getAndChoose(topic, testCluster));
+    Assertions.assertEquals(3, smoothWeight.getAndChoose(topic, testCluster));
+    Assertions.assertEquals(1, smoothWeight.getAndChoose(topic, testCluster));
   }
 }
diff --git a/app/src/test/java/org/astraea/partitioner/smooth/SmoothWeightRoundRobinTest.java b/app/src/test/java/org/astraea/partitioner/smooth/SmoothWeightRoundRobinTest.java
new file mode 100644
index 00000000..1fd6de80
--- /dev/null
+++ b/app/src/test/java/org/astraea/partitioner/smooth/SmoothWeightRoundRobinTest.java
@@ -0,0 +1,66 @@
+package org.astraea.partitioner.smooth;
+
+import java.util.List;
+import java.util.Map;
+import org.astraea.cost.ClusterInfo;
+import org.astraea.cost.FakeClusterInfo;
+import org.astraea.cost.NodeInfo;
+import org.astraea.cost.ReplicaInfo;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+public class SmoothWeightRoundRobinTest {
+  @Test
+  void testGetAndChoose() {
+    var topic = "test";
+    var smoothWeight = new SmoothWeightRoundRobin(Map.of(1, 5.0, 2, 3.0, 3, 1.0));
+    var testCluster = mockResult();
+
+    Assertions.assertEquals(1, smoothWeight.getAndChoose(topic, testCluster));
+    Assertions.assertEquals(2, smoothWeight.getAndChoose(topic, testCluster));
+    Assertions.assertEquals(3, smoothWeight.getAndChoose(topic, testCluster));
+    Assertions.assertEquals(1, smoothWeight.getAndChoose(topic, testCluster));
+    Assertions.assertEquals(2, smoothWeight.getAndChoose(topic, testCluster));
+    Assertions.assertEquals(3, smoothWeight.getAndChoose(topic, testCluster));
+    Assertions.assertEquals(1, smoothWeight.getAndChoose(topic, testCluster));
+  }
+
+  @Test
+  void testPartOfBrokerGetAndChoose() {
+    var topic = "test";
+    var smoothWeight = new SmoothWeightRoundRobin(Map.of(1, 5.0, 2, 3.0, 3, 1.0, 4, 1.0));
+    var testCluster = mockResult();
+
+    Assertions.assertEquals(1, smoothWeight.getAndChoose(topic, testCluster));
+    Assertions.assertEquals(2, smoothWeight.getAndChoose(topic, testCluster));
+    Assertions.assertEquals(3, smoothWeight.getAndChoose(topic, testCluster));
+    Assertions.assertEquals(1, smoothWeight.getAndChoose(topic, testCluster));
+    Assertions.assertEquals(2, smoothWeight.getAndChoose(topic, testCluster));
+    Assertions.assertEquals(3, smoothWeight.getAndChoose(topic, testCluster));
+    Assertions.assertEquals(1, smoothWeight.getAndChoose(topic, testCluster));
+  }
+
+  ClusterInfo mockResult() {
+    var re1 = Mockito.mock(ReplicaInfo.class);
+    var node1 = Mockito.mock(NodeInfo.class);
+    Mockito.when(re1.nodeInfo()).thenReturn(node1);
+    Mockito.when(node1.id()).thenReturn(1);
+
+    var re2 = Mockito.mock(ReplicaInfo.class);
+    var node2 = Mockito.mock(NodeInfo.class);
+    Mockito.when(re2.nodeInfo()).thenReturn(node2);
+    Mockito.when(node2.id()).thenReturn(2);
+
+    var re3 = Mockito.mock(ReplicaInfo.class);
+    var node3 = Mockito.mock(NodeInfo.class);
+    Mockito.when(re3.nodeInfo()).thenReturn(node3);
+    Mockito.when(node3.id()).thenReturn(3);
+    return new FakeClusterInfo() {
+      @Override
+      public List<ReplicaInfo> availableReplicaLeaders(String topic) {
+        return List.of(re1, re2, re3);
+      }
+    };
+  }
+}
diff --git a/app/src/test/java/org/astraea/performance/DataSupplierTest.java b/app/src/test/java/org/astraea/performance/DataSupplierTest.java
index 4d96f2b3..ee0707b7 100644
--- a/app/src/test/java/org/astraea/performance/DataSupplierTest.java
+++ b/app/src/test/java/org/astraea/performance/DataSupplierTest.java
@@ -1,7 +1,7 @@
 package org.astraea.performance;
 
 import java.util.concurrent.TimeUnit;
-import org.astraea.utils.DataUnit;
+import org.astraea.common.DataUnit;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
diff --git a/app/src/test/java/org/astraea/performance/ProducerExecutorTest.java b/app/src/test/java/org/astraea/performance/ProducerExecutorTest.java
index 20c42941..02a1bf0f 100644
--- a/app/src/test/java/org/astraea/performance/ProducerExecutorTest.java
+++ b/app/src/test/java/org/astraea/performance/ProducerExecutorTest.java
@@ -8,9 +8,9 @@ import java.util.concurrent.TimeUnit;
 import java.util.function.BiConsumer;
 import java.util.function.Supplier;
 import java.util.stream.Stream;
-import org.astraea.Utils;
 import org.astraea.admin.Admin;
 import org.astraea.admin.TopicPartition;
+import org.astraea.common.Utils;
 import org.astraea.concurrent.State;
 import org.astraea.producer.Producer;
 import org.astraea.service.RequireBrokerCluster;
diff --git a/app/src/test/java/org/astraea/producer/ProducerTest.java b/app/src/test/java/org/astraea/producer/ProducerTest.java
index c444da59..813136d6 100644
--- a/app/src/test/java/org/astraea/producer/ProducerTest.java
+++ b/app/src/test/java/org/astraea/producer/ProducerTest.java
@@ -9,7 +9,7 @@ import java.util.concurrent.ExecutionException;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
-import org.astraea.Utils;
+import org.astraea.common.Utils;
 import org.astraea.consumer.Consumer;
 import org.astraea.consumer.Deserializer;
 import org.astraea.consumer.Header;
diff --git a/app/src/test/java/org/astraea/service/Database.java b/app/src/test/java/org/astraea/service/Database.java
new file mode 100644
index 00000000..b1078bf1
--- /dev/null
+++ b/app/src/test/java/org/astraea/service/Database.java
@@ -0,0 +1,138 @@
+package org.astraea.service;
+
+import static com.wix.mysql.EmbeddedMysql.anEmbeddedMysql;
+import static com.wix.mysql.config.Charset.UTF8;
+import static com.wix.mysql.config.MysqldConfig.aMysqldConfig;
+
+import com.wix.mysql.distribution.Version;
+import java.io.Closeable;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.util.Calendar;
+import java.util.concurrent.TimeUnit;
+import org.astraea.common.Utils;
+
+public interface Database extends Closeable {
+  static Database mysql() {
+    return Database.builder().build();
+  }
+
+  static Builder builder() {
+    return new Builder();
+  }
+
+  /** @return hostname to connect */
+  String hostname();
+
+  /** @return port to connect */
+  int port();
+
+  /** @return database name of this mysql */
+  String databaseName();
+
+  /** @return username to log in this mysql */
+  String user();
+
+  /** @return password to log in this mysql */
+  String password();
+
+  /** @return full JDBC url */
+  String url();
+
+  /** @return create a new JDBC connection. Please close it manually */
+  Connection createConnection();
+
+  @Override
+  void close();
+
+  class Builder {
+    private Builder() {}
+
+    private String databaseName = Utils.randomString(5);
+    private String user = "user";
+    private String password = "password";
+    private int port = 0;
+
+    public Builder databaseName(String databaseName) {
+      this.databaseName = databaseName;
+      return this;
+    }
+
+    public Builder user(String user) {
+      this.user = user;
+      return this;
+    }
+
+    public Builder password(String password) {
+      this.password = password;
+      return this;
+    }
+
+    public Builder port(int port) {
+      this.port = port;
+      return this;
+    }
+
+    public Database build() {
+      var config =
+          aMysqldConfig(Version.v5_7_latest)
+              .withCharset(UTF8)
+              .withUser(user, password)
+              .withTimeZone(Calendar.getInstance().getTimeZone().getID())
+              .withTimeout(2, TimeUnit.MINUTES)
+              .withServerVariable("max_connect_errors", 666)
+              .withTempDir(Utils.createTempDirectory("embedded_mysql").getAbsolutePath())
+              .withPort(Utils.resolvePort(port))
+              // make mysql use " replace '
+              // see https://stackoverflow.com/questions/13884854/mysql-double-quoted-table-names
+              .withServerVariable("sql-mode", "ANSI_QUOTES")
+              .build();
+      var database = anEmbeddedMysql(config).addSchema(databaseName).start();
+      return new Database() {
+        @Override
+        public void close() {
+          database.stop();
+        }
+
+        @Override
+        public String hostname() {
+          return Utils.hostname();
+        }
+
+        @Override
+        public int port() {
+          return config.getPort();
+        }
+
+        @Override
+        public String databaseName() {
+          return databaseName;
+        }
+
+        @Override
+        public String user() {
+          return config.getUsername();
+        }
+
+        @Override
+        public String password() {
+          return config.getPassword();
+        }
+
+        @Override
+        public String url() {
+          return "jdbc:mysql://" + hostname() + ":" + port() + "/" + databaseName();
+        }
+
+        @Override
+        public Connection createConnection() {
+          try {
+            return DriverManager.getConnection(url(), user(), password());
+          } catch (Exception e) {
+            throw new RuntimeException(e);
+          }
+        }
+      };
+    }
+  }
+}
diff --git a/app/src/test/java/org/astraea/service/DatabaseTest.java b/app/src/test/java/org/astraea/service/DatabaseTest.java
new file mode 100644
index 00000000..6c9c3f6c
--- /dev/null
+++ b/app/src/test/java/org/astraea/service/DatabaseTest.java
@@ -0,0 +1,29 @@
+package org.astraea.service;
+
+import org.astraea.common.Utils;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class DatabaseTest {
+
+  @Test
+  void testConnectionInformation() {
+    var user = Utils.randomString(10);
+    var password = Utils.randomString(10);
+    var databaseName = Utils.randomString(10);
+    var port = Utils.availablePort();
+    try (var database =
+        Database.builder()
+            .user(user)
+            .password(password)
+            .databaseName(databaseName)
+            .port(port)
+            .build()) {
+      Assertions.assertEquals(Utils.hostname(), database.hostname());
+      Assertions.assertEquals(port, database.port());
+      Assertions.assertEquals(user, database.user());
+      Assertions.assertEquals(password, database.password());
+      Assertions.assertEquals(databaseName, database.databaseName());
+    }
+  }
+}
diff --git a/app/src/test/java/org/astraea/service/RequireBrokerCluster.java b/app/src/test/java/org/astraea/service/RequireBrokerCluster.java
index b278d59f..82b3c83c 100644
--- a/app/src/test/java/org/astraea/service/RequireBrokerCluster.java
+++ b/app/src/test/java/org/astraea/service/RequireBrokerCluster.java
@@ -2,7 +2,7 @@ package org.astraea.service;
 
 import java.util.Map;
 import java.util.Set;
-import org.astraea.Utils;
+import org.astraea.common.Utils;
 import org.junit.jupiter.api.AfterAll;
 
 /**
@@ -27,7 +27,7 @@ public abstract class RequireBrokerCluster extends RequireJmxServer {
 
   @AfterAll
   static void shutdownClusters() {
-    Utils.close(BROKER_CLUSTER);
-    Utils.close(ZOOKEEPER_CLUSTER);
+    Utils.swallowException(BROKER_CLUSTER::close);
+    Utils.swallowException(ZOOKEEPER_CLUSTER::close);
   }
 }
diff --git a/app/src/test/java/org/astraea/service/RequireJmxServer.java b/app/src/test/java/org/astraea/service/RequireJmxServer.java
index f591cfc9..5bd35248 100644
--- a/app/src/test/java/org/astraea/service/RequireJmxServer.java
+++ b/app/src/test/java/org/astraea/service/RequireJmxServer.java
@@ -10,7 +10,7 @@ import java.rmi.registry.LocateRegistry;
 import javax.management.remote.JMXConnectorServer;
 import javax.management.remote.JMXConnectorServerFactory;
 import javax.management.remote.JMXServiceURL;
-import org.astraea.Utils;
+import org.astraea.common.Utils;
 import org.junit.jupiter.api.AfterAll;
 
 public abstract class RequireJmxServer {
@@ -23,7 +23,7 @@ public abstract class RequireJmxServer {
 
   @AfterAll
   static void shutdownJmxServer() {
-    Utils.close(JMX_CONNECTOR_SERVER::stop);
+    Utils.packException(JMX_CONNECTOR_SERVER::stop);
   }
 
   /**
diff --git a/app/src/test/java/org/astraea/service/Services.java b/app/src/test/java/org/astraea/service/Services.java
index 1031d118..de300478 100644
--- a/app/src/test/java/org/astraea/service/Services.java
+++ b/app/src/test/java/org/astraea/service/Services.java
@@ -15,7 +15,7 @@ import org.apache.kafka.common.network.ListenerName;
 import org.apache.kafka.common.utils.SystemTime;
 import org.apache.zookeeper.server.NIOServerCnxnFactory;
 import org.apache.zookeeper.server.ZooKeeperServer;
-import org.astraea.Utils;
+import org.astraea.common.Utils;
 
 public final class Services {
 
diff --git a/app/src/test/java/org/astraea/web/BrokerHandlerTest.java b/app/src/test/java/org/astraea/web/BrokerHandlerTest.java
index 155978c5..2d4a0035 100644
--- a/app/src/test/java/org/astraea/web/BrokerHandlerTest.java
+++ b/app/src/test/java/org/astraea/web/BrokerHandlerTest.java
@@ -5,8 +5,8 @@ import java.util.NoSuchElementException;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
-import org.astraea.Utils;
 import org.astraea.admin.Admin;
+import org.astraea.common.Utils;
 import org.astraea.service.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
diff --git a/app/src/test/java/org/astraea/web/GroupHandlerTest.java b/app/src/test/java/org/astraea/web/GroupHandlerTest.java
index 4e442290..af667ef8 100644
--- a/app/src/test/java/org/astraea/web/GroupHandlerTest.java
+++ b/app/src/test/java/org/astraea/web/GroupHandlerTest.java
@@ -6,8 +6,8 @@ import java.util.NoSuchElementException;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
-import org.astraea.Utils;
 import org.astraea.admin.Admin;
+import org.astraea.common.Utils;
 import org.astraea.consumer.Consumer;
 import org.astraea.service.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
diff --git a/app/src/test/java/org/astraea/web/PipelineHandlerTest.java b/app/src/test/java/org/astraea/web/PipelineHandlerTest.java
new file mode 100644
index 00000000..7caa9223
--- /dev/null
+++ b/app/src/test/java/org/astraea/web/PipelineHandlerTest.java
@@ -0,0 +1,54 @@
+package org.astraea.web;
+
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import org.astraea.admin.Admin;
+import org.astraea.common.Utils;
+import org.astraea.producer.Producer;
+import org.astraea.service.RequireBrokerCluster;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+public class PipelineHandlerTest extends RequireBrokerCluster {
+
+  @Test
+  void testQueries() {
+    // this topic-partition has no producers/consumers
+    var tp = new PipelineHandler.TopicPartition("g", 1);
+    Assertions.assertTrue(
+        PipelineHandler.filter(Map.of(PipelineHandler.ACTIVE_KEY, "false")).test(tp));
+    Assertions.assertFalse(
+        PipelineHandler.filter(Map.of(PipelineHandler.ACTIVE_KEY, "true")).test(tp));
+    Assertions.assertTrue(PipelineHandler.filter(Map.of()).test(tp));
+
+    // add some producers/consumers
+    tp.from.add(Mockito.mock(PipelineHandler.Producer.class));
+    Assertions.assertFalse(
+        PipelineHandler.filter(Map.of(PipelineHandler.ACTIVE_KEY, "false")).test(tp));
+    Assertions.assertTrue(
+        PipelineHandler.filter(Map.of(PipelineHandler.ACTIVE_KEY, "true")).test(tp));
+    Assertions.assertTrue(PipelineHandler.filter(Map.of()).test(tp));
+  }
+
+  @Test
+  void testGetPipeline() throws InterruptedException, ExecutionException {
+    var topic = Utils.randomString(10);
+    try (var admin = Admin.of(bootstrapServers())) {
+      admin.creator().topic(topic).numberOfPartitions(1).create();
+      TimeUnit.SECONDS.sleep(2);
+      try (var producer = Producer.of(bootstrapServers())) {
+        producer.sender().topic(topic).value(new byte[10]).run().toCompletableFuture().get();
+        var handler = new PipelineHandler(admin);
+        var response =
+            Assertions.assertInstanceOf(
+                PipelineHandler.TopicPartitions.class, handler.get(Optional.empty(), Map.of()));
+        Assertions.assertNotEquals(0, response.topicPartitions.size());
+        Assertions.assertEquals(
+            1, response.topicPartitions.stream().filter(t -> t.topic.equals(topic)).count());
+      }
+    }
+  }
+}
diff --git a/app/src/test/java/org/astraea/web/ProducerHandlerTest.java b/app/src/test/java/org/astraea/web/ProducerHandlerTest.java
index 1f27e98a..11f96509 100644
--- a/app/src/test/java/org/astraea/web/ProducerHandlerTest.java
+++ b/app/src/test/java/org/astraea/web/ProducerHandlerTest.java
@@ -6,8 +6,8 @@ import java.util.Set;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
-import org.astraea.Utils;
 import org.astraea.admin.Admin;
+import org.astraea.common.Utils;
 import org.astraea.producer.Producer;
 import org.astraea.service.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
diff --git a/app/src/test/java/org/astraea/web/TopicHandlerTest.java b/app/src/test/java/org/astraea/web/TopicHandlerTest.java
index f5af95fb..96cb70dc 100644
--- a/app/src/test/java/org/astraea/web/TopicHandlerTest.java
+++ b/app/src/test/java/org/astraea/web/TopicHandlerTest.java
@@ -5,8 +5,8 @@ import java.util.NoSuchElementException;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
-import org.astraea.Utils;
 import org.astraea.admin.Admin;
+import org.astraea.common.Utils;
 import org.astraea.service.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;