Repository: skiptests/astraea
Pull request #256 (2022-02-24 12:15:39)

Introduce cost functions
related to #230

引入一個簡單的實作來開始cost function.

1. 現在只有先考慮throughput
2. 尚未添加測試，等討論一個階段後加入

chia7712: @chinghongfang 麻煩看一下這個概念實作，之後要由你接手~

Patch:
diff --git a/app/src/main/java/org/astraea/metrics/collector/BeanCollector.java b/app/src/main/java/org/astraea/metrics/collector/BeanCollector.java
index cf5ed419..540b47d3 100644
--- a/app/src/main/java/org/astraea/metrics/collector/BeanCollector.java
+++ b/app/src/main/java/org/astraea/metrics/collector/BeanCollector.java
@@ -1,12 +1,14 @@
 package org.astraea.metrics.collector;
 
 import java.time.Duration;
-import java.util.HashMap;
+import java.util.ArrayList;
+import java.util.Collection;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ConcurrentSkipListMap;
 import java.util.concurrent.locks.Lock;
@@ -70,7 +72,7 @@ public class BeanCollector {
     return new Register() {
       private String host;
       private int port = -1;
-      private Function<MBeanClient, HasBeanObject> getter;
+      private final List<Function<MBeanClient, HasBeanObject>> getters = new ArrayList<>();
 
       @Override
       public Register host(String host) {
@@ -85,8 +87,8 @@ public class BeanCollector {
       }
 
       @Override
-      public Register metricsGetter(Function<MBeanClient, HasBeanObject> getter) {
-        this.getter = Objects.requireNonNull(getter);
+      public Register metricsGetters(Collection<Function<MBeanClient, HasBeanObject>> getters) {
+        this.getters.addAll(Objects.requireNonNull(getters));
         return this;
       }
 
@@ -96,7 +98,7 @@ public class BeanCollector {
         var node = nodes.computeIfAbsent(nodeKey, ignored -> new Node(host, port));
         var receiver =
             new Receiver() {
-              private final Map<Long, HasBeanObject> objects = new HashMap<>();
+              private final Map<Long, HasBeanObject> objects = new ConcurrentHashMap<>();
 
               @Override
               public String host() {
@@ -136,9 +138,14 @@ public class BeanCollector {
                   try {
                     if (node.mBeanClient == null)
                       node.mBeanClient = clientCreator.apply(host, port);
-                    if (objects.size() >= numberOfObjectsPerNode)
-                      objects.keySet().stream().min((Long::compare)).ifPresent(objects::remove);
-                    objects.put(System.currentTimeMillis(), getter.apply(node.mBeanClient));
+                    getters.forEach(
+                        getter -> {
+                          if (objects.size() >= numberOfObjectsPerNode)
+                            objects.keySet().stream()
+                                .min((Long::compare))
+                                .ifPresent(objects::remove);
+                          objects.put(System.currentTimeMillis(), getter.apply(node.mBeanClient));
+                        });
                   } finally {
                     node.lock.unlock();
                   }
diff --git a/app/src/main/java/org/astraea/metrics/collector/Register.java b/app/src/main/java/org/astraea/metrics/collector/Register.java
index 5b27b604..a7bf78c5 100644
--- a/app/src/main/java/org/astraea/metrics/collector/Register.java
+++ b/app/src/main/java/org/astraea/metrics/collector/Register.java
@@ -1,5 +1,7 @@
 package org.astraea.metrics.collector;
 
+import java.util.Collection;
+import java.util.List;
 import java.util.function.Function;
 import org.astraea.metrics.HasBeanObject;
 import org.astraea.metrics.jmx.MBeanClient;
@@ -22,7 +24,15 @@ public interface Register {
    * @param getter to get metrics from MBeanClient
    * @return this register
    */
-  Register metricsGetter(Function<MBeanClient, HasBeanObject> getter);
+  default Register metricsGetter(Function<MBeanClient, HasBeanObject> getter) {
+    return metricsGetters(List.of(getter));
+  }
+
+  /**
+   * @param getters to get metrics from MBeanClient
+   * @return this register
+   */
+  Register metricsGetters(Collection<Function<MBeanClient, HasBeanObject>> getters);
 
   /** @return create a Receiver */
   Receiver build();
diff --git a/app/src/main/java/org/astraea/partitioner/ClusterInfo.java b/app/src/main/java/org/astraea/partitioner/ClusterInfo.java
index 8b6767e5..c3e9f87b 100644
--- a/app/src/main/java/org/astraea/partitioner/ClusterInfo.java
+++ b/app/src/main/java/org/astraea/partitioner/ClusterInfo.java
@@ -1,6 +1,7 @@
 package org.astraea.partitioner;
 
 import java.util.List;
+import java.util.NoSuchElementException;
 import java.util.stream.Collectors;
 
 public interface ClusterInfo {
@@ -28,6 +29,36 @@ public interface ClusterInfo {
     };
   }
 
+  /**
+   * find the node associated to specify node and port. Normally, the node + port should be unique
+   * in cluster.
+   *
+   * @param host hostname
+   * @param port client port
+   * @return the node information. It throws NoSuchElementException if specify node and port is not
+   *     associated to any node
+   */
+  default NodeInfo node(String host, int port) {
+    return nodes().stream()
+        .filter(n -> n.host().equals(host) && n.port() == port)
+        .findAny()
+        .orElseThrow(() -> new NoSuchElementException(host + ":" + port + " is nonexistent"));
+  }
+
+  /**
+   * find the node associated to node id.
+   *
+   * @param id node id
+   * @return the node information. It throws NoSuchElementException if specify node id is not
+   *     associated to any node
+   */
+  default NodeInfo node(int id) {
+    return nodes().stream()
+        .filter(n -> n.id() == id)
+        .findAny()
+        .orElseThrow(() -> new IllegalArgumentException(id + " is nonexistent"));
+  }
+
   /** @return The known set of nodes */
   List<NodeInfo> nodes();
 
diff --git a/app/src/main/java/org/astraea/partitioner/Configuration.java b/app/src/main/java/org/astraea/partitioner/Configuration.java
index 74000010..00c44a03 100644
--- a/app/src/main/java/org/astraea/partitioner/Configuration.java
+++ b/app/src/main/java/org/astraea/partitioner/Configuration.java
@@ -3,42 +3,50 @@ package org.astraea.partitioner;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
-import java.util.Objects;
+import java.util.NoSuchElementException;
+import java.util.Optional;
+import java.util.Set;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
 public interface Configuration {
 
-  static Configuration of(Map<String, ?> configs) {
+  static Configuration of(Map<String, String> configs) {
     return new Configuration() {
       @Override
-      public String string(String key) {
-        return Objects.requireNonNull(configs.get(key), key + " is nonexistent").toString();
+      public Optional<String> string(String key) {
+        return Optional.ofNullable(configs.get(key)).map(Object::toString);
       }
 
       @Override
       public List<String> list(String key, String separator) {
-        return Arrays.asList(string(key).split(separator));
+        return Arrays.asList(requireString(key).split(separator));
       }
 
       @Override
-      public <T> Map<String, T> map(
+      public <K, V> Map<K, V> map(
           String key,
           String listSeparator,
           String mapSeparator,
-          Function<String, T> valueConverter) {
-        Function<String, Map.Entry<String, T>> split =
+          Function<String, K> keyConverter,
+          Function<String, V> valueConverter) {
+        Function<String, Map.Entry<K, V>> split =
             s -> {
               var items = s.split(mapSeparator);
               if (items.length != 2)
                 throw new IllegalArgumentException(
                     "the value: " + s + " is using incorrect separator");
-              return Map.entry(items[0], valueConverter.apply(items[1]));
+              return Map.entry(keyConverter.apply(items[0]), valueConverter.apply(items[1]));
             };
         return list(key, listSeparator).stream()
             .map(split)
             .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
       }
+
+      @Override
+      public Set<Map.Entry<String, String>> entrySet() {
+        return configs.entrySet();
+      }
     };
   }
 
@@ -46,7 +54,23 @@ public interface Configuration {
    * @param key the key whose associated value is to be returned
    * @return string value. never null
    */
-  String string(String key);
+  Optional<String> string(String key);
+
+  /**
+   * @param key the key whose associated value is to be returned
+   * @return integer value. never null
+   */
+  default Optional<Integer> integer(String key) {
+    return string(key).map(Integer::parseInt);
+  }
+
+  /**
+   * @param key the key whose associated value is to be returned
+   * @return string value. never null
+   */
+  default String requireString(String key) {
+    return string(key).orElseThrow(() -> new NoSuchElementException(key + " is nonexistent"));
+  }
 
   /**
    * @param key the key whose associated value is to be returned
@@ -56,22 +80,48 @@ public interface Configuration {
   List<String> list(String key, String separator);
 
   /**
+   * parse the map structure from value associated to specify key
+   *
    * @param key the key whose associated value is to be returned
    * @param listSeparator to split string to multiple strings
    * @param mapSeparator to split multiple strings to map
-   * @return string map. never null
+   * @return map. never null
    */
   default Map<String, String> map(String key, String listSeparator, String mapSeparator) {
     return map(key, listSeparator, mapSeparator, s -> s);
   }
 
   /**
+   * parse the map structure from value associated to specify key
+   *
    * @param key the key whose associated value is to be returned
    * @param listSeparator to split string to multiple strings
    * @param mapSeparator to split multiple strings to map
-   * @param valueConverter used to convert string to specify type
-   * @return string map. never null
+   * @param valueConverter used to convert value string to specify type
+   * @return map. never null
    */
-  <T> Map<String, T> map(
-      String key, String listSeparator, String mapSeparator, Function<String, T> valueConverter);
+  default <T> Map<String, T> map(
+      String key, String listSeparator, String mapSeparator, Function<String, T> valueConverter) {
+    return map(key, listSeparator, mapSeparator, s -> s, valueConverter);
+  }
+
+  /**
+   * parse the map structure from value associated to specify key
+   *
+   * @param key the key whose associated value is to be returned
+   * @param listSeparator to split string to multiple strings
+   * @param mapSeparator to split multiple strings to map
+   * @param keyConverter used to convert key string to specify type
+   * @param valueConverter used to convert value string to specify type
+   * @return map. never null
+   */
+  <K, V> Map<K, V> map(
+      String key,
+      String listSeparator,
+      String mapSeparator,
+      Function<String, K> keyConverter,
+      Function<String, V> valueConverter);
+
+  /** @return a {@link Set} view of the mappings contained in this map. */
+  Set<Map.Entry<String, String>> entrySet();
 }
diff --git a/app/src/main/java/org/astraea/partitioner/Dispatcher.java b/app/src/main/java/org/astraea/partitioner/Dispatcher.java
index cfdddac8..772605c9 100644
--- a/app/src/main/java/org/astraea/partitioner/Dispatcher.java
+++ b/app/src/main/java/org/astraea/partitioner/Dispatcher.java
@@ -1,6 +1,7 @@
 package org.astraea.partitioner;
 
 import java.util.Map;
+import java.util.stream.Collectors;
 import org.apache.kafka.clients.producer.Partitioner;
 import org.apache.kafka.common.Cluster;
 
@@ -15,11 +16,23 @@ public interface Dispatcher extends Partitioner {
    */
   int partition(String topic, byte[] key, byte[] value, ClusterInfo clusterInfo);
 
-  void configure(Configuration config);
+  /**
+   * configure this dispatcher. This method is called only once.
+   *
+   * @param config configuration
+   */
+  default void configure(Configuration config) {}
+
+  /** close this dispatcher. This method is executed only once. */
+  @Override
+  default void close() {}
 
   @Override
   default void configure(Map<String, ?> configs) {
-    configure(Configuration.of(configs));
+    configure(
+        Configuration.of(
+            configs.entrySet().stream()
+                .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().toString()))));
   }
 
   @Override
diff --git a/app/src/main/java/org/astraea/partitioner/FlexibleRoundRobinDispatcher.java b/app/src/main/java/org/astraea/partitioner/FlexibleRoundRobinDispatcher.java
deleted file mode 100644
index 8ffb1595..00000000
--- a/app/src/main/java/org/astraea/partitioner/FlexibleRoundRobinDispatcher.java
+++ /dev/null
@@ -1,79 +0,0 @@
-package org.astraea.partitioner;
-
-import static org.astraea.Utils.overSecond;
-import static org.astraea.partitioner.nodeLoadMetric.PartitionerUtils.allPoisson;
-import static org.astraea.partitioner.nodeLoadMetric.PartitionerUtils.toPositive;
-
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.atomic.AtomicInteger;
-import org.astraea.partitioner.nodeLoadMetric.NodeLoadClient;
-
-public class FlexibleRoundRobinDispatcher implements Dispatcher {
-  public static final String JMX_SERVERS_KEY = "jmx_servers";
-  private static final double SLOP = 0.2;
-  private NodeLoadClient nodeLoadClient;
-  private final ConcurrentMap<String, AtomicInteger> topicCounterMap = new ConcurrentHashMap<>();
-  private long lastTime;
-  private Map<Integer, Double> allPoisson;
-  private double avgPoisson;
-
-  @Override
-  public int partition(String topic, byte[] key, byte[] value, ClusterInfo clusterInfo) {
-    var loadCount =
-        Objects.requireNonNull(
-            nodeLoadClient.loadSituation(clusterInfo), "OverLoadCount should not be null.");
-    updatePoissonIfNeed(loadCount);
-
-    var partitions = clusterInfo.partitions(topic);
-    var numPartitions = partitions.size();
-    var availablePartitions = clusterInfo.availablePartitions(topic);
-    if (!availablePartitions.isEmpty()) {
-      return choosePartition(availablePartitions, topic);
-    } else {
-      var nextValue = nextValue(topic);
-      // no partitions are available, give a non-available partition
-      return toPositive(nextValue) % numPartitions;
-    }
-  }
-
-  @Override
-  public void configure(Configuration config) {
-    nodeLoadClient = new NodeLoadClient(config.map(JMX_SERVERS_KEY, ",", ":", Integer::valueOf));
-  }
-
-  @Override
-  public void close() {}
-
-  private int nextValue(String topic) {
-    var counter = topicCounterMap.computeIfAbsent(topic, k -> new AtomicInteger(0));
-    return counter.getAndIncrement();
-  }
-
-  private void updatePoissonIfNeed(Map<Integer, Integer> loadCount) {
-    if (overSecond(lastTime, 1)) {
-      lastTime = System.currentTimeMillis();
-      allPoisson = allPoisson(loadCount);
-      avgPoisson = allPoisson.values().stream().reduce(0.0, Double::sum) / allPoisson.size();
-    }
-  }
-
-  private int choosePartition(List<PartitionInfo> availablePartitions, String topic) {
-    var nextValue = nextValue(topic);
-    var part = toPositive(nextValue) % availablePartitions.size();
-    var partitionInfo = availablePartitions.get(part);
-    if (chooseOrNot(allPoisson.get(partitionInfo.leader().id()))) {
-      return partitionInfo.partition();
-    } else {
-      // no partitions are available, give a non-available partition
-      return toPositive(nextValue) % availablePartitions.size();
-    }
-  }
-
-  private boolean chooseOrNot(double brokerPoisson) {
-    return !(brokerPoisson >= (1 + SLOP) * avgPoisson);
-  }
-}
diff --git a/app/src/main/java/org/astraea/partitioner/NodeId.java b/app/src/main/java/org/astraea/partitioner/NodeId.java
new file mode 100644
index 00000000..caa0fc53
--- /dev/null
+++ b/app/src/main/java/org/astraea/partitioner/NodeId.java
@@ -0,0 +1,18 @@
+package org.astraea.partitioner;
+
+/** We use this interface to replace the integer for more readable code. */
+@FunctionalInterface
+public interface NodeId extends Comparable<NodeInfo> {
+
+  static NodeId of(int id) {
+    return () -> id;
+  }
+
+  /** @return The id for this node */
+  int id();
+
+  @Override
+  default int compareTo(NodeInfo o) {
+    return Integer.compare(id(), o.id());
+  }
+}
diff --git a/app/src/main/java/org/astraea/partitioner/NodeInfo.java b/app/src/main/java/org/astraea/partitioner/NodeInfo.java
index 17cb880e..9a54c41f 100644
--- a/app/src/main/java/org/astraea/partitioner/NodeInfo.java
+++ b/app/src/main/java/org/astraea/partitioner/NodeInfo.java
@@ -1,6 +1,8 @@
 package org.astraea.partitioner;
 
-public interface NodeInfo {
+import java.util.Objects;
+
+public interface NodeInfo extends Comparable<NodeInfo>, NodeId {
 
   static NodeInfo of(org.apache.kafka.common.Node node) {
     return of(node.id(), node.host(), node.port());
@@ -22,15 +24,32 @@ public interface NodeInfo {
       public int port() {
         return port;
       }
+
+      @Override
+      public int hashCode() {
+        return Objects.hash(id, host, port);
+      }
+
+      @Override
+      public boolean equals(Object other) {
+        if (other instanceof NodeInfo) return compareTo((NodeInfo) other) == 0;
+        return false;
+      }
+
+      @Override
+      public int compareTo(NodeInfo other) {
+        int r = Integer.compare(id(), other.id());
+        if (r != 0) return r;
+        r = host().compareTo(other.host());
+        if (r != 0) return r;
+        return Integer.compare(port(), other.port());
+      }
     };
   }
 
   /** @return The host name for this node */
   String host();
 
-  /** @return The id for this node */
-  int id();
-
-  /** @return The client port for this node */
+  /** @return The client (kafka data, jmx, etc.) port for this node */
   int port();
 }
diff --git a/app/src/main/java/org/astraea/partitioner/StrictCostDispatcher.java b/app/src/main/java/org/astraea/partitioner/StrictCostDispatcher.java
new file mode 100644
index 00000000..e55073f8
--- /dev/null
+++ b/app/src/main/java/org/astraea/partitioner/StrictCostDispatcher.java
@@ -0,0 +1,115 @@
+package org.astraea.partitioner;
+
+import java.time.Duration;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Optional;
+import java.util.TreeMap;
+import java.util.stream.Collectors;
+import org.astraea.Utils;
+import org.astraea.metrics.collector.BeanCollector;
+import org.astraea.metrics.collector.Receiver;
+import org.astraea.partitioner.cost.CostFunction;
+
+/**
+ * this dispatcher scores the nodes by multiples cost functions. Each function evaluate the target
+ * node by different metrics. The default cost function ranks nodes by throughput. It means the node
+ * having lower throughput get higher score.
+ *
+ * <p>The requisite config is JMX port. Most cost functions need the JMX metrics to score nodes.
+ * Normally, all brokers use the same JMX port, so you can just define the `jmx.port=12345`. If one
+ * of brokers uses different JMX client port, you can define `broker.1000.jmx.port=11111` (`1000` is
+ * the broker id) to replace the value of `jmx.port`.
+ */
+public class StrictCostDispatcher implements Dispatcher {
+  public static final String JMX_PORT = "jmx.port";
+
+  private final BeanCollector beanCollector =
+      BeanCollector.builder().interval(Duration.ofSeconds(4)).build();
+  private final Collection<CostFunction> functions;
+  private Optional<Integer> jmxPortDefault = Optional.empty();
+  private final Map<NodeId, Integer> jmxPorts = new TreeMap<>();
+  private final Map<NodeId, Receiver> receivers = new TreeMap<>();
+
+  public StrictCostDispatcher() {
+    this(List.of(CostFunction.throughput()));
+  }
+
+  // visible for testing
+  StrictCostDispatcher(Collection<CostFunction> functions) {
+    this.functions = functions;
+  }
+
+  @Override
+  public int partition(String topic, byte[] key, byte[] value, ClusterInfo clusterInfo) {
+    var partitions = clusterInfo.availablePartitions(topic);
+    // just return first partition if there is no available partitions
+    if (partitions.isEmpty()) return 0;
+
+    // add new receivers for new brokers
+    partitions.stream()
+        .filter(p -> !receivers.containsKey(NodeId.of(p.leader().id())))
+        .forEach(
+            p ->
+                receivers.put(
+                    p.leader(),
+                    beanCollector
+                        .register()
+                        .host(p.leader().host())
+                        .port(jmxPort(p.leader().id()))
+                        .metricsGetters(
+                            functions.stream()
+                                .flatMap(c -> c.metricsGetters().stream())
+                                .collect(Collectors.toUnmodifiableList()))
+                        .build()));
+
+    // get latest beans for each node
+    var beans =
+        receivers.entrySet().stream()
+            .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().current()));
+
+    // get scores from all cost functions
+    var scores =
+        functions.stream()
+            .map(f -> f.cost(beans, clusterInfo))
+            .collect(Collectors.toUnmodifiableList());
+
+    // return the partition (node) having min score
+    return partitions.stream()
+        .map(
+            p ->
+                Map.entry(
+                    p, scores.stream().mapToDouble(s -> s.getOrDefault(p.leader(), 0.0D)).sum()))
+        .min(Map.Entry.comparingByValue())
+        .map(e -> e.getKey().partition())
+        .orElse(0);
+  }
+
+  @Override
+  public void configure(Configuration config) {
+    jmxPortDefault = config.integer(JMX_PORT);
+
+    // seeks for custom jmx ports.
+    config.entrySet().stream()
+        .filter(e -> e.getKey().startsWith("broker."))
+        .filter(e -> e.getKey().endsWith(JMX_PORT))
+        .forEach(
+            e ->
+                jmxPorts.put(
+                    NodeId.of(Integer.parseInt(e.getKey())), Integer.parseInt(e.getValue())));
+  }
+
+  // visible for testing
+  int jmxPort(int id) {
+    if (jmxPorts.containsKey(NodeId.of(id))) return jmxPorts.get(NodeId.of(id));
+    return jmxPortDefault.orElseThrow(
+        () -> new NoSuchElementException("broker: " + id + " does not have jmx port"));
+  }
+
+  @Override
+  public void close() {
+    receivers.values().forEach(Utils::close);
+  }
+}
diff --git a/app/src/main/java/org/astraea/partitioner/cost/CostFunction.java b/app/src/main/java/org/astraea/partitioner/cost/CostFunction.java
new file mode 100644
index 00000000..8a630150
--- /dev/null
+++ b/app/src/main/java/org/astraea/partitioner/cost/CostFunction.java
@@ -0,0 +1,29 @@
+package org.astraea.partitioner.cost;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import org.astraea.metrics.HasBeanObject;
+import org.astraea.metrics.jmx.MBeanClient;
+import org.astraea.partitioner.ClusterInfo;
+import org.astraea.partitioner.NodeId;
+
+public interface CostFunction {
+
+  static CostFunction throughput() {
+    return new ThroughputCost();
+  }
+
+  /**
+   * score all nodes according to passed beans and cluster information.
+   *
+   * @param beans beans of each node (key is node id)
+   * @param clusterInfo cluster information
+   * @return the score of each node. the score range is [0 - 1]
+   */
+  Map<NodeId, Double> cost(Map<NodeId, List<HasBeanObject>> beans, ClusterInfo clusterInfo);
+
+  /** @return the metrics getters. Those getters are used to fetch mbeans. */
+  Collection<Function<MBeanClient, HasBeanObject>> metricsGetters();
+}
diff --git a/app/src/main/java/org/astraea/partitioner/cost/ThroughputCost.java b/app/src/main/java/org/astraea/partitioner/cost/ThroughputCost.java
new file mode 100644
index 00000000..bf879b03
--- /dev/null
+++ b/app/src/main/java/org/astraea/partitioner/cost/ThroughputCost.java
@@ -0,0 +1,47 @@
+package org.astraea.partitioner.cost;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import org.astraea.metrics.HasBeanObject;
+import org.astraea.metrics.jmx.MBeanClient;
+import org.astraea.metrics.kafka.BrokerTopicMetricsResult;
+import org.astraea.metrics.kafka.KafkaMetrics;
+import org.astraea.partitioner.ClusterInfo;
+import org.astraea.partitioner.NodeId;
+
+public class ThroughputCost implements CostFunction {
+
+  @Override
+  public Map<NodeId, Double> cost(Map<NodeId, List<HasBeanObject>> beans, ClusterInfo clusterInfo) {
+    var score = score(beans);
+
+    var max = score.values().stream().mapToDouble(v -> v).max().orElse(1);
+
+    return clusterInfo.nodes().stream()
+        .collect(Collectors.toMap(n -> n, n -> score.getOrDefault(n, 0.0D) / max));
+  }
+
+  Map<NodeId, Double> score(Map<NodeId, List<HasBeanObject>> beans) {
+    // TODO: this implementation only consider the oneMinuteRate ...
+    return beans.entrySet().stream()
+        .collect(
+            Collectors.toMap(
+                Map.Entry::getKey,
+                e ->
+                    e.getValue().stream()
+                        .filter(b -> b instanceof BrokerTopicMetricsResult)
+                        .map(b -> (BrokerTopicMetricsResult) b)
+                        .mapToDouble(BrokerTopicMetricsResult::oneMinuteRate)
+                        .sum()));
+  }
+
+  @Override
+  public Collection<Function<MBeanClient, HasBeanObject>> metricsGetters() {
+    return List.of(
+        KafkaMetrics.BrokerTopic.BytesInPerSec::fetch,
+        KafkaMetrics.BrokerTopic.BytesOutPerSec::fetch);
+  }
+}


Test patch:
diff --git a/app/src/test/java/org/astraea/partitioner/ClusterInfoTest.java b/app/src/test/java/org/astraea/partitioner/ClusterInfoTest.java
new file mode 100644
index 00000000..5d2d6434
--- /dev/null
+++ b/app/src/test/java/org/astraea/partitioner/ClusterInfoTest.java
@@ -0,0 +1,33 @@
+package org.astraea.partitioner;
+
+import java.util.List;
+import org.apache.kafka.common.Cluster;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+public class ClusterInfoTest {
+
+  @Test
+  void testNode() {
+    var node = NodeInfoTest.node();
+    var partition = PartitionInfoTest.partitionInfo();
+    var kafkaCluster = Mockito.mock(Cluster.class);
+    Mockito.when(kafkaCluster.availablePartitionsForTopic(partition.topic()))
+        .thenReturn(List.of(partition));
+    Mockito.when(kafkaCluster.partitionsForTopic(partition.topic())).thenReturn(List.of(partition));
+    Mockito.when(kafkaCluster.nodes()).thenReturn(List.of(node));
+
+    var clusterInfo = ClusterInfo.of(kafkaCluster);
+
+    Assertions.assertEquals(1, clusterInfo.nodes().size());
+    Assertions.assertEquals(NodeInfo.of(node), clusterInfo.nodes().get(0));
+    Assertions.assertEquals(clusterInfo.nodes().get(0), clusterInfo.node(node.host(), node.port()));
+    Assertions.assertEquals(1, clusterInfo.availablePartitions(partition.topic()).size());
+    Assertions.assertEquals(1, clusterInfo.partitions(partition.topic()).size());
+    Assertions.assertEquals(
+        NodeInfo.of(node), clusterInfo.availablePartitions(partition.topic()).get(0).leader());
+    Assertions.assertEquals(
+        NodeInfo.of(node), clusterInfo.partitions(partition.topic()).get(0).leader());
+  }
+}
diff --git a/app/src/test/java/org/astraea/partitioner/ConfigurationTest.java b/app/src/test/java/org/astraea/partitioner/ConfigurationTest.java
new file mode 100644
index 00000000..a186aef3
--- /dev/null
+++ b/app/src/test/java/org/astraea/partitioner/ConfigurationTest.java
@@ -0,0 +1,30 @@
+package org.astraea.partitioner;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class ConfigurationTest {
+
+  @Test
+  void testString() {
+    var config = Configuration.of(Map.of("key", "value"));
+    Assertions.assertEquals(Optional.of("value"), config.string("key"));
+    Assertions.assertEquals("value", config.requireString("key"));
+  }
+
+  @Test
+  void testList() {
+    var config = Configuration.of(Map.of("key", "v0,v1"));
+    Assertions.assertEquals(List.of("v0", "v1"), config.list("key", ","));
+  }
+
+  @Test
+  void testMap() {
+    var config = Configuration.of(Map.of("key", "v0:0,v1:1"));
+    Assertions.assertEquals(
+        Map.of("v0", 0, "v1", 1), config.map("key", ",", ":", Integer::valueOf));
+  }
+}
diff --git a/app/src/test/java/org/astraea/partitioner/DispatcherTest.java b/app/src/test/java/org/astraea/partitioner/DispatcherTest.java
new file mode 100644
index 00000000..342d44a5
--- /dev/null
+++ b/app/src/test/java/org/astraea/partitioner/DispatcherTest.java
@@ -0,0 +1,36 @@
+package org.astraea.partitioner;
+
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class DispatcherTest {
+
+  @Test
+  void testNullKey() {
+    var count = new AtomicInteger();
+    var dispatcher =
+        new Dispatcher() {
+          @Override
+          public int partition(String topic, byte[] key, byte[] value, ClusterInfo clusterInfo) {
+            Assertions.assertEquals(0, Objects.requireNonNull(key).length);
+            Assertions.assertEquals(0, Objects.requireNonNull(value).length);
+            count.incrementAndGet();
+            return 0;
+          }
+
+          @Override
+          public void configure(Configuration config) {
+            count.incrementAndGet();
+          }
+        };
+    Assertions.assertEquals(0, count.get());
+    // it should not throw NPE
+    dispatcher.partition("t", null, null, null, null, null);
+    Assertions.assertEquals(1, count.get());
+    dispatcher.configure(Map.of("a", "b"));
+    Assertions.assertEquals(2, count.get());
+  }
+}
diff --git a/app/src/test/java/org/astraea/partitioner/NodeInfoTest.java b/app/src/test/java/org/astraea/partitioner/NodeInfoTest.java
new file mode 100644
index 00000000..d44e110f
--- /dev/null
+++ b/app/src/test/java/org/astraea/partitioner/NodeInfoTest.java
@@ -0,0 +1,22 @@
+package org.astraea.partitioner;
+
+import org.apache.kafka.common.Node;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class NodeInfoTest {
+
+  static Node node() {
+    return new Node(10, "host", 100);
+  }
+
+  @Test
+  void testAllGetters() {
+    var kafkaNode = node();
+    var node = NodeInfo.of(kafkaNode);
+
+    Assertions.assertEquals(kafkaNode.host(), node.host());
+    Assertions.assertEquals(kafkaNode.id(), node.id());
+    Assertions.assertEquals(kafkaNode.port(), node.port());
+  }
+}
diff --git a/app/src/test/java/org/astraea/partitioner/PartitionInfoTest.java b/app/src/test/java/org/astraea/partitioner/PartitionInfoTest.java
new file mode 100644
index 00000000..9a6159f1
--- /dev/null
+++ b/app/src/test/java/org/astraea/partitioner/PartitionInfoTest.java
@@ -0,0 +1,24 @@
+package org.astraea.partitioner;
+
+import org.apache.kafka.common.Node;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class PartitionInfoTest {
+
+  static org.apache.kafka.common.PartitionInfo partitionInfo() {
+    return new org.apache.kafka.common.PartitionInfo(
+        "ttt", 100, NodeInfoTest.node(), new Node[0], new Node[0]);
+  }
+
+  @Test
+  void testAllGetters() {
+    var kafkaPartition = partitionInfo();
+
+    var partitionInfo = PartitionInfo.of(kafkaPartition);
+
+    Assertions.assertEquals(kafkaPartition.topic(), partitionInfo.topic());
+    Assertions.assertEquals(kafkaPartition.partition(), partitionInfo.partition());
+    Assertions.assertEquals(NodeInfo.of(kafkaPartition.leader()), partitionInfo.leader());
+  }
+}
diff --git a/app/src/test/java/org/astraea/partitioner/StrictCostDispatcherTest.java b/app/src/test/java/org/astraea/partitioner/StrictCostDispatcherTest.java
new file mode 100644
index 00000000..fd7a03eb
--- /dev/null
+++ b/app/src/test/java/org/astraea/partitioner/StrictCostDispatcherTest.java
@@ -0,0 +1,25 @@
+package org.astraea.partitioner;
+
+import java.util.Map;
+import java.util.NoSuchElementException;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class StrictCostDispatcherTest {
+
+  @Test
+  void testClose() {
+    var dispatcher = new StrictCostDispatcher();
+    dispatcher.close();
+  }
+
+  @Test
+  void testConfigure() {
+    try (var dispatcher = new StrictCostDispatcher()) {
+      dispatcher.configure(Configuration.of(Map.of()));
+      Assertions.assertThrows(NoSuchElementException.class, () -> dispatcher.jmxPort(0));
+      dispatcher.configure(Configuration.of(Map.of(StrictCostDispatcher.JMX_PORT, "12345")));
+      Assertions.assertEquals(12345, dispatcher.jmxPort(0));
+    }
+  }
+}
diff --git a/app/src/test/java/org/astraea/partitioner/cost/ThroughputCostTest.java b/app/src/test/java/org/astraea/partitioner/cost/ThroughputCostTest.java
new file mode 100644
index 00000000..53ab3b20
--- /dev/null
+++ b/app/src/test/java/org/astraea/partitioner/cost/ThroughputCostTest.java
@@ -0,0 +1,42 @@
+package org.astraea.partitioner.cost;
+
+import java.util.List;
+import java.util.Map;
+import org.astraea.metrics.kafka.BrokerTopicMetricsResult;
+import org.astraea.partitioner.ClusterInfo;
+import org.astraea.partitioner.NodeInfo;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+public class ThroughputCostTest {
+
+  @Test
+  void testScore() {
+    var throughputCost = new ThroughputCost();
+
+    var node = NodeInfo.of(10, "host", 1000);
+    var bean = Mockito.mock(BrokerTopicMetricsResult.class);
+    Mockito.when(bean.oneMinuteRate()).thenReturn(100D);
+
+    var score = throughputCost.score(Map.of(node, List.of(bean)));
+    Assertions.assertEquals(1, score.size());
+    Assertions.assertEquals(100D, score.get(node));
+  }
+
+  @Test
+  void testCost() {
+    var throughputCost = new ThroughputCost();
+
+    var node = NodeInfo.of(10, "host", 1000);
+    var bean = Mockito.mock(BrokerTopicMetricsResult.class);
+    Mockito.when(bean.oneMinuteRate()).thenReturn(100D);
+
+    var cluster = Mockito.mock(ClusterInfo.class);
+    Mockito.when(cluster.nodes()).thenReturn(List.of(node));
+
+    var cost = throughputCost.cost(Map.of(node, List.of(bean)), cluster);
+    Assertions.assertEquals(1, cost.size());
+    Assertions.assertEquals(1, cost.get(node));
+  }
+}