Repository: skiptests/astraea
Pull request #27 (2021-09-23 15:55:17)

Implement poisson partitioner
[基於過去全局資訊預測未來最佳部署的partitioner(phase 1,phase 2)#8](https://github.com/skiptests/astraea/issues/8)

完成了poisson partitioner的基本功能
但有一個至關重要的問題我找不到太好的答案想要請教一下，我的該方法找不到很恰當的時間啟動`monitorOverLoad`。在我的設想中他應該伴隨一個producer整個周期。但將它放在client程式碼中顯然不是特別恰當，或許可以在partitoner中監測該方法是否有在運行，再做處理。但我不太了解這該怎麼寫。
解決這一問題後我會再更新partitoner的test文件。

關於單一節點及多節點情況區分，節點上下線，剛啟動收集數據需要一小段時間數據收集不完全導致算法的偏差等問題的優化及測試我會再逐步更新。

chia7712: @wycccccc 你不考慮先把checkstyle修好嗎XD
chia7712: > 但有一個至關重要的問題我找不到太好的答案想要請教一下，我的該方法找不到很恰當的時間啟動monitorOverLoad

跟著partitioner建立 然後跟著partitioner關閉 這樣不好嗎？如果是有效能問題，那我們可以晚一點再處理
wycccccc: 主要的一個原因在於我需要存儲overload的次數。
我把overload定義為偏離集群平均負載數值過大的情況。然後設想是每秒對整個集群都進行一次判定記錄每個node的情況，持續維護更新最近十秒的數據。
如果將其`monitorOverLoad`放入partitioner隨着他的開啟關閉，可能無法做到持久監控。效能問題確實是我考慮的出發點，因為等待jmx獲取數據再進行計算分配這一過程每次partitioner的速度會肉眼可見的慢下來。（應該是因為要等待client從server端拿到數據）這個點我認為是必須要去解決的。

>如果有些resource想要讓jvm上所有producer共用，那麼一個可能的方式是用java singleton的方式來設計partitioner，換言之，雖然每個producer都會建立一個partitioner，但實際上該partitioner都會將操作轉傳給一個static partitioner，然後該static partitioner要是thread-safe，如果你就可以很放心將成本很高（例如索取叢集資訊）的操作放在該static partitioner，此外，在每次關閉partitioner的時候，都要將計數器--，等到最後一個partitioner關閉時，他也必須去關閉static partitioner，然後等到下一個partitioner被建立時，又需要建立該static partitioner

有get到思路，我準備改寫一下`NodeLoadClient.class`,singleton感覺上是個不錯的解決方法，我目前想到的是用類似singleton的靜態內部類加載來處理。應該能夠解決上面我提到的這些問題。
提到的其餘問題我會在改寫的過程中一起解決。

wycccccc: 我按照之前的思路改了一下`NodeLoadClient.class`，我將更新jmx數據這件事另開了thread處理。他會在第一次partitioner中被開啟，因為是另外一個thread也不會因為這次partitioner結束而結束。它由搖籃系統維持。想問一下這樣做是否合適。我接下去會去寫一下partitioner的測試，各項小功能的測試已經完成，如果partitioner測試通過就可以開始和MbeanClient整合。
wycccccc: 我將metrics整合到了partitioner中，並修改了一些寫法，嘗試解決了上面提到的糟糕的設計並讓整體看起來更加清晰。不過還有些問題我想請教下意見
wycccccc: 我徹底的進行了一下修改，並且有整合singleton partitioner測試並通過。有關partitioner實機運行狀況是否正常的測試，我準備將他放在下個pr，這個目前確實有些臃腫了。
wycccccc: 我再去修改一下先別review
wycccccc: 上述問題都已解決。
chia7712: 那個建制錯誤是因為我在https://github.com/skiptests/astraea/pull/84把該public API刪掉了，我發了另一隻PR (#85）把該方法加回來
chia7712: @wycccccc 可否麻煩你rebase code?
chia7712: ```
admin.createTopic(topicName, 10);
```

這隻方法確定要刪掉了，因為partitions和replicas太容易混淆，要麻煩使用新的APIs

Patch:
diff --git a/app/build.gradle b/app/build.gradle
index d2922109..f35bed23 100644
--- a/app/build.gradle
+++ b/app/build.gradle
@@ -18,8 +18,9 @@ repositories {
 }
 
 def versions = [
-    kafka: project.properties['kafka.version'] ?: "2.8.1",
+    kafka: project.properties['kafka.version'] ?: "2.8.0",
     junit: project.properties['junit.version'] ?: "5.8.1",
+    mockito: project.properties['mockito.version'] ?: "3.12.4",
     jcommander: project.properties['jcommander.version'] ?: "1.81",
     slf4j: project.properties['slf4j.version'] ?: "1.7.32",
     zookeeper: project.properties['zookeeper.version'] ?: "3.5.9",
@@ -27,9 +28,12 @@ def versions = [
 
 dependencies {
     testImplementation "org.junit.jupiter:junit-jupiter:${versions["junit"]}"
+    testImplementation "org.mockito:mockito-core:${versions["mockito"]}"
+    testImplementation "org.mockito:mockito-inline:${versions["mockito"]}"
     testImplementation "org.apache.zookeeper:zookeeper:${versions["zookeeper"]}"
     testImplementation "org.apache.kafka:kafka_2.12:${versions["kafka"]}"
 
+
     implementation "org.apache.kafka:kafka-clients:${versions["kafka"]}"
     implementation "com.beust:jcommander:${versions["jcommander"]}"
     // we don't use slf4j actually, and it is used by kafka so we swallow the log.
diff --git a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/BrokersWeight.java b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/BrokersWeight.java
new file mode 100644
index 00000000..e9d779ef
--- /dev/null
+++ b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/BrokersWeight.java
@@ -0,0 +1,55 @@
+package org.astraea.partitioner.nodeLoadMetric;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class BrokersWeight {
+
+  /**
+   * Record the current weight of each node according to Poisson calculation and the weight after
+   * partitioner calculation.
+   */
+  private static HashMap<String, int[]> brokerHashMap = new HashMap<>();
+
+  private LoadPoisson loadPoisson;
+
+  public BrokersWeight(LoadPoisson loadPoisson) {
+    this.loadPoisson = loadPoisson;
+  }
+
+  /** Change the weight of the node according to the current Poisson. */
+  public synchronized void setBrokerHashMap() {
+    HashMap<String, Double> poissonMap = loadPoisson.setAllPoisson();
+
+    for (Map.Entry<String, Double> entry : poissonMap.entrySet()) {
+      if (!brokerHashMap.containsKey(entry.getKey())) {
+        brokerHashMap.put(entry.getKey(), new int[] {(int) ((1 - entry.getValue()) * 20), 0});
+      } else {
+        brokerHashMap.put(
+            entry.getKey(),
+            new int[] {(int) ((1 - entry.getValue()) * 20), brokerHashMap.get(entry.getKey())[1]});
+      }
+    }
+  }
+
+  public synchronized int getAllWeight() {
+    int allWeight = 0;
+    for (Map.Entry<String, int[]> entry : brokerHashMap.entrySet()) {
+      allWeight += entry.getValue()[0];
+    }
+    return allWeight;
+  }
+
+  public synchronized HashMap<String, int[]> getBrokerHashMap() {
+    return brokerHashMap;
+  }
+
+  public synchronized void setCurrentBrokerHashMap(HashMap<String, int[]> currentBrokerHashMap) {
+    brokerHashMap = currentBrokerHashMap;
+  }
+
+  // Only for test
+  void setBrokerHashMapValue(String x, int y) {
+    brokerHashMap.put(x, new int[] {0, y});
+  }
+}
diff --git a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/LoadPoisson.java b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/LoadPoisson.java
new file mode 100644
index 00000000..511cdaa8
--- /dev/null
+++ b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/LoadPoisson.java
@@ -0,0 +1,42 @@
+package org.astraea.partitioner.nodeLoadMetric;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class LoadPoisson {
+  private NodeLoadClient nodeLoadClient;
+
+  public LoadPoisson(NodeLoadClient nodeLoadClient) {
+    this.nodeLoadClient = nodeLoadClient;
+  }
+
+  public synchronized HashMap<String, Double> setAllPoisson() {
+    HashMap<String, Double> poissonMap = new HashMap<>();
+    int lambda = nodeLoadClient.getAvgLoadCount();
+    for (Map.Entry<String, Integer> entry : nodeLoadClient.getAllOverLoadCount().entrySet()) {
+      int x = nodeLoadClient.getBinOneCount(entry.getValue());
+      poissonMap.put(entry.getKey(), doPoisson(lambda, x));
+    }
+    return poissonMap;
+  }
+
+  public double doPoisson(int lambda, int x) {
+    double Probability = 0;
+    double ans = 0;
+
+    for (int i = 0; i <= x; i++) {
+      double j = Math.pow(lambda, i);
+      double e = Math.exp(-lambda);
+      long h = factorial(i);
+      Probability = (j * e) / h;
+      ans += Probability;
+    }
+
+    return ans;
+  }
+
+  public long factorial(long number) {
+    if (number <= 1) return 1;
+    else return number * factorial(number - 1);
+  }
+}
diff --git a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClient.java b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClient.java
new file mode 100644
index 00000000..009d5229
--- /dev/null
+++ b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClient.java
@@ -0,0 +1,94 @@
+package org.astraea.partitioner.nodeLoadMetric;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.concurrent.TimeUnit;
+import org.astraea.concurrent.ThreadPool;
+import org.astraea.metrics.jmx.MBeanClient;
+
+public class NodeLoadClient implements ThreadPool.Executor {
+
+  private final OverLoadNode overLoadNode;
+  private final Collection<NodeMetadata> nodeMetadataCollection = new ArrayList<>();
+
+  public NodeLoadClient(HashMap<String, String> jmxAddresses) throws IOException {
+    for (HashMap.Entry<String, String> entry : jmxAddresses.entrySet()) {
+      this.nodeMetadataCollection.add(
+          new NodeMetadata(entry.getKey(), createNodeMetrics(entry.getKey(), entry.getValue())));
+    }
+    this.overLoadNode = new OverLoadNode(this.nodeMetadataCollection);
+  }
+
+  public NodeMetrics createNodeMetrics(String key, String value) throws IOException {
+    return new NodeMetrics(key, value);
+  }
+
+  /** A thread that continuously updates metricsfor NodeLoadClient. */
+  @Override
+  public State execute() throws InterruptedException {
+    try {
+      refreshNodesMetrics();
+      overLoadNode.monitorOverLoad();
+      TimeUnit.SECONDS.sleep(1);
+    } catch (InterruptedException e) {
+      e.printStackTrace();
+    }
+    return null;
+  }
+
+  @Override
+  public void close() {
+    for (NodeMetadata nodeMetadata : nodeMetadataCollection) {
+      NodeMetrics nodeMetrics = nodeMetadata.getNodeMetrics();
+      MBeanClient mBeanClient = nodeMetrics.getKafkaMetricClient();
+      try {
+        mBeanClient.close();
+      } catch (Exception e) {
+        e.printStackTrace();
+      }
+    }
+  }
+
+  public synchronized HashMap<String, Integer> getAllOverLoadCount() {
+    HashMap<String, Integer> overLoadCount = new HashMap<>();
+    for (NodeMetadata nodeMetadata : nodeMetadataCollection) {
+      overLoadCount.put(nodeMetadata.getNodeID(), nodeMetadata.getOverLoadCount());
+    }
+    return overLoadCount;
+  }
+
+  public synchronized int getAvgLoadCount() {
+    double avgLoadCount = 0;
+    for (NodeMetadata nodeMetadata : nodeMetadataCollection) {
+      avgLoadCount += getBinOneCount(nodeMetadata.getOverLoadCount());
+    }
+    return nodeMetadataCollection.size() > 0
+        ? (int) avgLoadCount / nodeMetadataCollection.size()
+        : 0;
+  }
+
+  /** Get the number of times a node is overloaded. */
+  public static int getBinOneCount(int n) {
+    int index = 0;
+    int count = 0;
+    while (n > 0) {
+      int x = n & 1 << index;
+      if (x != 0) {
+        count++;
+        n = n - (1 << index);
+      }
+      index++;
+    }
+    return count;
+  }
+
+  public void refreshNodesMetrics() {
+    for (NodeMetadata nodeMetadata : nodeMetadataCollection) {
+      NodeMetrics nodeMetrics = nodeMetadata.getNodeMetrics();
+      nodeMetrics.refreshMetrics();
+      nodeMetadata.setTotalBytes(nodeMetrics.totalBytesPerSec());
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeMetadata.java b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeMetadata.java
new file mode 100644
index 00000000..a86a7a0a
--- /dev/null
+++ b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeMetadata.java
@@ -0,0 +1,40 @@
+package org.astraea.partitioner.nodeLoadMetric;
+
+/** Store information about each node */
+public class NodeMetadata {
+  private String nodeID;
+  private NodeMetrics nodeMetrics;
+  private double totalBytes;
+  private int overLoadCount;
+
+  NodeMetadata(String nodeID, NodeMetrics nodeMetrics) {
+    this.nodeID = nodeID;
+    this.nodeMetrics = nodeMetrics;
+    this.overLoadCount = 0;
+    this.totalBytes = 0.0;
+  }
+
+  public NodeMetrics getNodeMetrics() {
+    return this.nodeMetrics;
+  }
+
+  public void setOverLoadCount(int count) {
+    this.overLoadCount = count;
+  }
+
+  public void setTotalBytes(double bytes) {
+    this.totalBytes = bytes;
+  }
+
+  public double getTotalBytes() {
+    return this.totalBytes;
+  }
+
+  public String getNodeID() {
+    return this.nodeID;
+  }
+
+  public int getOverLoadCount() {
+    return this.overLoadCount;
+  }
+}
diff --git a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeMetrics.java b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeMetrics.java
new file mode 100644
index 00000000..92c857cd
--- /dev/null
+++ b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeMetrics.java
@@ -0,0 +1,62 @@
+package org.astraea.partitioner.nodeLoadMetric;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+import javax.management.remote.JMXServiceURL;
+import org.astraea.metrics.jmx.MBeanClient;
+import org.astraea.metrics.kafka.BrokerTopicMetricsResult;
+import org.astraea.metrics.kafka.KafkaMetrics;
+
+/** Responsible for connecting jmx according to the received address */
+public class NodeMetrics {
+  private final String JMX_URI_FORMAT = "service:jmx:rmi:///jndi/rmi://" + "%s" + "/jmxrmi";
+  private final JMXServiceURL serviceURL;
+  private final MBeanClient mBeanClient;
+  private final String nodeID;
+  private HashMap<String, Double> metricsValues;
+  private Collection<String> argumentTargetMetrics = new ArrayList<>();
+
+  NodeMetrics(String ID, String address) throws IOException {
+    argumentTargetMetrics.add("BytesInPerSec");
+    argumentTargetMetrics.add("BytesOutPerSec");
+    nodeID = ID;
+    if (Pattern.compile("^service:").matcher(address).find())
+      serviceURL = new JMXServiceURL(address);
+    else serviceURL = new JMXServiceURL(createJmxUrl(address));
+    mBeanClient = new MBeanClient(serviceURL);
+    metricsValues = new HashMap();
+  }
+
+  public String createJmxUrl(String address) {
+    return String.format(JMX_URI_FORMAT, address);
+  }
+
+  public void refreshMetrics() {
+    List<KafkaMetrics.BrokerTopic> metrics =
+        argumentTargetMetrics.stream()
+            .map(KafkaMetrics.BrokerTopic::of)
+            .collect(Collectors.toUnmodifiableList());
+
+    List<BrokerTopicMetricsResult> collect =
+        metrics.stream().map(x -> x.fetch(mBeanClient)).collect(Collectors.toUnmodifiableList());
+
+    for (BrokerTopicMetricsResult result : collect) {
+      metricsValues.put(
+          result.beanObject().getProperties().get("name"),
+          (Double) result.beanObject().getAttributes().get("MeanRate"));
+    }
+  }
+
+  public double totalBytesPerSec() {
+    return metricsValues.get("BytesInPerSec") + metricsValues.get("BytesOutPerSec");
+  }
+
+  public MBeanClient getKafkaMetricClient() {
+    return this.mBeanClient;
+  }
+}
diff --git a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/OverLoadNode.java b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/OverLoadNode.java
new file mode 100644
index 00000000..7552f78f
--- /dev/null
+++ b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/OverLoadNode.java
@@ -0,0 +1,100 @@
+package org.astraea.partitioner.nodeLoadMetric;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class OverLoadNode {
+  private double standardDeviation = 0;
+  private double avgBrokersMsgPerSec = 0;
+  private Collection<String> nodesID;
+  private int nodeNum;
+  private int mountCount = 0;
+  private HashMap<String, Double> eachBrokerMsgPerSec = new HashMap();
+  private Collection<NodeMetadata> nodesMetadata;
+
+  OverLoadNode(Collection<NodeMetadata> nodeMetadataCollection) {
+    this.nodesID =
+        nodeMetadataCollection.stream()
+            .map(NodeMetadata::getNodeID)
+            .collect(Collectors.toUnmodifiableList());
+    this.nodeNum = nodeMetadataCollection.size();
+    this.nodesMetadata = nodeMetadataCollection;
+  }
+
+  /** Monitor and update the number of overloads of each node. */
+  public void monitorOverLoad() {
+    setBrokersMsgPerSec();
+    setAvgBrokersMsgPerSec();
+    standardDeviationImperative();
+    for (NodeMetadata nodeMetadata : nodesMetadata) {
+      int ifOverLoad = 0;
+      if (nodeMetadata.getTotalBytes() > (avgBrokersMsgPerSec + standardDeviation)) {
+        ifOverLoad = 1;
+      }
+      nodeMetadata.setOverLoadCount(
+          setOverLoadCount(nodeMetadata.getOverLoadCount(), mountCount % 10, ifOverLoad));
+    }
+    mountCount++;
+  }
+
+  /**
+   * Use bit operations to record whether the node exceeds the load per second,the position of the
+   * number represents the recorded time.
+   */
+  public int setOverLoadCount(int overLoadCount, int roundCount, int ifOverLoad) {
+    int x = overLoadCount & 1 << roundCount;
+    if (x == ifOverLoad << roundCount) {
+      return overLoadCount;
+    } else {
+      if (ifOverLoad != 0) {
+        return overLoadCount | 1 << roundCount;
+      } else {
+        return overLoadCount - (int) Math.pow(2, roundCount);
+      }
+    }
+  }
+
+  public void setBrokersMsgPerSec() {
+    for (NodeMetadata nodeMetadata : nodesMetadata) {
+      eachBrokerMsgPerSec.put(nodeMetadata.getNodeID(), nodeMetadata.getTotalBytes());
+    }
+  }
+
+  public void setAvgBrokersMsgPerSec() {
+    double avg = 0;
+    for (Map.Entry<String, Double> entry : eachBrokerMsgPerSec.entrySet()) {
+      avg += entry.getValue();
+    }
+    this.avgBrokersMsgPerSec = avg / nodeNum;
+  }
+
+  public void standardDeviationImperative() {
+    double variance = 0;
+    for (Map.Entry<String, Double> entry : eachBrokerMsgPerSec.entrySet()) {
+      variance +=
+          (entry.getValue() - avgBrokersMsgPerSec) * (entry.getValue() - avgBrokersMsgPerSec);
+    }
+    this.standardDeviation = Math.sqrt(variance / nodeNum);
+  }
+
+  // Only for test
+  double getStandardDeviation() {
+    return this.standardDeviation;
+  }
+
+  // Only for test
+  public void setEachBrokerMsgPerSec(HashMap<String, Double> hashMap) {
+    this.eachBrokerMsgPerSec = hashMap;
+  }
+  // Only for test
+  double getAvgBrokersMsgPerSec() {
+    return this.avgBrokersMsgPerSec;
+  }
+
+  // Only for test
+  void setMountCount(int i) {
+    this.mountCount = i;
+  }
+}
diff --git a/app/src/main/java/org/astraea/partitioner/partitionerFactory/LinkPartitioner.java b/app/src/main/java/org/astraea/partitioner/partitionerFactory/LinkPartitioner.java
new file mode 100644
index 00000000..0404404b
--- /dev/null
+++ b/app/src/main/java/org/astraea/partitioner/partitionerFactory/LinkPartitioner.java
@@ -0,0 +1,128 @@
+package org.astraea.partitioner.partitionerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Random;
+import org.apache.kafka.clients.producer.Partitioner;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.PartitionInfo;
+import org.astraea.concurrent.ThreadPool;
+import org.astraea.partitioner.nodeLoadMetric.BrokersWeight;
+import org.astraea.partitioner.nodeLoadMetric.LoadPoisson;
+import org.astraea.partitioner.nodeLoadMetric.NodeLoadClient;
+
+public class LinkPartitioner implements Partitioner {
+
+  private static final SmoothPartitionerFactory FACTORY =
+      new SmoothPartitionerFactory(
+          Comparator.comparing(o -> o.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG).toString()));
+
+  private Partitioner partitioner;
+
+  @Override
+  public int partition(
+      String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
+    return partitioner.partition(topic, key, keyBytes, value, valueBytes, cluster);
+  }
+
+  @Override
+  public void onNewBatch(String topic, Cluster cluster, int prevPartition) {
+    partitioner.onNewBatch(topic, cluster, prevPartition);
+  }
+
+  @Override
+  public void close() {
+    partitioner.close();
+  }
+
+  @Override
+  public void configure(Map<String, ?> configs) throws NullPointerException {
+    Objects.requireNonNull(
+        (String) configs.get("jmx_servers"), "You must configure jmx_servers correctly");
+    partitioner = FACTORY.getOrCreate(ThreadSafeSmoothPartitioner.class, configs);
+  }
+
+  public Partitioner getPartitioner() {
+    return partitioner;
+  }
+
+  public static class ThreadSafeSmoothPartitioner implements Partitioner {
+    private NodeLoadClient nodeLoadClient;
+    private ThreadPool pool;
+
+    ThreadSafeSmoothPartitioner() {}
+
+    @Override
+    public int partition(
+        String topic,
+        Object key,
+        byte[] keyBytes,
+        Object value,
+        byte[] valueBytes,
+        Cluster cluster) {
+      LoadPoisson loadPoisson = new LoadPoisson(nodeLoadClient);
+      BrokersWeight brokersWeight = new BrokersWeight(loadPoisson);
+      brokersWeight.setBrokerHashMap();
+      Map.Entry<String, int[]> maxWeightServer = null;
+
+      int allWeight = brokersWeight.getAllWeight();
+      HashMap<String, int[]> currentBrokerHashMap = brokersWeight.getBrokerHashMap();
+
+      for (Map.Entry<String, int[]> item : currentBrokerHashMap.entrySet()) {
+        if (maxWeightServer == null || item.getValue()[1] > maxWeightServer.getValue()[1]) {
+          maxWeightServer = item;
+        }
+      }
+      assert maxWeightServer != null;
+      currentBrokerHashMap.put(
+          maxWeightServer.getKey(),
+          new int[] {maxWeightServer.getValue()[0], maxWeightServer.getValue()[1] - allWeight});
+      brokersWeight.setCurrentBrokerHashMap(currentBrokerHashMap);
+
+      ArrayList<Integer> partitionList = new ArrayList<>();
+      for (PartitionInfo partitionInfo :
+          cluster.partitionsForNode(Integer.parseInt(maxWeightServer.getKey()))) {
+        partitionList.add(partitionInfo.partition());
+      }
+      Random rand = new Random();
+
+      return partitionList.get(rand.nextInt(partitionList.size()));
+    }
+
+    @Override
+    public void close() {
+      pool.close();
+    }
+
+    @Override
+    public void configure(Map<String, ?> configs) {
+      try {
+        var jmxAddresses =
+            Objects.requireNonNull(
+                (String) configs.get("jmx_servers"), "You must configure jmx_servers correctly");
+        var list = Arrays.asList((jmxAddresses).split(","));
+        HashMap<String, String> mapAddress = new HashMap<>();
+        for (String str : list) {
+          var listAddress = Arrays.asList(str.split("@"));
+          mapAddress.put(listAddress.get(1), listAddress.get(0));
+        }
+        Objects.requireNonNull(
+            mapAddress, "You must configure jmx_servers correctly.(JmxAddress@NodeID)");
+        nodeLoadClient = new NodeLoadClient((mapAddress));
+      } catch (IOException e) {
+        throw new RuntimeException();
+      }
+      pool = ThreadPool.builder().executor(nodeLoadClient).build();
+    }
+  }
+
+  public static SmoothPartitionerFactory getFactory() {
+    return FACTORY;
+  }
+}
diff --git a/app/src/main/java/org/astraea/partitioner/partitionerFactory/SmoothPartitionerFactory.java b/app/src/main/java/org/astraea/partitioner/partitionerFactory/SmoothPartitionerFactory.java
new file mode 100644
index 00000000..ba3f5305
--- /dev/null
+++ b/app/src/main/java/org/astraea/partitioner/partitionerFactory/SmoothPartitionerFactory.java
@@ -0,0 +1,103 @@
+package org.astraea.partitioner.partitionerFactory;
+
+import java.util.Comparator;
+import java.util.Map;
+import java.util.TreeMap;
+import org.apache.kafka.clients.producer.Partitioner;
+import org.apache.kafka.common.Cluster;
+
+public class SmoothPartitionerFactory {
+  private final Object lock = new Object();
+  private final Map<Map<String, ?>, Integer> count;
+  private final Map<Map<String, ?>, Partitioner> instances;
+  private final Map<Map<String, ?>, Partitioner> smoothPartitionerMap;
+
+  /**
+   * create a factory with specific comparator.
+   *
+   * @param comparator used to compare the partitioners. There is no new producer if the comparator
+   *     returns 0 (equal).
+   */
+  SmoothPartitionerFactory(Comparator<Map<String, ?>> comparator) {
+    this.count = new TreeMap<>(comparator);
+    this.instances = new TreeMap<>(comparator);
+    this.smoothPartitionerMap = new TreeMap<>(comparator);
+  }
+
+  /**
+   * @param clz partitioner class
+   * @param configs used to initialize new partitioner
+   * @return create a new partitioner if there is no matched partitioner (checked by comparator).
+   *     Otherwise, it returns the existent partitioner.
+   */
+  Partitioner getOrCreate(Class<? extends Partitioner> clz, Map<String, ?> configs) {
+    synchronized (lock) {
+      var partitioner = instances.get(configs);
+      if (partitioner != null) {
+        count.put(configs, count.get(configs) + 1);
+        return partitioner;
+      }
+      return create(clz, configs);
+    }
+  }
+
+  private Partitioner create(Class<? extends Partitioner> clz, Map<String, ?> configs) {
+    try {
+      var partitioner = clz.getDeclaredConstructor().newInstance();
+      partitioner.configure(configs);
+      var proxy =
+          new Partitioner() {
+            @Override
+            public int partition(
+                String topic,
+                Object key,
+                byte[] keyBytes,
+                Object value,
+                byte[] valueBytes,
+                Cluster cluster) {
+              return partitioner.partition(topic, key, keyBytes, value, valueBytes, cluster);
+            }
+
+            @Override
+            public void onNewBatch(String topic, Cluster cluster, int prevPartition) {
+              partitioner.onNewBatch(topic, cluster, prevPartition);
+            }
+
+            @Override
+            public void close() {
+              synchronized (lock) {
+                var current = count.get(configs);
+                if (current == 1) {
+                  try {
+                    partitioner.close();
+                  } finally {
+                    count.remove(configs);
+                    instances.remove(configs);
+                    smoothPartitionerMap.remove(configs);
+                  }
+                } else count.put(configs, current - 1);
+              }
+            }
+
+            @Override
+            public void configure(Map<String, ?> configs) {
+              partitioner.configure(configs);
+            }
+          };
+      count.put(configs, 1);
+      instances.put(configs, proxy);
+      smoothPartitionerMap.put(configs, partitioner);
+      return proxy;
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  public Map<Map<String, ?>, Integer> getCount() {
+    return this.count;
+  }
+
+  public Map<Map<String, ?>, Partitioner> getSmoothPartitionerMap() {
+    return this.smoothPartitionerMap;
+  }
+}
diff --git a/docker/start_broker.sh b/docker/start_broker.sh
index a0700d99..f33c3cf8 100755
--- a/docker/start_broker.sh
+++ b/docker/start_broker.sh
@@ -183,4 +183,5 @@ docker run -d \
 echo "================================================="
 echo "broker address: ${address}:$broker_port"
 echo "jmx address: ${address}:$broker_jmx_port"
+echo "broker id: $broker_id"
 echo "================================================="


Test patch:
diff --git a/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClientTest.java b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClientTest.java
new file mode 100644
index 00000000..454f7835
--- /dev/null
+++ b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClientTest.java
@@ -0,0 +1,42 @@
+package org.astraea.partitioner.nodeLoadMetric;
+
+import static org.astraea.partitioner.nodeLoadMetric.NodeLoadClient.getBinOneCount;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class NodeLoadClientTest {
+  private HashMap<String, String> jmxAddresses;
+  private Collection<NodeMetadata> nodeMetadataCollection;
+
+  @BeforeEach
+  public void setUp() {
+    jmxAddresses = new HashMap<>();
+    jmxAddresses.put("0", "0.0.0.0");
+    jmxAddresses.put("1", "0.0.0.0");
+    jmxAddresses.put("2", "0.0.0.0");
+
+    nodeMetadataCollection = new ArrayList<>();
+    NodeMetrics nodeMetrics = mock(NodeMetrics.class);
+    NodeMetadata nodeMetadata0 = new NodeMetadata("0", nodeMetrics);
+    NodeMetadata nodeMetadata1 = new NodeMetadata("1", nodeMetrics);
+    NodeMetadata nodeMetadata2 = new NodeMetadata("2", nodeMetrics);
+    NodeMetadata nodeMetadata3 = new NodeMetadata("3", nodeMetrics);
+
+    nodeMetadataCollection.add(nodeMetadata0);
+    nodeMetadataCollection.add(nodeMetadata1);
+    nodeMetadataCollection.add(nodeMetadata2);
+    nodeMetadataCollection.add(nodeMetadata3);
+  }
+
+  @Test
+  public void testGetBinOneCount() {
+    assertEquals(getBinOneCount(7), 3);
+    assertEquals(getBinOneCount(10), 2);
+  }
+}
diff --git a/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/OverLoadNodeTest.java b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/OverLoadNodeTest.java
new file mode 100644
index 00000000..c0edb247
--- /dev/null
+++ b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/OverLoadNodeTest.java
@@ -0,0 +1,115 @@
+package org.astraea.partitioner.nodeLoadMetric;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class OverLoadNodeTest {
+  private Collection<NodeMetadata> nodeMetadataCollection;
+  private Collection<NodeMetadata> nodeMetadataCollection2;
+
+  @BeforeEach
+  public void setUp() {
+    nodeMetadataCollection = new ArrayList<>();
+    NodeMetrics nodeMetrics = mock(NodeMetrics.class);
+    NodeMetadata nodeMetadata0 = new NodeMetadata("0", nodeMetrics);
+    NodeMetadata nodeMetadata1 = new NodeMetadata("1", nodeMetrics);
+    NodeMetadata nodeMetadata2 = new NodeMetadata("2", nodeMetrics);
+    NodeMetadata nodeMetadata3 = new NodeMetadata("3", nodeMetrics);
+
+    nodeMetadataCollection.add(nodeMetadata0);
+    nodeMetadataCollection.add(nodeMetadata1);
+    nodeMetadataCollection.add(nodeMetadata2);
+    nodeMetadataCollection.add(nodeMetadata3);
+
+    NodeMetadata nodeMetadata20 = new NodeMetadata("0", nodeMetrics);
+    NodeMetadata nodeMetadata21 = new NodeMetadata("1", nodeMetrics);
+    NodeMetadata nodeMetadata22 = new NodeMetadata("2", nodeMetrics);
+    NodeMetadata nodeMetadata23 = new NodeMetadata("3", nodeMetrics);
+
+    nodeMetadataCollection2 = new ArrayList<>();
+    nodeMetadataCollection2.add(nodeMetadata20);
+    nodeMetadataCollection2.add(nodeMetadata21);
+    nodeMetadataCollection2.add(nodeMetadata22);
+    nodeMetadataCollection2.add(nodeMetadata23);
+  }
+
+  @Test
+  public void testStandardDeviationImperative() {
+    HashMap<String, Double> testHashMap = new HashMap<>();
+    testHashMap.put("0", 10.0);
+    testHashMap.put("1", 10.0);
+    testHashMap.put("2", 20.0);
+    testHashMap.put("3", 20.0);
+
+    OverLoadNode overLoadNode = new OverLoadNode(nodeMetadataCollection);
+    overLoadNode.setEachBrokerMsgPerSec(testHashMap);
+    overLoadNode.setAvgBrokersMsgPerSec();
+
+    assertEquals(overLoadNode.getAvgBrokersMsgPerSec(), 15);
+
+    overLoadNode.standardDeviationImperative();
+
+    assertEquals(overLoadNode.getStandardDeviation(), 5);
+  }
+
+  @Test
+  public void testSetOverLoadCount() {
+    Collection<NodeMetadata> nodeMetadataCollection = new ArrayList<>();
+
+    OverLoadNode overLoadNode = new OverLoadNode(nodeMetadataCollection);
+    assertEquals(overLoadNode.setOverLoadCount(0, 2, 1), 4);
+    assertEquals(overLoadNode.setOverLoadCount(31, 2, 1), 31);
+    assertEquals(overLoadNode.setOverLoadCount(31, 2, 0), 27);
+    assertEquals(overLoadNode.setOverLoadCount(20, 4, 0), 4);
+  }
+
+  @Test
+  public void testMonitorOverLoad() {
+    HashMap<String, Double> testHashMap = new HashMap<>();
+    testHashMap.put("0", 10.0);
+    testHashMap.put("1", 5.0);
+    testHashMap.put("2", 20.0);
+    testHashMap.put("3", 50.0);
+
+    OverLoadNode overLoadNode = new OverLoadNode(nodeMetadataCollection);
+
+    for (NodeMetadata nodeMetadata : nodeMetadataCollection) {
+      nodeMetadata.setTotalBytes(testHashMap.get(nodeMetadata.getNodeID()));
+    }
+
+    for (NodeMetadata nodeMetadata : nodeMetadataCollection2) {
+      nodeMetadata.setTotalBytes(testHashMap.get(nodeMetadata.getNodeID()));
+    }
+
+    for (int i = 0; i < 5; i++) {
+      overLoadNode.setMountCount(i);
+      overLoadNode.monitorOverLoad();
+    }
+
+    assertEquals(((NodeMetadata) nodeMetadataCollection.toArray()[3]).getOverLoadCount(), 31);
+
+    for (int i = 0; i < 15; i++) {
+      overLoadNode.setMountCount(i);
+      overLoadNode.monitorOverLoad();
+    }
+
+    assertEquals(((NodeMetadata) nodeMetadataCollection.toArray()[3]).getOverLoadCount(), 1023);
+
+    overLoadNode = new OverLoadNode(nodeMetadataCollection2);
+
+    for (int i = 0; i < 10; i++) {
+      overLoadNode.setMountCount(i);
+      overLoadNode.monitorOverLoad();
+    }
+
+    assertEquals(
+        ((NodeMetadata) nodeMetadataCollection.toArray()[3]).getOverLoadCount(),
+        ((NodeMetadata) nodeMetadataCollection2.toArray()[3]).getOverLoadCount());
+  }
+}
diff --git a/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/PartitionerTest.java b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/PartitionerTest.java
new file mode 100644
index 00000000..051532b8
--- /dev/null
+++ b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/PartitionerTest.java
@@ -0,0 +1,89 @@
+package org.astraea.partitioner.nodeLoadMetric;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.astraea.consumer.Builder;
+import org.astraea.consumer.Consumer;
+import org.astraea.consumer.Deserializer;
+import org.astraea.consumer.Header;
+import org.astraea.partitioner.partitionerFactory.LinkPartitioner;
+import org.astraea.producer.Producer;
+import org.astraea.producer.Serializer;
+import org.astraea.service.RequireBrokerCluster;
+import org.astraea.topic.TopicAdmin;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class PartitionerTest extends RequireBrokerCluster {
+  public final String brokerList = bootstrapServers();
+  TopicAdmin admin = TopicAdmin.of(bootstrapServers());
+  public final String topicName = "address";
+
+  public Properties initProConfig() {
+    Properties props = new Properties();
+    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, LinkPartitioner.class.getName());
+    props.put("jmx_servers", jmxServiceURL() + "@0");
+    return props;
+  }
+
+  @Test
+  public void testPartitioner() {
+    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(
+                initProConfig().entrySet().stream()
+                    .collect(Collectors.toMap(e -> e.getKey().toString(), Map.Entry::getValue)))
+            .build()) {
+      for (int i = 0; i < 20; i++) {
+        var metadata =
+            producer
+                .sender()
+                .topic(topicName)
+                .key(key)
+                .timestamp(timestamp)
+                .headers(List.of(header))
+                .run()
+                .toCompletableFuture()
+                .get();
+        Assertions.assertEquals(topicName, metadata.topic());
+        Assertions.assertEquals(timestamp, metadata.timestamp());
+      }
+    } catch (ExecutionException | InterruptedException e) {
+      e.printStackTrace();
+    }
+
+    try (var consumer =
+        Consumer.builder()
+            .brokers(bootstrapServers())
+            .offsetPolicy(Builder.OffsetPolicy.EARLIEST)
+            .topics(Set.of(topicName))
+            .keyDeserializer(Deserializer.STRING)
+            .build()) {
+      var records = consumer.poll(Duration.ofSeconds(10));
+      Assertions.assertEquals(20, records.size());
+      var record = records.iterator().next();
+      Assertions.assertEquals(topicName, record.topic());
+      Assertions.assertEquals("tainan", record.key());
+      Assertions.assertEquals(1, record.headers().size());
+      var actualHeader = record.headers().iterator().next();
+      Assertions.assertEquals(header.key(), actualHeader.key());
+      Assertions.assertArrayEquals(header.value(), actualHeader.value());
+    }
+  }
+}
diff --git a/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/SmoothPartitionerFactoryTest.java b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/SmoothPartitionerFactoryTest.java
new file mode 100644
index 00000000..7910381a
--- /dev/null
+++ b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/SmoothPartitionerFactoryTest.java
@@ -0,0 +1,250 @@
+package org.astraea.partitioner.nodeLoadMetric;
+
+import static java.util.Arrays.asList;
+import static org.astraea.partitioner.partitionerFactory.LinkPartitioner.getFactory;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mockConstruction;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.apache.kafka.clients.producer.Partitioner;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.astraea.partitioner.partitionerFactory.LinkPartitioner;
+import org.astraea.producer.Serializer;
+import org.astraea.service.RequireBrokerCluster;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedConstruction;
+
+public class SmoothPartitionerFactoryTest extends RequireBrokerCluster {
+  private String jmxAddresses;
+
+  private static final Node[] NODES =
+      new Node[] {
+        new Node(0, "localhost", 99), new Node(1, "localhost", 100), new Node(12, "localhost", 101)
+      };
+  private static final String TOPIC = "test";
+  // Intentionally make the partition list not in partition order to test the edge cases.
+  private static final List<PartitionInfo> PARTITIONS =
+      asList(
+          new PartitionInfo(TOPIC, 1, null, NODES, NODES),
+          new PartitionInfo(TOPIC, 2, NODES[1], NODES, NODES),
+          new PartitionInfo(TOPIC, 0, NODES[0], NODES, NODES));
+
+  @BeforeEach
+  void reset() {
+    jmxAddresses = "0.0.0.0,1.1.1.1,2.2.2.2";
+  }
+
+  public Properties initProConfig() {
+    Properties props = new Properties();
+    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers());
+    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
+    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
+    props.put(ProducerConfig.CLIENT_ID_CONFIG, "id1");
+    props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, LinkPartitioner.class.getName());
+    props.put("jmx_servers", jmxServiceURL() + "@0");
+    return props;
+  }
+
+  @Test
+  void testSingletonByProducer() {
+    try (MockedConstruction mocked =
+        mockConstruction(LinkPartitioner.ThreadSafeSmoothPartitioner.class)) {
+
+      var props = initProConfig();
+      // create multiples partitioners
+      var producers =
+          IntStream.range(0, 10)
+              .mapToObj(
+                  i ->
+                      org.astraea.producer.Producer.builder()
+                          .keySerializer(Serializer.STRING)
+                          .valueSerializer(Serializer.STRING)
+                          .configs(
+                              props.entrySet().stream()
+                                  .collect(
+                                      Collectors.toMap(
+                                          e -> e.getKey().toString(), Map.Entry::getValue)))
+                          .build())
+              .collect(Collectors.toList());
+
+      Assertions.assertEquals(getFactory().getSmoothPartitionerMap().size(), 1);
+      Partitioner partitioner = getFactory().getSmoothPartitionerMap().get(props);
+
+      // ThreadSafePartitioner is created only once
+      verify(partitioner).configure(any());
+      verify(partitioner, never()).partition(anyString(), any(), any(), any(), any(), any());
+      verify(partitioner, never()).close();
+
+      producers.forEach(org.astraea.producer.Producer::close);
+      // ThreadSafePartitioner is closed only once
+      verify(partitioner).configure(any());
+      verify(partitioner, never()).partition(anyString(), any(), any(), any(), any(), any());
+      verify(partitioner).close();
+    }
+  }
+
+  @Test
+  void testThrowException() {
+    var configs =
+        Map.of(
+            ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:11111", "jmx_server", jmxAddresses);
+
+    LinkPartitioner partitioner = new LinkPartitioner();
+    Assertions.assertThrows(RuntimeException.class, () -> partitioner.configure(configs))
+        .getMessage();
+  }
+
+  @Test
+  void testSingleton() throws InterruptedException {
+
+    var configs =
+        Map.of(
+            ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
+            "localhost:11111",
+            "jmx_servers",
+            jmxAddresses);
+
+    var executor = Executors.newFixedThreadPool(10);
+    var partitioners = new ArrayList<LinkPartitioner>();
+    Partitioner currentSmoothPartitioner = null;
+
+    // create partitions by multi-threads
+    IntStream.range(0, 10)
+        .forEach(
+            i -> {
+              executor.execute(
+                  () -> {
+                    try (MockedConstruction mocked =
+                        mockConstruction(LinkPartitioner.ThreadSafeSmoothPartitioner.class)) {
+                      var partitioner = new LinkPartitioner();
+                      partitioner.configure(configs);
+                      synchronized (partitioners) {
+                        partitioners.add(partitioner);
+                      }
+                    }
+                  });
+            });
+    executor.shutdown();
+    Assertions.assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS));
+
+    for (LinkPartitioner partitioner : partitioners) {
+      if (currentSmoothPartitioner == null) {
+        currentSmoothPartitioner = partitioner.getPartitioner();
+      } else {
+        Assertions.assertEquals(currentSmoothPartitioner, partitioner.getPartitioner());
+      }
+    }
+
+    Partitioner partitioner10 = getFactory().getSmoothPartitionerMap().get(configs);
+    verify(partitioner10).configure(any());
+    verify(partitioner10, never()).partition(anyString(), any(), any(), any(), any(), any());
+    verify(partitioner10, never()).close();
+
+    final Cluster cluster =
+        new Cluster(
+            "clusterId",
+            asList(NODES),
+            PARTITIONS,
+            Collections.<String>emptySet(),
+            Collections.<String>emptySet());
+
+    // try to call partition
+    partitioners.get(0).partition("test", null, null, null, null, cluster);
+    partitioners.get(1).partition("test", null, null, null, null, cluster);
+    partitioners.get(2).partition("test", null, null, null, null, cluster);
+    verify(partitioner10).configure(any());
+    verify(partitioner10, times(3)).partition(anyString(), any(), any(), any(), any(), any());
+    verify(partitioner10, never()).close();
+
+    // ThreadSafePartitioner is not closed if not all PassToProducers are closed.
+    var executor2 = Executors.newFixedThreadPool(partitioners.size() - 1);
+
+    partitioners.subList(1, partitioners.size()).forEach(p -> executor2.execute(p::close));
+
+    executor2.shutdown();
+
+    Assertions.assertTrue(executor2.awaitTermination(30, TimeUnit.SECONDS));
+    verify(partitioner10).configure(any());
+    verify(partitioner10, times(3)).partition(anyString(), any(), any(), any(), any(), any());
+    verify(partitioner10, never()).close();
+
+    // ok, all are closed
+    partitioners.get(0).close();
+    verify(partitioner10).configure(any());
+    verify(partitioner10, times(3)).partition(anyString(), any(), any(), any(), any(), any());
+    verify(partitioner10).close();
+    Assertions.assertNull(getFactory().getSmoothPartitionerMap().get(configs));
+
+    try (MockedConstruction mocked =
+        mockConstruction(LinkPartitioner.ThreadSafeSmoothPartitioner.class)) {
+      // let us create it again
+      var partitioner = new LinkPartitioner();
+      partitioner.configure(configs);
+      partitioner10 = getFactory().getSmoothPartitionerMap().get(configs);
+      verify(partitioner10).configure(any());
+      verify(partitioner10, never()).partition(anyString(), any(), any(), any(), any(), any());
+      verify(partitioner10, never()).close();
+
+      // close it
+      partitioner.close();
+      verify(partitioner10).configure(any());
+      verify(partitioner10, never()).partition(anyString(), any(), any(), any(), any(), any());
+      verify(partitioner10).close();
+    }
+
+    try (MockedConstruction mocked =
+        mockConstruction(LinkPartitioner.ThreadSafeSmoothPartitioner.class)) {
+      // create two ThreadSafePartitioner with two configs
+
+      var partitioner0 = new LinkPartitioner();
+      var props0 =
+          Map.of(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "value0", "jmx_servers", jmxAddresses);
+      partitioner0.configure(props0);
+
+      var smoothPartitioner0 = getFactory().getSmoothPartitionerMap().get(props0);
+
+      verify(smoothPartitioner0, times(1)).configure(any());
+      verify(smoothPartitioner0, never()).partition(anyString(), any(), any(), any(), any(), any());
+      verify(smoothPartitioner0, never()).close();
+
+      var partitioner1 = new LinkPartitioner();
+      var props1 =
+          Map.of(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "value1", "jmx_servers", jmxAddresses);
+      partitioner1.configure(props1);
+      var smoothPartitioner1 = getFactory().getSmoothPartitionerMap().get(props1);
+      Assertions.assertNotEquals(smoothPartitioner0, smoothPartitioner1);
+
+      verify(smoothPartitioner1, times(1)).configure(any());
+      verify(smoothPartitioner1, never()).partition(anyString(), any(), any(), any(), any(), any());
+      verify(smoothPartitioner1, never()).close();
+
+      // close them
+      partitioner0.close();
+      verify(smoothPartitioner0).configure(any());
+      verify(smoothPartitioner0, never()).partition(anyString(), any(), any(), any(), any(), any());
+      verify(smoothPartitioner0).close();
+      partitioner1.close();
+      verify(smoothPartitioner1).configure(any());
+      verify(smoothPartitioner1, never()).partition(anyString(), any(), any(), any(), any(), any());
+      verify(smoothPartitioner1).close();
+    }
+  }
+}
diff --git a/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/TestBrokersWeight.java b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/TestBrokersWeight.java
new file mode 100644
index 00000000..7eb86e31
--- /dev/null
+++ b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/TestBrokersWeight.java
@@ -0,0 +1,33 @@
+package org.astraea.partitioner.nodeLoadMetric;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.HashMap;
+import org.junit.jupiter.api.Test;
+
+public class TestBrokersWeight {
+  LoadPoisson loadPoisson = mock(LoadPoisson.class);
+
+  @Test
+  public void testSetBrokerHashMap() {
+    HashMap<String, Double> poissonMap = new HashMap<>();
+    poissonMap.put("0", 0.5);
+    poissonMap.put("1", 0.8);
+    poissonMap.put("2", 0.3);
+
+    when(loadPoisson.setAllPoisson()).thenReturn(poissonMap);
+
+    BrokersWeight brokersWeight = new BrokersWeight(loadPoisson);
+
+    brokersWeight.setBrokerHashMap();
+
+    assertEquals(brokersWeight.getBrokerHashMap().get("0")[0], 10);
+    assertEquals(brokersWeight.getBrokerHashMap().get("1")[0], 3);
+
+    brokersWeight.setBrokerHashMapValue("0", 8);
+    brokersWeight.setBrokerHashMap();
+    assertEquals(brokersWeight.getBrokerHashMap().get("0")[1], 8);
+  }
+}
diff --git a/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/TestLoadPoisson.java b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/TestLoadPoisson.java
new file mode 100644
index 00000000..d2c86cb0
--- /dev/null
+++ b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/TestLoadPoisson.java
@@ -0,0 +1,56 @@
+package org.astraea.partitioner.nodeLoadMetric;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.HashMap;
+import org.junit.jupiter.api.Test;
+
+public class TestLoadPoisson {
+  public NodeLoadClient nodeLoadClient;
+
+  @Test
+  public void testFactorial() {
+    nodeLoadClient = mock(NodeLoadClient.class);
+    LoadPoisson loadPoisson = new LoadPoisson(nodeLoadClient);
+
+    assertEquals(loadPoisson.factorial(3), 6);
+    assertEquals(loadPoisson.factorial(5), 120);
+  }
+
+  @Test
+  public void testDoPoisson() {
+    nodeLoadClient = mock(NodeLoadClient.class);
+    LoadPoisson loadPoisson = new LoadPoisson(nodeLoadClient);
+
+    assertEquals(loadPoisson.doPoisson(10, 15), 0.9512595966960214);
+    assertEquals(loadPoisson.doPoisson(5, 5), 0.6159606548330632);
+    assertEquals(loadPoisson.doPoisson(5, 8), 0.9319063652781515);
+    assertEquals(loadPoisson.doPoisson(5, 2), 0.12465201948308113);
+  }
+
+  @Test
+  public void testSetAllPoisson() {
+    HashMap<String, Integer> testNodesLoadCount = new HashMap<>();
+    testNodesLoadCount.put("0", 10);
+    testNodesLoadCount.put("1", 31);
+    testNodesLoadCount.put("2", 255);
+
+    HashMap<String, Double> poissonMap;
+    HashMap<String, Double> testPoissonMap = new HashMap<>();
+    testPoissonMap.put("0", 0.12465201948308113);
+    testPoissonMap.put("1", 0.6159606548330632);
+    testPoissonMap.put("2", 0.9319063652781515);
+
+    NodeLoadClient nodeLoadClient = mock(NodeLoadClient.class);
+    when(nodeLoadClient.getAllOverLoadCount()).thenReturn(testNodesLoadCount);
+    when(nodeLoadClient.getAvgLoadCount()).thenReturn(5);
+
+    LoadPoisson loadPoisson = new LoadPoisson(nodeLoadClient);
+
+    poissonMap = loadPoisson.setAllPoisson();
+
+    assertEquals(poissonMap, testPoissonMap);
+  }
+}