Repository: skiptests/astraea (java)

Issue:
[WEB] Remove default `costWeights` value from web service
每個 Kafka 叢集遭遇的負載問題不太一樣，基本上沒有一個適合的固定預設值能夠涵蓋所有情境，對此提供一個 `costWeights` 預設值不太妥當，我們應該移除該預設值。

1. 移除 `costWeights` 的預設值
2. 更新 documentation
3. 有個測試確保整個行為

Pull request description:
[WEB] Make `costWeights` a mandatory argument in `POST /balancer`
Resolve #1387.

移除 `POST /balancer` 的  `costWeights` 的預設值，因為不存在一個固定的預設優化項目能涵蓋所有叢集的問題。

Patch:
diff --git a/app/src/main/java/org/astraea/app/web/BalancerHandler.java b/app/src/main/java/org/astraea/app/web/BalancerHandler.java
index 9d97befa..05b33018 100644
--- a/app/src/main/java/org/astraea/app/web/BalancerHandler.java
+++ b/app/src/main/java/org/astraea/app/web/BalancerHandler.java
@@ -67,8 +67,6 @@ import org.astraea.common.metrics.collector.MetricSensor;
 
 class BalancerHandler implements Handler {
 
-  static final HasClusterCost DEFAULT_CLUSTER_COST_FUNCTION =
-      HasClusterCost.of(Map.of(new ReplicaSizeCost(), 1.0, new ReplicaLeaderCost(), 1.0));
   static final HasMoveCost DEFAULT_MOVE_COST_FUNCTIONS =
       HasMoveCost.of(
           List.of(new ReplicaNumberCost(), new ReplicaLeaderCost(), new ReplicaSizeCost()));
@@ -359,7 +357,7 @@ class BalancerHandler implements Handler {
 
   static HasClusterCost getClusterCost(BalancerPostRequest request) {
     var costWeights = request.costWeights;
-    if (costWeights.isEmpty()) return DEFAULT_CLUSTER_COST_FUNCTION;
+    if (costWeights.isEmpty()) throw new IllegalArgumentException("costWeights is not specified");
     var costWeightMap =
         costWeights.stream()
             .flatMap(cw -> cw.cost.map(c -> Map.entry(c, cw.weight.orElse(1D))).stream())
diff --git a/common/src/main/java/org/astraea/common/balancer/algorithms/GreedyBalancer.java b/common/src/main/java/org/astraea/common/balancer/algorithms/GreedyBalancer.java
index 73b0d963..03d5cf9b 100644
--- a/common/src/main/java/org/astraea/common/balancer/algorithms/GreedyBalancer.java
+++ b/common/src/main/java/org/astraea/common/balancer/algorithms/GreedyBalancer.java
@@ -177,12 +177,12 @@ public class GreedyBalancer implements Balancer {
     var currentMinCost =
         new DoubleAccumulator((l, r) -> Double.isNaN(r) ? l : Math.min(l, r), initialCost.value());
     MBeanRegister.local()
-        .setDomainName("astraea.balancer")
-        .addProperty("id", config.executionId())
-        .addProperty("algorithm", GreedyBalancer.class.getSimpleName())
-        .addProperty("run", Integer.toString(run.getAndIncrement()))
-        .addAttribute("Iteration", Long.class, currentIteration::sum)
-        .addAttribute("MinCost", Double.class, currentMinCost::get)
+        .domainName("astraea.balancer")
+        .property("id", config.executionId())
+        .property("algorithm", GreedyBalancer.class.getSimpleName())
+        .property("run", Integer.toString(run.getAndIncrement()))
+        .attribute("Iteration", Long.class, currentIteration::sum)
+        .attribute("MinCost", Double.class, currentMinCost::get)
         .register();
 
     while (true) {
diff --git a/common/src/main/java/org/astraea/common/metrics/MBeanRegister.java b/common/src/main/java/org/astraea/common/metrics/MBeanRegister.java
index 671d040a..2ab742fa 100644
--- a/common/src/main/java/org/astraea/common/metrics/MBeanRegister.java
+++ b/common/src/main/java/org/astraea/common/metrics/MBeanRegister.java
@@ -68,7 +68,7 @@ public class MBeanRegister {
      *
      * @return this.
      */
-    public LocalRegister setDomainName(String domainName) {
+    public LocalRegister domainName(String domainName) {
       this.domainName = domainName;
       return this;
     }
@@ -78,7 +78,7 @@ public class MBeanRegister {
      *
      * @return this.
      */
-    public LocalRegister addProperty(String key, String value) {
+    public LocalRegister property(String key, String value) {
       this.properties.put(key, value);
       return this;
     }
@@ -88,7 +88,7 @@ public class MBeanRegister {
      *
      * @return this.
      */
-    public LocalRegister setDescription(String description) {
+    public LocalRegister description(String description) {
       this.description = description;
       return this;
     }
@@ -102,8 +102,7 @@ public class MBeanRegister {
      *     thread-safe.
      * @return this.
      */
-    public <T> LocalRegister addAttribute(
-        String name, Class<T> attributeClass, Supplier<T> source) {
+    public <T> LocalRegister attribute(String name, Class<T> attributeClass, Supplier<T> source) {
       Objects.requireNonNull(name);
       Objects.requireNonNull(attributeClass);
       Objects.requireNonNull(source);
diff --git a/common/src/main/java/org/astraea/common/partitioner/StrictCostDispatcher.java b/common/src/main/java/org/astraea/common/partitioner/StrictCostDispatcher.java
index b0a49848..2a507d54 100644
--- a/common/src/main/java/org/astraea/common/partitioner/StrictCostDispatcher.java
+++ b/common/src/main/java/org/astraea/common/partitioner/StrictCostDispatcher.java
@@ -59,11 +59,11 @@ public class StrictCostDispatcher implements Dispatcher {
 
   // visible for testing
   final MetricCollector metricCollector =
-      MetricCollector.builder().interval(Duration.ofSeconds(4)).build();
+      MetricCollector.builder().interval(Duration.ofMillis(1500)).build();
 
-  Duration roundRobinLease;
+  Duration roundRobinLease = Duration.ofSeconds(4);
 
-  HasBrokerCost costFunction = HasBrokerCost.EMPTY;
+  HasBrokerCost costFunction = new NodeLatencyCost();
   Function<Integer, Optional<Integer>> jmxPortGetter = (id) -> Optional.empty();
 
   final int[] roundRobin = new int[ROUND_ROBIN_LENGTH];
@@ -155,44 +155,19 @@ public class StrictCostDispatcher implements Dispatcher {
   @Override
   public void configure(Configuration config) {
     var configuredFunctions = parseCostFunctionWeight(config);
-    configure(
-        configuredFunctions.isEmpty() ? Map.of(new NodeLatencyCost(), 1D) : configuredFunctions,
-        config.integer(JMX_PORT),
-        PartitionerUtils.parseIdJMXPort(config),
-        config
-            .string(ROUND_ROBIN_LEASE_KEY)
-            .map(Utils::toDuration)
-            // The duration of updating beans is 4 seconds, so
-            // the default duration of updating RR is 4 seconds.
-            .orElse(Duration.ofSeconds(4)));
-  }
-
-  /**
-   * configure this StrictCostDispatcher. This method is extracted for testing.
-   *
-   * @param functions cost functions used by this dispatcher.
-   * @param jmxPortDefault jmx port by default
-   * @param customJmxPort jmx port for each node
-   */
-  void configure(
-      Map<HasBrokerCost, Double> functions,
-      Optional<Integer> jmxPortDefault,
-      Map<Integer, Integer> customJmxPort,
-      Duration roundRobinLease) {
-    this.costFunction = HasBrokerCost.of(functions);
-    this.jmxPortGetter = id -> Optional.ofNullable(customJmxPort.get(id)).or(() -> jmxPortDefault);
+    if (!configuredFunctions.isEmpty()) this.costFunction = HasBrokerCost.of(configuredFunctions);
+    var customJmxPort = PartitionerUtils.parseIdJMXPort(config);
+    var defaultJmxPort = config.integer(JMX_PORT);
+    this.jmxPortGetter = id -> Optional.ofNullable(customJmxPort.get(id)).or(() -> defaultJmxPort);
+    config
+        .string(ROUND_ROBIN_LEASE_KEY)
+        .map(Utils::toDuration)
+        .ifPresent(d -> this.roundRobinLease = d);
 
     // put local mbean client first
-    if (!metricCollector.listIdentities().contains(-1)) {
-      this.costFunction
-          .fetcher()
-          .ifPresent(
-              fetcher -> {
-                metricCollector.registerLocalJmx(-1);
-                metricCollector.addFetcher(fetcher);
-              });
-      this.roundRobinLease = roundRobinLease;
-    }
+    if (!metricCollector.listIdentities().contains(-1)) metricCollector.registerLocalJmx(-1);
+
+    this.costFunction.fetcher().ifPresent(metricCollector::addFetcher);
   }
 
   /**
diff --git a/docs/web_server/web_api_balancer_chinese.md b/docs/web_server/web_api_balancer_chinese.md
index ceaafb88..23b3d0db 100644
--- a/docs/web_server/web_api_balancer_chinese.md
+++ b/docs/web_server/web_api_balancer_chinese.md
@@ -14,11 +14,11 @@ POST /balancer
 
 | 名稱                | 說明                                                         | 預設值                                                   |
 |-------------------|------------------------------------------------------------|-------------------------------------------------------|
+| costWeights       | (必填) 指定要優化的目標以及對應權重                                        | 無                                                     |
 | topics            | (選填) 只嘗試搬移指定的 topics                                       | 無，除了內部 topics 以外的都作為候選對象                              |
 | timeout           | (選填) 指定產生時間                                                | 3s                                                    |
 | balancer          | (選填) 欲使用的負載優化計劃搜尋演算法                                       | org.astraea.common.balancer.algorithms.GreedyBalancer |
 | balancerConfig    | (選填) 搜尋演算法的實作細節參數，此為一個 JSON Object 內含一系列的 key/value String | 無                                                     |
-| costWeights       | (選填) 指定要優化的目標以及權重                                          | ReplicaSizeCost,ReplicaLeaderCost權重皆為1                |
  | maxMigratedSize   | (選填) 設定最大可搬移的log size                                      | 無 　                                                   |
  | maxMigratedLeader | (選填) 設定最大可搬移的leader 數量                                     | 無                                                     |
 


Test patch:
diff --git a/app/src/test/java/org/astraea/app/web/BalancerHandlerTest.java b/app/src/test/java/org/astraea/app/web/BalancerHandlerTest.java
index a4aa643c..894a7422 100644
--- a/app/src/test/java/org/astraea/app/web/BalancerHandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/BalancerHandlerTest.java
@@ -21,6 +21,7 @@ import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.NoSuchElementException;
@@ -93,6 +94,8 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
       List.of(costWeight(IncreasingCost.class.getName(), 1));
   private static final List<BalancerHandler.CostWeight> defaultDecreasing =
       List.of(costWeight(DecreasingCost.class.getName(), 1));
+  private static final Channel defaultPostPlan =
+      httpRequest(Map.of(COST_WEIGHT_KEY, defaultDecreasing));
 
   @Test
   @Timeout(value = 60)
@@ -392,16 +395,14 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
               BalancerHandler.PostPlanResponse.class,
               handler
                   .post(
-                      Channel.ofRequest(
-                          JsonConverter.defaultConverter()
-                              .toJson(
-                                  Map.of(
-                                      COST_WEIGHT_KEY,
-                                      defaultIncreasing,
-                                      TIMEOUT_KEY,
-                                      "996ms",
-                                      BALANCER_IMPLEMENTATION_KEY,
-                                      GreedyBalancer.class.getName()))))
+                      httpRequest(
+                          Map.of(
+                              COST_WEIGHT_KEY,
+                              defaultIncreasing,
+                              TIMEOUT_KEY,
+                              "996ms",
+                              BALANCER_IMPLEMENTATION_KEY,
+                              GreedyBalancer.class.getName())))
                   .toCompletableFuture()
                   .join());
       Utils.sleep(Duration.ofSeconds(5));
@@ -423,10 +424,7 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
                   CompletionException.class,
                   () ->
                       handler
-                          .put(
-                              Channel.ofRequest(
-                                  JsonConverter.defaultConverter()
-                                      .toJson(Map.of("id", progress.id))))
+                          .put(httpRequest(Map.of("id", progress.id)))
                           .toCompletableFuture()
                           .join())
               .getCause(),
@@ -456,12 +454,7 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
       var response =
           Assertions.assertInstanceOf(
               BalancerHandler.PutPlanResponse.class,
-              handler
-                  .put(
-                      Channel.ofRequest(
-                          JsonConverter.defaultConverter().toJson(Map.of("id", thePlanId))))
-                  .toCompletableFuture()
-                  .join());
+              handler.put(httpRequest(Map.of("id", thePlanId))).toCompletableFuture().join());
       Utils.sleep(Duration.ofSeconds(1));
 
       // assert
@@ -487,13 +480,7 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
       // no such plan id
       Assertions.assertThrows(
           IllegalArgumentException.class,
-          () ->
-              handler
-                  .put(
-                      Channel.ofRequest(
-                          JsonConverter.defaultConverter().toJson(Map.of("id", "no such plan"))))
-                  .toCompletableFuture()
-                  .join(),
+          () -> handler.put(httpRequest(Map.of("id", "no such plan"))).toCompletableFuture().join(),
           "The requested plan doesn't exists");
     }
   }
@@ -528,9 +515,7 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
                   executor.submit(
                       () -> {
                         // the plan
-                        final var request =
-                            Channel.ofRequest(
-                                JsonConverter.defaultConverter().toJson(Map.of("id", progress.id)));
+                        final var request = httpRequest(Map.of("id", progress.id));
                         // use cyclic barrier to ensure all threads are ready to work
                         Utils.packException(() -> barrier.await());
                         // send the put request
@@ -574,24 +559,13 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
       var plan1 = submitPlanGeneration(handler, Map.of());
 
       Assertions.assertDoesNotThrow(
-          () ->
-              handler
-                  .put(
-                      Channel.ofRequest(
-                          JsonConverter.defaultConverter().toJson(Map.of("id", plan0.id))))
-                  .toCompletableFuture()
-                  .join());
+          () -> handler.put(httpRequest(Map.of("id", plan0.id))).toCompletableFuture().join());
       Assertions.assertInstanceOf(
           IllegalStateException.class,
           Assertions.assertThrows(
                   CompletionException.class,
                   () ->
-                      handler
-                          .put(
-                              Channel.ofRequest(
-                                  JsonConverter.defaultConverter().toJson(Map.of("id", plan1.id))))
-                          .toCompletableFuture()
-                          .join())
+                      handler.put(httpRequest(Map.of("id", plan1.id))).toCompletableFuture().join())
               .getCause());
     }
   }
@@ -647,10 +621,7 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
                   CompletionException.class,
                   () ->
                       handler
-                          .put(
-                              Channel.ofRequest(
-                                  JsonConverter.defaultConverter()
-                                      .toJson(Map.of("id", theReport.id))))
+                          .put(httpRequest(Map.of("id", theReport.id)))
                           .toCompletableFuture()
                           .join())
               .getCause());
@@ -692,21 +663,21 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
       Mockito.when(admin.clusterInfo(Mockito.any()))
           .thenAnswer((invoke) -> CompletableFuture.completedFuture(clusterHasFuture));
       Assertions.assertThrows(
-          IllegalStateException.class, () -> new BalancerHandler(admin).post(Channel.EMPTY));
+          IllegalStateException.class, () -> new BalancerHandler(admin).post(defaultPostPlan));
 
       Mockito.when(admin.clusterInfo(Mockito.any()))
           .thenAnswer((invoke) -> CompletableFuture.completedFuture(clusterHasAdding));
       Assertions.assertThrows(
-          IllegalStateException.class, () -> new BalancerHandler(admin).post(Channel.EMPTY));
+          IllegalStateException.class, () -> new BalancerHandler(admin).post(defaultPostPlan));
 
       Mockito.when(admin.clusterInfo(Mockito.any()))
           .thenAnswer((invoke) -> CompletableFuture.completedFuture(clusterHasRemoving));
       Assertions.assertThrows(
-          IllegalStateException.class, () -> new BalancerHandler(admin).post(Channel.EMPTY));
+          IllegalStateException.class, () -> new BalancerHandler(admin).post(defaultPostPlan));
 
       Mockito.when(admin.clusterInfo(Mockito.any()))
           .thenAnswer((invoke) -> CompletableFuture.completedFuture(base));
-      Assertions.assertDoesNotThrow(() -> new BalancerHandler(admin).post(Channel.EMPTY));
+      Assertions.assertDoesNotThrow(() -> new BalancerHandler(admin).post(defaultPostPlan));
     }
   }
 
@@ -740,10 +711,7 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
                   CompletionException.class,
                   () ->
                       handler
-                          .put(
-                              Channel.ofRequest(
-                                  JsonConverter.defaultConverter()
-                                      .toJson(Map.of("id", theProgress.id))))
+                          .put(httpRequest(Map.of("id", theProgress.id)))
                           .toCompletableFuture()
                           .join(),
                   "The cluster state has changed, prevent the plan from execution")
@@ -791,12 +759,7 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
       var response =
           Assertions.assertInstanceOf(
               BalancerHandler.PutPlanResponse.class,
-              handler
-                  .put(
-                      Channel.ofRequest(
-                          JsonConverter.defaultConverter().toJson(Map.of("id", progress.id))))
-                  .toCompletableFuture()
-                  .join());
+              handler.put(httpRequest(Map.of("id", progress.id))).toCompletableFuture().join());
       Assertions.assertNotNull(response.id, "The plan should be executed");
 
       // not done yet
@@ -842,7 +805,10 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
       var post =
           Assertions.assertInstanceOf(
               BalancerHandler.PostPlanResponse.class,
-              handler.post(Channel.EMPTY).toCompletableFuture().join());
+              handler
+                  .post(httpRequest(Map.of(COST_WEIGHT_KEY, defaultDecreasing)))
+                  .toCompletableFuture()
+                  .join());
       Utils.waitFor(
           () ->
               ((BalancerHandler.PlanExecutionProgress)
@@ -866,12 +832,7 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
       var response =
           Assertions.assertInstanceOf(
               BalancerHandler.PutPlanResponse.class,
-              handler
-                  .put(
-                      Channel.ofRequest(
-                          JsonConverter.defaultConverter().toJson(Map.of("id", post.id))))
-                  .toCompletableFuture()
-                  .join());
+              handler.put(httpRequest(Map.of("id", post.id))).toCompletableFuture().join());
       Assertions.assertNotNull(response.id, "The plan should be executed");
 
       // exception
@@ -900,13 +861,7 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
       // plan doesn't exists
       Assertions.assertThrows(
           IllegalArgumentException.class,
-          () ->
-              handler
-                  .put(
-                      Channel.ofRequest(
-                          JsonConverter.defaultConverter().toJson(Map.of("id", "no such plan"))))
-                  .toCompletableFuture()
-                  .join(),
+          () -> handler.put(httpRequest(Map.of("id", "no such plan"))).toCompletableFuture().join(),
           "This plan doesn't exists");
     }
   }
@@ -923,13 +878,7 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
               Map.of(COST_WEIGHT_KEY, defaultDecreasing, TOPICS_KEY, String.join(",", topics)));
 
       Assertions.assertDoesNotThrow(
-          () ->
-              handler
-                  .put(
-                      Channel.ofRequest(
-                          JsonConverter.defaultConverter().toJson(Map.of("id", progress.id))))
-                  .toCompletableFuture()
-                  .join(),
+          () -> handler.put(httpRequest(Map.of("id", progress.id))).toCompletableFuture().join(),
           "Schedule the rebalance task");
 
       // Wait until the migration occurred
@@ -946,13 +895,7 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
       }
 
       Assertions.assertDoesNotThrow(
-          () ->
-              handler
-                  .put(
-                      Channel.ofRequest(
-                          JsonConverter.defaultConverter().toJson(Map.of("id", progress.id))))
-                  .toCompletableFuture()
-                  .join(),
+          () -> handler.put(httpRequest(Map.of("id", progress.id))).toCompletableFuture().join(),
           "Idempotent behavior");
     }
   }
@@ -1006,14 +949,22 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
       var clusterInfo =
           admin.topicNames(false).thenCompose(admin::clusterInfo).toCompletableFuture().join();
       {
-        // default
-        var postRequest =
-            BalancerHandler.parsePostRequestWrapper(new BalancerPostRequest(), clusterInfo);
+        // no cost weight
+        Assertions.assertThrows(
+            IllegalArgumentException.class,
+            () -> BalancerHandler.parsePostRequestWrapper(new BalancerPostRequest(), clusterInfo),
+            "CostWeights must be specified");
+      }
+      {
+        // minimal
+        var request = new BalancerPostRequest();
+        request.costWeights = List.of(costWeight(DecreasingCost.class.getName(), 1));
+        var postRequest = BalancerHandler.parsePostRequestWrapper(request, clusterInfo);
         var config = postRequest.algorithmConfig;
         Assertions.assertTrue(config.config().entrySet().isEmpty());
         Assertions.assertInstanceOf(HasClusterCost.class, config.clusterCostFunction());
-        Assertions.assertEquals(
-            BalancerHandler.DEFAULT_CLUSTER_COST_FUNCTION, config.clusterCostFunction());
+        Assertions.assertTrue(config.clusterCostFunction().toString().contains("DecreasingCost"));
+        Assertions.assertTrue(config.clusterCostFunction().toString().contains("weight 1"));
         Assertions.assertEquals(TIMEOUT_DEFAULT, postRequest.executionTime.toSeconds());
         Assertions.assertTrue(
             clusterInfo.topics().stream().allMatch(t -> config.topicFilter().test(t)));
@@ -1088,10 +1039,7 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
     try (var admin = Admin.of(bootstrapServers())) {
       var costFunction = Collections.singleton(costWeight(TimeoutCost.class.getName(), 1));
       var handler = new BalancerHandler(admin, (ignore) -> Optional.of(jmxServiceURL().getPort()));
-      var channel =
-          Channel.ofRequest(
-              JsonConverter.defaultConverter()
-                  .toJson(Map.of(TIMEOUT_KEY, "10", COST_WEIGHT_KEY, costFunction)));
+      var channel = httpRequest(Map.of(TIMEOUT_KEY, "10", COST_WEIGHT_KEY, costFunction));
       var post =
           (BalancerHandler.PostPlanResponse) handler.post(channel).toCompletableFuture().join();
       Utils.sleep(Duration.ofSeconds(11));
@@ -1285,12 +1233,13 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
   /** Submit the plan and wait until it generated. */
   private BalancerHandler.PlanExecutionProgress submitPlanGeneration(
       BalancerHandler handler, Map<String, Object> requestBody) {
+    if (!requestBody.containsKey(COST_WEIGHT_KEY)) {
+      requestBody = new HashMap<>(requestBody);
+      requestBody.put(COST_WEIGHT_KEY, defaultDecreasing);
+    }
     var post =
         (BalancerHandler.PostPlanResponse)
-            handler
-                .post(Channel.ofRequest(JsonConverter.defaultConverter().toJson(requestBody)))
-                .toCompletableFuture()
-                .join();
+            handler.post(httpRequest(requestBody)).toCompletableFuture().join();
     Utils.waitFor(
         () -> {
           var progress =
@@ -1411,4 +1360,8 @@ public class BalancerHandlerTest extends RequireBrokerCluster {
     cw.weight = Optional.of(weight);
     return cw;
   }
+
+  private static Channel httpRequest(Map<String, ?> payload) {
+    return Channel.ofRequest(JsonConverter.defaultConverter().toJson(payload));
+  }
 }
diff --git a/common/src/test/java/org/astraea/common/metrics/MBeanRegisterTest.java b/common/src/test/java/org/astraea/common/metrics/MBeanRegisterTest.java
index 9103710b..ce1b8d3c 100644
--- a/common/src/test/java/org/astraea/common/metrics/MBeanRegisterTest.java
+++ b/common/src/test/java/org/astraea/common/metrics/MBeanRegisterTest.java
@@ -19,11 +19,27 @@ package org.astraea.common.metrics;
 import java.util.Map;
 import java.util.UUID;
 import java.util.function.Supplier;
+import org.astraea.common.metrics.client.producer.ProducerMetrics;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 class MBeanRegisterTest {
 
+  @Test
+  void testNodeMetrics() {
+    MBeanRegister.local()
+        .domainName("kafka.producer")
+        .property("type", "producer-node-metrics")
+        .property("node-id", "100")
+        .property("client-id", "xxxx")
+        .attribute("incoming-byte-rate", Double.class, () -> 10D)
+        .register();
+
+    var metrics = ProducerMetrics.nodes(MBeanClient.local());
+    Assertions.assertEquals(1, metrics.size());
+    Assertions.assertEquals(10D, metrics.iterator().next().incomingByteRate());
+  }
+
   @Test
   void testBuilder() {
     try (MBeanClient client = MBeanClient.local()) {
@@ -36,11 +52,11 @@ class MBeanRegisterTest {
 
       // register
       MBeanRegister.local()
-          .setDomainName(domainName)
-          .setDescription("Hello World")
-          .addProperty("id", id)
-          .addAttribute("Name", String.class, () -> "Robert")
-          .addAttribute("Age", Integer.class, () -> 43)
+          .domainName(domainName)
+          .description("Hello World")
+          .property("id", id)
+          .attribute("Name", String.class, () -> "Robert")
+          .attribute("Age", Integer.class, () -> 43)
           .register();
       Assertions.assertEquals(domainName, bean.get().domainName());
       Assertions.assertEquals(Map.of("id", id), bean.get().properties());
diff --git a/common/src/test/java/org/astraea/common/partitioner/StrictCostDispatcherPerfTest.java b/common/src/test/java/org/astraea/common/partitioner/StrictCostDispatcherPerfTest.java
new file mode 100644
index 00000000..de1016f4
--- /dev/null
+++ b/common/src/test/java/org/astraea/common/partitioner/StrictCostDispatcherPerfTest.java
@@ -0,0 +1,144 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.common.partitioner;
+
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.astraea.common.Configuration;
+import org.astraea.common.Utils;
+import org.astraea.common.admin.ClusterInfo;
+import org.astraea.common.admin.NodeInfo;
+import org.astraea.common.admin.ReplicaInfo;
+import org.astraea.common.metrics.MBeanRegister;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class StrictCostDispatcherPerfTest {
+
+  private static AtomicLong createMetric(int brokerId) {
+    var latency = new AtomicLong(100);
+    MBeanRegister.local()
+        .domainName("kafka.producer")
+        .property("type", "producer-node-metrics")
+        .property("node-id", String.valueOf(brokerId))
+        .property("client-id", "xxxx")
+        .attribute("request-latency-avg", Double.class, () -> (double) latency.get())
+        .register();
+    return latency;
+  }
+
+  @Test
+  void test() {
+    var node0 = NodeInfo.of(0, "node0", 2222);
+    var node1 = NodeInfo.of(1, "node1", 2222);
+    var node2 = NodeInfo.of(2, "node2", 2222);
+    var clusterInfo =
+        ClusterInfo.of(
+            List.of(node0, node1, node2),
+            List.of(
+                ReplicaInfo.of("topic", 0, node0, true, true, false),
+                ReplicaInfo.of("topic", 1, node1, true, true, false),
+                ReplicaInfo.of("topic", 2, node2, true, true, false)));
+
+    var node0Latency = createMetric(0);
+    var node1Latency = createMetric(1);
+    var node2Latency = createMetric(2);
+
+    var key = "key".getBytes(StandardCharsets.UTF_8);
+    var value = "value".getBytes(StandardCharsets.UTF_8);
+    try (var dispatcher = new StrictCostDispatcher()) {
+      dispatcher.configure(Configuration.of(Map.of("round.robin.lease", "2s")));
+
+      Supplier<Map<Integer, List<Integer>>> resultSupplier =
+          () -> {
+            var result =
+                IntStream.range(0, 1000)
+                    .mapToObj(ignored -> dispatcher.partition("topic", key, value, clusterInfo))
+                    .collect(Collectors.groupingBy(i -> i));
+
+            var keys =
+                Arrays.stream(dispatcher.roundRobin)
+                    .boxed()
+                    .collect(Collectors.groupingBy(i -> i))
+                    .keySet();
+            Assertions.assertEquals(3, keys.size(), "keys: " + keys);
+
+            Assertions.assertEquals(3, result.size());
+            result.values().forEach(v -> Assertions.assertNotEquals(0, v.size()));
+            return result;
+          };
+
+      // first run
+      node0Latency.set(100);
+      node1Latency.set(10);
+      node2Latency.set(1);
+      Utils.sleep(Duration.ofSeconds(2));
+
+      var result = resultSupplier.get();
+
+      Assertions.assertTrue(
+          result.get(2).size() > result.get(1).size(),
+          "broker 0: "
+              + result.get(0).size()
+              + " broker 1: "
+              + result.get(1).size()
+              + " broker 2: "
+              + result.get(2).size());
+
+      Assertions.assertTrue(
+          result.get(1).size() > result.get(0).size(),
+          "broker 0: "
+              + result.get(0).size()
+              + " broker 1: "
+              + result.get(1).size()
+              + " broker 2: "
+              + result.get(2).size());
+
+      // second run
+      node0Latency.set(77);
+      node1Latency.set(3);
+      node2Latency.set(999);
+      Utils.sleep(Duration.ofSeconds(3));
+
+      result = resultSupplier.get();
+      Assertions.assertTrue(
+          result.get(1).size() > result.get(0).size(),
+          "broker 0: "
+              + result.get(0).size()
+              + " broker 1: "
+              + result.get(1).size()
+              + " broker 2: "
+              + result.get(2).size());
+
+      Assertions.assertTrue(
+          result.get(0).size() > result.get(2).size(),
+          "broker 0: "
+              + result.get(0).size()
+              + " broker 1: "
+              + result.get(1).size()
+              + " broker 2: "
+              + result.get(2).size());
+    }
+  }
+}
diff --git a/common/src/test/java/org/astraea/common/partitioner/StrictCostDispatcherTest.java b/common/src/test/java/org/astraea/common/partitioner/StrictCostDispatcherTest.java
index c800aef3..68260a75 100644
--- a/common/src/test/java/org/astraea/common/partitioner/StrictCostDispatcherTest.java
+++ b/common/src/test/java/org/astraea/common/partitioner/StrictCostDispatcherTest.java
@@ -95,7 +95,7 @@ public class StrictCostDispatcherTest {
   @Test
   void testNoAvailableBrokers() {
     try (var dispatcher = new StrictCostDispatcher()) {
-      dispatcher.configure(Map.of(), Optional.empty(), Map.of(), Duration.ofSeconds(10));
+      dispatcher.configure(Configuration.EMPTY);
       Assertions.assertEquals(
           0, dispatcher.partition("topic", new byte[0], new byte[0], ClusterInfo.empty()));
     }
@@ -106,7 +106,7 @@ public class StrictCostDispatcherTest {
     var nodeInfo = NodeInfo.of(10, "host", 11111);
     var replicaInfo = ReplicaInfo.of("topic", 10, nodeInfo, true, true, false);
     try (var dispatcher = new StrictCostDispatcher()) {
-      dispatcher.configure(Map.of(), Optional.empty(), Map.of(), Duration.ofSeconds(10));
+      dispatcher.configure(Configuration.EMPTY);
       Assertions.assertEquals(
           10,
           dispatcher.partition(
@@ -152,6 +152,15 @@ public class StrictCostDispatcherTest {
         () -> StrictCostDispatcher.parseCostFunctionWeight(config2));
   }
 
+  public static class DumbHasBrokerCost implements HasBrokerCost {
+
+    @Override
+    public BrokerCost brokerCost(
+        ClusterInfo<? extends ReplicaInfo> clusterInfo, ClusterBean clusterBean) {
+      return Map::of;
+    }
+  }
+
   @Test
   void testCostFunctionWithoutFetcher() {
     HasBrokerCost costFunction = (clusterInfo, bean) -> Mockito.mock(BrokerCost.class);
@@ -159,8 +168,7 @@ public class StrictCostDispatcherTest {
     var replicaInfo1 =
         ReplicaInfo.of("topic", 1, NodeInfo.of(12, "host2", 11111), true, true, true);
     try (var dispatcher = new StrictCostDispatcher()) {
-      dispatcher.configure(
-          Map.of(costFunction, 1D), Optional.empty(), Map.of(), Duration.ofSeconds(10));
+      dispatcher.configure(Configuration.of((Map.of(DumbHasBrokerCost.class.getName(), "1"))));
       dispatcher.partition(
           "topic",
           new byte[0],
@@ -175,19 +183,15 @@ public class StrictCostDispatcherTest {
     try (var dispatcher = new StrictCostDispatcher()) {
 
       // pass due to local mbean
-      dispatcher.configure(
-          Map.of(new NodeThroughputCost(), 1D), Optional.empty(), Map.of(), Duration.ofSeconds(10));
-
-      // pass due to default port
-      dispatcher.configure(
-          Map.of(new NodeThroughputCost(), 1D), Optional.of(111), Map.of(), Duration.ofSeconds(10));
+      dispatcher.configure(Configuration.of(Map.of(NodeThroughputCost.class.getName(), "1")));
+    }
+  }
 
-      // pass due to specify port
-      dispatcher.configure(
-          Map.of(new NodeThroughputCost(), 1D),
-          Optional.empty(),
-          Map.of(222, 111),
-          Duration.ofSeconds(10));
+  public static class MyFunction implements HasBrokerCost {
+    @Override
+    public BrokerCost brokerCost(
+        ClusterInfo<? extends ReplicaInfo> clusterInfo, ClusterBean clusterBean) {
+      return () -> Map.of(22, 10D);
     }
   }
 
@@ -196,16 +200,7 @@ public class StrictCostDispatcherTest {
     var brokerId = 22;
     var partitionId = 123;
     try (var dispatcher = new StrictCostDispatcher()) {
-      var costFunction =
-          new HasBrokerCost() {
-            @Override
-            public BrokerCost brokerCost(
-                ClusterInfo<? extends ReplicaInfo> clusterInfo, ClusterBean clusterBean) {
-              return () -> Map.of(brokerId, 10D);
-            }
-          };
-      dispatcher.configure(
-          Map.of(costFunction, 1D), Optional.empty(), Map.of(), Duration.ofSeconds(10));
+      dispatcher.configure(Configuration.of(Map.of(MyFunction.class.getName(), "1")));
 
       var replicaInfo0 =
           ReplicaInfo.of(