Repository: skiptests/astraea
Pull request #138 (2021-12-03 18:34:29)

Integrate BeanColloctor into Partitioner
繼續 #113 的完善
將BeanColloctor整合進了Partitioner中。
簡化了使用者配置Jmx addresses的行為，不再需要將NodeID與Jmx addresses一同配置。
Code更加乾淨了。

wycccccc: 還有一些小問題需要請教 #137
一些程式中的`print()`會在bug解決後刪除
NodeLoadClient還會再補充一些測試。
wycccccc: #137 
> 另一個角度是說 這個問題先緩一緩 先把重構的部份處理好再回來

這個問題應該緩不了，因為我們的測試環境就是一個host上有三個node。算是上述提到的半個情景，我有嘗試解決，現在可以支持同一個host上有多個node的情況。
wycccccc: 緊接着 #112 進行討論
我試圖創建一個`Map<producer,partitioner>`來將producer與partitoner進行匹配方便控制partitioner中的state。
我盡力想通過producer調出partitioner，或是通過partitioner匹配到producer。但似乎都做不到。
很可惜現在我們的使用者將需要多配置一個參數producerID
每一個producerID都需要唯一，幫助進行partitioner與producer的匹配。

使用者目前實現dependency的方法為。
```
var dependencyClient = new DependencyClient();
dependencyClient.initializeDependency(props);
dependencyClient.beginDependency(props);
producer.send();
dependencyClient.finishDependency(props);
```
wycccccc: 如果producerID的做法沒有問題的話，會有以下一個問題
現在的producerID為了程序正常運轉是必須需要設置的，但是實際上只有當用dependency的時候才需要這個參數。
為了讓使用者更自由確定是否設置producerID，還需要加一套producerID的防呆機制。
我先將這個問題記錄在這裡，如果確定producerID是可行的未來方向。我再開始處理這個問題。
chia7712: > 如果producerID的做法沒有問題的話，會有以下一個問題
現在的producerID為了程序正常運轉是必須需要設置的，但是實際上只有當用dependency的時候才需要這個參數。
為了讓使用者更自由確定是否設置producerID，還需要加一套producerID的防呆機制。
我先將這個問題記錄在這裡，如果確定producerID是可行的未來方向。我再開始處理這個問題。

麻煩開另一個議題來追蹤，這應該跟這隻PR的關係不大
wycccccc: 上述的建議都已修改，現在代碼應該精煉了不少，再麻煩學長review。
wycccccc: 我提議能夠先將此PR meger，因為我已基於這支PR完成了partitoner v1.0.0,添加了不少新東西。再在此push顯得過於臃腫，不太適合繼續在此幹活了。

Patch:
diff --git a/app/src/main/java/org/astraea/Utils.java b/app/src/main/java/org/astraea/Utils.java
index eaf3c062..cdd34473 100644
--- a/app/src/main/java/org/astraea/Utils.java
+++ b/app/src/main/java/org/astraea/Utils.java
@@ -94,6 +94,23 @@ public final class Utils {
     throw new RuntimeException("Timeout to wait procedure");
   }
 
+  /**
+   * Get the field of the object.
+   *
+   * @param object reflected object.
+   * @param fieldName reflected field name.
+   * @return Required field.
+   */
+  public static Object requireField(Object object, String fieldName) {
+    try {
+      var field = object.getClass().getDeclaredField(fieldName);
+      field.setAccessible(true);
+      return field.get(object);
+    } catch (NoSuchFieldException | IllegalAccessException e) {
+      throw new IllegalArgumentException(e);
+    }
+  }
+
   public static int requirePositive(int value) {
     if (value <= 0)
       throw new IllegalArgumentException("the value: " + value + " must be bigger than zero");
diff --git a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/BrokersWeight.java b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/BrokersWeight.java
deleted file mode 100644
index e9d779ef..00000000
--- a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/BrokersWeight.java
+++ /dev/null
@@ -1,55 +0,0 @@
-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
index 511cdaa8..b94c6723 100644
--- a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/LoadPoisson.java
+++ b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/LoadPoisson.java
@@ -3,40 +3,41 @@ package org.astraea.partitioner.nodeLoadMetric;
 import java.util.HashMap;
 import java.util.Map;
 
+/** do poisson for node's load situation */
 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));
-    }
+  public HashMap<Integer, Double> allPoisson(Map<Integer, Integer> overLoadCount) {
+    var poissonMap = new HashMap<Integer, Double>();
+    var lambda = avgLoadCount(overLoadCount);
+    overLoadCount.forEach((nodeID, count) -> poissonMap.put(nodeID, doPoisson(lambda, count)));
     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);
+  // visible for test
+  double doPoisson(int lambda, int x) {
+    var Probability = 0.0;
+    var ans = 0.0;
+    var i = 0;
+    while (i <= x) {
+      var j = Math.pow(lambda, i);
+      var e = Math.exp(-lambda);
+      var h = factorial(i);
       Probability = (j * e) / h;
       ans += Probability;
+      i++;
     }
-
     return ans;
   }
 
-  public long factorial(long number) {
+  // visible for test
+  long factorial(long number) {
     if (number <= 1) return 1;
     else return number * factorial(number - 1);
   }
+
+  private int avgLoadCount(Map<Integer, Integer> overLoadCount) {
+    var avgLoadCount =
+        overLoadCount.values().stream().mapToDouble(Integer::doubleValue).average().orElse(0);
+    return (int) avgLoadCount;
+  }
 }
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 5afe5c2c..03542570 100644
--- a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClient.java
+++ b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClient.java
@@ -1,89 +1,237 @@
 package org.astraea.partitioner.nodeLoadMetric;
 
+import static java.lang.Double.sum;
+
 import java.io.IOException;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
 import java.util.ArrayList;
