Repository: skiptests/astraea
Pull request #213 (2022-02-08 07:33:34)

Record size distribution
Record 大小可使用分佈模型。

Performance tool多了新的參數`--size.distribution`，並且把原本參數`--fixed.size`合併進去，若要使用原本的`--fixed.size`功能，可以打上`--size.distribution fixed`。

範例，
```
performance --bootstrap.servers localhost:9092 --size.distribution zipfian
```

chia7712: 不好意思，要麻煩一下處理衝突

Patch:
diff --git a/README.md b/README.md
index 20d5a504..4c26b33c 100644
--- a/README.md
+++ b/README.md
@@ -112,7 +112,7 @@ For example. Assume you have two Kafka brokers,
 You can execute the following command to create a Prometheus instance that fetches data from the above 4 exporters.
 
 ```shell
-./docker/start_prometheus.sh start 192.168.0.1:10558,192.168.0.2:10558 192.168.0.1:9100,192.168.0.2:9100
+./docker/start_prometheus.sh start
 ```
 
 The console will show the http address of prometheus service, also some hints for you to set up Grafana. See next [section](#run-grafana) for
@@ -129,6 +129,16 @@ prometheus-9090
 [INFO] =================================================
 ```
 
+There are two ways to change the prometheus configuration.
+
+```shell
+# refresh prometheus config by command-line arguments
+./docker/start_prometheus.sh refresh host1:1111,host2:1111 host1:2222,host2:2222
+
+# refresh prometheus config by specific file
+./docker/start_prometheus.sh refresh ./path/to/config/file/prometheus.yml
+```
+
 #### Update Prometheus configuration
 
 To update your Prometheus configuration, there are two ways.
@@ -220,11 +230,14 @@ Run the benchmark from source
   The time units can be "days", "day", "h", "m", "s", "ms", "us", "ns".
   e.g. "--run.until 1m" or "--run.until 89242records" Default: 1000records
 8. --record.size: the (bound of) record size in byte. Default: 1 KiB
-9. --fixed.size: the flag to let all records have the same size
-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)
+9. --prop.file: the path to property file.
+10. --partitioner: the partitioner to use in producers
+11. --createCSV: put the metrics into a csv file if this flag is set. Default: false
+12. --compression: the compression algorithm used by producer. Available algorithm are none, gzip, snappy, lz4, and zstd. Default: (NONE)
+13. --jmx.servers: the jmx server addresses of the brokers 
+14. --key.distribution: name of the distribution on key. Available distribution names: "uniform", "zipfian", "latest", "fixed". Default: (No key)
+15. --size.distribution: name of the distribution on value size. Available distribution names: "uniform", "zipfian", "latest", "fixed". Default: "uniform"
+16. --specify.broker: list of broker IDs to produce records to. Default: (Do Not Specify)
 
 ---
 
diff --git a/app/src/main/java/org/astraea/partitioner/ClusterInfo.java b/app/src/main/java/org/astraea/cost/ClusterInfo.java
similarity index 53%
rename from app/src/main/java/org/astraea/partitioner/ClusterInfo.java
rename to app/src/main/java/org/astraea/cost/ClusterInfo.java
index c3e9f87b..c6bccbe4 100644
--- a/app/src/main/java/org/astraea/partitioner/ClusterInfo.java
+++ b/app/src/main/java/org/astraea/cost/ClusterInfo.java
@@ -1,8 +1,14 @@
-package org.astraea.partitioner;
+package org.astraea.cost;
 
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.stream.Collectors;
+import org.astraea.metrics.HasBeanObject;
 
 public interface ClusterInfo {
 
@@ -26,6 +32,58 @@ public interface ClusterInfo {
             .map(PartitionInfo::of)
             .collect(Collectors.toUnmodifiableList());
       }
+
+      @Override
+      public Collection<HasBeanObject> beans(int brokerId) {
+        return List.of();
+      }
+
+      @Override
+      public Map<Integer, Collection<HasBeanObject>> allBeans() {
+        return Map.of();
+      }
+    };
+  }
+
+  /**
+   * merge the beans into cluster information
+   *
+   * @param cluster cluster information
+   * @param beans extra beans
+   * @return a new cluster information with extra beans
+   */
+  static ClusterInfo of(ClusterInfo cluster, Map<Integer, Collection<HasBeanObject>> beans) {
+    var all = new HashMap<Integer, List<HasBeanObject>>();
+    cluster
+        .allBeans()
+        .forEach((key, value) -> all.computeIfAbsent(key, k -> new ArrayList<>()).addAll(value));
+    beans.forEach((key, value) -> all.computeIfAbsent(key, k -> new ArrayList<>()).addAll(value));
+    return new ClusterInfo() {
+
+      @Override
+      public List<NodeInfo> nodes() {
+        return cluster.nodes();
+      }
+
+      @Override
+      public List<PartitionInfo> availablePartitions(String topic) {
+        return cluster.availablePartitions(topic);
+      }
+
+      @Override
+      public List<PartitionInfo> partitions(String topic) {
+        return cluster.partitions(topic);
+      }
+
+      @Override
+      public Collection<HasBeanObject> beans(int brokerId) {
+        return all.getOrDefault(brokerId, List.of());
+      }
+
+      @Override
+      public Map<Integer, Collection<HasBeanObject>> allBeans() {
+        return Collections.unmodifiableMap(all);
+      }
     };
   }
 
@@ -77,4 +135,14 @@ public interface ClusterInfo {
    * @return A list of partitions
    */
   List<PartitionInfo> partitions(String topic);
+
+  /**
+   * @param brokerId broker id
+   * @return return the metrics of broker. It returns empty collection if the broker id is
+   *     nonexistent
+   */
+  Collection<HasBeanObject> beans(int brokerId);
+
+  /** @return all beans of all brokers */
+  Map<Integer, Collection<HasBeanObject>> allBeans();
 }
diff --git a/app/src/main/java/org/astraea/partitioner/cost/CostFunction.java b/app/src/main/java/org/astraea/cost/CostFunction.java
similarity index 68%
rename from app/src/main/java/org/astraea/partitioner/cost/CostFunction.java
rename to app/src/main/java/org/astraea/cost/CostFunction.java
index 81475d73..58a57bdc 100644
--- a/app/src/main/java/org/astraea/partitioner/cost/CostFunction.java
+++ b/app/src/main/java/org/astraea/cost/CostFunction.java
@@ -1,10 +1,7 @@
-package org.astraea.partitioner.cost;
+package org.astraea.cost;
 
-import java.util.List;
 import java.util.Map;
-import org.astraea.metrics.HasBeanObject;
 import org.astraea.metrics.collector.Fetcher;
-import org.astraea.partitioner.ClusterInfo;
 
 public interface CostFunction {
 
@@ -19,7 +16,7 @@ public interface CostFunction {
    * @param clusterInfo cluster information
    * @return the score of each node. the score range is [0 - 1]
    */
-  Map<Integer, Double> cost(Map<Integer, List<HasBeanObject>> beans, ClusterInfo clusterInfo);
+  Map<Integer, Double> cost(ClusterInfo clusterInfo);
 
   /** @return the metrics getters. Those getters are used to fetch mbeans. */
   Fetcher fetcher();
diff --git a/app/src/main/java/org/astraea/partitioner/NodeInfo.java b/app/src/main/java/org/astraea/cost/NodeInfo.java
similarity index 97%
rename from app/src/main/java/org/astraea/partitioner/NodeInfo.java
rename to app/src/main/java/org/astraea/cost/NodeInfo.java
index f0f8029a..f7633d4a 100644
--- a/app/src/main/java/org/astraea/partitioner/NodeInfo.java
+++ b/app/src/main/java/org/astraea/cost/NodeInfo.java
@@ -1,4 +1,4 @@
-package org.astraea.partitioner;
+package org.astraea.cost;
 
 import java.util.Objects;
 
diff --git a/app/src/main/java/org/astraea/partitioner/PartitionInfo.java b/app/src/main/java/org/astraea/cost/PartitionInfo.java
similarity index 95%
rename from app/src/main/java/org/astraea/partitioner/PartitionInfo.java
rename to app/src/main/java/org/astraea/cost/PartitionInfo.java
index 197b77b5..50d3ffca 100644
--- a/app/src/main/java/org/astraea/partitioner/PartitionInfo.java
+++ b/app/src/main/java/org/astraea/cost/PartitionInfo.java
@@ -1,4 +1,4 @@
-package org.astraea.partitioner;
+package org.astraea.cost;
 
 public interface PartitionInfo {
 
diff --git a/app/src/main/java/org/astraea/partitioner/cost/ThroughputCost.java b/app/src/main/java/org/astraea/cost/ThroughputCost.java
similarity index 79%
rename from app/src/main/java/org/astraea/partitioner/cost/ThroughputCost.java
rename to app/src/main/java/org/astraea/cost/ThroughputCost.java
index 7464f92b..92670ae5 100644
--- a/app/src/main/java/org/astraea/partitioner/cost/ThroughputCost.java
+++ b/app/src/main/java/org/astraea/cost/ThroughputCost.java
@@ -1,5 +1,6 @@
-package org.astraea.partitioner.cost;
+package org.astraea.cost;
 
+import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
@@ -7,15 +8,12 @@ import org.astraea.metrics.HasBeanObject;
 import org.astraea.metrics.collector.Fetcher;
 import org.astraea.metrics.kafka.BrokerTopicMetricsResult;
 import org.astraea.metrics.kafka.KafkaMetrics;
-import org.astraea.partitioner.ClusterInfo;
-import org.astraea.partitioner.NodeInfo;
 
 public class ThroughputCost implements CostFunction {
 
   @Override
-  public Map<Integer, Double> cost(
-      Map<Integer, List<HasBeanObject>> beans, ClusterInfo clusterInfo) {
-    var score = score(beans);
+  public Map<Integer, Double> cost(ClusterInfo clusterInfo) {
+    var score = score(clusterInfo.allBeans());
 
     var max = score.values().stream().mapToDouble(v -> v).max().orElse(1);
 
@@ -24,7 +22,7 @@ public class ThroughputCost implements CostFunction {
         .collect(Collectors.toMap(n -> n, n -> score.getOrDefault(n, 0.0D) / max));
   }
 
-  Map<Integer, Double> score(Map<Integer, List<HasBeanObject>> beans) {
+  Map<Integer, Double> score(Map<Integer, Collection<HasBeanObject>> beans) {
     // TODO: this implementation only consider the oneMinuteRate ...
     return beans.entrySet().stream()
         .collect(
diff --git a/app/src/main/java/org/astraea/metrics/collector/Receiver.java b/app/src/main/java/org/astraea/metrics/collector/Receiver.java
index ccaa103e..95e4eecc 100644
--- a/app/src/main/java/org/astraea/metrics/collector/Receiver.java
+++ b/app/src/main/java/org/astraea/metrics/collector/Receiver.java
@@ -1,6 +1,6 @@
 package org.astraea.metrics.collector;
 
-import java.util.List;
+import java.util.Collection;
 import org.astraea.metrics.HasBeanObject;
 
 /**
@@ -20,7 +20,7 @@ public interface Receiver extends AutoCloseable {
    *
    * @return current mbeans.
    */
-  List<HasBeanObject> current();
+  Collection<HasBeanObject> current();
 
   @Override
   void close();
diff --git a/app/src/main/java/org/astraea/partitioner/Dispatcher.java b/app/src/main/java/org/astraea/partitioner/Dispatcher.java
index 772605c9..199b0a25 100644
--- a/app/src/main/java/org/astraea/partitioner/Dispatcher.java
+++ b/app/src/main/java/org/astraea/partitioner/Dispatcher.java
@@ -4,6 +4,7 @@ import java.util.Map;
 import java.util.stream.Collectors;
 import org.apache.kafka.clients.producer.Partitioner;
 import org.apache.kafka.common.Cluster;
+import org.astraea.cost.ClusterInfo;
 
 public interface Dispatcher extends Partitioner {
   /**
diff --git a/app/src/main/java/org/astraea/partitioner/StrictCostDispatcher.java b/app/src/main/java/org/astraea/partitioner/StrictCostDispatcher.java
index d23be8d1..55128ee5 100644
--- a/app/src/main/java/org/astraea/partitioner/StrictCostDispatcher.java
+++ b/app/src/main/java/org/astraea/partitioner/StrictCostDispatcher.java
@@ -9,10 +9,12 @@ import java.util.Optional;
 import java.util.TreeMap;
 import java.util.stream.Collectors;
 import org.astraea.Utils;
+import org.astraea.cost.ClusterInfo;
+import org.astraea.cost.CostFunction;
+import org.astraea.cost.PartitionInfo;
 import org.astraea.metrics.collector.BeanCollector;
 import org.astraea.metrics.collector.Fetcher;
 import org.astraea.metrics.collector.Receiver;
-import org.astraea.partitioner.cost.CostFunction;
 
 /**
  * this dispatcher scores the nodes by multiples cost functions. Each function evaluate the target
@@ -68,7 +70,7 @@ public class StrictCostDispatcher implements Dispatcher {
     // get scores from all cost functions
     var scores =
         functions.stream()
-            .map(f -> f.cost(beans, clusterInfo))
+            .map(f -> f.cost(ClusterInfo.of(clusterInfo, beans)))
             .collect(Collectors.toUnmodifiableList());
 
     return bestPartition(partitions, scores).map(e -> e.getKey().partition()).orElse(0);
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 34281d9c..202890e4 100644
--- a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClient.java
+++ b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClient.java
@@ -9,6 +9,8 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
+import java.util.TreeMap;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicReference;
@@ -16,10 +18,11 @@ import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.stream.Collectors;
 import org.astraea.Utils;
+import org.astraea.cost.ClusterInfo;
+import org.astraea.cost.NodeInfo;
 import org.astraea.metrics.collector.Receiver;
 import org.astraea.metrics.java.HasJvmMemory;
 import org.astraea.metrics.kafka.KafkaMetrics;
-import org.astraea.partitioner.ClusterInfo;
 
 /**
  * this clas is responsible for obtaining jmx metrics from BeanCollector and calculating the
@@ -27,9 +30,8 @@ import org.astraea.partitioner.ClusterInfo;
  */
 public class NodeLoadClient {
 
-  private final List<Receiver> receiverList;
-  private final Map<String, Integer> currentJmxAddresses;
   private final Lock lock = new ReentrantLock();
+  private final Optional<Integer> jmxPortDefault;
   private static final ReceiverFactory RECEIVER_FACTORY = new ReceiverFactory();
   private long lastTime = -1;
   // visible for testing
@@ -38,14 +40,16 @@ public class NodeLoadClient {
   private int referenceBrokerID;
   private int totalInput;
   private int totalOutput;
+  private Map<Integer, Integer> jmxOfBrokerID;
+  private Map<String, Integer> jmxAddress;
 
   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)$";
 
-  public NodeLoadClient(Map<String, Integer> jmxAddresses) {
-    this.receiverList = RECEIVER_FACTORY.receiversList(jmxAddresses);
-    receiverList.forEach(receiver -> Utils.waitFor(() -> receiver.current().size() > 0));
-    currentJmxAddresses = jmxAddresses;
+  public NodeLoadClient(Map<Integer, Integer> jmxAddresses, int jmxPortDefault) {
+    if (jmxPortDefault != -1) this.jmxPortDefault = Optional.of(jmxPortDefault);
+    else this.jmxPortDefault = Optional.empty();
+    jmxOfBrokerID = jmxAddresses;
   }
 
   /**
@@ -69,6 +73,29 @@ public class NodeLoadClient {
     return currentLoadNode;
   }
 
+  Map<String, Integer> jmxAddress(ClusterInfo cluster) {
+    var iterator = jmxOfBrokerID.keySet().iterator();
+    var currentJmxAddresses = new TreeMap<String, Integer>();
+    var idHost = cluster.nodes().stream().collect(Collectors.toMap(NodeInfo::id, NodeInfo::host));
+    while (iterator.hasNext()) {
+      var id = iterator.next();
+      if (idHost.containsKey(id)) {
+        currentJmxAddresses.put(ipAddress(idHost.get(id)), jmxOfBrokerID.get(id));
+        idHost.remove(id);
+      }
+    }
+
+    if (idHost.size() > 0) {
+      jmxPortDefault.ifPresentOrElse(
+          port -> idHost.forEach((key, value) -> currentJmxAddresses.put(ipAddress(value), port)),
+          () -> {
+            var notFindJMX = new ArrayList<>(idHost.keySet());
+            throw new RuntimeException(notFindJMX + "jmx servers not found.");
+          });
+    }
+    return currentJmxAddresses;
+  }
+
   /** @param cluster the cluster from the partitioner */
   private void nodesOverLoad(ClusterInfo cluster) {
     if (lastTime == -1) {
@@ -78,6 +105,9 @@ public class NodeLoadClient {
               .map(node -> new Broker(node.id(), node.host(), node.port()))
               .collect(Collectors.toList());
       referenceBrokerID = brokers.stream().findFirst().get().brokerID;
+      jmxAddress = jmxAddress(cluster);
+      var receiverList = RECEIVER_FACTORY.receiversList(jmxAddress);
+      receiverList.forEach(receiver -> Utils.waitFor(() -> receiver.current().size() > 0));
       var nodeIDReceiver = new HashMap<Integer, List<Receiver>>();
       receiverList.forEach(
           receiver -> {
@@ -128,7 +158,8 @@ public class NodeLoadClient {
                       .metrics
                       .get(KafkaMetrics.BrokerTopic.BytesInPerSec.name())
                       .current()
-                      .get(0)
+                      .iterator()
+                      .next()
                       .beanObject()
                       .getAttributes()
                       .get("Count")
@@ -137,9 +168,10 @@ public class NodeLoadClient {
               Long.parseLong(
                   broker
                       .metrics
-                      .get(KafkaMetrics.BrokerTopic.BytesInPerSec.name())
+                      .get(KafkaMetrics.BrokerTopic.BytesOutPerSec.name())
                       .current()
-                      .get(0)
+                      .iterator()
+                      .next()
                       .beanObject()
                       .getAttributes()
                       .get("Count")
@@ -236,7 +268,7 @@ public class NodeLoadClient {
   }
 
   public void close() {
-    RECEIVER_FACTORY.close(currentJmxAddresses);
+    RECEIVER_FACTORY.close(jmxAddress);
   }
 
   private List<Integer> brokerIDOfReceiver(String host) {
diff --git a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/PartitionerUtils.java b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/PartitionerUtils.java
index 9036314c..b557cd74 100644
--- a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/PartitionerUtils.java
+++ b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/PartitionerUtils.java
@@ -1,7 +1,10 @@
 package org.astraea.partitioner.nodeLoadMetric;
 
+import java.io.FileInputStream;
+import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Properties;
 
 /** do poisson for node's load situation */
 public class PartitionerUtils {
@@ -45,20 +48,13 @@ public class PartitionerUtils {
     return (int) Math.round(avgLoadCount);
   }
 
-  /**
-   * A cheap way to deterministically convert a number to a positive value. When the input is
-   * positive, the original value is returned. When the input number is negative, the returned
-   * positive value is the original value bit AND against 0x7fffffff which is not its absolute
-   * value.
-   *
-   * <p>Note: changing this method in the future will possibly cause partition selection not to be
-   * compatible with the existing messages already placed on a partition since it is used in
-   * producer's {@link org.apache.kafka.clients.producer.internals.DefaultPartitioner}
-   *
-   * @param number a given number
-   * @return a positive number.
-   */
-  public static int toPositive(int number) {
-    return number & 0x7fffffff;
+  public static Properties partitionerConfig(Map<String, ?> configs) {
+    var properties = new Properties();
+    try {
+      properties.load(new FileInputStream((String) configs.get("partitioner.config")));
+    } catch (IOException e) {
+      e.printStackTrace();
+    }
+    return properties;
   }
 }
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 bdc8bc98..fb6f7ac6 100644
--- a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/ReceiverFactory.java
+++ b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/ReceiverFactory.java
@@ -69,9 +69,9 @@ public class ReceiverFactory {
     return receiversList;
   }
 
-  public void close(Map<String, Integer> jmxAddresses) {
+  public void close(Map<String, Integer> jmxAddress) {
     synchronized (lock) {
-      jmxAddresses.forEach(
+      jmxAddress.forEach(
           (host, port) -> {
             if (count.containsKey(nodeKey(host, port))) {
               if (count.get(nodeKey(host, port)) == 1) {
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 b1632239..b9c39017 100644
--- a/app/src/main/java/org/astraea/partitioner/smoothPartitioner/SmoothWeightPartitioner.java
+++ b/app/src/main/java/org/astraea/partitioner/smoothPartitioner/SmoothWeightPartitioner.java
@@ -2,22 +2,23 @@ package org.astraea.partitioner.smoothPartitioner;
 
 import static org.astraea.Utils.overSecond;
 import static org.astraea.partitioner.nodeLoadMetric.PartitionerUtils.allPoisson;
+import static org.astraea.partitioner.nodeLoadMetric.PartitionerUtils.partitionerConfig;
 import static org.astraea.partitioner.nodeLoadMetric.PartitionerUtils.weightPoisson;
 
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.Iterator;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Random;
+import java.util.TreeMap;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
 import org.apache.kafka.clients.producer.Partitioner;
 import org.apache.kafka.common.Cluster;
-import org.astraea.partitioner.ClusterInfo;
+import org.astraea.cost.ClusterInfo;
+import org.astraea.partitioner.Configuration;
 import org.astraea.partitioner.nodeLoadMetric.NodeLoadClient;
 
 /**
@@ -26,14 +27,14 @@ import org.astraea.partitioner.nodeLoadMetric.NodeLoadClient;
  * Finally, the result of poisson is used as the weight to perform smooth weighted RoundRobin.
  */
 public class SmoothWeightPartitioner implements Partitioner {
+  private static final String JMX_PORT = "jmx.port";
 
-  /**
-   * Record the current weight of each node according to Poisson calculation and the weight after
-   * partitioner calculation.
-   */
+  //  Record the current weight of each node according to Poisson calculation and the weight after
+  // partitioner calculation.
   private final ConcurrentMap<Integer, SmoothWeightServer> brokersWeight =
       new ConcurrentHashMap<>();
 
+  private final Map<Integer, Integer> jmxPorts = new TreeMap<>();
   private NodeLoadClient nodeLoadClient;
   private long lastTime = -1;
   private final Random rand = new Random();
@@ -49,7 +50,7 @@ public class SmoothWeightPartitioner implements Partitioner {
 
     SmoothWeightServer maxWeightServer = null;
 
-    Iterator<Integer> iterator = brokersWeight.keySet().iterator();
+    var iterator = brokersWeight.keySet().iterator();
     var currentWrightSum = weightSum;
     while (iterator.hasNext()) {
       var smoothWeightServer = brokersWeight.get(iterator.next());
@@ -80,19 +81,26 @@ public class SmoothWeightPartitioner implements Partitioner {
 
   @Override
   public void configure(Map<String, ?> configs) {
-    var jmxAddresses =
-        Objects.requireNonNull(
-            configs.get("jmx_servers").toString(), "You must configure jmx_servers correctly");
-    var list = Arrays.asList((jmxAddresses).split(","));
-    HashMap<String, Integer> mapAddress = new HashMap<>();
-    list.forEach(
-        str ->
-            mapAddress.put(
-                Arrays.asList(str.split(":")).get(0),
-                Integer.parseInt(Arrays.asList(str.split(":")).get(1))));
-    Objects.requireNonNull(mapAddress, "You must configure jmx_servers correctly.");
-
-    nodeLoadClient = new NodeLoadClient(mapAddress);
+    var properties = partitionerConfig(configs);
+    var config =
+        Configuration.of(
+            properties.entrySet().stream()
+                .collect(
+                    Collectors.toMap(e -> e.getKey().toString(), e -> e.getValue().toString())));
+    var jmxPort = config.integer(JMX_PORT);
+
+    jmxPort = config.integer(JMX_PORT);
+
+    // seeks for custom jmx ports.
+    config.entrySet().stream()
+        .filter(e -> e.getKey().startsWith("broker."))
+        .filter(e -> e.getKey().endsWith(JMX_PORT))
+        .forEach(
+            e ->
+                jmxPorts.put(
+                    Integer.parseInt(e.getKey().split("\\.")[1]), Integer.parseInt(e.getValue())));
+
+    nodeLoadClient = new NodeLoadClient(jmxPorts, jmxPort.orElse(-1));
   }
 
   /** Change the weight of the node according to the current Poisson. */
diff --git a/app/src/main/java/org/astraea/performance/Distribution.java b/app/src/main/java/org/astraea/performance/Distribution.java
deleted file mode 100644
index 83392220..00000000
--- a/app/src/main/java/org/astraea/performance/Distribution.java
+++ /dev/null
@@ -1,85 +0,0 @@
-package org.astraea.performance;
-
-import com.beust.jcommander.ParameterException;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Random;
-import java.util.stream.IntStream;
-import org.astraea.argument.Field;
-
-/** Randomly generate a long number with respect to some distribution */
-public interface Distribution {
-
-  long get();
-
-  class DistributionField extends Field<Distribution> {
-    @Override
-    public Distribution convert(String rawArgument) {
-      var args = rawArgument.split(":");
-      var name = rawArgument.split(":")[0];
-
-      switch (name) {
-        case "fixed":
-          return fixed();
-        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\".");
-      }
-    }
-  }
-
-  static Distribution fixed() {
-    var rand = new Random();
-    return fixed(rand.nextLong());
-  }
-
-  static Distribution fixed(long value) {
-    return () -> value;
-  }
-
-  /** 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/DistributionType.java b/app/src/main/java/org/astraea/performance/DistributionType.java
new file mode 100644
index 00000000..15918a7e
--- /dev/null
+++ b/app/src/main/java/org/astraea/performance/DistributionType.java
@@ -0,0 +1,97 @@
+package org.astraea.performance;
+
+import com.beust.jcommander.ParameterException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Random;
+import java.util.function.Supplier;
+import java.util.stream.IntStream;
+import org.astraea.argument.Field;
+
+/**
+ * Random distribution generator. Example: {@code Supplier<long> uniformDistribution =
+ * Distribution.UNIFORM.create(100); while (true){ // the value will in range [0, 100)
+ * uniformDistribution.get(); } }
+ */
+public enum DistributionType {
+  FIXED("fixed") {
+    @Override
+    public Supplier<Long> create(int n) {
+      return () -> (long) n;
+    }
+  },
+
+  UNIFORM("uniform") {
+    @Override
+    public Supplier<Long> create(int n) {
+      var rand = new Random();
+      return () -> (long) rand.nextInt(n);
+    }
+  },
+
+  /** A distribution for providing different random value every 2 seconds */
+  LATEST("latest") {
+    @Override
+    public Supplier<Long> create(int n) {
+      var rand = new Random();
+      return () -> {
+        var time = System.currentTimeMillis();
+        rand.setSeed(time - time % 2000);
+        return (long) rand.nextInt(n);
+      };
+    }
+  },
+
+  /**
+   * 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
+   */
+  ZIPFIAN("zipfian") {
+    @Override
+    public Supplier<Long> create(int n) {
+      var cumulativeDensityTable = new ArrayList<Double>();
+      var rand = new Random();
+      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;
+      };
+    }
+  };
+
+  public final String name;
+
+  abstract Supplier<Long> create(int n);
+
+  DistributionType(String name) {
+    this.name = name;
+  }
+
+  /**
+   * convert(String): Accept lower-case name only e.g. "fixed", "uniform", "latest" and "zipfian"
+   * are legal e.g. "Fixed" and "UNIFORM" are illegal
+   */
+  static class DistributionTypeField extends Field<DistributionType> {
+    @Override
+    public DistributionType convert(String name) {
+      return Arrays.stream(DistributionType.values())
+          .filter(distribute -> distribute.name.equals(name))
+          .findFirst()
+          .orElseThrow(
+              () ->
+                  new ParameterException(
+                      "Unknown distribution \""
+                          + name
+                          + "\". use \"fixed\" \"uniform\", \"latest\", \"zipfian\"."));
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/performance/Manager.java b/app/src/main/java/org/astraea/performance/Manager.java
index e388ff2b..d544e56a 100644
--- a/app/src/main/java/org/astraea/performance/Manager.java
+++ b/app/src/main/java/org/astraea/performance/Manager.java
@@ -7,6 +7,7 @@ import java.util.Random;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Supplier;
 import org.astraea.utils.DataSize;
 import org.astraea.utils.DataUnit;
 
@@ -23,7 +24,7 @@ public class Manager {
   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 Supplier<Long> keyDistribution;
   private final RandomContent randomContent;
 
   /**
@@ -51,8 +52,12 @@ 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);
+    this.keyDistribution = argument.keyDistributionType.create(100000);
+    this.randomContent =
+        new RandomContent(
+            argument.recordSize,
+            argument.sizeDistributionType.create(
+                argument.recordSize.measurement(DataUnit.Byte).intValue()));
   }
 
   /**
@@ -109,23 +114,24 @@ public class Manager {
 
   /** Randomly choose a key according to the distribution. */
   public byte[] getKey() {
-    return (String.valueOf(distribution.get())).getBytes();
+    return (String.valueOf(keyDistribution.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 Supplier<Long> distribution;
     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
+     * @param distribution Determine whether to fix the size of random generated content or random
+     *     size with specified distribution
      */
-    public RandomContent(DataSize dataSize, boolean fixedSize) {
+    public RandomContent(DataSize dataSize, Supplier<Long> distribution) {
       this.dataSize = dataSize;
-      this.fixedSize = fixedSize;
+      this.distribution = distribution;
       content = new byte[dataSize.measurement(DataUnit.Byte).intValue()];
     }
 
@@ -133,8 +139,8 @@ public class Manager {
       // 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);
+      return Arrays.copyOfRange(
+          content, (int) (distribution.get() % 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 8b521746..08d3ed16 100644
--- a/app/src/main/java/org/astraea/performance/Performance.java
+++ b/app/src/main/java/org/astraea/performance/Performance.java
@@ -194,8 +194,10 @@ public class Performance {
             .timestamp(start)
             .run()
             .whenComplete(
-                (m, e) ->
-                    observer.accept(System.currentTimeMillis() - start, m.serializedValueSize()));
+                (m, e) -> {
+                  if (e == null)
+                    observer.accept(System.currentTimeMillis() - start, m.serializedValueSize());
+                });
         return State.RUNNING;
       }
 
@@ -271,11 +273,6 @@ public class Performance {
         converter = ExeTime.Field.class)
     ExeTime exeTime = ExeTime.of("1000records");
 
-    @Parameter(
-        names = {"--fixed.size"},
-        description = "boolean: send fixed size records if this flag is set")
-    boolean fixedSize = false;
-
     @Parameter(
         names = {"--record.size"},
         description = "DataSize: size of each record. e.g. \"500KiB\"",
@@ -285,7 +282,7 @@ public class Performance {
     @Parameter(
         names = {"--jmx.servers"},
         description =
-            "String: server to get jmx metrics <jmx_server>@<broker_id>[,<jmx_server>@<broker_id>]*",
+            "String: server to get jmx metrics <broker_id>.<jmx_serverPort>[,<broker_id>.<jmx_serverPort>]*",
         validateWith = NonEmptyStringField.class)
     String jmxServers = "";
 
@@ -312,7 +309,7 @@ public class Performance {
 
     @Parameter(
         names = {"--createCSV"},
-        description = "create the metrics into a csv file if this flag is set")
+        description = "put the metrics into a csv file if this flag is set")
     boolean createCSV = false;
 
     @Parameter(
@@ -326,8 +323,15 @@ public class Performance {
         names = {"--key.distribution"},
         description =
             "String: Distribution name. Available distribution names: \"fixed\" \"uniform\", \"zipfian\", \"latest\". Default: uniform",
-        converter = Distribution.DistributionField.class)
-    Distribution distribution = Distribution.uniform();
+        converter = DistributionType.DistributionTypeField.class)
+    DistributionType keyDistributionType = DistributionType.UNIFORM;
+
+    @Parameter(
+        names = {"--size.distribution"},
+        description =
+            "String: Distribution name. Available distribution names: \"uniform\", \"zipfian\", \"latest\", \"fixed\". Default: \"uniform\"",
+        converter = DistributionType.DistributionTypeField.class)
+    DistributionType sizeDistributionType = DistributionType.UNIFORM;
 
     @Parameter(
         names = {"--specify.broker"},
diff --git a/app/src/main/java/org/astraea/performance/Tracker.java b/app/src/main/java/org/astraea/performance/Tracker.java
index 9bbab9e2..b197dbf8 100644
--- a/app/src/main/java/org/astraea/performance/Tracker.java
+++ b/app/src/main/java/org/astraea/performance/Tracker.java
@@ -76,7 +76,7 @@ public class Tracker implements Executor {
           "  producer[%d] average publish latency: %.3f ms%n", i, result.averageLatencies.get(i));
     }
     System.out.println("\n");
-    return percentage >= 100D;
+    return manager.producedDone();
   }
 
   private boolean logConsumers(Result result) {
diff --git a/config/partitionerConfig.properties b/config/partitionerConfig.properties
new file mode 100644
index 00000000..e31beda9
--- /dev/null
+++ b/config/partitionerConfig.properties
@@ -0,0 +1,20 @@
+#The properties are used to configure SmoothWeightPartitioner
+#
+#If broker.id.jmx.port is configured, then the jmx port of these brokers
+# will be configured first. If there are brokers that are not configured
+# to jmx port, then jmx.port will be used for configuration. And if any
+# broker is not configured, and jmx.port is not configured at the same
+# time, then an error will be reported.
+#
+###########################Smooth Partitioner Basics############################
+
+#If no jmx port is specified for the broker,jmx.port will be used.
+jmx.port=Default;
+
+# list of brokers used for jmx port about the rest of the cluster
+# format: broker.{brokerID}.jmx.port=jmx.port
+#broker.0.jmx.port=9999
+#broker.1.jmx.port=8888
+#......
+bootstrap.servers=localhost:9092
+broker.id.jmx.port=Default;
diff --git a/docker/start_node_exporter.sh b/docker/start_node_exporter.sh
index 30b30a34..703bf5b0 100755
--- a/docker/start_node_exporter.sh
+++ b/docker/start_node_exporter.sh
@@ -56,10 +56,10 @@ function run_node_exporter() {
 container_id="$(run_node_exporter)"
 info Container ID of node_exporter: "$container_id"
 
-if [[ "$(curl --connect-timeout 1 --retry 3 -s "http://0.0.0.0:$PORT")" == "" ]] || [[ "$(docker_container_running "$container_id")" == "false" ]]; then
-  error Cannot access node_exporter metric server: "http://0.0.0.0:$PORT". Something go wrong!
+if curl --connect-timeout 1 --retry 3 --retry-connrefused -s "http://$address:$PORT/metrics" > /dev/null; then
+  info node_exporter running at "http://$address:$PORT"
+else
+  error Cannot access node_exporter metric server: "http://$address:$PORT/metrics". Something go wrong!
   error Execute \"docker logs "$container_id"\" to see what\'s going on.
   exit 3
 fi
-
-info node_exporter running at "http://$address:$PORT"
\ No newline at end of file
diff --git a/docker/start_prometheus.sh b/docker/start_prometheus.sh
index c1e61511..de0bacd4 100755
--- a/docker/start_prometheus.sh
+++ b/docker/start_prometheus.sh
@@ -7,9 +7,33 @@ function showHelp() {
   echo "COMMAND: "
   echo "    start <kafka-broker-jmx-addresses> <node-exporter-addresses>    create/start the prometheus docker instance"
   echo "    refresh <kafka-broker-jmx-addresses> <node-exporter-addresses>  refresh and apply the prometheus config"
+  echo "    refresh <config-file>                                           refresh and apply the prometheus config"
   echo "    help                                                            show this dialog"
 }
 
+function gradle_project_directory {
+    local script_folder="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" &>/dev/null && pwd)"
+    local target_folder="$script_folder"
+
+    # attempt to find the nearest parent directory that contains the gradlew file
+    # we recognize such directory as a Gradle repo that contains the build directory
+    while [ "$target_folder" != "/" ]; do
+        target_folder="$(dirname $target_folder)"
+
+        if [ -f "$target_folder/gradlew" ]; then
+            break
+        fi
+    done
+
+    if [ "$target_folder" == "/" ]; then
+        echo "Cannot find the gradle directory from $script_folder."
+        echo "Is the directory exists?"
+        exit 5
+    fi
+
+    echo "$target_folder"
+}
+
 # ===============================[checks]===============================
 
 if [[ "$(which docker)" == "" ]]; then
@@ -46,11 +70,16 @@ function wrap_address() {
 prometheus_port="${PROMETHEUS_PORT:-9090}"
 image_name=prom/prometheus:v2.32.1
 container_name="prometheus-${prometheus_port}"
-file=/tmp/prometheus-${prometheus_port}.yml
-scrape_interval="5s"
+file_prefix="$(gradle_project_directory)/build/docker/prometheus"
+file="${file_prefix}/prometheus-${prometheus_port}.yml"
+temp_file="${file_prefix}/prometheus-${prometheus_port}-editing.yml"
+scrape_interval="${scrape_interval:-200ms}"
+scrape_timeout="${scrape_timeout:-200ms}"
 volume_name_1="prometheus-${prometheus_port}-etc"
 volume_name_2="prometheus-${prometheus_port}-prometheus"
 
+mkdir -p "$file_prefix"
+
 function write_config() {
   kafka_jmx_addresses="$(wrap_address "$1")"
   node_exporter_addresses="$(wrap_address "$2")"
@@ -65,11 +94,13 @@ scrape_configs:
 
   - job_name: 'kafka'
     scrape_interval: $scrape_interval
+    scrape_timeout: $scrape_timeout
     static_configs:
       - targets: [$kafka_jmx_addresses]
 
   - job_name: 'node'
     scrape_interval: $scrape_interval
+    scrape_timeout: $scrape_timeout
     static_configs:
       - targets: [$node_exporter_addresses]
 EOT
@@ -91,8 +122,8 @@ function info() {
 }
 
 function refresh_config_from_file() {
-  info "Refresh prometheus configuration from config file $file"
-  docker cp "$file" "$container_name:/etc/prometheus/prometheus.yml"
+  info "Refresh prometheus configuration from config file $1"
+  docker cp "$1" "$container_name:/etc/prometheus/prometheus.yml"
   docker kill --signal="SIGHUP" "$container_name"
 }
 
@@ -134,19 +165,18 @@ function main() {
   info "================================================="
 }
 
-
 # shellcheck disable=SC2199
 if [[ "$1" == "start" ]]; then
   # shellcheck disable=SC2068
   main "$2" "$3"
-elif [[ "$1" == "refresh" ]] && [[ "$2" != "" ]]; then
+elif [[ "$1" == "refresh" ]] && [[ "$#" -eq 3 ]]; then
   # shellcheck disable=SC2068
   refresh_config "$2" "$3"
-elif [[ "$1" == "refresh" ]]; then
-  refresh_config_from_file
+elif [[ "$1" == "refresh" ]] && [[ "$#" -eq 2 ]]; then
+  refresh_config_from_file "$2"
 elif [[ "$1" == "help" ]]; then
   showHelp
 else
   info Unknown argument: "$@"
   showHelp
-fi
\ No newline at end of file
+fi
diff --git a/gradle/wrapper/gradle-wrapper.properties b/gradle/wrapper/gradle-wrapper.properties
index 976bbe0a..7086e5cb 100644
--- a/gradle/wrapper/gradle-wrapper.properties
+++ b/gradle/wrapper/gradle-wrapper.properties
@@ -1,6 +1,6 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
 # please update the gradle version in gradlew too
-distributionUrl=https\://services.gradle.org/distributions/gradle-7.4-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-7.4.1-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
diff --git a/gradlew b/gradlew
index 017e6341..5abce60e 100755
--- a/gradlew
+++ b/gradlew
@@ -116,7 +116,7 @@ esac
 
 # gradle-wrapper.jar is not in git repo anymore, so we have to download it now.
 if [ ! -e "$APP_HOME/gradle/wrapper/gradle-wrapper.jar" ]; then
-  if ! curl -s -S --retry 3 -L -o "$APP_HOME/gradle/wrapper/gradle-wrapper.jar" "https://raw.githubusercontent.com/gradle/gradle/v7.4.0/gradle/wrapper/gradle-wrapper.jar"; then
+  if ! curl -s -S --retry 3 -L -o "$APP_HOME/gradle/wrapper/gradle-wrapper.jar" "https://raw.githubusercontent.com/gradle/gradle/v7.4.1/gradle/wrapper/gradle-wrapper.jar"; then
     rm -f "$APP_HOME/gradle/wrapper/gradle-wrapper.jar"
     echo "failed to download gradle-wrapper.jar"
     exit 1


Test patch:
diff --git a/app/src/test/java/org/astraea/partitioner/ClusterInfoTest.java b/app/src/test/java/org/astraea/cost/ClusterInfoTest.java
similarity index 61%
rename from app/src/test/java/org/astraea/partitioner/ClusterInfoTest.java
rename to app/src/test/java/org/astraea/cost/ClusterInfoTest.java
index 5d2d6434..de532ed5 100644
--- a/app/src/test/java/org/astraea/partitioner/ClusterInfoTest.java
+++ b/app/src/test/java/org/astraea/cost/ClusterInfoTest.java
@@ -1,7 +1,10 @@
-package org.astraea.partitioner;
+package org.astraea.cost;
 
+import java.util.Collection;
 import java.util.List;
+import java.util.Map;
 import org.apache.kafka.common.Cluster;
+import org.astraea.metrics.HasBeanObject;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
@@ -30,4 +33,22 @@ public class ClusterInfoTest {
     Assertions.assertEquals(
         NodeInfo.of(node), clusterInfo.partitions(partition.topic()).get(0).leader());
   }
+
+  @Test
+  void testEmptyBeans() {
+    var clusterInfo = ClusterInfo.of(Mockito.mock(org.apache.kafka.common.Cluster.class));
+    Assertions.assertEquals(0, clusterInfo.allBeans().size());
+    Assertions.assertEquals(0, clusterInfo.beans(19).size());
+  }
+
+  @Test
+  void testBeans() {
+    var beans = Map.of(1, (Collection<HasBeanObject>) List.of(Mockito.mock(HasBeanObject.class)));
+    var origin = Mockito.mock(ClusterInfo.class);
+    Mockito.when(origin.allBeans()).thenReturn(Map.of());
+    var clusterInfo = ClusterInfo.of(origin, beans);
+    Assertions.assertEquals(1, clusterInfo.allBeans().size());
+    Assertions.assertEquals(0, clusterInfo.beans(19).size());
+    Assertions.assertEquals(1, clusterInfo.beans(1).size());
+  }
 }
diff --git a/app/src/test/java/org/astraea/partitioner/NodeInfoTest.java b/app/src/test/java/org/astraea/cost/NodeInfoTest.java
similarity index 93%
rename from app/src/test/java/org/astraea/partitioner/NodeInfoTest.java
rename to app/src/test/java/org/astraea/cost/NodeInfoTest.java
index d44e110f..61b6434e 100644
--- a/app/src/test/java/org/astraea/partitioner/NodeInfoTest.java
+++ b/app/src/test/java/org/astraea/cost/NodeInfoTest.java
@@ -1,4 +1,4 @@
-package org.astraea.partitioner;
+package org.astraea.cost;
 
 import org.apache.kafka.common.Node;
 import org.junit.jupiter.api.Assertions;
diff --git a/app/src/test/java/org/astraea/partitioner/PartitionInfoTest.java b/app/src/test/java/org/astraea/cost/PartitionInfoTest.java
similarity index 95%
rename from app/src/test/java/org/astraea/partitioner/PartitionInfoTest.java
rename to app/src/test/java/org/astraea/cost/PartitionInfoTest.java
index 9a6159f1..3a3aac74 100644
--- a/app/src/test/java/org/astraea/partitioner/PartitionInfoTest.java
+++ b/app/src/test/java/org/astraea/cost/PartitionInfoTest.java
@@ -1,4 +1,4 @@
-package org.astraea.partitioner;
+package org.astraea.cost;
 
 import org.apache.kafka.common.Node;
 import org.junit.jupiter.api.Assertions;
diff --git a/app/src/test/java/org/astraea/partitioner/cost/ThroughputCostTest.java b/app/src/test/java/org/astraea/cost/ThroughputCostTest.java
similarity index 84%
rename from app/src/test/java/org/astraea/partitioner/cost/ThroughputCostTest.java
rename to app/src/test/java/org/astraea/cost/ThroughputCostTest.java
index 48cb0d9d..820ded3d 100644
--- a/app/src/test/java/org/astraea/partitioner/cost/ThroughputCostTest.java
+++ b/app/src/test/java/org/astraea/cost/ThroughputCostTest.java
@@ -1,10 +1,8 @@
-package org.astraea.partitioner.cost;
+package org.astraea.cost;
 
 import java.util.List;
 import java.util.Map;
 import org.astraea.metrics.kafka.BrokerTopicMetricsResult;
-import org.astraea.partitioner.ClusterInfo;
-import org.astraea.partitioner.NodeInfo;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
@@ -34,8 +32,9 @@ public class ThroughputCostTest {
 
     var cluster = Mockito.mock(ClusterInfo.class);
     Mockito.when(cluster.nodes()).thenReturn(List.of(node));
+    Mockito.when(cluster.allBeans()).thenReturn(Map.of());
 
-    var cost = throughputCost.cost(Map.of(10, List.of(bean)), cluster);
+    var cost = throughputCost.cost(ClusterInfo.of(cluster, Map.of(10, List.of(bean))));
     Assertions.assertEquals(1, cost.size());
     Assertions.assertEquals(1, cost.get(10));
   }
diff --git a/app/src/test/java/org/astraea/metrics/collector/BeanCollectorTest.java b/app/src/test/java/org/astraea/metrics/collector/BeanCollectorTest.java
index 522383a3..df7f3fd9 100644
--- a/app/src/test/java/org/astraea/metrics/collector/BeanCollectorTest.java
+++ b/app/src/test/java/org/astraea/metrics/collector/BeanCollectorTest.java
@@ -75,7 +75,7 @@ public class BeanCollectorTest {
 
     var c0 = receiver.current();
     Assertions.assertEquals(1, c0.size());
-    var firstObject = c0.get(0);
+    var firstObject = c0.iterator().next();
     sleep(1);
 
     var c1 = receiver.current();
diff --git a/app/src/test/java/org/astraea/partitioner/DispatcherTest.java b/app/src/test/java/org/astraea/partitioner/DispatcherTest.java
index 342d44a5..d15917f4 100644
--- a/app/src/test/java/org/astraea/partitioner/DispatcherTest.java
+++ b/app/src/test/java/org/astraea/partitioner/DispatcherTest.java
@@ -3,6 +3,7 @@ package org.astraea.partitioner;
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.atomic.AtomicInteger;
+import org.astraea.cost.ClusterInfo;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
diff --git a/app/src/test/java/org/astraea/partitioner/StrictCostDispatcherTest.java b/app/src/test/java/org/astraea/partitioner/StrictCostDispatcherTest.java
index f194629a..f471e71f 100644
--- a/app/src/test/java/org/astraea/partitioner/StrictCostDispatcherTest.java
+++ b/app/src/test/java/org/astraea/partitioner/StrictCostDispatcherTest.java
@@ -5,10 +5,12 @@ import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.function.Function;
 import java.util.stream.Collectors;
-import org.astraea.metrics.HasBeanObject;
+import org.astraea.cost.ClusterInfo;
+import org.astraea.cost.CostFunction;
+import org.astraea.cost.NodeInfo;
+import org.astraea.cost.PartitionInfo;
 import org.astraea.metrics.collector.Fetcher;
 import org.astraea.metrics.collector.Receiver;
-import org.astraea.partitioner.cost.CostFunction;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
@@ -65,9 +67,8 @@ public class StrictCostDispatcherTest {
     var costFunction =
         new CostFunction() {
           @Override
-          public Map<Integer, Double> cost(
-              Map<Integer, List<HasBeanObject>> beans, ClusterInfo clusterInfo) {
-            return beans.keySet().stream()
+          public Map<Integer, Double> cost(ClusterInfo clusterInfo) {
+            return clusterInfo.allBeans().keySet().stream()
                 .collect(
                     Collectors.toMap(Function.identity(), id -> id.equals(n0.id()) ? 0.9D : 0.5D));
           }
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 3c754173..fd25eb11 100644
--- a/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClientTest.java
+++ b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClientTest.java
@@ -2,15 +2,14 @@ package org.astraea.partitioner.nodeLoadMetric;
 
 import static org.mockito.Mockito.when;
 
-import java.io.IOException;
 import java.lang.reflect.Field;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.astraea.Utils;
-import org.astraea.partitioner.ClusterInfo;
-import org.astraea.partitioner.NodeInfo;
+import org.astraea.cost.ClusterInfo;
+import org.astraea.cost.NodeInfo;
 import org.astraea.service.RequireBrokerCluster;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.Assertions;
@@ -24,10 +23,10 @@ public class NodeLoadClientTest extends RequireBrokerCluster {
   private NodeLoadClient nodeLoadClient;
 
   @BeforeAll
-  void setUp() throws IOException {
-    var map = new HashMap<String, Integer>();
-    map.put(jmxServiceURL().getHost(), jmxServiceURL().getPort());
-    nodeLoadClient = new NodeLoadClient(map);
+  void setUp() {
+    var map = new HashMap<Integer, Integer>();
+    map.put(0, jmxServiceURL().getPort());
+    nodeLoadClient = new NodeLoadClient(map, jmxServiceURL().getPort());
   }
 
   @AfterAll
@@ -89,7 +88,6 @@ public class NodeLoadClientTest extends RequireBrokerCluster {
           nodes.add(NodeInfo.of(count.get(), hostPort[0], Integer.parseInt(hostPort[1])));
           count.getAndIncrement();
         });
-
     var cluster = Mockito.mock(ClusterInfo.class);
     when(cluster.nodes()).thenReturn(nodes);
     var load = nodeLoadClient.loadSituation(cluster);
diff --git a/app/src/test/java/org/astraea/partitioner/smoothPartitioner/PartitionerTest.java b/app/src/test/java/org/astraea/partitioner/smoothPartitioner/PartitionerTest.java
index af24a097..c655bfa4 100644
--- a/app/src/test/java/org/astraea/partitioner/smoothPartitioner/PartitionerTest.java
+++ b/app/src/test/java/org/astraea/partitioner/smoothPartitioner/PartitionerTest.java
@@ -6,6 +6,9 @@ import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
 import java.lang.reflect.Field;
 import java.time.Duration;
 import java.util.HashMap;
@@ -49,12 +52,24 @@ public class PartitionerTest extends RequireBrokerCluster {
     props.put(ProducerConfig.CLIENT_ID_CONFIG, "id1");
     props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, SmoothWeightPartitioner.class.getName());
     props.put("producerID", 1);
-    props.put("jmx_servers", jmxServiceURL().getHost() + ":" + jmxServiceURL().getPort());
+    var file = new File(PartitionerTest.class.getResource("").getPath() + "PartitionerConfigTest");
+    try {
+      var fileWriter = new FileWriter(file);
+      fileWriter.write("jmx.port=" + jmxServiceURL().getPort());
+      fileWriter.flush();
+      fileWriter.close();
+    } catch (IOException e) {
+      e.printStackTrace();
+    }
+    props.put(
+        "partitioner.config",
+        PartitionerTest.class.getResource("").getPath() + "PartitionerConfigTest");
     return props;
   }
 
   @Test
   void testSetBrokerHashMap() {
+    initProConfig();
     var nodeLoadClient = Mockito.mock(NodeLoadClient.class);
     when(nodeLoadClient.thoughPutComparison(anyInt())).thenReturn(1.0);
     var poissonMap = new HashMap<Integer, Double>();
@@ -218,6 +233,57 @@ public class PartitionerTest extends RequireBrokerCluster {
     Assertions.assertEquals(smoothWeightServer.originalWeight(), -9);
   }
 
+  @Test
+  void testJmxConfig() {
+    var props = initProConfig();
+    var file = new File(PartitionerTest.class.getResource("").getPath() + "PartitionerConfigTest");
+    try {
+      var fileWriter = new FileWriter(file);
+      fileWriter.write(
+          "broker.0.jmx.port="
+              + jmxServiceURL().getPort()
+              + "\n"
+              + "broker.1.jmx.port="
+              + jmxServiceURL().getPort()
+              + "\n"
+              + "broker.2.jmx.port="
+              + jmxServiceURL().getPort()
+              + "\n");
+      fileWriter.flush();
+      fileWriter.close();
+    } catch (IOException e) {
+      e.printStackTrace();
+    }
+    var topicName = "addressN";
+    admin.creator().topic(topicName).numberOfPartitions(10).create();
+    var key = "tainan";
+    var timestamp = System.currentTimeMillis() + 10;
+    var header = Header.of("a", "b".getBytes());
+
+    try (var producer =
+        Producer.builder()
+            .keySerializer(Serializer.STRING)
+            .configs(
+                props.entrySet().stream()
+                    .collect(Collectors.toMap(e -> e.getKey().toString(), Map.Entry::getValue)))
+            .build()) {
+      var metadata =
+          producer
+              .sender()
+              .topic(topicName)
+              .key(key)
+              .timestamp(timestamp)
+              .headers(List.of(header))
+              .run()
+              .toCompletableFuture()
+              .get();
+      assertEquals(topicName, metadata.topic());
+      assertEquals(timestamp, metadata.timestamp());
+    } catch (ExecutionException | InterruptedException e) {
+      e.printStackTrace();
+    }
+  }
+
   private Executor producerExecutor(
       Producer<String, byte[]> producer, String topic, String key, Header header, long timeStamp) {
     return new Executor() {
diff --git a/app/src/test/java/org/astraea/performance/DistributionTest.java b/app/src/test/java/org/astraea/performance/DistributionTest.java
index 6ff1e06f..a227e19a 100644
--- a/app/src/test/java/org/astraea/performance/DistributionTest.java
+++ b/app/src/test/java/org/astraea/performance/DistributionTest.java
@@ -1,5 +1,6 @@
 package org.astraea.performance;
 
+import java.util.Random;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import org.junit.jupiter.api.Assertions;
@@ -9,7 +10,7 @@ public class DistributionTest {
 
   @Test
   void testFixed() {
-    var distribution = Distribution.fixed();
+    var distribution = DistributionType.FIXED.create(new Random().nextInt());
     Assertions.assertEquals(
         1,
         IntStream.range(0, 10)
@@ -18,9 +19,16 @@ public class DistributionTest {
             .size());
   }
 
+  @Test
+  void testUniform() {
+    var distribution = DistributionType.UNIFORM.create(5);
+    Assertions.assertTrue(distribution.get() < 5);
+    Assertions.assertTrue(distribution.get() >= 0);
+  }
+
   @Test
   void testLatest() throws InterruptedException {
-    var distribution = Distribution.latest();
+    var distribution = DistributionType.LATEST.create(Integer.MAX_VALUE);
     Assertions.assertEquals(distribution.get(), distribution.get());
 
     long first = distribution.get();
@@ -31,7 +39,8 @@ public class DistributionTest {
 
   @Test
   void testZipfian() {
-    var distribution = Distribution.zipfian(5);
+    var distribution = DistributionType.ZIPFIAN.create(5);
     Assertions.assertTrue(distribution.get() < 5);
+    Assertions.assertTrue(distribution.get() >= 0);
   }
 }
diff --git a/app/src/test/java/org/astraea/performance/ManagerTest.java b/app/src/test/java/org/astraea/performance/ManagerTest.java
index a6ef7934..050795a1 100644
--- a/app/src/test/java/org/astraea/performance/ManagerTest.java
+++ b/app/src/test/java/org/astraea/performance/ManagerTest.java
@@ -58,19 +58,19 @@ public class ManagerTest {
   void testGetKey() {
     var argument = new Performance.Argument();
 
-    argument.distribution = Distribution.uniform();
+    argument.keyDistributionType = DistributionType.UNIFORM;
     var manager = new Manager(argument, List.of(), List.of());
     Assertions.assertTrue(manager.getKey().length > 0);
 
-    argument.distribution = Distribution.zipfian(10);
+    argument.keyDistributionType = DistributionType.ZIPFIAN;
     manager = new Manager(argument, List.of(), List.of());
     Assertions.assertTrue(manager.getKey().length > 0);
 
-    argument.distribution = Distribution.latest();
+    argument.keyDistributionType = DistributionType.LATEST;
     manager = new Manager(argument, List.of(), List.of());
     Assertions.assertTrue(manager.getKey().length > 0);
 
-    argument.distribution = Distribution.fixed();
+    argument.keyDistributionType = DistributionType.FIXED;
     manager = new Manager(argument, List.of(), List.of());
     Assertions.assertTrue(manager.getKey().length > 0);
   }
diff --git a/app/src/test/java/org/astraea/performance/PerformanceTest.java b/app/src/test/java/org/astraea/performance/PerformanceTest.java
index a753be5e..686d1e17 100644
--- a/app/src/test/java/org/astraea/performance/PerformanceTest.java
+++ b/app/src/test/java/org/astraea/performance/PerformanceTest.java
@@ -31,7 +31,7 @@ public class PerformanceTest extends RequireBrokerCluster {
     var param = new Performance.Argument();
     param.brokers = bootstrapServers();
     param.topic = topicName;
-    param.fixedSize = true;
+    param.sizeDistributionType = DistributionType.FIXED;
     param.exeTime = ExeTime.of("100records");
     param.specifyBroker = List.of(0);
     param.consumers = 0;
@@ -74,7 +74,7 @@ public class PerformanceTest extends RequireBrokerCluster {
     var param = new Performance.Argument();
     param.brokers = bootstrapServers();
     param.topic = topicName;
-    param.fixedSize = true;
+    param.sizeDistributionType = DistributionType.FIXED;
     param.exeTime = ExeTime.of("100records");
     param.specifyBroker = List.of(0, 1);
     param.consumers = 0;
@@ -112,7 +112,7 @@ public class PerformanceTest extends RequireBrokerCluster {
     var param = new Performance.Argument();
     param.brokers = bootstrapServers();
     param.topic = "testProducerExecutor-" + System.currentTimeMillis();
-    param.fixedSize = true;
+    param.sizeDistributionType = DistributionType.FIXED;
     param.consumers = 0;
     try (Executor executor =
         Performance.producerExecutor(
@@ -133,7 +133,7 @@ public class PerformanceTest extends RequireBrokerCluster {
     Metrics metrics = new Metrics();
     var topicName = "testConsumerExecutor-" + System.currentTimeMillis();
     var param = new Performance.Argument();
-    param.fixedSize = true;
+    param.sizeDistributionType = DistributionType.FIXED;
     try (Executor executor =
         Performance.consumerExecutor(
             Consumer.builder().topics(Set.of(topicName)).brokers(bootstrapServers()).build(),