Repository: skiptests/astraea
Pull request #181 (2022-01-05 12:35:01)

Optimize partitioner algorithm
這次更新的內容是 #8 中優化部分的內容。
概述主要實現的內容：
1.對thoughput相關metrics做了歸一化處理，不再是之前簡單的相加比較。

2.之前有一個設計上的失誤。需要每秒最多進行一次load的計算，而之前在程式中表現出來是partitioner每次發送都計算，現已修正。

3.metrics和程式邏輯中需要的與broker相關的變量都被放入了`Broker`中統一管理。

4.partitioner對於異格機器有了應變的能力，哪怕每台機子的thoughPut能力不同，一樣可以做到loadBalance。（具體效果未知
）

5.增加了jvmUsage，當超過80%用量時，會強制減少該broker可能被partitioner分配到的次數。

===============================================================================
在實際環境中測試了一下，程式還存在一些問題待改進
- [x] Broker應該直接存儲BrokerID
- [x] loadSituation()被多線程調用不是很安全
- [x] 對於多jmx server無法正確全部配置

確認正常運行後我再請學長開始review


Patch:
diff --git a/.gitignore b/.gitignore
index 6bc550bc..3835bffb 100644
--- a/.gitignore
+++ b/.gitignore
@@ -15,7 +15,6 @@ zookeeper.dockerfile
 broker.dockerfile
 kafka_tool.dockerfile
 spark.dockerfile
-astraea.dockerfile
 
 # we don't put binary file to git repo
 gradle-wrapper.jar
\ No newline at end of file
diff --git a/README.md b/README.md
index 9aff1486..286e0cb0 100644
--- a/README.md
+++ b/README.md
@@ -221,6 +221,7 @@ Run the benchmark from source
 10. --prop.file: the path to property file.
 11. --partitioner: the partitioner to use in producers
 12. --jmx.servers: the jmx server addresses of the brokers 
+13. --key.distribution: Name of the distribution. Available distribution names: "uniform", "zipfian", "latest". Default: (No key)
 
 ---
 
diff --git a/app/src/main/java/org/astraea/App.java b/app/src/main/java/org/astraea/App.java
index a049a25a..23312892 100644
--- a/app/src/main/java/org/astraea/App.java
+++ b/app/src/main/java/org/astraea/App.java
@@ -5,6 +5,7 @@ import java.lang.reflect.InvocationTargetException;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
+import org.astraea.automation.Automation;
 import org.astraea.metrics.MetricExplorer;
 import org.astraea.performance.Performance;
 import org.astraea.topic.ReplicaCollie;
@@ -20,7 +21,8 @@ public class App {
           "replica", ReplicaCollie.class,
           "score", PartitionScore.class,
           "performance", Performance.class,
-          "monitor", ReplicaSyncingMonitor.class);
+          "monitor", ReplicaSyncingMonitor.class,
+          "automation", Automation.class);
 
   static void execute(Map<String, Class<?>> mains, List<String> args) throws Throwable {
 
diff --git a/app/src/main/java/org/astraea/Utils.java b/app/src/main/java/org/astraea/Utils.java
index cdd34473..b022c9ec 100644
--- a/app/src/main/java/org/astraea/Utils.java
+++ b/app/src/main/java/org/astraea/Utils.java
@@ -7,6 +7,10 @@ import java.net.InetAddress;
 import java.net.UnknownHostException;
 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.TimeUnit;
 import java.util.function.Supplier;
 import java.util.stream.Stream;
@@ -111,6 +115,22 @@ public final class Utils {
     }
   }
 
+  /**
+   * 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");
diff --git a/app/src/main/java/org/astraea/automation/Automation.java b/app/src/main/java/org/astraea/automation/Automation.java
new file mode 100644
index 00000000..50fcd198
--- /dev/null
+++ b/app/src/main/java/org/astraea/automation/Automation.java
@@ -0,0 +1,97 @@
+package org.astraea.automation;
+
+import com.beust.jcommander.Parameter;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.ExecutionException;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.KafkaAdminClient;
+import org.astraea.argument.ArgumentUtil;
+import org.astraea.performance.Performance;
+
+/**
+ * By configuring the parameters in config/automation.properties, control the execution times of
+ * performance and its configuration parameters.
+ *
+ * <ol>
+ *   <li>--file: The address of the automation.properties in config folder.
+ * </ol>
+ */
+public class Automation {
+  private static final List<String> performanceProperties =
+      List.of(
+          "--bootstrap.servers",
+          "--record.size",
+          "--run.until",
+          "--compression",
+          "--consumers",
+          "--fixed.size",
+          "--jmx.servers",
+          "--partitioner",
+          "--partitions",
+          "--producers",
+          "--prop.file",
+          "--replicas",
+          "--topic");
+
+  public static void main(String[] args) {
+    try {
+      var properties = new Properties();
+      var arg = ArgumentUtil.parseArgument(new automationArgument(), args);
+      properties.load(new FileInputStream(arg.address));
+      var whetherDeleteTopic = properties.getProperty("--whetherDeleteTopic").equals("true");
+      var bootstrap = properties.getProperty("--bootstrap.servers");
+      var config = new Properties();
+      config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap);
+
+      var i = 0;
+      var times = 0;
+      if (properties.getProperty("--time").equals("Default")) times = 5;
+      else times = Integer.parseInt(properties.getProperty("--time"));
+
+      while (i < times) {
+        var str =
+            Performance.execute(
+                ArgumentUtil.parseArgument(
+                    new Performance.Argument(), performanceArgs(properties)));
+        i++;
+        if (whetherDeleteTopic) {
+          try (final AdminClient adminClient = KafkaAdminClient.create(config)) {
+            var topicName = str.get();
+            adminClient.deleteTopics(List.of(topicName));
+          }
+        }
+        System.out.println("=============== " + i + " time Performance Complete! ===============");
+      }
+    } catch (IOException | InterruptedException | ExecutionException e) {
+      e.printStackTrace();
+    }
+  }
+
+  @SuppressWarnings("unChecked")
+  private static String[] performanceArgs(Properties properties) {
+    var args = new ArrayList<String>();
+    performanceProperties.forEach(
+        str -> {
+          var property = properties.getProperty(str);
+          if (property != null && !property.equals("Default")) {
+            args.add(str);
+            args.add(property);
+          }
+        });
+    var strings = new String[args.size()];
+    return args.toArray(strings);
+  }
+
+  private static class automationArgument {
+    @Parameter(
+        names = {"--file"},
+        description = "String: automation.properties address",
+        validateWith = ArgumentUtil.NotEmptyString.class)
+    String address = "";
+  }
+}
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 dac0c30f..cf5ed419 100644
--- a/app/src/main/java/org/astraea/metrics/collector/BeanCollector.java
+++ b/app/src/main/java/org/astraea/metrics/collector/BeanCollector.java
@@ -1,7 +1,12 @@
 package org.astraea.metrics.collector;
 
 import java.time.Duration;
-import java.util.*;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ConcurrentSkipListMap;
 import java.util.concurrent.locks.Lock;
@@ -121,7 +126,7 @@ public class BeanCollector {
                 }
               }
 