-import java.util.Collection;
+import java.util.Comparator;
 import java.util.HashMap;
-import java.util.concurrent.TimeUnit;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.Node;
 import org.astraea.Utils;
-import org.astraea.concurrent.ThreadPool;
+import org.astraea.metrics.collector.Receiver;
+import org.astraea.metrics.kafka.KafkaMetrics;
 
-public class NodeLoadClient implements ThreadPool.Executor {
+/**
+ * this clas is responsible for obtaining jmx metrics from BeanCollector and calculating the
+ * overload status of each node through them.
+ */
+public class NodeLoadClient {
 
-  private final OverLoadNode overLoadNode;
-  private final Collection<NodeMetadata> nodeMetadataCollection = new ArrayList<>();
+  private final List<Receiver> receiverList;
 
-  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);
-  }
+  private Map<Integer, Map<String, Receiver>> eachNodeIDMetrics;
 
-  public NodeMetrics createNodeMetrics(String key, String value) throws IOException {
-    return new NodeMetrics(key, value);
-  }
+  private final Map<String, Integer> currentJmxAddresses;
 
-  /** 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;
+  private static final ReceiverFactory RECEIVER_FACTORY = new ReceiverFactory();
+
+  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) throws IOException {
+    this.receiverList = RECEIVER_FACTORY.receiversList(jmxAddresses);
+    receiverList.forEach(receiver -> Utils.waitFor(() -> receiver.current().size() > 0));
+    currentJmxAddresses = jmxAddresses;
   }
 
-  @Override
-  public void close() {
-    for (NodeMetadata nodeMetadata : nodeMetadataCollection) {
-      NodeMetrics nodeMetrics = nodeMetadata.getNodeMetrics();
-      Utils.close(nodeMetrics.getKafkaMetricClient());
+  /**
+   * @param cluster the cluster from the partitioner
+   * @return each node overLoad count in preset time
+   */
+  public Map<Integer, Integer> nodesOverLoad(Cluster cluster) throws UnknownHostException {
+    var nodeIDReceiver = new HashMap<Integer, List<Receiver>>();
+    var addresses =
+        cluster.nodes().stream()
+            .collect(Collectors.toMap(Node::id, node -> Map.entry(node.host(), node.port())));
+
+    for (Map.Entry<Integer, Map.Entry<String, Integer>> hostPort : addresses.entrySet()) {
+      List<Receiver> matchingNode;
+      var host = ipAddress(hostPort);
+      matchingNode =
+          receiverList.stream()
+              .filter(
+                  receiver ->
+                      (Objects.equals(nodeIDReceiver.size(), 0)
+                              || (nodeIDReceiver.values().stream()
+                                      .map(
+                                          n ->
+                                              Objects.equals(
+                                                  n.stream().findAny().get().host(), host))
+                                      .anyMatch(n -> n.equals(true))
+                                  && nodeIDReceiver.values().stream()
+                                      .map(
+                                          n ->
+                                              !Objects.equals(
+                                                  n.stream().findAny().get().port(),
+                                                  hostPort.getValue().getValue()))
+                                      .anyMatch(n -> n.equals(true))))
+                          && Objects.equals(receiver.host(), host))
+              .collect(Collectors.toList());
+
+      if (matchingNode.size() > 0) {
+        if (!nodeIDReceiver.containsKey(hostPort.getKey()))
+          nodeIDReceiver.put(hostPort.getKey(), matchingNode);
+        else matchingNode.forEach(node -> nodeIDReceiver.get(hostPort.getKey()).add(node));
+      }
     }
-  }
 