-              private void tryUpdate() {
+              private synchronized void tryUpdate() {
                 var needUpdate =
                     objects.keySet().stream()
                         .max((Long::compare))
diff --git a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClient.java b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClient.java
index 03542570..95909263 100644
--- a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClient.java
+++ b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClient.java
@@ -1,24 +1,21 @@
 package org.astraea.partitioner.nodeLoadMetric;
 
-import static java.lang.Double.sum;
-
 import java.io.IOException;
 import java.net.InetAddress;
 import java.net.UnknownHostException;
+import java.time.Duration;
 import java.util.ArrayList;
-import java.util.Comparator;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicReference;
-import java.util.function.Function;
 import java.util.stream.Collectors;
-import java.util.stream.IntStream;
 import org.apache.kafka.common.Cluster;
-import org.apache.kafka.common.Node;
 import org.astraea.Utils;
 import org.astraea.metrics.collector.Receiver;
+import org.astraea.metrics.java.HasJvmMemory;
 import org.astraea.metrics.kafka.KafkaMetrics;
 
 /**
@@ -28,12 +25,15 @@ import org.astraea.metrics.kafka.KafkaMetrics;
 public class NodeLoadClient {
 
   private final List<Receiver> receiverList;
-
-  private Map<Integer, Map<String, Receiver>> eachNodeIDMetrics;
-
   private final Map<String, Integer> currentJmxAddresses;
-
   private static final ReceiverFactory RECEIVER_FACTORY = new ReceiverFactory();
+  private long lastTime = -1;
+  // visible for testing
+  List<Broker> brokers;
+  private Map<Integer, Integer> currentLoadNode;
+  private int referenceBrokerID;
+  private CountDownLatch countDownLatch;
+  private boolean notInMethod = true;
 
   private static final String regex =
       "((25[0-5]|2[0-4]\\d|1\\d{2}|[1-9]?\\d)\\.){3}" + "(25[0-5]|2[0-4]\\d|1\\d{2}|[1-9]?\\d)$";
@@ -45,193 +45,294 @@ public class NodeLoadClient {
   }
 
   /**
-   * @param cluster the cluster from the partitioner
-   * @return each node overLoad count in preset time
+   * When the interval between getting the load twice exceeds one second, update the load and
+   * return, otherwise directly return to the existing load.
+   *
+   * @param cluster from partitioner
+   * @return each node load count in preset time
    */
-  public Map<Integer, Integer> nodesOverLoad(Cluster cluster) throws UnknownHostException {
-    var nodeIDReceiver = new HashMap<Integer, List<Receiver>>();
-    var addresses =
-        cluster.nodes().stream()
-            .collect(Collectors.toMap(Node::id, node -> Map.entry(node.host(), node.port())));
-
-    for (Map.Entry<Integer, Map.Entry<String, Integer>> hostPort : addresses.entrySet()) {
-      List<Receiver> matchingNode;
-      var host = ipAddress(hostPort);
-      matchingNode =
-          receiverList.stream()
-              .filter(
-                  receiver ->
-                      (Objects.equals(nodeIDReceiver.size(), 0)
-                              || (nodeIDReceiver.values().stream()
-                                      .map(
-                                          n ->
-                                              Objects.equals(
-                                                  n.stream().findAny().get().host(), host))
-                                      .anyMatch(n -> n.equals(true))
-                                  && nodeIDReceiver.values().stream()
-                                      .map(
-                                          n ->
-                                              !Objects.equals(
-                                                  n.stream().findAny().get().port(),
-                                                  hostPort.getValue().getValue()))
-                                      .anyMatch(n -> n.equals(true))))
-                          && Objects.equals(receiver.host(), host))
-              .collect(Collectors.toList());
-
-      if (matchingNode.size() > 0) {
-        if (!nodeIDReceiver.containsKey(hostPort.getKey()))
-          nodeIDReceiver.put(hostPort.getKey(), matchingNode);
-        else matchingNode.forEach(node -> nodeIDReceiver.get(hostPort.getKey()).add(node));
-      }
+  public synchronized Map<Integer, Integer> loadSituation(Cluster cluster)
+      throws UnknownHostException {
+    if (overOneSecond() && notInMethod) {
+      notInMethod = false;
+      nodesOverLoad(cluster);
+    }
+    try {
+      countDownLatch.await();
+    } catch (InterruptedException e) {
+      e.printStackTrace();
     }
+    return currentLoadNode;
+  }
 
-    eachNodeIDMetrics = metricsNameForReceiver(nodeIDReceiver);
-    var eachBrokerMsgPerSec = brokersMsgPerSec();
-    var overLoadCount = new HashMap<Integer, Integer>();
-    eachBrokerMsgPerSec.keySet().forEach(e -> overLoadCount.put(e, 0));
-    var i = 0;
-    var minRecordSec =
-        eachBrokerMsgPerSec.values().stream()
-            .map(n -> n.size())
-            .min(Comparator.comparing(Integer::intValue))
-            .get();
-    while (i < minRecordSec) {
-      var avg = avgMsgSec(i, eachBrokerMsgPerSec);
-      var eachMsg = brokersMsgSec(i, eachBrokerMsgPerSec);
-      var standardDeviation = standardDeviationImperative(eachMsg, avg);
-      eachMsg.forEach(
-          (nodeID, msg) -> {
-            if (msg > (avg + standardDeviation)) {
-              overLoadCount.put(nodeID, overLoadCount.get(nodeID) + 1);
-            }
+  /** @param cluster the cluster from the partitioner */
+  private synchronized void nodesOverLoad(Cluster cluster) {
+    var nodes = cluster.nodes();
+    if (lastTime == -1) {
+      countDownLatch = new CountDownLatch(1);
+      brokers =
+          nodes.stream()
+              .map(node -> new Broker(node.id(), node.host(), node.port()))
+              .collect(Collectors.toList());
+      referenceBrokerID = brokers.stream().findFirst().get().brokerID;
+      var nodeIDReceiver = new HashMap<Integer, List<Receiver>>();
+      receiverList.forEach(
+          receiver -> {
+            var brokersID = brokerIDOfReceiver(receiver.host());
+            brokersID.forEach(
+                brokerID -> {
+                  if (nodeIDReceiver.containsKey(brokerID)) {
+                    nodeIDReceiver.get(brokerID).add(receiver);
+                  } else {
+                    nodeIDReceiver.put(brokerID, new ArrayList<>(List.of(receiver)));
+                  }
+                });
           });
-      i++;
+      metricsNameForReceiver(nodeIDReceiver);
     }
-    return overLoadCount;
-  }
-
-  /** @return data transfer per second per node */
-  private Map<Integer, List<Double>> brokersMsgPerSec() {
-    var eachMsg = new HashMap<Integer, List<Double>>();
-
-    for (Map.Entry<Integer, Map<String, Receiver>> entry : eachNodeIDMetrics.entrySet()) {
-      List<Double> sumList = new ArrayList<>();
-      var outMsg = entry.getValue().get(KafkaMetrics.BrokerTopic.BytesOutPerSec.toString());
-      var inMsg = entry.getValue().get(KafkaMetrics.BrokerTopic.BytesInPerSec.toString());
-      var outMsgBean = outMsg.current();
-      var inMsgBean = inMsg.current();
-      IntStream.range(0, Math.min(outMsgBean.size(), inMsgBean.size()))
-          .forEach(
-              i ->
-                  sumList.add(
-                      sum(
-                          Double.parseDouble(
-                              outMsgBean
-                                  .get(i)
-                                  .beanObject()
-                                  .getAttributes()
-                                  .get("MeanRate")
-                                  .toString()),
-                          Double.parseDouble(
-                              inMsgBean
-                                  .get(i)
-                                  .beanObject()
-                                  .getAttributes()
-                                  .get("MeanRate")
-                                  .toString()))));
-      eachMsg.put(entry.getKey(), sumList);
-    }
-    return eachMsg;
+
+    brokersMsgPerSec();
+
+    var totalInput = brokers.stream().map(broker -> broker.input).reduce(0.0, Double::sum);
+    var totalOutput = brokers.stream().map(broker -> broker.output).reduce(0.0, Double::sum);
+
+    AtomicReference<Double> totalBrokerSituation = new AtomicReference<>(0.0);
+    brokers.forEach(
+        broker -> {
+          broker.maxThoughPut();
+          broker.inputNormalized(totalInput);
+          broker.outputNormalized(totalOutput);
+          broker.thoughPutComparison(findBroker(referenceBrokerID).maxThoughPut);
+          broker.brokerSituation();
+          totalBrokerSituation.updateAndGet(v -> v + broker.brokerSituation);
+        });
+
+    var totalSituation = totalBrokerSituation.get();
+
+    brokers.forEach(broker -> broker.brokerSituationNormalized(totalSituation));
+
+    var avg = totalSituation / brokers.size();
+
+    // TODO Regarding countermeasures against cluster conditions
+    var standardDeviation = standardDeviationImperative(avg);
+
+    brokers.forEach(broker -> broker.load(brokerLoad(broker.brokerSituation, avg)));
+    currentLoadNode =
+        brokers.stream().collect(Collectors.toMap(broker -> broker.brokerID, Broker::totalLoad));
+    lastTime = System.currentTimeMillis();
+    countDownLatch.countDown();
+    notInMethod = true;
   }
 
-  public List<Double> avgBrokersMsgPerSec(Map<Integer, List<Double>> eachMsg) {
-    return IntStream.range(0, eachMsg.values().stream().findFirst().get().size())
-        .mapToDouble(i -> eachMsg.values().stream().map(s -> s.get(i)).reduce(0.0, Double::sum))
-        .map(sum -> sum / eachMsg.size())
-        .boxed()
-        .collect(Collectors.toList());
+  private void brokersMsgPerSec() {
+    brokers.forEach(
+        broker -> {
+          broker.output =
+              Double.parseDouble(
+                  broker
+                      .metrics
+                      .get(KafkaMetrics.BrokerTopic.BytesOutPerSec.name())
+                      .current()
+                      .get(0)
+                      .beanObject()
+                      .getAttributes()
+                      .get("MeanRate")
+                      .toString());
+          broker.input =
+              Double.parseDouble(
+                  broker
+                      .metrics
+                      .get(KafkaMetrics.BrokerTopic.BytesInPerSec.name())
+                      .current()
+                      .get(0)
+                      .beanObject()
+                      .getAttributes()
+                      .get("MeanRate")
+                      .toString());
+        });
   }
 
-  public double standardDeviationImperative(
-      Map<Integer, Double> eachMsg, double avgBrokersMsgPerSec) {
+  public double standardDeviationImperative(double avgBrokersSituation) {
     var variance = new AtomicReference<>(0.0);
-    eachMsg.forEach(
-        (nodeID, msg) ->
+    brokers.forEach(
+        broker ->
             variance.updateAndGet(
-                v -> v + (msg - avgBrokersMsgPerSec) * (msg - avgBrokersMsgPerSec)));
-    return Math.sqrt(variance.get() / eachMsg.size());
+                v ->
+                    v
+                        + (broker.brokerSituationNormalized - avgBrokersSituation)
+                            * (broker.brokerSituationNormalized - avgBrokersSituation)));
+    return Math.sqrt(variance.get() / brokers.size());
   }
 
-  /**
-   * @param nodeIDReceiver nodes metrics that exists in topic
-   * @return the name and corresponding data of the metrics obtained by each node
-   */
-  public Map<Integer, Map<String, Receiver>> metricsNameForReceiver(
-      HashMap<Integer, List<Receiver>> nodeIDReceiver) {
-
-    var result = new HashMap<Integer, Map<String, Receiver>>();
-
-    nodeIDReceiver.forEach(
-        (nodeID, receivers) -> {
-          result.put(
-              nodeID,
-              receivers.stream()
-                  .map(
-                      receiver ->
-                          receiver.current().stream()
-                              .map(
-                                  bean ->
-                                      bean.beanObject().getProperties().values().stream()
-                                          .findAny()
-                                          .get())
-                              .findAny()
-                              .get())
-                  .collect(Collectors.toList())
-                  .stream()
-                  .collect(
-                      Collectors.toMap(
-                          Function.identity(),
-                          name ->
-                              receivers.stream()
-                                  .filter(
-                                      mbean ->
-                                          mbean.current().stream()
-                                              .findAny()
-                                              .get()
-                                              .beanObject()
-                                              .getProperties()
-                                              .values()
-                                              .stream()
-                                              .findAny()
-                                              .get()
-                                              .equals(name))
-                                  .findAny()
-                                  .get())));
-        });
-    return result;
+  /** @param brokerIDReceiver nodes metrics that exists in topic */
+  private void metricsNameForReceiver(HashMap<Integer, List<Receiver>> brokerIDReceiver) {
+    brokerIDReceiver.forEach(
+        (brokerID, receivers) ->
+            receivers.stream()
+                .map(
+                    receiver ->
+                        receiver.current().stream()
+                            .map(
+                                bean ->
+                                    bean.beanObject().getProperties().values().stream()
+                                        .findAny()
+                                        .get())
+                            .findAny()
+                            .get())
+                .collect(Collectors.toList())
+                .forEach(
+                    name ->
+                        findBroker(brokerID)
+                            .metrics
+                            .put(
+                                name,
+                                receivers.stream()
+                                    .filter(
+                                        mbean ->
+                                            mbean.current().stream()
+                                                .findAny()
+                                                .get()
+                                                .beanObject()
+                                                .getProperties()
+                                                .values()
+                                                .stream()
+                                                .findAny()
+                                                .get()
+                                                .equals(name))
+                                    .findAny()
+                                    .get())));
   }
 
-  private Map<Integer, Double> brokersMsgSec(int index, Map<Integer, List<Double>> eachMsg) {
-    return eachMsg.entrySet().stream()
-        .collect(Collectors.toMap(Map.Entry::getKey, entry -> entry.getValue().get(index)));
+  private Broker findBroker(int brokerID) {
+    return brokers.stream().filter(broker -> broker.brokerID == brokerID).findAny().get();
   }
 
-  private double avgMsgSec(int index, Map<Integer, List<Double>> eachMsg) {
-    return avgBrokersMsgPerSec(eachMsg).get(index);
-  }
-
-  private String ipAddress(Map.Entry<Integer, Map.Entry<String, Integer>> hostPort)
-      throws UnknownHostException {
-    var host = "-1.-1.-1.-1";
-    if (!hostPort.getValue().getKey().matches(regex)) {
-      host = InetAddress.getByName(hostPort.getValue().getKey()).toString().split("/")[1];
+  private String ipAddress(String host) {
+    var correctHost = "-1.-1.-1.-1";
+    if (notIPAddress(host)) {
+      try {
+        correctHost = String.valueOf(InetAddress.getByName(host)).split("/")[1];
+      } catch (UnknownHostException e) {
+        e.printStackTrace();
+      }
     } else {
-      host = hostPort.getValue().getKey();
+      correctHost = host;
     }
-    return host;
+    return correctHost;
+  }
+
+  private boolean notIPAddress(String host) {
+    return !host.matches(regex);
+  }
+
+  private boolean overOneSecond() {
+    return lastTime + Duration.ofSeconds(1).toMillis() <= System.currentTimeMillis();
   }
 
   public void close() {
     RECEIVER_FACTORY.close(currentJmxAddresses);
   }
+
+  private List<Integer> brokerIDOfReceiver(String host) {
+    return brokers.stream()
+        .filter(broker -> Objects.equals(ipAddress(broker.host), host))
+        .map(broker -> broker.brokerID)
+        .collect(Collectors.toList());
+  }
+
+  // visible of test
+  int brokerLoad(double brokerSituation, double avg) {
+    var initialization = 5;
+    var loadDeviationRate = (brokerSituation - avg) / avg;
+    var load = (int) Math.round(initialization + loadDeviationRate * 10);
+    if (load > 10) load = 10;
+    else if (load < 0) load = 0;
+    return load;
+  }
+
+  public double thoughPutComparison(int brokerID) {
+    return findBroker(brokerID).thoughPutComparison;
+  }
+
+  public double memoryUsage(int brokerID) {
+    return findBroker(brokerID).memoryUsage();
+  }
+
+  // visible of test
+  static class Broker {
+    private final int brokerID;
+    private final String host;
+    private final int port;
+    private int count = 0;
+    private final Map<Integer, Integer> load = new HashMap<>();
+    private final Map<String, Receiver> metrics = new HashMap<>();
+    private double input;
+    private double output;
+    private double maxThoughPut;
+    private double brokerSituation;
+    private double thoughPutComparison;
+    private double inputNormalized;
+    private double outputNormalized;
+    private double brokerSituationNormalized;
+    private static final double inputWeights = 0.5;
+    private static final double outputWeights = 0.5;
+
+    // visible of test
+    Broker(int brokerID, String host, int port) {
+      this.brokerID = brokerID;
+      this.host = host;
+      this.port = port;
+    }
+
+    private int totalLoad() {
+      return load.values().stream().reduce(0, Integer::sum);
+    }
+
+    private void load(Integer currentLoad) {
+      this.load.put(count % 10, currentLoad);
+      count++;
+    }
+
+    private void brokerSituationNormalized(double totalSituation) {
+      this.brokerSituationNormalized = brokerSituation / totalSituation;
+    }
+
+    private void inputNormalized(double brokersInput) {
+      this.inputNormalized = (input + 1) / (brokersInput + 1);
+    }
+
+    private void outputNormalized(double brokersOutput) {
+      this.outputNormalized = (output + 1) / (brokersOutput + 1);
+    }
+
+    private void brokerSituation() {
+      this.brokerSituation =
+          (inputNormalized * inputWeights + outputNormalized * outputWeights) / thoughPutComparison;
+    }
+
+    private void maxThoughPut() {
+      this.maxThoughPut = Math.max(maxThoughPut, input + output);
+    }
+
+    private void thoughPutComparison(double referenceThoughPut) {
+      if (count >= 20) {
+        this.thoughPutComparison =
+            (maxThoughPut - referenceThoughPut) / (referenceThoughPut + 1) + 1;
+      } else if (count >= 10) {
+        this.thoughPutComparison =
+            (maxThoughPut - referenceThoughPut) / (referenceThoughPut + 1) / 2 + 1;
+      } else if (count >= 0) {
+        this.thoughPutComparison = 1;
+      }
+    }
+
+    private double memoryUsage() {
+      if (count < 10) {
+        return 0.1;
+      } else {
+        var jvm = (HasJvmMemory) metrics.get("Memory").current().stream().findAny().get();
+        return (jvm.heapMemoryUsage().getUsed() + 0.0) / (jvm.heapMemoryUsage().getMax() + 1);
+      }
+    }
+  }
 }
diff --git a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/ReceiverFactory.java b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/ReceiverFactory.java
index ac309631..eb230376 100644
--- a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/ReceiverFactory.java
+++ b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/ReceiverFactory.java
@@ -23,7 +23,7 @@ public class ReceiverFactory {
     beanCollector =
         BeanCollector.builder()
             .interval(Duration.ofSeconds(1))
-            .numberOfObjectsPerNode(10)
+            .numberOfObjectsPerNode(1)
             .clientCreator(MBeanClient::jndi)
             .build();
   }
@@ -51,6 +51,13 @@ public class ReceiverFactory {
                       .port(port)
                       .metricsGetter(KafkaMetrics.BrokerTopic.BytesOutPerSec::fetch)
                       .build());
+              receiversList.add(
+                  beanCollector
+                      .register()
+                      .host(host)
+                      .port(port)
+                      .metricsGetter(KafkaMetrics.Host::jvmMemory)
+                      .build());
               count.put(nodeKey(host, port), 1);
             } else {
               count.put(nodeKey(host, port), count.get(nodeKey(host, port)) + 1);
diff --git a/app/src/main/java/org/astraea/partitioner/smoothPartitioner/DependencyClient.java b/app/src/main/java/org/astraea/partitioner/smoothPartitioner/DependencyClient.java
index 777dfc7c..5667d1a7 100644
--- a/app/src/main/java/org/astraea/partitioner/smoothPartitioner/DependencyClient.java
+++ b/app/src/main/java/org/astraea/partitioner/smoothPartitioner/DependencyClient.java
@@ -1,6 +1,14 @@
 package org.astraea.partitioner.smoothPartitioner;
 
-// TODO
+/**
+ * The interface for the {@link SmoothWeightPartitioner}
+ *
+ * @see SmoothWeightPartitioner Enable users to send dependent data.
+ */
 public interface DependencyClient {
+  void startDependency();
+
   void finishDependency();
+
+  String state();
 }
diff --git a/app/src/main/java/org/astraea/partitioner/smoothPartitioner/SmoothWeightPartitioner.java b/app/src/main/java/org/astraea/partitioner/smoothPartitioner/SmoothWeightPartitioner.java
index 4942327a..287af45e 100644
--- a/app/src/main/java/org/astraea/partitioner/smoothPartitioner/SmoothWeightPartitioner.java
+++ b/app/src/main/java/org/astraea/partitioner/smoothPartitioner/SmoothWeightPartitioner.java
@@ -27,7 +27,8 @@ import org.astraea.partitioner.nodeLoadMetric.NodeLoadClient;
  * <pre>{@code
  * KafkaProducer producer = new KafkaProducer(props);
  *
- * var dependencyControl = SmoothWeightPartitioner.beginDependency(producer);
+ * var dependencyControl = SmoothWeightPartitioner.dependencyClient(producer);
+ * dependencyControl.startDependency();
  * try{
  *     producer.send();
  *     producer.send();
@@ -43,10 +44,10 @@ public class SmoothWeightPartitioner implements Partitioner, DependencyClient {
    * Record the current weight of each node according to Poisson calculation and the weight after
    * partitioner calculation.
    */
-  private Map<Integer, int[]> brokerHashMap = new HashMap<>();
+  private final Map<Integer, int[]> brokersWeight = new HashMap<>();
 
   private NodeLoadClient nodeLoadClient;
-  private static final DependencyManager dependencyManager = new DependencyManager();
+  private DependencyManager dependencyManager;
   /**
    * This parameter only works in dependency mode.And it will be specified when the dependency is
    * executed for the first time, and no changes will be made afterwards.
@@ -56,51 +57,55 @@ public class SmoothWeightPartitioner implements Partitioner, DependencyClient {
   @Override
   public int partition(
       String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
-    Map<Integer, Integer> overLoadCount;
+    Map<Integer, Integer> loadCount;
     var rand = new Random();
     switch (dependencyManager.currentState) {
-      case Start_Dependency:
-        overLoadCount = overLoadCount(cluster);
-        Objects.requireNonNull(overLoadCount, "OverLoadCount should not be null.");
-        var minOverLoadNode =
-            overLoadCount.entrySet().stream().min(Map.Entry.comparingByValue()).get();
+      case START_DEPENDENCY:
+        loadCount = loadCount(cluster);
+        Objects.requireNonNull(loadCount, "OverLoadCount should not be null.");
+        var minOverLoadNode = loadCount.entrySet().stream().min(Map.Entry.comparingByValue()).get();
         var minOverLoadPartition = cluster.partitionsForNode(minOverLoadNode.getKey());
         targetPartition =
             minOverLoadPartition.get(rand.nextInt(minOverLoadPartition.size())).partition();
-        dependencyManager.transitionTo(DependencyManager.State.IN_Dependency);
+        dependencyManager.transitionTo(DependencyManager.State.IN_DEPENDENCY);
         return targetPartition;
-      case IN_Dependency:
+      case IN_DEPENDENCY:
         return targetPartition;
       default:
-        overLoadCount = overLoadCount(cluster);
+        loadCount = loadCount(cluster);
         var loadPoisson = new LoadPoisson();
 
-        Objects.requireNonNull(overLoadCount, "OverLoadCount should not be null.");
-        brokerHashMap(loadPoisson.allPoisson(overLoadCount));
+        Objects.requireNonNull(loadCount, "OverLoadCount should not be null.");
+        brokersWeight(loadPoisson.allPoisson(loadCount));
         AtomicReference<Map.Entry<Integer, int[]>> maxWeightServer = new AtomicReference<>();
         var allWeight = allNodesWeight();
-        var currentBrokerHashMap = brokerHashMap();
-        currentBrokerHashMap.forEach(
+        brokersWeight.forEach(
             (nodeID, weight) -> {
               if (maxWeightServer.get() == null
                   || weight[1] > maxWeightServer.get().getValue()[1]) {
                 maxWeightServer.set(Map.entry(nodeID, weight));
               }
-              currentBrokerHashMap.put(nodeID, new int[] {weight[0], weight[1] + weight[0]});
+              brokersWeight.put(nodeID, new int[] {weight[0], weight[1] + weight[0]});
             });
         Objects.requireNonNull(maxWeightServer.get(), "MaxWeightServer should not be null.");
-        currentBrokerHashMap.put(
+        brokersWeight.put(
             maxWeightServer.get().getKey(),
             new int[] {
               maxWeightServer.get().getValue()[0], maxWeightServer.get().getValue()[1] - allWeight
             });
-        currentBrokerHashMap(currentBrokerHashMap);
+        brokersWeight
+            .keySet()
+            .forEach(
+                brokerID -> {
+                  if (memoryWarning(brokerID)) {
+                    subBrokerWeight(brokerID, 100);
+                  }
+                });
 
         ArrayList<Integer> partitionList = new ArrayList<>();
         cluster
             .partitionsForNode(maxWeightServer.get().getKey())
             .forEach(partitionInfo -> partitionList.add(partitionInfo.partition()));
-
         return partitionList.get(rand.nextInt(partitionList.size()));
     }
   }
@@ -126,6 +131,7 @@ public class SmoothWeightPartitioner implements Partitioner, DependencyClient {
       Objects.requireNonNull(mapAddress, "You must configure jmx_servers correctly.");
 
       nodeLoadClient = new NodeLoadClient(mapAddress);
+      dependencyManager = new DependencyManager();
     } catch (IOException e) {
       throw new RuntimeException();
     }
@@ -133,38 +139,37 @@ public class SmoothWeightPartitioner implements Partitioner, DependencyClient {
 
   /** Change the weight of the node according to the current Poisson. */
   // visible for test
-  public synchronized void brokerHashMap(Map<Integer, Double> poissonMap) {
+  public synchronized void brokersWeight(Map<Integer, Double> poissonMap) {
     poissonMap.forEach(
         (key, value) -> {
-          if (!brokerHashMap.containsKey(key)) {
-            brokerHashMap.put(key, new int[] {(int) ((1 - value) * 20), 0});
+          var thoughPutAbility = nodeLoadClient.thoughPutComparison(key);
+          if (!brokersWeight.containsKey(key)) {
+            brokersWeight.put(key, new int[] {(int) ((1 - value) * 20 * thoughPutAbility), 0});
           } else {
-            brokerHashMap.put(key, new int[] {(int) ((1 - value) * 20), brokerHashMap.get(key)[1]});
+            brokersWeight.put(
+                key,
+                new int[] {(int) ((1 - value) * 20 * thoughPutAbility), brokersWeight.get(key)[1]});
           }
         });
   }
 
-  private synchronized int allNodesWeight() {
-    return brokerHashMap.values().stream().mapToInt(vs -> vs[0]).sum();
-  }
-
-  // visible for test
-  public synchronized Map<Integer, int[]> brokerHashMap() {
-    return brokerHashMap;
+  private void subBrokerWeight(int brokerID, int subNumber) {
+    brokersWeight.put(
+        brokerID,
+        new int[] {brokersWeight.get(brokerID)[0], brokersWeight.get(brokerID)[1] - subNumber});
   }
 
-  private synchronized void currentBrokerHashMap(Map<Integer, int[]> currentBrokerHashMap) {
-    brokerHashMap = currentBrokerHashMap;
+  private synchronized int allNodesWeight() {
+    return brokersWeight.values().stream().mapToInt(vs -> vs[0]).sum();
   }
 
-  // visible for test
-  public void brokerHashMapValue(Integer x, int y) {
-    brokerHashMap.put(x, new int[] {0, y});
+  public static DependencyClient dependencyClient(KafkaProducer<?, ?> producer) {
+    return (DependencyClient) Utils.requireField(producer, "partitioner");
   }
 
-  public static synchronized DependencyClient beginDependency(KafkaProducer<?, ?> producer) {
-    dependencyManager.beginDependency();
-    return (DependencyClient) Utils.requireField(producer, "partitioner");
+  @Override
+  public synchronized void startDependency() {
+    dependencyManager.startDependency();
   }
 
   @Override
@@ -172,22 +177,31 @@ public class SmoothWeightPartitioner implements Partitioner, DependencyClient {
     dependencyManager.finishDependency();
   }
 
-  private Map<Integer, Integer> overLoadCount(Cluster cluster) {
+  private boolean memoryWarning(int brokerID) {
+    return nodeLoadClient.memoryUsage(brokerID) >= 0.8;
+  }
+
+  private Map<Integer, Integer> loadCount(Cluster cluster) {
     try {
-      return nodeLoadClient.nodesOverLoad(cluster);
+      return nodeLoadClient.loadSituation(cluster);
     } catch (UnknownHostException e) {
       throw new IllegalArgumentException(e);
     }
   }
 
+  // visible of testing
+  public String state() {
+    return dependencyManager.currentState.toString();
+  }
+
   /**
    * Store the state related to the partitioner dependency and avoid wrong transitions between them.
    */
   private static class DependencyManager {
     private volatile State currentState = State.UNINITIALIZED;
 
-    private synchronized void beginDependency() {
-      transitionTo(State.Start_Dependency);
+    private synchronized void startDependency() {
+      transitionTo(State.START_DEPENDENCY);
     }
 
     private synchronized void finishDependency() {
@@ -196,19 +210,19 @@ public class SmoothWeightPartitioner implements Partitioner, DependencyClient {
 
     private enum State {
       UNINITIALIZED,
-      Start_Dependency,
-      IN_Dependency,
+      START_DEPENDENCY,
+      IN_DEPENDENCY,
       FATAL_ERROR;
 
       private boolean isTransitionValid(
           DependencyManager.State source, DependencyManager.State target) {
         switch (target) {
           case UNINITIALIZED:
-            return source == Start_Dependency || source == IN_Dependency;
-          case Start_Dependency:
+            return source == START_DEPENDENCY || source == IN_DEPENDENCY;
+          case START_DEPENDENCY:
             return source == UNINITIALIZED;
-          case IN_Dependency:
-            return source == Start_Dependency;
+          case IN_DEPENDENCY:
+            return source == START_DEPENDENCY;
           default:
             return true;
         }
@@ -216,12 +230,16 @@ public class SmoothWeightPartitioner implements Partitioner, DependencyClient {
     }
 
     private void transitionTo(State target) {
-      if (!currentState.isTransitionValid(currentState, target)) {
-        throw new KafkaException(
-            "Invalid transition attempted from state "
-                + currentState.name()
-                + " to state "
-                + target.name());
+      try {
+        if (!currentState.isTransitionValid(currentState, target)) {
+          throw new KafkaException(
+              "Invalid transition attempted from state "
+                  + currentState.name()
+                  + " to state "
+                  + target.name());
+        }
+      } catch (KafkaException e) {
+        // TODO
       }
       currentState = target;
     }
diff --git a/app/src/main/java/org/astraea/performance/Distribution.java b/app/src/main/java/org/astraea/performance/Distribution.java
new file mode 100644
index 00000000..8148aad6
--- /dev/null
+++ b/app/src/main/java/org/astraea/performance/Distribution.java
@@ -0,0 +1,74 @@
+package org.astraea.performance;
+
+import com.beust.jcommander.IStringConverter;
+import com.beust.jcommander.ParameterException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import java.util.stream.IntStream;
+
+/** Randomly generate a long number with respect to some distribution */
+public interface Distribution {
+
+  long get();
+
+  class DistributionConverter implements IStringConverter<Distribution> {
+    @Override
+    public Distribution convert(String rawArgument) {
+      var args = rawArgument.split(":");
+      var name = rawArgument.split(":")[0];
+
+      switch (name) {
+        case "uniform":
+          if (args.length > 1) return uniform(Integer.parseInt(args[1]));
+          return uniform();
+        case "latest":
+          return latest();
+        case "zipfian":
+          if (args.length > 1) return zipfian(Integer.parseInt(args[1]));
+          return zipfian(30);
+        default:
+          throw new ParameterException(
+              "Unknown distribution \"" + name + "\". use \"uniform\", \"latest\", \"zipfian\".");
+      }
+    }
+  }
+
+  /** A distribution for providing a random long number from range [0, 2147483647) */
+  static Distribution uniform() {
+    return uniform(Integer.MAX_VALUE);
+  }
+
+  /** A distribution for providing a random long number from range [0, N) */
+  static Distribution uniform(int N) {
+    var rand = new Random();
+    return () -> (long) (rand.nextInt(N));
+  }
+
+  /** A distribution for providing different value every 2 seconds */
+  static Distribution latest() {
+    return () -> System.currentTimeMillis() / 2000L;
+  }
+
+  /**
+   * Building a zipfian distribution with PDF: 1/k/H_N, where H_N is the Nth harmonic number (= 1/1
+   * + 1/2 + ... + 1/N); k is the key id
+   */
+  static Distribution zipfian(int N) {
+    var rand = new Random();
+    final List<Double> cumulativeDensityTable = new ArrayList<>();
+    var H_N = IntStream.range(1, N + 1).mapToDouble(k -> 1D / k).sum();
+    cumulativeDensityTable.add(1D / H_N);
+    IntStream.range(1, N)
+        .forEach(
+            i ->
+                cumulativeDensityTable.add(cumulativeDensityTable.get(i - 1) + 1D / (i + 1) / H_N));
+    return () -> {
+      final double randNum = rand.nextDouble();
+      for (int i = 0; i < cumulativeDensityTable.size(); ++i) {
+        if (randNum < cumulativeDensityTable.get(i)) return (long) i;
+      }
+      return (long) cumulativeDensityTable.size() - 1L;
+    };
+  }
+}
diff --git a/app/src/main/java/org/astraea/performance/FileWriter.java b/app/src/main/java/org/astraea/performance/FileWriter.java
new file mode 100644
index 00000000..f59c00a8
--- /dev/null
+++ b/app/src/main/java/org/astraea/performance/FileWriter.java
@@ -0,0 +1,117 @@
+package org.astraea.performance;
+
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.stream.IntStream;
+import org.astraea.concurrent.ThreadPool;
+
+public class FileWriter implements ThreadPool.Executor {
+  private final Manager manager;
+  private final Tracker tracker;
+  private BufferedWriter writer;
+  private String CSVName;
+
+  public FileWriter(Manager manager, Tracker tracker) {
+    this.manager = manager;
+    this.tracker = tracker;
+    initFileWriter();
+  }
+
+  @Override
+  public ThreadPool.Executor.State execute() throws InterruptedException {
+    if (logToCSV()) return State.DONE;
+    Thread.sleep(1000);
+    return State.RUNNING;
+  }
+
+  public String CSVName() {
+    return CSVName;
+  }
+
+  private void initFileWriter() {
+    CSVName = "Performance" + new SimpleDateFormat("yyyyMMddHHmmss").format(new Date()) + ".csv";
+    try {
+      writer = new BufferedWriter(new java.io.FileWriter(CSVName));
+      writer.write(
+          "Time \\ Name, Consumed/Produced, Output throughput (MiB/sec), Input throughput (MiB/sec), "
+              + "Publish max latency (ms), Publish min latency (ms), "
+              + "End-to-end max latency (ms), End-to-end min latency (ms)");
+      IntStream.range(0, tracker.producerResult().bytes.size())
+          .forEach(
+              i -> {
+                try {
+                  writer.write(
+                      ",Producer["
+                          + i
+                          + "] average throughput (MB/sec), Producer["
+                          + i
+                          + "] average publish latency (ms)");
+                } catch (IOException ignore) {
+                }
+              });
+      IntStream.range(0, tracker.consumerResult().bytes.size())
+          .forEach(
+              i -> {
+                try {
+                  writer.write(
+                      ",Consumer["
+                          + i
+                          + "] average throughput (MB/sec), Consumer["
+                          + i
+                          + "] average ene-to-end latency (ms)");
+                } catch (IOException ignore) {
+                }
+              });
+      writer.newLine();
+    } catch (IOException ignore) {
+      writer = null;
+    }
+  }
+
+  private boolean logToCSV() {
+    var producerResult = tracker.producerResult();
+    var consumerResult = tracker.consumerResult();
+    if (writer == null || producerResult.completedRecords == 0L) return false;
+    var duration = tracker.duration();
+    var producerPercentage =
+        Math.min(
+            100D,
+            manager.exeTime().percentage(producerResult.completedRecords, duration.toMillis()));
+    var consumerPercentage = consumerResult.completedRecords * 100D / manager.producedRecords();
+    try {
+      writer.write(
+          duration.toHoursPart()
+              + "h"
+              + duration.toMinutesPart()
+              + "m"
+              + duration.toSecondsPart()
+              + "s");
+      writer.write(String.format(",%.2f%% / %.2f%%", consumerPercentage, producerPercentage));
+      writer.write("," + producerResult.averageBytes(duration));
+      writer.write("," + consumerResult.averageBytes(duration));
+      writer.write("," + producerResult.maxLatency + "," + producerResult.minLatency);
+      writer.write("," + consumerResult.maxLatency + "," + consumerResult.minLatency);
+      for (int i = 0; i < producerResult.bytes.size(); ++i) {
+        writer.write("," + Tracker.avg(duration, producerResult.bytes.get(i)));
+        writer.write("," + producerResult.averageLatencies.get(i));
+      }
+      for (int i = 0; i < consumerResult.bytes.size(); ++i) {
+        writer.write("," + Tracker.avg(duration, consumerResult.bytes.get(i)));
+        writer.write("," + consumerResult.averageLatencies.get(i));
+      }
+      writer.newLine();
+    } catch (IOException ignore) {
+    }
+    return manager.producedDone() && manager.consumedDone();
+  }
+
+  @Override
+  public void close() {
+    try {
+      writer.close();
+    } catch (IOException ignore) {
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/performance/Manager.java b/app/src/main/java/org/astraea/performance/Manager.java
index e84b809f..2fb59b53 100644
--- a/app/src/main/java/org/astraea/performance/Manager.java
+++ b/app/src/main/java/org/astraea/performance/Manager.java
@@ -1,8 +1,6 @@
 package org.astraea.performance;
 
-import java.util.List;
-import java.util.Optional;
-import java.util.Random;
+import java.util.*;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
@@ -16,15 +14,14 @@ import org.astraea.utils.DataUnit;
  */
 public class Manager {
   private final ExeTime exeTime;
-  private final boolean fixedSize;
-  private final DataSize dataSize;
   private final CountDownLatch getAssignment;
   private final AtomicInteger producerClosed;
-  private final Random rand = new Random();
 
   private final List<Metrics> producerMetrics, consumerMetrics;
   private final long start = System.currentTimeMillis();
   private final AtomicLong payloadNum = new AtomicLong(0);
+  private final Distribution distribution;
+  private final RandomContent randomContent;
 
   /**
    * Used to manage producing/consuming.
@@ -43,9 +40,7 @@ public class Manager {
    */
   public Manager(
       Performance.Argument argument, List<Metrics> producerMetrics, List<Metrics> consumerMetrics) {
-    this.fixedSize = argument.fixedSize;
-    this.dataSize = argument.recordSize;
-    if (dataSize.greaterThan(DataUnit.Byte.of(Integer.MAX_VALUE)))
+    if (argument.recordSize.greaterThan(DataUnit.Byte.of(Integer.MAX_VALUE)))
       throw new IllegalArgumentException(
           "Record size should be smaller than or equal to 2147483648 (Integer.MAX_VALUE) bytes");
     this.getAssignment = new CountDownLatch(argument.consumers);
@@ -53,6 +48,8 @@ public class Manager {
     this.producerMetrics = producerMetrics;
     this.consumerMetrics = consumerMetrics;
     this.exeTime = argument.exeTime;
+    this.distribution = argument.distribution;
+    this.randomContent = new RandomContent(argument.recordSize, argument.fixedSize);
   }
 
   /**
@@ -66,11 +63,7 @@ public class Manager {
     if (exeTime.percentage(payloadNum.getAndIncrement(), System.currentTimeMillis() - start)
         >= 100D) return Optional.empty();
 
-    byte[] payload =
-        (this.fixedSize)
-            ? new byte[dataSize.measurement(DataUnit.Byte).intValue()]
-            : new byte[rand.nextInt(dataSize.measurement(DataUnit.Byte).intValue()) + 1];
-    rand.nextBytes(payload);
+    var payload = randomContent.getContent();
     return Optional.of(payload);
   }
 
@@ -109,4 +102,36 @@ public class Manager {
   public boolean consumedDone() {
     return producedDone() && consumedRecords() >= producedRecords();
   }
+
+  /** Randomly choose a key according to the distribution. */
+  public Optional<byte[]> getKey() {
+    if (distribution == null) return Optional.empty();
+    return Optional.of((String.valueOf(distribution.get())).getBytes());
+  }
+
+  /** Randomly generate content before {@link #getContent()} is called. */
+  private static class RandomContent {
+    private final Random rand = new Random();
+    private final DataSize dataSize;
+    private final boolean fixedSize;
+    private final byte[] content;
+
+    /**
+     * @param dataSize The size of each random generated content in bytes.
+     * @param fixedSize Determine whether to fix the size of random generated content
+     */
+    public RandomContent(DataSize dataSize, boolean fixedSize) {
+      this.dataSize = dataSize;
+      this.fixedSize = fixedSize;
+      content = new byte[dataSize.measurement(DataUnit.Byte).intValue()];
+    }
+
+    public byte[] getContent() {
+      // Randomly change one position of the content;
+      content[rand.nextInt(dataSize.measurement(DataUnit.Byte).intValue())] =
+          (byte) rand.nextInt(256);
+      if (fixedSize) return Arrays.copyOf(content, content.length);
+      else return Arrays.copyOfRange(content, rand.nextInt(content.length), content.length);
+    }
+  }
 }
diff --git a/app/src/main/java/org/astraea/performance/Performance.java b/app/src/main/java/org/astraea/performance/Performance.java
index 17e7cbe3..5a4cd928 100644
--- a/app/src/main/java/org/astraea/performance/Performance.java
+++ b/app/src/main/java/org/astraea/performance/Performance.java
@@ -5,9 +5,10 @@ import com.beust.jcommander.Parameter;
 import com.beust.jcommander.ParameterException;
 import java.io.IOException;
 import java.time.Duration;
-import java.util.Map;
-import java.util.Set;
+import java.util.*;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
 import java.util.function.BiConsumer;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
@@ -52,14 +53,16 @@ import org.astraea.utils.DataUnit;
  * To avoid records being produced too fast, producer wait for one millisecond after each send.
  */
 public class Performance {
-
+  /** Used in Automation, to achieve the end of one Performance and then start another. */
   public static void main(String[] args)
       throws InterruptedException, IOException, ExecutionException {
     execute(ArgumentUtil.parseArgument(new Argument(), args));
   }
 
-  public static void execute(final Argument param)
+  public static Future<String> execute(final Argument param)
       throws InterruptedException, IOException, ExecutionException {
+    List<Integer> partitions;
+    var future = new CompletableFuture<String>();
     try (var topicAdmin = TopicAdmin.of(param.props())) {
       topicAdmin
           .creator()
@@ -67,6 +70,8 @@ public class Performance {
           .numberOfPartitions(param.partitions)
           .topic(param.topic)
           .create();
+
+      partitions = partition(param, topicAdmin);
     }
 
     var consumerMetrics =
@@ -80,6 +85,8 @@ public class Performance {
 
     var manager = new Manager(param, producerMetrics, consumerMetrics);
     var tracker = new Tracker(producerMetrics, consumerMetrics, manager);
+    Collection<ThreadPool.Executor> fileWriter =
+        (param.createCSV) ? List.of(new FileWriter(manager, tracker)) : List.of();
     var groupId = "groupId-" + System.currentTimeMillis();
     try (ThreadPool threadPool =
         ThreadPool.builder()
@@ -110,11 +117,15 @@ public class Performance {
                                     .build(),
                                 param,
                                 producerMetrics.get(i),
+                                partitions,
                                 manager))
                     .collect(Collectors.toUnmodifiableList()))
             .executor(tracker)
+            .executors(fileWriter)
             .build()) {
       threadPool.waitAll();
+      future.complete(param.topic);
+      return future;
     }
   }
 
@@ -158,13 +169,14 @@ public class Performance {
       Producer<byte[], byte[]> producer,
       Argument param,
       BiConsumer<Long, Long> observer,
+      List<Integer> partitions,
       Manager manager) {
     return new ThreadPool.Executor() {
       @Override
       public State execute() throws InterruptedException {
         // Wait for all consumers get assignment.
         manager.awaitPartitionAssignment();
-
+        var rand = new Random();
         var payload = manager.payload();
         if (payload.isEmpty()) return State.DONE;
 
@@ -172,6 +184,8 @@ public class Performance {
         producer
             .sender()
             .topic(param.topic)
+            .partition(partitions.get(rand.nextInt(partitions.size())))
+            .key(manager.getKey().orElse(null))
             .value(payload.get())
             .timestamp(start)
             .run()
@@ -192,7 +206,22 @@ public class Performance {
     };
   }
 
-  static class Argument extends BasicArgumentWithPropFile {
+  // visible for test
+  static List<Integer> partition(Argument param, TopicAdmin topicAdmin) {
+    if (positiveSpecifyBroker(param)) {
+      return topicAdmin
+          .partitionsOfBrokers(Set.of(param.topic), new HashSet<>(param.specifyBroker))
+          .stream()
+          .map(topicPartition -> topicPartition.partition())
+          .collect(Collectors.toList());
+    } else return List.of(-1);
+  }
+
+  private static boolean positiveSpecifyBroker(Argument param) {
+    return param.specifyBroker.stream().allMatch(broker -> broker >= 0);
+  }
+
+  public static class Argument extends BasicArgumentWithPropFile {
 
     @Parameter(
         names = {"--topic"},
@@ -266,12 +295,31 @@ public class Performance {
       return props;
     }
 
+    @Parameter(
+        names = {"--createCSV"},
+        description = "create the metrics into a csv file if this flag is set")
+    boolean createCSV = false;
+
     @Parameter(
         names = {"--compression"},
         description =
             "String: the compression algorithm used by producer. Available algorithm are none, gzip, snappy, lz4, and zstd",
         converter = CompressionArgument.class)
     CompressionType compression = CompressionType.NONE;
+
+    @Parameter(
+        names = {"--key.distribution"},
+        description =
+            "String: Distribution name. Available distribution names: \"uniform\", \"zipfian\", \"latest\". Default: (No key)",
+        converter = Distribution.DistributionConverter.class)
+    Distribution distribution = null;
+
+    @Parameter(
+        names = {"--specify.broker"},
+        description =
+            "String: Used with SpecifyBrokerPartitioner to specify the brokers that partitioner can send.",
+        validateWith = ArgumentUtil.NotEmptyString.class)
+    List<Integer> specifyBroker = List.of(-1);
   }
 
   static class CompressionArgument implements IStringConverter<CompressionType> {
diff --git a/app/src/main/java/org/astraea/performance/Tracker.java b/app/src/main/java/org/astraea/performance/Tracker.java
index 33380550..fbd25def 100644
--- a/app/src/main/java/org/astraea/performance/Tracker.java
+++ b/app/src/main/java/org/astraea/performance/Tracker.java
@@ -1,9 +1,7 @@
 package org.astraea.performance;
 
 import java.time.Duration;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
+import java.util.*;
 import org.astraea.concurrent.ThreadPool;
 
 /** Print out the given metrics. */
@@ -11,36 +9,41 @@ public class Tracker implements ThreadPool.Executor {
   private final List<Metrics> producerData;
   private final List<Metrics> consumerData;
   private final Manager manager;
+  // Snapshot of metrics
+  private Result producerResult;
+  private Result consumerResult;
   long start = 0L;
 
   public Tracker(List<Metrics> producerData, List<Metrics> consumerData, Manager manager) {
     this.producerData = producerData;
     this.consumerData = consumerData;
+    this.producerResult = result(producerData);
+    this.consumerResult = result(consumerData);
     this.manager = manager;
   }
 
   @Override
   public State execute() throws InterruptedException {
-    if (logProducers() & logConsumers()) return State.DONE;
+    producerResult = result(producerData);
+    consumerResult = result(consumerData);
+    if (logProducers(producerResult) & logConsumers(consumerResult)) return State.DONE;
     // Log after waiting for one second
     Thread.sleep(1000);
     return State.RUNNING;
   }
 
-  private Duration duration() {
+  Duration duration() {
     if (start == 0L) start = System.currentTimeMillis() - Duration.ofSeconds(1).toMillis();
     return Duration.ofMillis(System.currentTimeMillis() - start);
   }
 
-  private static double avg(Duration duration, long value) {
+  static double avg(Duration duration, long value) {
     return duration.toSeconds() <= 0
         ? 0
         : ((double) (value / duration.toSeconds())) / 1024D / 1024D;
   }
 
-  private boolean logProducers() {
-    var result = result(producerData);
-
+  private boolean logProducers(Result result) {
     if (result.completedRecords == 0) return false;
 
     var duration = duration();
@@ -70,10 +73,9 @@ public class Tracker implements ThreadPool.Executor {
     return percentage >= 100D;
   }
 
-  private boolean logConsumers() {
+  private boolean logConsumers(Result result) {
     // there is no consumer, so we just complete this log.
     if (consumerData.isEmpty()) return true;
-    var result = result(consumerData);
     if (result.completedRecords == 0) return false;
     var duration = duration();
 
@@ -94,6 +96,14 @@ public class Tracker implements ThreadPool.Executor {
     return manager.producedDone() && percentage >= 100D;
   }
 
+  public Result producerResult() {
+    return producerResult;
+  }
+
+  public Result consumerResult() {
+    return consumerResult;
+  }
+
   private static Result result(List<Metrics> metrics) {
     var completed = 0;
     var bytes = new ArrayList<Long>();
@@ -115,7 +125,7 @@ public class Tracker implements ThreadPool.Executor {
         max);
   }
 
-  private static class Result {
+  static class Result {
     public final long completedRecords;
     public final List<Long> bytes;
     public final List<Double> averageLatencies;
diff --git a/app/src/main/java/org/astraea/producer/Builder.java b/app/src/main/java/org/astraea/producer/Builder.java
index d7a68918..44b9184c 100644
--- a/app/src/main/java/org/astraea/producer/Builder.java
+++ b/app/src/main/java/org/astraea/producer/Builder.java
@@ -86,7 +86,7 @@ public class Builder<Key, Value> {
 
           @Override
           public Sender<Key, Value> partition(int partition) {
-            this.partition = partition;
+            if (partition >= 0) this.partition = partition;
             return this;
           }
 
diff --git a/app/src/main/java/org/astraea/topic/Builder.java b/app/src/main/java/org/astraea/topic/Builder.java
index 24b754f7..d7f143a4 100644
--- a/app/src/main/java/org/astraea/topic/Builder.java
+++ b/app/src/main/java/org/astraea/topic/Builder.java
@@ -211,6 +211,14 @@ public class Builder {
                   .collect(Collectors.toSet()));
     }
 
+    @Override
+    public List<TopicPartition> partitionsOfBrokers(Set<String> topics, Set<Integer> brokersID) {
+      return replicas(topics).entrySet().stream()
+          .filter(e -> e.getValue().stream().anyMatch(r -> brokersID.contains(r.broker())))
+          .map(Map.Entry::getKey)
+          .collect(Collectors.toList());
+    }
+
     @Override
     public Map<TopicPartition, List<Replica>> replicas(Set<String> topics) {
       var replicaInfos =
diff --git a/app/src/main/java/org/astraea/topic/TopicAdmin.java b/app/src/main/java/org/astraea/topic/TopicAdmin.java
index d2cc1ffc..ae5e49a8 100644
--- a/app/src/main/java/org/astraea/topic/TopicAdmin.java
+++ b/app/src/main/java/org/astraea/topic/TopicAdmin.java
@@ -57,6 +57,13 @@ public interface TopicAdmin extends Closeable {
   /** @return all brokers' ids */
   Set<Integer> brokerIds();
 
+  /**
+   * @param topics topic names
+   * @param brokersID brokers ID
+   * @return the partitions of brokers
+   */
+  List<TopicPartition> partitionsOfBrokers(Set<String> topics, Set<Integer> brokersID);
+
   /**
    * @param brokers a Set containing broker's ID
    * @return all log directory
diff --git a/config/automation.properties b/config/automation.properties
new file mode 100644
index 00000000..6ecd7247
--- /dev/null
+++ b/config/automation.properties
@@ -0,0 +1,24 @@
+#Defaults is to use the default parameters
+
+#You can set as many times as you want it to run.
+--time = Default
+
+#This parameter must be configured so that the performance can be run.
+--bootstrap.servers = Default
+
+#Set it to true to delete the generated topic every time you perform a performance.
+--whetherDeleteTopic = false
+
+#Optional
+--record.size = Default
+--run.until = Default
+--compression = Default
+--consumers = Default
+--fixed.size = Default
+--jmx.servers = Default
+--partitioner = Default
+--partitions = Default
+--producers = Default
+--prop.file =Default
+--replicas = Default
+--topic = Default
diff --git a/docker/start_broker.sh b/docker/start_broker.sh
index a826cdfa..05aa1e59 100755
--- a/docker/start_broker.sh
+++ b/docker/start_broker.sh
@@ -4,15 +4,16 @@
 
 declare -r USER=astraea
 declare -r VERSION=${REVISION:-${VERSION:-2.8.1}}
-declare -r REPO=${REPO:-astraea/broker}
+declare -r REPO=${REPO:-ghcr.io/skiptests/astraea/broker}
 declare -r IMAGE_NAME="$REPO:$VERSION"
+declare -r BUILD=${BUILD:-false}
 declare -r RUN=${RUN:-true}
 declare -r DOCKER_FOLDER=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" &>/dev/null && pwd)
 declare -r DOCKERFILE=$DOCKER_FOLDER/broker.dockerfile
 declare -r DATA_FOLDER_IN_CONTAINER_PREFIX="/tmp/log-folder"
 declare -r EXPORTER_VERSION="0.16.1"
-declare -r EXPORTER_PORT="$(($(($RANDOM % 10000)) + 10000))"
-declare -r BROKER_PORT="$(($(($RANDOM % 10000)) + 10000))"
+declare -r EXPORTER_PORT=${EXPORTER_PORT:-$(($(($RANDOM % 10000)) + 10000))}
+declare -r BROKER_PORT=${BROKER_PORT:-$(($(($RANDOM % 10000)) + 10000))}
 declare -r CONTAINER_NAME="broker-$BROKER_PORT"
 declare -r BROKER_JMX_PORT="$(($(($RANDOM % 10000)) + 10000))"
 declare -r ADMIN_NAME="admin"
@@ -45,7 +46,8 @@ function showHelp() {
   echo "    HEAP_OPTS=\"-Xmx2G -Xms2G\"                set broker JVM memory"
   echo "    REVISION=trunk                           set revision of kafka source code to build container"
   echo "    VERSION=2.8.1                            set version of kafka distribution"
-  echo "    RUN=false                                set false if you want to build image only"
+  echo "    BUILD=false                              set true if you want to build image locally"
+  echo "    RUN=false                                set false if you want to build/pull image only"
   echo "    DATA_FOLDERS=/tmp/folder1,/tmp/folder2   set host folders used by broker"
 }
 
@@ -147,7 +149,22 @@ function generateDockerfile() {
 
 function buildImageIfNeed() {
   if [[ "$(docker images -q $IMAGE_NAME 2>/dev/null)" == "" ]]; then
-    docker build --no-cache -t "$IMAGE_NAME" -f "$DOCKERFILE" "$DOCKER_FOLDER"
+    local needToBuild="true"
+    if [[ "$BUILD" == "false" ]]; then
+      docker pull $IMAGE_NAME 2>/dev/null
+      if [[ "$?" == "0" ]]; then
+        needToBuild="false"
+      else
+        echo "Can't find $IMAGE_NAME from repo. Will build $IMAGE_NAME on the local"
+      fi
+    fi
+    if [[ "$needToBuild" == "true" ]]; then
+      generateDockerfile
+      docker build --no-cache -t "$IMAGE_NAME" -f "$DOCKERFILE" "$DOCKER_FOLDER"
+      if [[ "$?" != "0" ]]; then
+        exit 2
+      fi
+    fi
   fi
 }
 
@@ -201,7 +218,7 @@ function generateMountCommand() {
     declare -i count=0
 
     for folder in "${folders[@]}"; do
-      mount="-v $folder:$DATA_FOLDER_IN_CONTAINER_PREFIX-$count"
+      mount="$mount -v $folder:$DATA_FOLDER_IN_CONTAINER_PREFIX-$count"
       count=$((count + 1))
     done
   fi
diff --git a/docker/start_astraea.sh b/docker/start_kafka_tool.sh
similarity index 50%
rename from docker/start_astraea.sh
rename to docker/start_kafka_tool.sh
index 8d387d2c..d2afe920 100755
--- a/docker/start_astraea.sh
+++ b/docker/start_kafka_tool.sh
@@ -2,12 +2,24 @@
 
 # ===============================[global variables]===============================
 
-IMAGE_NAME="astraea/astraea:latest"
-DOCKER_FOLDER=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" &>/dev/null && pwd)
-DOCKER_FILE=$DOCKER_FOLDER/astraea.dockerfile
+declare -r VERSION=${REVISION:-${VERSION:-main}}
+declare -r REPO=${REPO:-ghcr.io/skiptests/astraea/kafka-tool}
+declare -r IMAGE_NAME="$REPO:$VERSION"
+declare -r DOCKER_FOLDER=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" &>/dev/null && pwd)
+declare -r DOCKERFILE=$DOCKER_FOLDER/kafka_tool.dockerfile
+declare -r BUILD=${BUILD:-false}
+declare -r RUN=${RUN:-true}
 
 # ===================================[functions]===================================
 
+function showHelp() {
+  echo "Usage: [ENV] start_kafka_tool.sh"
+  echo "ENV: "
+  echo "    REPO=astraea/kafka-tool    set the docker repo"
+  echo "    BUILD=false                set true if you want to build image locally"
+  echo "    RUN=false                  set false if you want to build/pull image only"
+}
+
 function checkDocker() {
   if [[ "$(which docker)" == "" ]]; then
     echo "you have to install docker"
@@ -31,14 +43,30 @@ RUN git clone https://github.com/skiptests/astraea
 
 # pre-build project to collect all dependencies
 WORKDIR /tmp/astraea
+RUN git checkout $VERSION
 RUN ./gradlew clean build -x test --no-daemon
 RUN cp \$(find ./app/build/libs/ -maxdepth 1 -type f -name app-*-all.jar) /tmp/app.jar
-" >"$DOCKER_FILE"
+" >"$DOCKERFILE"
 }
 
 function buildImageIfNeed() {
   if [[ "$(docker images -q $IMAGE_NAME 2>/dev/null)" == "" ]]; then
-    docker build --no-cache -t "$IMAGE_NAME" -f "$DOCKER_FILE" "$DOCKER_FOLDER"
+    local needToBuild="true"
+    if [[ "$BUILD" == "false" ]]; then
+      docker pull $IMAGE_NAME 2>/dev/null
+      if [[ "$?" == "0" ]]; then
+        needToBuild="false"
+      else
+        echo "Can't find $IMAGE_NAME from repo. Will build $IMAGE_NAME on the local"
+      fi
+    fi
+    if [[ "$needToBuild" == "true" ]]; then
+      generateDockerfile
+      docker build --no-cache -t "$IMAGE_NAME" -f "$DOCKERFILE" "$DOCKER_FOLDER"
+      if [[ "$?" != "0" ]]; then
+        exit 2
+      fi
+    fi
   fi
 }
 
@@ -54,8 +82,15 @@ function runContainer() {
 checkDocker
 generateDockerfile
 buildImageIfNeed
+
+if [[ "$RUN" != "true" ]]; then
+  echo "docker image: $IMAGE_NAME is created"
+  exit 0
+fi
+
 if [[ -n "$1" ]]; then
   runContainer "$*"
 else
+  showHelp
   runContainer "help"
 fi
diff --git a/docker/start_spark.sh b/docker/start_spark.sh
index 402ee4bc..7ae25739 100755
--- a/docker/start_spark.sh
+++ b/docker/start_spark.sh
@@ -4,11 +4,12 @@
 
 declare -r USER=astraea
 declare -r VERSION=${REVISION:-${VERSION:-3.1.2}}
-declare -r REPO=${REPO:-astraea/spark}
+declare -r REPO=${REPO:-ghcr.io/skiptests/astraea/spark}
 declare -r IMAGE_NAME="$REPO:$VERSION"
+declare -r BUILD=${BUILD:-false}
 declare -r RUN=${RUN:-true}
-declare -r SPARK_PORT="$(($(($RANDOM % 10000)) + 10000))"
-declare -r SPARK_UI_PORT="$(($(($RANDOM % 10000)) + 10000))"
+declare -r SPARK_PORT=${SPARK_PORT:-$(($(($RANDOM % 10000)) + 10000))}
+declare -r SPARK_UI_PORT=${SPARK_UI_PORT:-$(($(($RANDOM % 10000)) + 10000))}
 declare -r IVY_VERSION=2.5.0
 declare -r DELTA_VERSION=${DELTA_VERSION:-1.0.0}
 declare -r PYTHON_KAFKA_VERSION=${PYTHON_KAFKA_VERSION:-1.7.0}
@@ -27,12 +28,12 @@ function showHelp() {
   echo "Optional Arguments: "
   echo "    master-url=spar://node00:1111    start a spark worker. Or start a spark master if master-url is not defined"
   echo "ENV: "
-  echo "    REPO=astraea/spark               set the docker repo"
   echo "    VERSION=3.1.2                    set version of spark distribution"
   echo "    DELTA_VERSION=1.0.0              set version of delta distribution"
-  echo "    PYTHON_KAFKA_VERSION=1.7.0           set version of confluent kafka distribution"
-  echo "    RUN=false                        set false if you want to build image only"
-  echo "    PYTHON_DEPS=delta-spark=1.0.0    those dependencies will be pre-installed in the docker image"
+  echo "    PYTHON_KAFKA_VERSION=1.7.0       set version of confluent kafka distribution"
+  echo "    BUILD=false                      set true if you want to build image locally"
+  echo "    RUN=false                        set false if you want to build/pull image only"
+  echo "    PYTHON_DEPS=delta-spark=1.0.0    set the python dependencies which are pre-installed in the docker image"
 }
 
 function checkDocker() {
@@ -146,7 +147,22 @@ function generateDockerfile() {
 
 function buildImageIfNeed() {
   if [[ "$(docker images -q $IMAGE_NAME 2>/dev/null)" == "" ]]; then
-    docker build --no-cache -t "$IMAGE_NAME" -f "$DOCKERFILE" "$DOCKER_FOLDER"
+    local needToBuild="true"
+    if [[ "$BUILD" == "false" ]]; then
+      docker pull $IMAGE_NAME 2>/dev/null
+      if [[ "$?" == "0" ]]; then
+        needToBuild="false"
+      else
+        echo "Can't find $IMAGE_NAME from repo. Will build $IMAGE_NAME on the local"
+      fi
+    fi
+    if [[ "$needToBuild" == "true" ]]; then
+      generateDockerfile
+      docker build --no-cache -t "$IMAGE_NAME" -f "$DOCKERFILE" "$DOCKER_FOLDER"
+      if [[ "$?" != "0" ]]; then
+        exit 2
+      fi
+    fi
   fi
 }
 
diff --git a/docker/start_zookeeper.sh b/docker/start_zookeeper.sh
index 84f487f1..37271c57 100755
--- a/docker/start_zookeeper.sh
+++ b/docker/start_zookeeper.sh
@@ -4,14 +4,15 @@
 
 declare -r USER=astraea
 declare -r VERSION=${VERSION:-3.7.0}
-declare -r REPO=${REPO:-astraea/zookeeper}
+declare -r REPO=${REPO:-ghcr.io/skiptests/astraea/zookeeper}
 declare -r IMAGE_NAME="$REPO:$VERSION"
-declare -r PORT="$(($(($RANDOM % 10000)) + 10000))"
+declare -r ZOOKEEPER_PORT=${ZOOKEEPER_PORT:-$(($(($RANDOM % 10000)) + 10000))}
+declare -r BUILD=${BUILD:-false}
 declare -r RUN=${RUN:-true}
 declare -r DOCKER_FOLDER=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" &>/dev/null && pwd)
 declare -r DOCKERFILE=$DOCKER_FOLDER/zookeeper.dockerfile
 declare -r DATA_FOLDER_IN_CONTAINER="/tmp/zookeeper-dir"
-declare -r CONTAINER_NAME="zookeeper-$PORT"
+declare -r CONTAINER_NAME="zookeeper-$ZOOKEEPER_PORT"
 declare -r ADDRESS=$([[ "$(which ipconfig)" != "" ]] && ipconfig getifaddr en0 || hostname -i)
 
 # ===================================[functions]===================================
@@ -21,7 +22,8 @@ function showHelp() {
   echo "ENV: "
   echo "    REPO=astraea/zk            set the docker repo"
   echo "    VERSION=3.7.0              set version of zookeeper distribution"
-  echo "    RUN=false                  set false if you want to build image only"
+  echo "    BUILD=false                set true if you want to build image locally"
+  echo "    RUN=false                  set false if you want to build/pull image only"
   echo "    DATA_FOLDER=/tmp/folder1   set host folders used by zookeeper"
 }
 
@@ -68,7 +70,22 @@ RUN echo "clientPort=2181" >> ./conf/zoo.cfg
 
 function buildImageIfNeed() {
   if [[ "$(docker images -q $IMAGE_NAME 2>/dev/null)" == "" ]]; then
-    docker build --no-cache -t "$IMAGE_NAME" -f "$DOCKERFILE" "$DOCKER_FOLDER"
+    local needToBuild="true"
+    if [[ "$BUILD" == "false" ]]; then
+      docker pull $IMAGE_NAME 2>/dev/null
+      if [[ "$?" == "0" ]]; then
+        needToBuild="false"
+      else
+        echo "Can't find $IMAGE_NAME from repo. Will build $IMAGE_NAME on the local"
+      fi
+    fi
+    if [[ "$needToBuild" == "true" ]]; then
+      generateDockerfile
+      docker build --no-cache -t "$IMAGE_NAME" -f "$DOCKERFILE" "$DOCKER_FOLDER"
+      if [[ "$?" != "0" ]]; then
+        exit 2
+      fi
+    fi
   fi
 }
 
@@ -80,7 +97,6 @@ if [[ "$1" == "help" ]]; then
 fi
 
 checkDocker
-generateDockerfile
 buildImageIfNeed
 
 if [[ "$RUN" != "true" ]]; then
@@ -95,15 +111,15 @@ if [[ -n "$DATA_FOLDER" ]]; then
   echo "mount $DATA_FOLDER to container"
   docker run -d \
     --name $CONTAINER_NAME \
-    -p $PORT:2181 \
+    -p $ZOOKEEPER_PORT:2181 \
     -v "$DATA_FOLDER":$DATA_FOLDER_IN_CONTAINER \
     $IMAGE_NAME ./bin/zkServer.sh start-foreground
 else
   docker run -d \
-    -p $PORT:2181 \
+    -p $ZOOKEEPER_PORT:2181 \
     $IMAGE_NAME ./bin/zkServer.sh start-foreground
 fi
 
 echo "================================================="
-echo "run $DOCKER_FOLDER/start_broker.sh zookeeper.connect=$ADDRESS:$PORT to join kafka broker"
+echo "run $DOCKER_FOLDER/start_broker.sh zookeeper.connect=$ADDRESS:$ZOOKEEPER_PORT to join kafka broker"
 echo "================================================="


Test patch:
diff --git a/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/DependencyPartitionTest.java b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/DependencyPartitionTest.java
index ea2c5fb2..1e8ac86d 100644
--- a/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/DependencyPartitionTest.java
+++ b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/DependencyPartitionTest.java
@@ -46,7 +46,8 @@ public class DependencyPartitionTest extends RequireBrokerCluster {
         ps.entrySet().stream()
             .collect(Collectors.toMap(e -> e.getKey().toString(), Map.Entry::getValue));
     try (var producer = Producer.builder().configs(props).build()) {
-      var dependencyClient = SmoothWeightPartitioner.beginDependency(producer.kafkaProducer());
+      var dependencyClient = SmoothWeightPartitioner.dependencyClient(producer.kafkaProducer());
+      dependencyClient.startDependency();
       var targetPartition = 0;
       var i = 0;
       while (i < 20) {
@@ -86,4 +87,18 @@ public class DependencyPartitionTest extends RequireBrokerCluster {
       Assertions.assertArrayEquals(header.value(), actualHeader.value());
     }
   }
+
+  @Test
+  void testDependencyStateChange() {
+    var props =
+        initProConfig().entrySet().stream()
+            .collect(Collectors.toMap(e -> e.getKey().toString(), Map.Entry::getValue));
+    var producer1 = Producer.builder().configs(props).build();
+    var producer2 = Producer.builder().configs(props).build();
+    var client1 = SmoothWeightPartitioner.dependencyClient(producer1.kafkaProducer());
+    var client2 = SmoothWeightPartitioner.dependencyClient(producer2.kafkaProducer());
+    client1.startDependency();
+    Assertions.assertEquals(client1.state(), "START_DEPENDENCY");
+    Assertions.assertEquals(client2.state(), "UNINITIALIZED");
+  }
 }
diff --git a/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClientTest.java b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClientTest.java
index 85c8692c..8fc7e8b2 100644
--- a/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClientTest.java
+++ b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClientTest.java
@@ -1,31 +1,31 @@
 package org.astraea.partitioner.nodeLoadMetric;
 
+import static org.mockito.Mockito.when;
+
 import java.io.IOException;
-import java.util.Arrays;
+import java.lang.reflect.Field;
+import java.net.UnknownHostException;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import org.apache.kafka.clients.producer.ProducerConfig;
-import org.apache.kafka.common.serialization.ByteArraySerializer;
-import org.astraea.partitioner.smoothPartitioner.SmoothWeightPartitioner;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.Node;
 import org.astraea.service.RequireBrokerCluster;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestInstance;
+import org.mockito.Mockito;
 
 @TestInstance(TestInstance.Lifecycle.PER_CLASS)
 public class NodeLoadClientTest extends RequireBrokerCluster {
-  private final String brokerList = bootstrapServers();
   private NodeLoadClient nodeLoadClient;
-  private String jmxAddress;
-  private Map props;
 
   @BeforeAll
   void setUp() throws IOException {
-    props = initProConfig();
     var map = new HashMap<String, Integer>();
     map.put(jmxServiceURL().getHost(), jmxServiceURL().getPort());
     nodeLoadClient = new NodeLoadClient(map);
@@ -36,25 +36,27 @@ public class NodeLoadClientTest extends RequireBrokerCluster {
     nodeLoadClient.close();
   }
 
-  Properties initProConfig() {
-    Properties props = new Properties();
-    jmxAddress = jmxServiceURL().getHost() + ":" + jmxServiceURL().getPort();
-    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
-    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
-    props.put(ProducerConfig.CLIENT_ID_CONFIG, "id1");
-    props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, SmoothWeightPartitioner.class.getName());
-    props.put("jmx_servers", jmxAddress);
-    return props;
+  private Field field(Object object, String fieldName) {
+    Field field = null;
+    try {
+      field = object.getClass().getDeclaredField(fieldName);
+      field.setAccessible(true);
+    } catch (NoSuchFieldException e) {
+      e.printStackTrace();
+    }
+    return field;
   }
 
   @Test
-  void testAvgBrokersMsgPerSec() {
-    var testMap = new HashMap<Integer, List<Double>>();
-    testMap.put(0, Arrays.asList(20.0, 30.0, 40.0));
-    testMap.put(1, Arrays.asList(30.0, 20.0, 30.0));
-    testMap.put(2, Arrays.asList(40.0, 40.0, 20.0));
-    Assertions.assertEquals(
-        nodeLoadClient.avgBrokersMsgPerSec(testMap), Arrays.asList(30.0, 30.0, 30.0));
+  void testBrokerLoad() {
+    List<NodeLoadClient.Broker> brokers = new ArrayList<>();
+    brokers.add(new NodeLoadClient.Broker(0, "0.0.0.0", 111));
+    brokers.add(new NodeLoadClient.Broker(1, "0.0.0.0", 222));
+    brokers.add(new NodeLoadClient.Broker(2, "0.0.0.0", 333));
+    setBrokers(brokers);
+    Assertions.assertEquals(nodeLoadClient.brokerLoad(0.37, 1.0 / 3), 6);
+    Assertions.assertEquals(nodeLoadClient.brokerLoad(0.01, 1.0 / 3), 0);
+    Assertions.assertEquals(nodeLoadClient.brokerLoad(0.8, 1.0 / 3), 10);
   }
 
   @Test
@@ -65,7 +67,73 @@ public class NodeLoadClientTest extends RequireBrokerCluster {
     testMap.put(2, 25.0);
     testMap.put(3, 20.0);
     testMap.put(4, 20.0);
-    Assertions.assertEquals(
-        nodeLoadClient.standardDeviationImperative(testMap, 20), 3.1622776601683795);
+
+    List<NodeLoadClient.Broker> brokers = new ArrayList<>();
+    brokers.add(0, setSituationNormalized(0, "0.0.0.0", 111, 15.0));
+    brokers.add(1, setSituationNormalized(0, "0.0.0.0", 222, 20.0));
+    brokers.add(2, setSituationNormalized(0, "0.0.0.0", 333, 25.0));
+    brokers.add(3, setSituationNormalized(0, "0.0.0.0", 444, 20.0));
+    brokers.add(4, setSituationNormalized(0, "0.0.0.0", 555, 20.0));
+    setBrokers(brokers);
+
+    Assertions.assertEquals(nodeLoadClient.standardDeviationImperative(20), 3.1622776601683795);
+  }
+
+  @Test
+  void testLoadSituation() throws UnknownHostException {
+    var bootstrapServers = List.of(bootstrapServers().split(","));
+    List<Node> nodes = new ArrayList<>();
+    AtomicInteger count = new AtomicInteger(0);
+    bootstrapServers.forEach(
+        str -> {
+          var hostPort = str.split(":");
+          nodes.add(new Node(count.get(), hostPort[0], Integer.parseInt(hostPort[1])));
+          count.getAndIncrement();
+        });
+
+    Cluster cluster = Mockito.mock(Cluster.class);
+    when(cluster.nodes()).thenReturn(nodes);
+    var load = nodeLoadClient.loadSituation(cluster);
+    Assertions.assertEquals(load.get(0), 5);
+    Assertions.assertEquals(load.get(1), 5);
+    Assertions.assertEquals(load.get(2), 5);
+    load = nodeLoadClient.loadSituation(cluster);
+    Assertions.assertEquals(load.get(0), 5);
+    Assertions.assertEquals(load.get(1), 5);
+    Assertions.assertEquals(load.get(2), 5);
+    sleep(1);
+    load = nodeLoadClient.loadSituation(cluster);
+    Assertions.assertEquals(load.get(0), 10);
+    Assertions.assertEquals(load.get(1), 10);
+    Assertions.assertEquals(load.get(2), 10);
+  }
+
+  private NodeLoadClient.Broker setSituationNormalized(
+      int brokerID, String host, int port, double situation) {
+    var broker = new NodeLoadClient.Broker(brokerID, host, port);
+    var brokerSituation = field(broker, "brokerSituationNormalized");
+    try {
+      brokerSituation.set(broker, situation);
+    } catch (IllegalAccessException e) {
+      e.printStackTrace();
+    }
+    return broker;
+  }
+
+  private void setBrokers(List<NodeLoadClient.Broker> brokers) {
+    var brokersField = field(nodeLoadClient, "brokers");
+    try {
+      brokersField.set(nodeLoadClient, brokers);
+    } catch (IllegalAccessException e) {
+      e.printStackTrace();
+    }
+  }
+
+  private static void sleep(int seconds) {
+    try {
+      TimeUnit.SECONDS.sleep(seconds);
+    } catch (InterruptedException e) {
+      throw new RuntimeException(e);
+    }
   }
 }
diff --git a/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/PartitionerTest.java b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/PartitionerTest.java
index 3869d56c..6bbbca23 100644
--- a/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/PartitionerTest.java
+++ b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/PartitionerTest.java
@@ -1,7 +1,11 @@
 package org.astraea.partitioner.nodeLoadMetric;
 
+import static org.astraea.Utils.requireField;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.Mockito.when;
 
+import java.lang.reflect.Field;
 import java.time.Duration;
 import java.util.HashMap;
 import java.util.List;
@@ -9,9 +13,12 @@ import java.util.Map;
 import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.astraea.concurrent.ThreadPool;
 import org.astraea.consumer.Consumer;
 import org.astraea.consumer.Deserializer;
 import org.astraea.consumer.Header;
@@ -22,11 +29,13 @@ import org.astraea.service.RequireBrokerCluster;
 import org.astraea.topic.TopicAdmin;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+import org.mockito.Mockito;
 
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
 public class PartitionerTest extends RequireBrokerCluster {
   private final String brokerList = bootstrapServers();
   TopicAdmin admin = TopicAdmin.of(bootstrapServers());
-  private final String topicName = "address";
 
   private Properties initProConfig() {
     Properties props = new Properties();
@@ -40,8 +49,31 @@ public class PartitionerTest extends RequireBrokerCluster {
     return props;
   }
 
+  @Test
+  void testSetBrokerHashMap() {
+    var nodeLoadClient = Mockito.mock(NodeLoadClient.class);
+    when(nodeLoadClient.thoughPutComparison(anyInt())).thenReturn(1.0);
+    var poissonMap = new HashMap<Integer, Double>();
+    poissonMap.put(0, 0.5);
+    poissonMap.put(1, 0.8);
+    poissonMap.put(2, 0.3);
+
+    var smoothWeightPartitioner = new SmoothWeightPartitioner();
+    setNodeLoadClient(nodeLoadClient, smoothWeightPartitioner);
+    smoothWeightPartitioner.brokersWeight(poissonMap);
+
+    var brokerWeight = (Map<Integer, int[]>) requireField(smoothWeightPartitioner, "brokersWeight");
+    assertEquals(brokerWeight.get(0)[0], 10);
+    assertEquals(brokerWeight.get(1)[0], 3);
+
+    brokerWeight.put(0, new int[] {0, 8});
+    smoothWeightPartitioner.brokersWeight(poissonMap);
+    assertEquals(brokerWeight.get(0)[1], 8);
+  }
+
   @Test
   void testPartitioner() {
+    var topicName = "address";
     admin.creator().topic(topicName).numberOfPartitions(10).create();
     var key = "tainan";
     var timestamp = System.currentTimeMillis() + 10;
@@ -54,7 +86,7 @@ public class PartitionerTest extends RequireBrokerCluster {
                     .collect(Collectors.toMap(e -> e.getKey().toString(), Map.Entry::getValue)))
             .build()) {
       var i = 0;
-      while (i < 20) {
+      while (i < 300) {
         var metadata =
             producer
                 .sender()
@@ -72,7 +104,7 @@ public class PartitionerTest extends RequireBrokerCluster {
     } catch (ExecutionException | InterruptedException e) {
       e.printStackTrace();
     }
-
+    sleep(1);
     try (var consumer =
         Consumer.builder()
             .brokers(bootstrapServers())
@@ -80,8 +112,12 @@ public class PartitionerTest extends RequireBrokerCluster {
             .topics(Set.of(topicName))
             .keyDeserializer(Deserializer.STRING)
             .build()) {
-      var records = consumer.poll(Duration.ofSeconds(10));
-      assertEquals(20, records.size());
+      var records = consumer.poll(Duration.ofSeconds(20));
+      var recordsCount = records.size();
+      while (recordsCount < 300) {
+        recordsCount += consumer.poll(Duration.ofSeconds(20)).size();
+      }
+      assertEquals(300, recordsCount);
       var record = records.iterator().next();
       assertEquals(topicName, record.topic());
       assertEquals("tainan", record.key());
@@ -93,21 +129,117 @@ public class PartitionerTest extends RequireBrokerCluster {
   }
 
   @Test
-  void testSetBrokerHashMap() {
-    var poissonMap = new HashMap<Integer, Double>();
-    poissonMap.put(0, 0.5);
-    poissonMap.put(1, 0.8);
-    poissonMap.put(2, 0.3);
+  void testMultipleProducer() {
+    var topicName = "addr";
+    admin.creator().topic(topicName).numberOfPartitions(10).create();
+    var key = "tainan";
+    var timestamp = System.currentTimeMillis() + 10;
+    var header = Header.of("a", "b".getBytes());
+    try (ThreadPool threadPool =
+        ThreadPool.builder()
+            .executors(
+                IntStream.range(0, 10)
+                    .mapToObj(
+                        i ->
+                            producerExecutor(
+                                Producer.builder()
+                                    .keySerializer(Serializer.STRING)
+                                    .configs(
+                                        initProConfig().entrySet().stream()
+                                            .collect(
+                                                Collectors.toMap(
+                                                    e -> e.getKey().toString(),
+                                                    Map.Entry::getValue)))
+                                    .build(),
+                                topicName,
+                                key,
+                                header,
+                                timestamp))
+                    .collect(Collectors.toUnmodifiableList()))
+            .build()) {
+      threadPool.waitAll();
+    }
+    try (var consumer =
+        Consumer.builder()
+            .brokers(bootstrapServers())
+            .fromBeginning()
+            .topics(Set.of(topicName))
+            .keyDeserializer(Deserializer.STRING)
+            .build()) {
+      var records = consumer.poll(Duration.ofSeconds(20));
+      var recordsCount = records.size();
+      while (recordsCount < 1000) {
+        recordsCount += consumer.poll(Duration.ofSeconds(20)).size();
+      }
+      assertEquals(1000, recordsCount);
+      var record = records.iterator().next();
+      assertEquals(topicName, record.topic());
+      assertEquals("tainan", record.key());
+      assertEquals(1, record.headers().size());
+      var actualHeader = record.headers().iterator().next();
+      assertEquals(header.key(), actualHeader.key());
+      Assertions.assertArrayEquals(header.value(), actualHeader.value());
+    }
+  }
 
-    var smoothWeightPartitioner = new SmoothWeightPartitioner();
+  private ThreadPool.Executor producerExecutor(
+      Producer<String, byte[]> producer, String topic, String key, Header header, long timeStamp) {
+    return new ThreadPool.Executor() {
+      int i = 0;
+
+      @Override
+      public State execute() throws InterruptedException {
+        if (i > 99) return State.DONE;
+        try {
+          producer
+              .sender()
+              .topic(topic)
+              .key(key)
+              .timestamp(timeStamp)
+              .headers(List.of(header))
+              .run()
+              .toCompletableFuture()
+              .get();
+        } catch (ExecutionException e) {
+          e.printStackTrace();
+        }
+        i++;
+        return State.RUNNING;
+      }
+
+      @Override
+      public void close() {
+        producer.close();
+      }
+    };
+  }
 
-    smoothWeightPartitioner.brokerHashMap(poissonMap);
+  private void setNodeLoadClient(
+      NodeLoadClient nodeLoadClient, SmoothWeightPartitioner partitioner) {
+    var target = field(partitioner, "nodeLoadClient");
+    try {
+      target.set(partitioner, nodeLoadClient);
+    } catch (IllegalAccessException e) {
+      e.printStackTrace();
+    }
+  }
 
-    assertEquals(smoothWeightPartitioner.brokerHashMap().get(0)[0], 10);
-    assertEquals(smoothWeightPartitioner.brokerHashMap().get(1)[0], 3);
+  private Field field(Object object, String fieldName) {
+    Field field = null;
+    try {
+      field = object.getClass().getDeclaredField(fieldName);
+      field.setAccessible(true);
+    } catch (NoSuchFieldException e) {
+      e.printStackTrace();
+    }
+    return field;
+  }
 
-    smoothWeightPartitioner.brokerHashMapValue(0, 8);
-    smoothWeightPartitioner.brokerHashMap(poissonMap);
-    assertEquals(smoothWeightPartitioner.brokerHashMap().get(0)[1], 8);
+  private static void sleep(int seconds) {
+    try {
+      TimeUnit.SECONDS.sleep(seconds);
+    } catch (InterruptedException e) {
+      throw new RuntimeException(e);
+    }
   }
 }
diff --git a/app/src/test/java/org/astraea/performance/DistributionTest.java b/app/src/test/java/org/astraea/performance/DistributionTest.java
new file mode 100644
index 00000000..2cb38401
--- /dev/null
+++ b/app/src/test/java/org/astraea/performance/DistributionTest.java
@@ -0,0 +1,23 @@
+package org.astraea.performance;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class DistributionTest {
+  @Test
+  void testLatest() throws InterruptedException {
+    var distribution = Distribution.latest();
+    Assertions.assertEquals(distribution.get(), distribution.get());
+
+    long first = distribution.get();
+    Thread.sleep(2000);
+
+    Assertions.assertNotEquals(first, distribution.get());
+  }
+
+  @Test
+  void testZipfian() {
+    var distribution = Distribution.zipfian(5);
+    Assertions.assertTrue(distribution.get() < 5);
+  }
+}
diff --git a/app/src/test/java/org/astraea/performance/ManagerTest.java b/app/src/test/java/org/astraea/performance/ManagerTest.java
index 37b42817..57f72668 100644
--- a/app/src/test/java/org/astraea/performance/ManagerTest.java
+++ b/app/src/test/java/org/astraea/performance/ManagerTest.java
@@ -53,4 +53,25 @@ public class ManagerTest {
     // Assertion failed with probability < 1/102400
     Assertions.assertFalse(same);
   }
+
+  @Test
+  void testGetKey() {
+    var argument = new Performance.Argument();
+
+    argument.distribution = Distribution.uniform();
+    var manager = new Manager(argument, List.of(), List.of());
+    Assertions.assertTrue(manager.getKey().isPresent());
+
+    argument.distribution = Distribution.zipfian(10);
+    manager = new Manager(argument, List.of(), List.of());
+    Assertions.assertTrue(manager.getKey().isPresent());
+
+    argument.distribution = Distribution.latest();
+    manager = new Manager(argument, List.of(), List.of());
+    Assertions.assertTrue(manager.getKey().isPresent());
+
+    argument.distribution = null;
+    manager = new Manager(argument, List.of(), List.of());
+    Assertions.assertTrue(manager.getKey().isEmpty());
+  }
 }
diff --git a/app/src/test/java/org/astraea/performance/PerformanceTest.java b/app/src/test/java/org/astraea/performance/PerformanceTest.java
index a2e50993..852ab5ae 100644
--- a/app/src/test/java/org/astraea/performance/PerformanceTest.java
+++ b/app/src/test/java/org/astraea/performance/PerformanceTest.java
@@ -1,22 +1,114 @@
 package org.astraea.performance;
 
+import static org.astraea.performance.Performance.partition;
+
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
+import org.apache.kafka.common.TopicPartition;
 import org.astraea.Utils;
 import org.astraea.concurrent.ThreadPool;
 import org.astraea.consumer.Consumer;
 import org.astraea.producer.Producer;
 import org.astraea.service.RequireBrokerCluster;
+import org.astraea.topic.TopicAdmin;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 public class PerformanceTest extends RequireBrokerCluster {
 
+  @Test
+  void testSpecifyBrokerProducerExecutor() {
+    var admin = TopicAdmin.of(bootstrapServers());
+    var topicName = "testConsumerExecutor-" + System.currentTimeMillis();
+    admin.creator().topic(topicName).numberOfPartitions(10).create();
+    Utils.waitFor(() -> admin.publicTopicNames().contains(topicName));
+
+    var metrics = new Metrics();
+    var param = new Performance.Argument();
+    param.brokers = bootstrapServers();
+    param.topic = topicName;
+    param.fixedSize = true;
+    param.exeTime = ExeTime.of("100records");
+    param.specifyBroker = List.of(0);
+    param.consumers = 0;
+    param.partitions = 10;
+    try (ThreadPool.Executor executor =
+        Performance.producerExecutor(
+            Producer.builder().brokers(bootstrapServers()).build(),
+            param,
+            metrics,
+            partition(param, admin),
+            new Manager(param, List.of(), List.of()))) {
+      ThreadPool threadPool = ThreadPool.builder().executor(executor).build();
+      threadPool.waitAll();
+      threadPool.close();
+
+      Utils.waitFor(() -> metrics.num() == 100);
+      var offsets = admin.offsets(Set.of(topicName));
+      var partitions =
+          offsets.entrySet().stream()
+              .filter(entry -> entry.getValue().latest() > 0)
+              .map(entry -> entry.getKey().partition())
+              .collect(Collectors.toList());
+      var partitionsOfBrokers =
+          admin.partitionsOfBrokers(Set.of(topicName), Set.of(0)).stream()
+              .map(TopicPartition::partition)
+              .collect(Collectors.toSet());
+      partitions.forEach(
+          partition -> Assertions.assertTrue(partitionsOfBrokers.contains(partition)));
+    }
+  }
+
+  @Test
+  void testMultipleSpecifyBrokersProducerExecutor() {
+    var admin = TopicAdmin.of(bootstrapServers());
+    var topicName = "testConsumerExecutor-" + System.currentTimeMillis();
+    admin.creator().topic(topicName).numberOfPartitions(10).create();
+    Utils.waitFor(() -> admin.publicTopicNames().contains(topicName));
+
+    var metrics = new Metrics();
+    var param = new Performance.Argument();
+    param.brokers = bootstrapServers();
+    param.topic = topicName;
+    param.fixedSize = true;
+    param.exeTime = ExeTime.of("100records");
+    param.specifyBroker = List.of(0, 1);
+    param.consumers = 0;
+    param.partitions = 10;
+    try (ThreadPool.Executor executor =
+        Performance.producerExecutor(
+            Producer.builder().brokers(bootstrapServers()).build(),
+            param,
+            metrics,
+            partition(param, admin),
+            new Manager(param, List.of(), List.of()))) {
+      ThreadPool threadPool = ThreadPool.builder().executor(executor).build();
+      threadPool.waitAll();
+      threadPool.close();
+
+      Utils.waitFor(() -> metrics.num() == 100);
+      var offsets = admin.offsets(Set.of(topicName));
+      var partitions =
+          offsets.entrySet().stream()
+              .filter(entry -> entry.getValue().latest() > 0)
+              .map(entry -> entry.getKey().partition())
+              .collect(Collectors.toList());
+      var partitionsOfBrokers =
+          admin.partitionsOfBrokers(Set.of(topicName), Set.of(0, 1)).stream()
+              .map(TopicPartition::partition)
+              .collect(Collectors.toSet());
+      partitions.forEach(
+          partition -> Assertions.assertTrue(partitionsOfBrokers.contains(partition)));
+    }
+  }
+
   @Test
   void testProducerExecutor() throws InterruptedException {
     var metrics = new Metrics();
     var param = new Performance.Argument();
+    param.brokers = bootstrapServers();
     param.topic = "testProducerExecutor-" + System.currentTimeMillis();
     param.fixedSize = true;
     param.consumers = 0;
@@ -25,6 +117,7 @@ public class PerformanceTest extends RequireBrokerCluster {
             Producer.builder().brokers(bootstrapServers()).build(),
             param,
             metrics,
+            List.of(-1),
             new Manager(param, List.of(), List.of()))) {
       executor.execute();