-  public synchronized HashMap<String, Integer> getAllOverLoadCount() {
-    HashMap<String, Integer> overLoadCount = new HashMap<>();
-    for (NodeMetadata nodeMetadata : nodeMetadataCollection) {
-      overLoadCount.put(nodeMetadata.getNodeID(), nodeMetadata.getOverLoadCount());
+    eachNodeIDMetrics = metricsNameForReceiver(nodeIDReceiver);
+    var eachBrokerMsgPerSec = brokersMsgPerSec();
+    var overLoadCount = new HashMap<Integer, Integer>();
+    eachBrokerMsgPerSec.keySet().forEach(e -> overLoadCount.put(e, 0));
+    var i = 0;
+    var minRecordSec =
+        eachBrokerMsgPerSec.values().stream()
+            .map(n -> n.size())
+            .min(Comparator.comparing(Integer::intValue))
+            .get();
+    while (i < minRecordSec) {
+      var avg = avgMsgSec(i, eachBrokerMsgPerSec);
+      var eachMsg = brokersMsgSec(i, eachBrokerMsgPerSec);
+      var standardDeviation = standardDeviationImperative(eachMsg, avg);
+      eachMsg.forEach(
+          (nodeID, msg) -> {
+            if (msg > (avg + standardDeviation)) {
+              overLoadCount.put(nodeID, overLoadCount.get(nodeID) + 1);
+            }
+          });
+      i++;
     }
     return overLoadCount;
   }
 
-  public synchronized int getAvgLoadCount() {
-    double avgLoadCount = 0;
-    for (NodeMetadata nodeMetadata : nodeMetadataCollection) {
-      avgLoadCount += getBinOneCount(nodeMetadata.getOverLoadCount());
+  /** @return data transfer per second per node */
+  private Map<Integer, List<Double>> brokersMsgPerSec() {
+    var eachMsg = new HashMap<Integer, List<Double>>();
+
+    for (Map.Entry<Integer, Map<String, Receiver>> entry : eachNodeIDMetrics.entrySet()) {
+      List<Double> sumList = new ArrayList<>();
+      var outMsg = entry.getValue().get(KafkaMetrics.BrokerTopic.BytesOutPerSec.toString());
+      var inMsg = entry.getValue().get(KafkaMetrics.BrokerTopic.BytesInPerSec.toString());
+      var outMsgBean = outMsg.current();
+      var inMsgBean = inMsg.current();
+      IntStream.range(0, Math.min(outMsgBean.size(), inMsgBean.size()))
+          .forEach(
+              i ->
+                  sumList.add(
+                      sum(
+                          Double.parseDouble(
+                              outMsgBean
+                                  .get(i)
+                                  .beanObject()
+                                  .getAttributes()
+                                  .get("MeanRate")
+                                  .toString()),
+                          Double.parseDouble(
+                              inMsgBean
+                                  .get(i)
+                                  .beanObject()
+                                  .getAttributes()
+                                  .get("MeanRate")
+                                  .toString()))));
+      eachMsg.put(entry.getKey(), sumList);
     }
-    return nodeMetadataCollection.size() > 0
-        ? (int) avgLoadCount / nodeMetadataCollection.size()
-        : 0;
+    return eachMsg;
   }
 
-  /** 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 List<Double> avgBrokersMsgPerSec(Map<Integer, List<Double>> eachMsg) {
+    return IntStream.range(0, eachMsg.values().stream().findFirst().get().size())
+        .mapToDouble(i -> eachMsg.values().stream().map(s -> s.get(i)).reduce(0.0, Double::sum))
+        .map(sum -> sum / eachMsg.size())
+        .boxed()
+        .collect(Collectors.toList());
+  }
+
+  public double standardDeviationImperative(
+      Map<Integer, Double> eachMsg, double avgBrokersMsgPerSec) {
+    var variance = new AtomicReference<>(0.0);
+    eachMsg.forEach(
+        (nodeID, msg) ->
+            variance.updateAndGet(
+                v -> v + (msg - avgBrokersMsgPerSec) * (msg - avgBrokersMsgPerSec)));
+    return Math.sqrt(variance.get() / eachMsg.size());
+  }
+
+  /**
+   * @param nodeIDReceiver nodes metrics that exists in topic
+   * @return the name and corresponding data of the metrics obtained by each node
+   */
+  public Map<Integer, Map<String, Receiver>> metricsNameForReceiver(
+      HashMap<Integer, List<Receiver>> nodeIDReceiver) {
+
+    var result = new HashMap<Integer, Map<String, Receiver>>();
+
+    nodeIDReceiver.forEach(
+        (nodeID, receivers) -> {
+          result.put(
+              nodeID,
+              receivers.stream()
+                  .map(
+                      receiver ->
+                          receiver.current().stream()
+                              .map(
+                                  bean ->
+                                      bean.beanObject().getProperties().values().stream()
+                                          .findAny()
+                                          .get())
+                              .findAny()
+                              .get())
+                  .collect(Collectors.toList())
+                  .stream()
+                  .collect(
+                      Collectors.toMap(
+                          Function.identity(),
+                          name ->
+                              receivers.stream()
+                                  .filter(
+                                      mbean ->
+                                          mbean.current().stream()
+                                              .findAny()
+                                              .get()
+                                              .beanObject()
+                                              .getProperties()
+                                              .values()
+                                              .stream()
+                                              .findAny()
+                                              .get()
+                                              .equals(name))
+                                  .findAny()
+                                  .get())));
+        });
+    return result;
   }
 
-  public void refreshNodesMetrics() {
-    for (NodeMetadata nodeMetadata : nodeMetadataCollection) {
-      NodeMetrics nodeMetrics = nodeMetadata.getNodeMetrics();
-      nodeMetrics.refreshMetrics();
-      nodeMetadata.setTotalBytes(nodeMetrics.totalBytesPerSec());
+  private Map<Integer, Double> brokersMsgSec(int index, Map<Integer, List<Double>> eachMsg) {
+    return eachMsg.entrySet().stream()
+        .collect(Collectors.toMap(Map.Entry::getKey, entry -> entry.getValue().get(index)));
+  }
+
+  private double avgMsgSec(int index, Map<Integer, List<Double>> eachMsg) {
+    return avgBrokersMsgPerSec(eachMsg).get(index);
+  }
+
+  private String ipAddress(Map.Entry<Integer, Map.Entry<String, Integer>> hostPort)
+      throws UnknownHostException {
+    var host = "-1.-1.-1.-1";
+    if (!hostPort.getValue().getKey().matches(regex)) {
+      host = InetAddress.getByName(hostPort.getValue().getKey()).toString().split("/")[1];
+    } else {
+      host = hostPort.getValue().getKey();
     }
+    return host;
+  }
+
+  public void close() {
+    RECEIVER_FACTORY.close(currentJmxAddresses);
   }
 }
diff --git a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeMetadata.java b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeMetadata.java
deleted file mode 100644
index a86a7a0a..00000000
--- a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeMetadata.java
+++ /dev/null
@@ -1,40 +0,0 @@
-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
deleted file mode 100644
index df3c1e3b..00000000
--- a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeMetrics.java
+++ /dev/null
@@ -1,62 +0,0 @@
-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 = MBeanClient.of(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
deleted file mode 100644
index 7552f78f..00000000
--- a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/OverLoadNode.java
+++ /dev/null
@@ -1,100 +0,0 @@
-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/nodeLoadMetric/ReceiverFactory.java b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/ReceiverFactory.java
new file mode 100644
index 00000000..ac309631
--- /dev/null
+++ b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/ReceiverFactory.java
@@ -0,0 +1,91 @@
+package org.astraea.partitioner.nodeLoadMetric;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.astraea.metrics.collector.BeanCollector;
+import org.astraea.metrics.collector.Receiver;
+import org.astraea.metrics.jmx.MBeanClient;
+import org.astraea.metrics.kafka.KafkaMetrics;
+
+/** create a single BeanCollector to avoid waste of resources */
+public class ReceiverFactory {
+  private final Object lock = new Object();
+  private final List<Receiver> receiversList = new ArrayList<>();
+  private final BeanCollector beanCollector;
+  private final Map<String, Integer> count = new HashMap<>();
+
+  /** create a factory with specific comparator. * */
+  public ReceiverFactory() {
+    beanCollector =
+        BeanCollector.builder()
+            .interval(Duration.ofSeconds(1))
+            .numberOfObjectsPerNode(10)
+            .clientCreator(MBeanClient::jndi)
+            .build();
+  }
+
+  /**
+   * @return create new receivers,if the receiver of the node does not exist.Then return
+   *     receiversList for all node's Receivers.Otherwise, it returns the existent receiversList.
+   */
+  public List<Receiver> receiversList(Map<String, Integer> jmxAddresses) {
+    synchronized (lock) {
+      jmxAddresses.forEach(
+          (host, port) -> {
+            if (!count.containsKey(nodeKey(host, port))) {
+              receiversList.add(
+                  beanCollector
+                      .register()
+                      .host(host)
+                      .port(port)
+                      .metricsGetter(KafkaMetrics.BrokerTopic.BytesInPerSec::fetch)
+                      .build());
+              receiversList.add(
+                  beanCollector
+                      .register()
+                      .host(host)
+                      .port(port)
+                      .metricsGetter(KafkaMetrics.BrokerTopic.BytesOutPerSec::fetch)
+                      .build());
+              count.put(nodeKey(host, port), 1);
+            } else {
+              count.put(nodeKey(host, port), count.get(nodeKey(host, port)) + 1);
+            }
+          });
+    }
+    return receiversList;
+  }
+
+  public void close(Map<String, Integer> jmxAddresses) {
+    synchronized (lock) {
+      jmxAddresses.forEach(
+          (host, port) -> {
+            if (count.containsKey(nodeKey(host, port))) {
+              if (count.get(nodeKey(host, port)) == 1) {
+                receiversList.stream()
+                    .filter(receiver -> receiver.host().equals(host) && receiver.port() == port)
+                    .forEach(Receiver::close);
+                count.remove(nodeKey(host, port));
+              } else {
+                count.put(nodeKey(host, port), count.get(nodeKey(host, port)) - 1);
+              }
+            }
+          });
+    }
+  }
+
+  private String nodeKey(String host, Integer port) {
+    return host + ":" + port;
+  }
+
+  // visible for testing
+  int factoryCount(Map<String, Integer> jmxAddresses) {
+    var testCount = new AtomicInteger(-1);
+    jmxAddresses.forEach((host, port) -> testCount.set(count.getOrDefault(nodeKey(host, port), 0)));
+    return testCount.get();
+  }
+}
diff --git a/app/src/main/java/org/astraea/partitioner/partitionerFactory/LinkPartitioner.java b/app/src/main/java/org/astraea/partitioner/partitionerFactory/LinkPartitioner.java
deleted file mode 100644
index d3b40786..00000000
--- a/app/src/main/java/org/astraea/partitioner/partitionerFactory/LinkPartitioner.java
+++ /dev/null
@@ -1,130 +0,0 @@
-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;
-        }
-        currentBrokerHashMap.put(
-            item.getKey(), new int[] {item.getValue()[0], item.getValue()[1] + item.getValue()[0]});
-      }
-      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
deleted file mode 100644
index ba3f5305..00000000
--- a/app/src/main/java/org/astraea/partitioner/partitionerFactory/SmoothPartitionerFactory.java
+++ /dev/null
@@ -1,103 +0,0 @@
-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/app/src/main/java/org/astraea/partitioner/smoothPartitioner/DependencyClient.java b/app/src/main/java/org/astraea/partitioner/smoothPartitioner/DependencyClient.java
new file mode 100644
index 00000000..777dfc7c
--- /dev/null
+++ b/app/src/main/java/org/astraea/partitioner/smoothPartitioner/DependencyClient.java
@@ -0,0 +1,6 @@
+package org.astraea.partitioner.smoothPartitioner;
+
+// TODO
+public interface DependencyClient {
+  void finishDependency();
+}
diff --git a/app/src/main/java/org/astraea/partitioner/smoothPartitioner/SmoothWeightPartitioner.java b/app/src/main/java/org/astraea/partitioner/smoothPartitioner/SmoothWeightPartitioner.java
new file mode 100644
index 00000000..4942327a
--- /dev/null
+++ b/app/src/main/java/org/astraea/partitioner/smoothPartitioner/SmoothWeightPartitioner.java
@@ -0,0 +1,229 @@
+package org.astraea.partitioner.smoothPartitioner;
+
+import java.io.IOException;
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Partitioner;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.KafkaException;
+import org.astraea.Utils;
+import org.astraea.partitioner.nodeLoadMetric.LoadPoisson;
+import org.astraea.partitioner.nodeLoadMetric.NodeLoadClient;
+
+/**
+ * Based on the jmx metrics obtained from Kafka, it records the load status of the node over a
+ * period of time. Predict the future status of each node through the poisson of the load status.
+ * Finally, the result of poisson is used as the weight to perform smooth weighted RoundRobin.
+ *
+ * <p>SmoothWeightPartitioner also allows users to make producers send dependency data.
+ *
+ * <pre>{@code
+ * KafkaProducer producer = new KafkaProducer(props);
+ *
+ * var dependencyControl = SmoothWeightPartitioner.beginDependency(producer);
+ * try{
+ *     producer.send();
+ *     producer.send();
+ *     producer.send();
+ * } finally{
+ *     dependencyControl.finishDependency();
+ * }
+ * }</pre>
+ */
+public class SmoothWeightPartitioner implements Partitioner, DependencyClient {
+
+  /**
+   * Record the current weight of each node according to Poisson calculation and the weight after
+   * partitioner calculation.
+   */
+  private Map<Integer, int[]> brokerHashMap = new HashMap<>();
+
+  private NodeLoadClient nodeLoadClient;
+  private static final DependencyManager dependencyManager = new DependencyManager();
+  /**
+   * This parameter only works in dependency mode.And it will be specified when the dependency is
+   * executed for the first time, and no changes will be made afterwards.
+   */
+  private int targetPartition;
+
+  @Override
+  public int partition(
+      String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
+    Map<Integer, Integer> overLoadCount;
+    var rand = new Random();
+    switch (dependencyManager.currentState) {
+      case Start_Dependency:
+        overLoadCount = overLoadCount(cluster);
+        Objects.requireNonNull(overLoadCount, "OverLoadCount should not be null.");
+        var minOverLoadNode =
+            overLoadCount.entrySet().stream().min(Map.Entry.comparingByValue()).get();
+        var minOverLoadPartition = cluster.partitionsForNode(minOverLoadNode.getKey());
+        targetPartition =
+            minOverLoadPartition.get(rand.nextInt(minOverLoadPartition.size())).partition();
+        dependencyManager.transitionTo(DependencyManager.State.IN_Dependency);
+        return targetPartition;
+      case IN_Dependency:
+        return targetPartition;
+      default:
+        overLoadCount = overLoadCount(cluster);
+        var loadPoisson = new LoadPoisson();
+
+        Objects.requireNonNull(overLoadCount, "OverLoadCount should not be null.");
+        brokerHashMap(loadPoisson.allPoisson(overLoadCount));
+        AtomicReference<Map.Entry<Integer, int[]>> maxWeightServer = new AtomicReference<>();
+        var allWeight = allNodesWeight();
+        var currentBrokerHashMap = brokerHashMap();
+        currentBrokerHashMap.forEach(
+            (nodeID, weight) -> {
+              if (maxWeightServer.get() == null
+                  || weight[1] > maxWeightServer.get().getValue()[1]) {
+                maxWeightServer.set(Map.entry(nodeID, weight));
+              }
+              currentBrokerHashMap.put(nodeID, new int[] {weight[0], weight[1] + weight[0]});
+            });
+        Objects.requireNonNull(maxWeightServer.get(), "MaxWeightServer should not be null.");
+        currentBrokerHashMap.put(
+            maxWeightServer.get().getKey(),
+            new int[] {
+              maxWeightServer.get().getValue()[0], maxWeightServer.get().getValue()[1] - allWeight
+            });
+        currentBrokerHashMap(currentBrokerHashMap);
+
+        ArrayList<Integer> partitionList = new ArrayList<>();
+        cluster
+            .partitionsForNode(maxWeightServer.get().getKey())
+            .forEach(partitionInfo -> partitionList.add(partitionInfo.partition()));
+
+        return partitionList.get(rand.nextInt(partitionList.size()));
+    }
+  }
+
+  @Override
+  public void close() {
+    nodeLoadClient.close();
+  }
+
+  @Override
+  public void configure(Map<String, ?> configs) {
+    try {
+      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);
+    } catch (IOException e) {
+      throw new RuntimeException();
+    }
+  }
+
+  /** Change the weight of the node according to the current Poisson. */
+  // visible for test
+  public synchronized void brokerHashMap(Map<Integer, Double> poissonMap) {
+    poissonMap.forEach(
+        (key, value) -> {
+          if (!brokerHashMap.containsKey(key)) {
+            brokerHashMap.put(key, new int[] {(int) ((1 - value) * 20), 0});
+          } else {
+            brokerHashMap.put(key, new int[] {(int) ((1 - value) * 20), brokerHashMap.get(key)[1]});
+          }
+        });
+  }
+
+  private synchronized int allNodesWeight() {
+    return brokerHashMap.values().stream().mapToInt(vs -> vs[0]).sum();
+  }
+
+  // visible for test
+  public synchronized Map<Integer, int[]> brokerHashMap() {
+    return brokerHashMap;
+  }
+
+  private synchronized void currentBrokerHashMap(Map<Integer, int[]> currentBrokerHashMap) {
+    brokerHashMap = currentBrokerHashMap;
+  }
+
+  // visible for test
+  public void brokerHashMapValue(Integer x, int y) {
+    brokerHashMap.put(x, new int[] {0, y});
+  }
+
+  public static synchronized DependencyClient beginDependency(KafkaProducer<?, ?> producer) {
+    dependencyManager.beginDependency();
+    return (DependencyClient) Utils.requireField(producer, "partitioner");
+  }
+
+  @Override
+  public synchronized void finishDependency() {
+    dependencyManager.finishDependency();
+  }
+
+  private Map<Integer, Integer> overLoadCount(Cluster cluster) {
+    try {
+      return nodeLoadClient.nodesOverLoad(cluster);
+    } catch (UnknownHostException e) {
+      throw new IllegalArgumentException(e);
+    }
+  }
+
+  /**
+   * Store the state related to the partitioner dependency and avoid wrong transitions between them.
+   */
+  private static class DependencyManager {
+    private volatile State currentState = State.UNINITIALIZED;
+
+    private synchronized void beginDependency() {
+      transitionTo(State.Start_Dependency);
+    }
+
+    private synchronized void finishDependency() {
+      transitionTo(State.UNINITIALIZED);
+    }
+
+    private enum State {
+      UNINITIALIZED,
+      Start_Dependency,
+      IN_Dependency,
+      FATAL_ERROR;
+
+      private boolean isTransitionValid(
+          DependencyManager.State source, DependencyManager.State target) {
+        switch (target) {
+          case UNINITIALIZED:
+            return source == Start_Dependency || source == IN_Dependency;
+          case Start_Dependency:
+            return source == UNINITIALIZED;
+          case IN_Dependency:
+            return source == Start_Dependency;
+          default:
+            return true;
+        }
+      }
+    }
+
+    private void transitionTo(State target) {
+      if (!currentState.isTransitionValid(currentState, target)) {
+        throw new KafkaException(
+            "Invalid transition attempted from state "
+                + currentState.name()
+                + " to state "
+                + target.name());
+      }
+      currentState = target;
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/producer/Builder.java b/app/src/main/java/org/astraea/producer/Builder.java
index df4fea0b..d7a68918 100644
--- a/app/src/main/java/org/astraea/producer/Builder.java
+++ b/app/src/main/java/org/astraea/producer/Builder.java
@@ -1,6 +1,10 @@
 package org.astraea.producer;
 
-import java.util.*;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import org.apache.kafka.clients.producer.KafkaProducer;
@@ -117,6 +121,11 @@ public class Builder<Key, Value> {
         kafkaProducer.flush();
       }
 
+      @Override
+      public KafkaProducer<Key, Value> kafkaProducer() {
+        return kafkaProducer;
+      }
+
       @Override
       public void close() {
         kafkaProducer.close();
diff --git a/app/src/main/java/org/astraea/producer/Producer.java b/app/src/main/java/org/astraea/producer/Producer.java
index fc0813e0..a4051338 100644
--- a/app/src/main/java/org/astraea/producer/Producer.java
+++ b/app/src/main/java/org/astraea/producer/Producer.java
@@ -1,6 +1,7 @@
 package org.astraea.producer;
 
 import java.util.Map;
+import org.apache.kafka.clients.producer.KafkaProducer;
 
 /** An interface for sending records. */
 public interface Producer<Key, Value> extends AutoCloseable {
@@ -11,6 +12,8 @@ public interface Producer<Key, Value> extends AutoCloseable {
 
   void close();
 
+  KafkaProducer<Key, Value> kafkaProducer();
+
   static Builder<byte[], byte[]> builder() {
     return new Builder<>();
   }


Test patch:
diff --git a/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricsTest.java b/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricsTest.java
index a64c9370..0a006df0 100644
--- a/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricsTest.java
+++ b/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricsTest.java
@@ -1,6 +1,8 @@
 package org.astraea.metrics.kafka;
 
-import static org.junit.jupiter.api.Assertions.*;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.condition.OS.LINUX;
 import static org.junit.jupiter.api.condition.OS.WINDOWS;
 
@@ -9,7 +11,7 @@ import java.lang.management.ManagementFactory;
 import java.util.Arrays;
 import java.util.Set;
 import java.util.stream.Collectors;
-import javax.management.*;
+import javax.management.MBeanServer;
 import javax.management.remote.JMXConnectorServer;
 import javax.management.remote.JMXConnectorServerFactory;
 import javax.management.remote.JMXServiceURL;
diff --git a/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/DependencyPartitionTest.java b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/DependencyPartitionTest.java
new file mode 100644
index 00000000..ea2c5fb2
--- /dev/null
+++ b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/DependencyPartitionTest.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.Consumer;
+import org.astraea.consumer.Deserializer;
+import org.astraea.consumer.Header;
+import org.astraea.partitioner.smoothPartitioner.SmoothWeightPartitioner;
+import org.astraea.producer.Producer;
+import org.astraea.service.RequireBrokerCluster;
+import org.astraea.topic.TopicAdmin;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class DependencyPartitionTest extends RequireBrokerCluster {
+  private final String brokerList = bootstrapServers();
+  TopicAdmin admin = TopicAdmin.of(bootstrapServers());
+  private final String topicName = "address";
+
+  private Properties initProConfig() {
+    Properties props = new Properties();
+    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
+    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, SmoothWeightPartitioner.class.getName());
+    props.put("jmx_servers", jmxServiceURL().getHost() + ":" + jmxServiceURL().getPort());
+    return props;
+  }
+
+  @Test
+  void testDependencyPartitioner() {
+    admin.creator().topic(topicName).numberOfPartitions(10).create();
+    var ps = initProConfig();
+    var timestamp = System.currentTimeMillis() + 10;
+    var header = Header.of("a", "b".getBytes());
+    ps.put("producerID", 2);
+    var props =
+        ps.entrySet().stream()
+            .collect(Collectors.toMap(e -> e.getKey().toString(), Map.Entry::getValue));
+    try (var producer = Producer.builder().configs(props).build()) {
+      var dependencyClient = SmoothWeightPartitioner.beginDependency(producer.kafkaProducer());
+      var targetPartition = 0;
+      var i = 0;
+      while (i < 20) {
+        var metadata =
+            producer
+                .sender()
+                .topic(topicName)
+                .timestamp(timestamp)
+                .headers(List.of(header))
+                .run()
+                .toCompletableFuture()
+                .get();
+        if (i > 0) Assertions.assertEquals(targetPartition, metadata.partition());
+        targetPartition = metadata.partition();
+        Assertions.assertEquals(topicName, metadata.topic());
+        Assertions.assertEquals(timestamp, metadata.timestamp());
+        i++;
+      }
+      dependencyClient.finishDependency();
+    } catch (ExecutionException | InterruptedException e) {
+      e.printStackTrace();
+    }
+    try (var consumer =
+        Consumer.builder()
+            .brokers(bootstrapServers())
+            .fromBeginning()
+            .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(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/LoadPoissonTest.java b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/LoadPoissonTest.java
new file mode 100644
index 00000000..a87672e5
--- /dev/null
+++ b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/LoadPoissonTest.java
@@ -0,0 +1,47 @@
+package org.astraea.partitioner.nodeLoadMetric;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.HashMap;
+import org.junit.jupiter.api.Test;
+
+public class LoadPoissonTest {
+
+  @Test
+  void testFactorial() {
+    LoadPoisson loadPoisson = new LoadPoisson();
+
+    assertEquals(loadPoisson.factorial(3), 6);
+    assertEquals(loadPoisson.factorial(5), 120);
+  }
+
+  @Test
+  void testDoPoisson() {
+    LoadPoisson loadPoisson = new LoadPoisson();
+
+    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
+  void testSetAllPoisson() {
+    HashMap<Integer, Integer> testNodesLoadCount = new HashMap<>();
+    testNodesLoadCount.put(0, 2);
+    testNodesLoadCount.put(1, 5);
+    testNodesLoadCount.put(2, 8);
+
+    HashMap<Integer, Double> poissonMap;
+    HashMap<Integer, Double> testPoissonMap = new HashMap<>();
+    testPoissonMap.put(0, 0.12465201948308113);
+    testPoissonMap.put(1, 0.6159606548330632);
+    testPoissonMap.put(2, 0.9319063652781515);
+
+    LoadPoisson loadPoisson = new LoadPoisson();
+
+    poissonMap = loadPoisson.allPoisson(testNodesLoadCount);
+
+    assertEquals(poissonMap, testPoissonMap);
+  }
+}
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 454f7835..85c8692c 100644
--- a/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClientTest.java
+++ b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClientTest.java
@@ -1,42 +1,71 @@
 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.io.IOException;
+import java.util.Arrays;
 import java.util.HashMap;
-import org.junit.jupiter.api.BeforeEach;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.astraea.partitioner.smoothPartitioner.SmoothWeightPartitioner;
+import org.astraea.service.RequireBrokerCluster;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+public class NodeLoadClientTest extends RequireBrokerCluster {
+  private final String brokerList = bootstrapServers();
+  private NodeLoadClient nodeLoadClient;
+  private String jmxAddress;
+  private Map props;
 
-public class NodeLoadClientTest {
-  private HashMap<String, String> jmxAddresses;
-  private Collection<NodeMetadata> nodeMetadataCollection;
+  @BeforeAll
+  void setUp() throws IOException {
+    props = initProConfig();
+    var map = new HashMap<String, Integer>();
+    map.put(jmxServiceURL().getHost(), jmxServiceURL().getPort());
+    nodeLoadClient = new NodeLoadClient(map);
+  }
 
-  @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");
+  @AfterAll
+  void tearDown() {
+    nodeLoadClient.close();
+  }
 
-    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);
+  Properties initProConfig() {
+    Properties props = new Properties();
+    jmxAddress = jmxServiceURL().getHost() + ":" + jmxServiceURL().getPort();
+    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
+    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
+    props.put(ProducerConfig.CLIENT_ID_CONFIG, "id1");
+    props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, SmoothWeightPartitioner.class.getName());
+    props.put("jmx_servers", jmxAddress);
+    return props;
+  }
 
-    nodeMetadataCollection.add(nodeMetadata0);
-    nodeMetadataCollection.add(nodeMetadata1);
-    nodeMetadataCollection.add(nodeMetadata2);
-    nodeMetadataCollection.add(nodeMetadata3);
+  @Test
+  void testAvgBrokersMsgPerSec() {
+    var testMap = new HashMap<Integer, List<Double>>();
+    testMap.put(0, Arrays.asList(20.0, 30.0, 40.0));
+    testMap.put(1, Arrays.asList(30.0, 20.0, 30.0));
+    testMap.put(2, Arrays.asList(40.0, 40.0, 20.0));
+    Assertions.assertEquals(
+        nodeLoadClient.avgBrokersMsgPerSec(testMap), Arrays.asList(30.0, 30.0, 30.0));
   }
 
   @Test
-  public void testGetBinOneCount() {
-    assertEquals(getBinOneCount(7), 3);
-    assertEquals(getBinOneCount(10), 2);
+  void testStandardDeviationImperative() {
+    var testMap = new HashMap<Integer, Double>();
+    testMap.put(0, 15.0);
+    testMap.put(1, 20.0);
+    testMap.put(2, 25.0);
+    testMap.put(3, 20.0);
+    testMap.put(4, 20.0);
+    Assertions.assertEquals(
+        nodeLoadClient.standardDeviationImperative(testMap, 20), 3.1622776601683795);
   }
 }
diff --git a/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/OverLoadNodeTest.java b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/OverLoadNodeTest.java
deleted file mode 100644
index c0edb247..00000000
--- a/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/OverLoadNodeTest.java
+++ /dev/null
@@ -1,115 +0,0 @@
-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
index b7062ea1..3869d56c 100644
--- a/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/PartitionerTest.java
+++ b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/PartitionerTest.java
@@ -1,6 +1,9 @@
 package org.astraea.partitioner.nodeLoadMetric;
 
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
 import java.time.Duration;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
@@ -12,7 +15,7 @@ import org.apache.kafka.common.serialization.ByteArraySerializer;
 import org.astraea.consumer.Consumer;
 import org.astraea.consumer.Deserializer;
 import org.astraea.consumer.Header;
-import org.astraea.partitioner.partitionerFactory.LinkPartitioner;
+import org.astraea.partitioner.smoothPartitioner.SmoothWeightPartitioner;
 import org.astraea.producer.Producer;
 import org.astraea.producer.Serializer;
 import org.astraea.service.RequireBrokerCluster;
@@ -21,24 +24,25 @@ import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 public class PartitionerTest extends RequireBrokerCluster {
-  public final String brokerList = bootstrapServers();
+  private final String brokerList = bootstrapServers();
   TopicAdmin admin = TopicAdmin.of(bootstrapServers());
-  public final String topicName = "address";
+  private final String topicName = "address";
 
-  public Properties initProConfig() {
+  private Properties initProConfig() {
     Properties props = new Properties();
     props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
+    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");
+    props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, SmoothWeightPartitioner.class.getName());
+    props.put("producerID", 1);
+    props.put("jmx_servers", jmxServiceURL().getHost() + ":" + jmxServiceURL().getPort());
     return props;
   }
 
   @Test
-  public void testPartitioner() {
+  void testPartitioner() {
     admin.creator().topic(topicName).numberOfPartitions(10).create();
-
     var key = "tainan";
     var timestamp = System.currentTimeMillis() + 10;
     var header = Header.of("a", "b".getBytes());
@@ -49,7 +53,8 @@ public class PartitionerTest extends RequireBrokerCluster {
                 initProConfig().entrySet().stream()
                     .collect(Collectors.toMap(e -> e.getKey().toString(), Map.Entry::getValue)))
             .build()) {
-      for (int i = 0; i < 20; i++) {
+      var i = 0;
+      while (i < 20) {
         var metadata =
             producer
                 .sender()
@@ -60,8 +65,9 @@ public class PartitionerTest extends RequireBrokerCluster {
                 .run()
                 .toCompletableFuture()
                 .get();
-        Assertions.assertEquals(topicName, metadata.topic());
-        Assertions.assertEquals(timestamp, metadata.timestamp());
+        assertEquals(topicName, metadata.topic());
+        assertEquals(timestamp, metadata.timestamp());
+        i++;
       }
     } catch (ExecutionException | InterruptedException e) {
       e.printStackTrace();
@@ -75,14 +81,33 @@ public class PartitionerTest extends RequireBrokerCluster {
             .keyDeserializer(Deserializer.STRING)
             .build()) {
       var records = consumer.poll(Duration.ofSeconds(10));
-      Assertions.assertEquals(20, records.size());
+      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());
+      assertEquals(topicName, record.topic());
+      assertEquals("tainan", record.key());
+      assertEquals(1, record.headers().size());
       var actualHeader = record.headers().iterator().next();
-      Assertions.assertEquals(header.key(), actualHeader.key());
+      assertEquals(header.key(), actualHeader.key());
       Assertions.assertArrayEquals(header.value(), actualHeader.value());
     }
   }
+
+  @Test
+  void testSetBrokerHashMap() {
+    var poissonMap = new HashMap<Integer, Double>();
+    poissonMap.put(0, 0.5);
+    poissonMap.put(1, 0.8);
+    poissonMap.put(2, 0.3);
+
+    var smoothWeightPartitioner = new SmoothWeightPartitioner();
+
+    smoothWeightPartitioner.brokerHashMap(poissonMap);
+
+    assertEquals(smoothWeightPartitioner.brokerHashMap().get(0)[0], 10);
+    assertEquals(smoothWeightPartitioner.brokerHashMap().get(1)[0], 3);
+
+    smoothWeightPartitioner.brokerHashMapValue(0, 8);
+    smoothWeightPartitioner.brokerHashMap(poissonMap);
+    assertEquals(smoothWeightPartitioner.brokerHashMap().get(0)[1], 8);
+  }
 }
diff --git a/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/ReceiverFactoryTest.java b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/ReceiverFactoryTest.java
new file mode 100644
index 00000000..e0519480
--- /dev/null
+++ b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/ReceiverFactoryTest.java
@@ -0,0 +1,31 @@
+package org.astraea.partitioner.nodeLoadMetric;
+
+import java.util.Map;
+import org.astraea.service.RequireBrokerCluster;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class ReceiverFactoryTest extends RequireBrokerCluster {
+  @Test
+  void testSingleton() {
+    var jmxAddress = Map.of(jmxServiceURL().getHost(), jmxServiceURL().getPort());
+    var FACTORY = new ReceiverFactory();
+    var bc1 = FACTORY.receiversList(jmxAddress);
+    var bc2 = FACTORY.receiversList(jmxAddress);
+    Assertions.assertEquals(bc1, bc2);
+    var bc3 = FACTORY.receiversList(jmxAddress);
+    var bc4 = FACTORY.receiversList(jmxAddress);
+    Assertions.assertEquals(FACTORY.factoryCount(jmxAddress), 4);
+    FACTORY.close(jmxAddress);
+    Assertions.assertEquals(FACTORY.factoryCount(jmxAddress), 3);
+    FACTORY.close(jmxAddress);
+    FACTORY.close(jmxAddress);
+    Assertions.assertEquals(FACTORY.factoryCount(jmxAddress), 1);
+    FACTORY.close(jmxAddress);
+    Assertions.assertEquals(FACTORY.factoryCount(jmxAddress), 0);
+    var bc5 = FACTORY.receiversList(jmxAddress);
+    Assertions.assertEquals(FACTORY.factoryCount(jmxAddress), 1);
+    FACTORY.close(jmxAddress);
+    Assertions.assertEquals(FACTORY.factoryCount(jmxAddress), 0);
+  }
+}
diff --git a/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/SmoothPartitionerFactoryTest.java b/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/SmoothPartitionerFactoryTest.java
deleted file mode 100644
index 7910381a..00000000
--- a/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/SmoothPartitionerFactoryTest.java
+++ /dev/null
@@ -1,250 +0,0 @@
-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
deleted file mode 100644
index 7eb86e31..00000000
--- a/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/TestBrokersWeight.java
+++ /dev/null
@@ -1,33 +0,0 @@
-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
deleted file mode 100644
index d2c86cb0..00000000
--- a/app/src/test/java/org/astraea/partitioner/nodeLoadMetric/TestLoadPoisson.java
+++ /dev/null
@@ -1,56 +0,0 @@
-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);
-  }
-}