Repository: skiptests/astraea
Pull request #469 (2022-07-09 16:18:25)

Introduce NodeLatencyCost depending on local metrics to be default co…
1. `NodeLatencyCost`透過本地producer metrics來衡量節點延遲
2. `StrictCostDispatcher`的預設cost function改用`NodeLatencyCost`


Patch:
diff --git a/app/src/main/java/org/astraea/app/cost/NodeLatencyCost.java b/app/src/main/java/org/astraea/app/cost/NodeLatencyCost.java
new file mode 100644
index 00000000..0ecbab2f
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/cost/NodeLatencyCost.java
@@ -0,0 +1,54 @@
+/*
+ * 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.app.cost;
+
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import org.astraea.app.admin.ClusterInfo;
+import org.astraea.app.metrics.HasBeanObject;
+import org.astraea.app.metrics.KafkaMetrics;
+import org.astraea.app.metrics.collector.Fetcher;
+import org.astraea.app.metrics.producer.HasProducerNodeMetrics;
+
+public class NodeLatencyCost implements HasBrokerCost {
+
+  @Override
+  public BrokerCost brokerCost(ClusterInfo clusterInfo) {
+    var result =
+        clusterInfo.clusterBean().all().entrySet().stream()
+            .collect(
+                Collectors.toMap(
+                    Map.Entry::getKey,
+                    e ->
+                        e.getValue().stream()
+                            .filter(b -> b instanceof HasProducerNodeMetrics)
+                            .map(b -> (HasProducerNodeMetrics) b)
+                            .mapToDouble(HasProducerNodeMetrics::requestLatencyAvg)
+                            .sum()));
+    return () -> result;
+  }
+
+  @Override
+  public Optional<Fetcher> fetcher() {
+    return Optional.of(
+        client ->
+            KafkaMetrics.Producer.nodes(client).values().stream()
+                .map(b -> (HasBeanObject) b)
+                .collect(Collectors.toUnmodifiableList()));
+  }
+}
diff --git a/app/src/main/java/org/astraea/app/metrics/KafkaMetrics.java b/app/src/main/java/org/astraea/app/metrics/KafkaMetrics.java
index ddb0cb00..7e48cb9e 100644
--- a/app/src/main/java/org/astraea/app/metrics/KafkaMetrics.java
+++ b/app/src/main/java/org/astraea/app/metrics/KafkaMetrics.java
@@ -19,6 +19,7 @@ package org.astraea.app.metrics;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Map;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 import org.astraea.app.metrics.broker.BrokerTopicMetricsResult;
 import org.astraea.app.metrics.broker.HasValue;
@@ -400,6 +401,36 @@ public final class KafkaMetrics {
               Collectors.toUnmodifiableMap(b -> b.properties().get("client-id"), b -> () -> b));
     }
 
+    /**
+     * collect HasProducerNodeMetrics from all producers.
+     *
+     * @param mBeanClient to query metrics
+     * @return key is broker id, and value is associated to broker metrics recorded by all producers
+     */
+    public static Map<Integer, Collection<HasProducerNodeMetrics>> nodes(MBeanClient mBeanClient) {
+      Function<String, Integer> brokerId =
+          string -> Integer.parseInt(string.substring(string.indexOf("-") + 1));
+      return mBeanClient
+          .queryBeans(
+              BeanQuery.builder()
+                  .domainName("kafka.producer")
+                  .property("type", "producer-node-metrics")
+                  .property("node-id", "*")
+                  .property("client-id", "*")
+                  .build())
+          .stream()
+          .collect(Collectors.groupingBy(b -> brokerId.apply(b.properties().get("node-id"))))
+          .entrySet()
+          .stream()
+          .collect(
+              Collectors.toMap(
+                  Map.Entry::getKey,
+                  e ->
+                      e.getValue().stream()
+                          .map(b -> (HasProducerNodeMetrics) (() -> b))
+                          .collect(Collectors.toUnmodifiableList())));
+    }
+
     /**
      * topic metrics traced by producer
      *
diff --git a/app/src/main/java/org/astraea/app/partitioner/RoundRobin.java b/app/src/main/java/org/astraea/app/partitioner/RoundRobin.java
index d920587c..e007e18f 100644
--- a/app/src/main/java/org/astraea/app/partitioner/RoundRobin.java
+++ b/app/src/main/java/org/astraea/app/partitioner/RoundRobin.java
@@ -30,8 +30,8 @@ import java.util.stream.Collectors;
  */
 public interface RoundRobin<E> {
 
-  static <E> RoundRobin<E> smoothWeighted(Map<E, Double> weights) {
-    return new SmoothWeightedRoundRobin<>(weights);
+  static <E> RoundRobin<E> smooth(Map<E, Double> scores) {
+    return new SmoothRoundRobin<>(scores);
   }
 
   /**
@@ -42,44 +42,44 @@ public interface RoundRobin<E> {
    */
   Optional<E> next(Set<E> availableTargets);
 
-  class SmoothWeightedRoundRobin<E> implements RoundRobin<E> {
+  class SmoothRoundRobin<E> implements RoundRobin<E> {
 
-    private final Map<E, Double> effectiveWeights;
-    private volatile Map<E, Double> currentWeights;
+    private final Map<E, Double> effectiveScores;
+    private volatile Map<E, Double> currentScores;
 
-    private SmoothWeightedRoundRobin(Map<E, Double> weights) {
-      this.effectiveWeights = Collections.unmodifiableMap(weights);
-      this.currentWeights =
-          weights.keySet().stream()
+    private SmoothRoundRobin(Map<E, Double> scores) {
+      this.effectiveScores = Collections.unmodifiableMap(scores);
+      this.currentScores =
+          scores.keySet().stream()
               .collect(Collectors.toUnmodifiableMap(Function.identity(), ignored -> 0D));
     }
 
     @Override
     public Optional<E> next(Set<E> availableTargets) {
       // no data no answer
-      if (effectiveWeights.isEmpty() || availableTargets.isEmpty()) return Optional.empty();
+      if (effectiveScores.isEmpty() || availableTargets.isEmpty()) return Optional.empty();
 
-      // 1) calculate the sum of all effective weights
-      var sum = effectiveWeights.values().stream().mapToDouble(d -> d).sum();
-      // 2) add effective weight to each current weight
-      var nextWeights =
-          currentWeights.entrySet().stream()
+      // 1) calculate the sum of all effective scores
+      var sum = effectiveScores.values().stream().mapToDouble(d -> d).sum();
+      // 2) add effective score to each current score
+      var nextScores =
+          currentScores.entrySet().stream()
               .collect(
                   Collectors.toMap(
                       Map.Entry::getKey,
-                      e -> effectiveWeights.getOrDefault(e.getKey(), 0D) + e.getValue()));
+                      e -> effectiveScores.getOrDefault(e.getKey(), 0D) + e.getValue()));
       // 3) get the E which has max value
       var maxObj =
-          nextWeights.entrySet().stream()
+          nextScores.entrySet().stream()
               .filter(e -> availableTargets.contains(e.getKey()))
               .max(Map.Entry.comparingByValue())
               .map(Map.Entry::getKey);
-      // 4) subtract weight from max (effective weight)
+      // 4) subtract score from max (effective score)
       maxObj.ifPresent(
           o -> {
-            nextWeights.put(o, nextWeights.get(o) - sum);
-            // 5) update current weights by next weights
-            currentWeights = Collections.unmodifiableMap(nextWeights);
+            nextScores.put(o, nextScores.get(o) - sum);
+            // 5) update current scores by next scores
+            currentScores = Collections.unmodifiableMap(nextScores);
           });
       return maxObj;
     }
diff --git a/app/src/main/java/org/astraea/app/partitioner/StrictCostDispatcher.java b/app/src/main/java/org/astraea/app/partitioner/StrictCostDispatcher.java
index b55fda60..179fb03c 100644
--- a/app/src/main/java/org/astraea/app/partitioner/StrictCostDispatcher.java
+++ b/app/src/main/java/org/astraea/app/partitioner/StrictCostDispatcher.java
@@ -31,7 +31,7 @@ import org.astraea.app.admin.ReplicaInfo;
 import org.astraea.app.common.Utils;
 import org.astraea.app.cost.CostFunction;
 import org.astraea.app.cost.HasBrokerCost;
-import org.astraea.app.cost.ReplicaLeaderCost;
+import org.astraea.app.cost.NodeLatencyCost;
 import org.astraea.app.metrics.collector.BeanCollector;
 import org.astraea.app.metrics.collector.Fetcher;
 import org.astraea.app.metrics.collector.Receiver;
@@ -41,10 +41,11 @@ import org.astraea.app.metrics.collector.Receiver;
  * node by different metrics. The default cost function ranks nodes by replica leader. It means the
  * node having lower replica leaders get higher score.
  *
- * <p>The requisite config is JMX port. Most cost functions need the JMX metrics to score nodes.
+ * <p>The important 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`.
+ * the broker id) to replace the value of `jmx.port`. If the jmx port is undefined, only local mbean
+ * client is created for each cost function
  *
  * <p>You can configure the cost functions you want to use. By giving the name of that cost function
  * and its weight. For example,
@@ -58,7 +59,7 @@ public class StrictCostDispatcher implements Dispatcher {
       BeanCollector.builder().interval(Duration.ofSeconds(4)).build();
 
   // The cost-functions we consider and the weight of them. It is visible for test
-  Map<CostFunction, Double> functions = Map.of(new ReplicaLeaderCost.NoMetrics(), 1D);
+  Map<CostFunction, Double> functions = Map.of();
 
   // all-in-one fetcher referenced to cost functions
   Optional<Fetcher> fetcher;
@@ -127,6 +128,22 @@ public class StrictCostDispatcher implements Dispatcher {
         .orElse(0);
   }
 
+  /**
+   * The value of cost returned from cost function is conflict to score, since the higher cost
+   * represents lower score. This helper reverses the cost by subtracting the cost from "max cost".
+   *
+   * @param cost to convert
+   * @return weights
+   */
+  static Map<Integer, Double> costToScore(Map<Integer, Double> cost) {
+    var max = cost.values().stream().max(Double::compare);
+    return max.map(
+            m ->
+                cost.entrySet().stream()
+                    .collect(Collectors.toMap(Map.Entry::getKey, e -> m - e.getValue())))
+        .orElse(cost);
+  }
+
   /**
    * create new Round-Robin based.
    *
@@ -136,34 +153,30 @@ public class StrictCostDispatcher implements Dispatcher {
    */
   static RoundRobin<Integer> newRoundRobin(
       Map<CostFunction, Double> costFunctions, ClusterInfo clusterInfo) {
-    return RoundRobin.smoothWeighted(
+    var weightedCost =
         costFunctions.entrySet().stream()
             .filter(e -> e.getKey() instanceof HasBrokerCost)
-            .map(e -> Map.entry((HasBrokerCost) e.getKey(), e.getValue()))
             .flatMap(
                 functionWeight ->
-                    functionWeight
-                        .getKey()
-                        // Execute all cost functions
-                        .brokerCost(clusterInfo)
-                        .value()
-                        .entrySet()
-                        .stream()
-                        // Weight on cost functions result
-                        .map(
-                            IdScore ->
-                                Map.entry(
-                                    IdScore.getKey(),
-                                    IdScore.getValue() * functionWeight.getValue())))
+                    ((HasBrokerCost) functionWeight.getKey())
+                        .brokerCost(clusterInfo).value().entrySet().stream()
+                            .map(
+                                idAndCost ->
+                                    Map.entry(
+                                        idAndCost.getKey(),
+                                        idAndCost.getValue() * functionWeight.getValue())))
             .collect(
                 Collectors.toMap(
-                    Map.Entry::getKey, Map.Entry::getValue, Double::sum, HashMap::new)));
+                    Map.Entry::getKey, Map.Entry::getValue, Double::sum, HashMap::new));
+    return RoundRobin.smooth(costToScore(weightedCost));
   }
 
   @Override
   public void configure(Configuration config) {
+    var configuredFunctions = parseCostFunctionWeight(config);
+
     configure(
-        parseCostFunctionWeight(config),
+        configuredFunctions.isEmpty() ? Map.of(new NodeLatencyCost(), 1D) : configuredFunctions,
         config.integer(JMX_PORT),
         PartitionerUtils.parseIdJMXPort(config));
   }
@@ -179,17 +192,18 @@ public class StrictCostDispatcher implements Dispatcher {
       Map<CostFunction, Double> functions,
       Optional<Integer> jmxPortDefault,
       Map<Integer, Integer> customJmxPort) {
-    if (!functions.isEmpty()) this.functions = functions;
-    this.fetcher = Fetcher.of(functions.keySet());
+    this.functions = functions;
+    this.fetcher = Fetcher.of(this.functions.keySet());
     this.jmxPortGetter =
         id ->
             Optional.ofNullable(customJmxPort.get(id))
                 .or(() -> jmxPortDefault)
                 .orElseThrow(
                     () -> new NoSuchElementException("broker: " + id + " does not have jmx port"));
-    if (fetcher.isPresent() && jmxPortDefault.isEmpty() && customJmxPort.isEmpty())
-      throw new IllegalArgumentException(
-          "JMX port is empty but the cost functions need metrics from JMX server");
+
+    // put local mbean client first
+    this.fetcher.ifPresent(
+        f -> receivers.put(-1, beanCollector.register().local().fetcher(f).build()));
   }
 
   /**


Test patch:
diff --git a/app/src/test/java/org/astraea/app/cost/NodeLatencyCostTest.java b/app/src/test/java/org/astraea/app/cost/NodeLatencyCostTest.java
new file mode 100644
index 00000000..54562c2a
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/cost/NodeLatencyCostTest.java
@@ -0,0 +1,65 @@
+/*
+ * 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.app.cost;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.stream.Collectors;
+import org.astraea.app.admin.Admin;
+import org.astraea.app.admin.ClusterBean;
+import org.astraea.app.admin.ClusterInfo;
+import org.astraea.app.common.Utils;
+import org.astraea.app.metrics.HasBeanObject;
+import org.astraea.app.metrics.KafkaMetrics;
+import org.astraea.app.metrics.jmx.MBeanClient;
+import org.astraea.app.producer.Producer;
+import org.astraea.app.service.RequireBrokerCluster;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+public class NodeLatencyCostTest extends RequireBrokerCluster {
+
+  @Test
+  void testCost() {
+    var brokerId = brokerIds().iterator().next();
+    var topic = Utils.randomString(10);
+    try (var admin = Admin.of(bootstrapServers());
+        var producer = Producer.of(bootstrapServers())) {
+      admin.creator().topic(topic).numberOfPartitions(1).create();
+      Utils.sleep(Duration.ofSeconds(3));
+      producer.sender().topic(Utils.randomString(10)).value(new byte[100]).run();
+      producer.flush();
+
+      var beans = KafkaMetrics.Producer.node(MBeanClient.local(), brokerId);
+      var clusterBean =
+          ClusterBean.of(
+              Map.of(
+                  brokerId,
+                  beans.values().stream()
+                      .map(b -> (HasBeanObject) b)
+                      .collect(Collectors.toUnmodifiableList())));
+      var clusterInfo = Mockito.mock(ClusterInfo.class);
+      Mockito.when(clusterInfo.clusterBean()).thenReturn(clusterBean);
+
+      var function = new NodeLatencyCost();
+      var cost = function.brokerCost(clusterInfo);
+      Assertions.assertEquals(1, cost.value().size());
+      Assertions.assertNotNull(cost.value().get(brokerId));
+    }
+  }
+}
diff --git a/app/src/test/java/org/astraea/app/metrics/producer/HasProducerNodeMetricsTest.java b/app/src/test/java/org/astraea/app/metrics/producer/HasProducerNodeMetricsTest.java
index ae8f0ec2..fab789f9 100644
--- a/app/src/test/java/org/astraea/app/metrics/producer/HasProducerNodeMetricsTest.java
+++ b/app/src/test/java/org/astraea/app/metrics/producer/HasProducerNodeMetricsTest.java
@@ -16,37 +16,90 @@
  */
 package org.astraea.app.metrics.producer;
 
+import java.time.Duration;
+import java.util.Collection;
+import java.util.Set;
 import java.util.concurrent.ExecutionException;
+import org.astraea.app.admin.Admin;
+import org.astraea.app.admin.TopicPartition;
 import org.astraea.app.common.Utils;
 import org.astraea.app.metrics.KafkaMetrics;
 import org.astraea.app.metrics.jmx.MBeanClient;
 import org.astraea.app.producer.Producer;
-import org.astraea.app.service.RequireSingleBrokerCluster;
+import org.astraea.app.service.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
-public class HasProducerNodeMetricsTest extends RequireSingleBrokerCluster {
+public class HasProducerNodeMetricsTest extends RequireBrokerCluster {
 
   @Test
-  void testAttributes() throws ExecutionException, InterruptedException {
+  void testSingleBroker() throws ExecutionException, InterruptedException {
     var topic = Utils.randomString(10);
-    try (var producer = Producer.of(bootstrapServers())) {
+    try (var admin = Admin.of(bootstrapServers());
+        var producer = Producer.of(bootstrapServers())) {
+      admin.creator().topic(topic).numberOfPartitions(1).create();
+      Utils.sleep(Duration.ofSeconds(3));
+      var owner = admin.replicas(Set.of(topic)).get(new TopicPartition(topic, 0)).get(0).broker();
       producer.sender().topic(topic).run().toCompletableFuture().get();
-      var metrics = KafkaMetrics.Producer.node(MBeanClient.local(), brokerIds().iterator().next());
+      var metrics = KafkaMetrics.Producer.node(MBeanClient.local(), owner);
       Assertions.assertEquals(1, metrics.size());
-      var producerNodeMetrics = metrics.get("producer-1");
-      Assertions.assertNotEquals(0D, producerNodeMetrics.incomingByteRate());
-      Assertions.assertNotEquals(0D, producerNodeMetrics.incomingByteTotal());
-      Assertions.assertNotEquals(0D, producerNodeMetrics.outgoingByteRate());
-      Assertions.assertNotEquals(0D, producerNodeMetrics.outgoingByteTotal());
-      Assertions.assertNotEquals(0D, producerNodeMetrics.requestLatencyAvg());
-      Assertions.assertNotEquals(0D, producerNodeMetrics.requestLatencyMax());
-      Assertions.assertNotEquals(0D, producerNodeMetrics.requestRate());
-      Assertions.assertNotEquals(0D, producerNodeMetrics.requestSizeAvg());
-      Assertions.assertNotEquals(0D, producerNodeMetrics.requestSizeMax());
-      Assertions.assertNotEquals(0D, producerNodeMetrics.requestTotal());
-      Assertions.assertNotEquals(0D, producerNodeMetrics.responseRate());
-      Assertions.assertNotEquals(0D, producerNodeMetrics.responseTotal());
+      check(metrics.get("producer-1"));
     }
   }
+
+  @Test
+  void testMultiBrokers() throws ExecutionException, InterruptedException {
+    var topic = Utils.randomString(10);
+    try (var admin = Admin.of(bootstrapServers());
+        var producer = Producer.of(bootstrapServers())) {
+      admin.creator().topic(topic).numberOfPartitions(3).create();
+      Utils.sleep(Duration.ofSeconds(3));
+      var owner = admin.replicas(Set.of(topic)).get(new TopicPartition(topic, 0)).get(0).broker();
+      producer
+          .sender()
+          .topic(topic)
+          .value(new byte[10])
+          .partition(0)
+          .run()
+          .toCompletableFuture()
+          .get();
+      producer
+          .sender()
+          .topic(topic)
+          .value(new byte[10])
+          .partition(1)
+          .run()
+          .toCompletableFuture()
+          .get();
+      producer
+          .sender()
+          .topic(topic)
+          .value(new byte[10])
+          .partition(2)
+          .run()
+          .toCompletableFuture()
+          .get();
+      var metrics = KafkaMetrics.Producer.nodes(MBeanClient.local());
+      Assertions.assertNotEquals(1, metrics.size());
+      Assertions.assertTrue(metrics.keySet().containsAll(brokerIds()));
+      metrics.values().stream()
+          .flatMap(Collection::stream)
+          .forEach(HasProducerNodeMetricsTest::check);
+    }
+  }
+
+  private static void check(HasProducerNodeMetrics metrics) {
+    Assertions.assertNotEquals(0D, metrics.incomingByteRate());
+    Assertions.assertNotEquals(0D, metrics.incomingByteTotal());
+    Assertions.assertNotEquals(0D, metrics.outgoingByteRate());
+    Assertions.assertNotEquals(0D, metrics.outgoingByteTotal());
+    Assertions.assertNotEquals(0D, metrics.requestLatencyAvg());
+    Assertions.assertNotEquals(0D, metrics.requestLatencyMax());
+    Assertions.assertNotEquals(0D, metrics.requestRate());
+    Assertions.assertNotEquals(0D, metrics.requestSizeAvg());
+    Assertions.assertNotEquals(0D, metrics.requestSizeMax());
+    Assertions.assertNotEquals(0D, metrics.requestTotal());
+    Assertions.assertNotEquals(0D, metrics.responseRate());
+    Assertions.assertNotEquals(0D, metrics.responseTotal());
+  }
 }
diff --git a/app/src/test/java/org/astraea/app/partitioner/RoundRobinTest.java b/app/src/test/java/org/astraea/app/partitioner/RoundRobinTest.java
index b0eddde5..0f58f5a5 100644
--- a/app/src/test/java/org/astraea/app/partitioner/RoundRobinTest.java
+++ b/app/src/test/java/org/astraea/app/partitioner/RoundRobinTest.java
@@ -25,10 +25,10 @@ import org.junit.jupiter.api.Test;
 public class RoundRobinTest {
 
   @Test
-  void testSmoothWeighted() {
-    var weight = Map.of(0, 10D, 1, 5D);
+  void testSmoothRoundRobin() {
+    var scores = Map.of(0, 10D, 1, 5D);
 
-    var rr = RoundRobin.smoothWeighted(weight);
+    var rr = RoundRobin.smooth(scores);
 
     // nothing to pick
     Assertions.assertEquals(Optional.empty(), rr.next(Set.of()));
@@ -39,7 +39,7 @@ public class RoundRobinTest {
     Assertions.assertEquals(Optional.of(1), rr.next(Set.of(1)));
 
     // reset
-    rr = RoundRobin.smoothWeighted(weight);
+    rr = RoundRobin.smooth(scores);
     Assertions.assertEquals(Optional.of(0), rr.next(Set.of(0, 1)));
     Assertions.assertEquals(Optional.of(1), rr.next(Set.of(0, 1)));
     Assertions.assertEquals(Optional.of(0), rr.next(Set.of(0, 1)));
diff --git a/app/src/test/java/org/astraea/app/partitioner/StrictCostDispatcherTest.java b/app/src/test/java/org/astraea/app/partitioner/StrictCostDispatcherTest.java
index 3b846d6a..dbd08999 100644
--- a/app/src/test/java/org/astraea/app/partitioner/StrictCostDispatcherTest.java
+++ b/app/src/test/java/org/astraea/app/partitioner/StrictCostDispatcherTest.java
@@ -209,26 +209,27 @@ public class StrictCostDispatcherTest {
     Mockito.when(clusterInfo.availableReplicaLeaders(Mockito.anyString()))
         .thenReturn(List.of(replicaInfo0, replicaInfo1, replicaInfo2));
     Mockito.when(clusterInfo.clusterBean()).thenReturn(ClusterBean.of(Map.of()));
-    // there is no receivers by default
-    Assertions.assertEquals(0, dispatcher.receivers.size());
+    // there is one local receiver by default
+    Assertions.assertEquals(1, dispatcher.receivers.size());
+    Assertions.assertEquals(-1, dispatcher.receivers.keySet().iterator().next());
 
     // generate two receivers since there are two brokers (hosting three replicas)
     dispatcher.partition("topic", new byte[0], new byte[0], clusterInfo);
-    Assertions.assertEquals(2, dispatcher.receivers.size());
+    Assertions.assertEquals(3, dispatcher.receivers.size());
     Assertions.assertEquals(2, count.get());
 
     // all brokers have receivers already so no new receiver is born
     dispatcher.partition("topic", new byte[0], new byte[0], clusterInfo);
-    Assertions.assertEquals(2, dispatcher.receivers.size());
+    Assertions.assertEquals(3, dispatcher.receivers.size());
     Assertions.assertEquals(2, count.get());
   }
 
   @Test
   void testEmptyJmxPort() {
     var dispatcher = new StrictCostDispatcher();
-    Assertions.assertThrows(
-        IllegalArgumentException.class,
-        () -> dispatcher.configure(Map.of(new ThroughputCost(), 1D), Optional.empty(), Map.of()));
+
+    // pass due to local mbean
+    dispatcher.configure(Map.of(new ThroughputCost(), 1D), Optional.empty(), Map.of());
 
     // pass due to default port
     dispatcher.configure(Map.of(new ThroughputCost(), 1D), Optional.of(111), Map.of());
@@ -268,7 +269,15 @@ public class StrictCostDispatcherTest {
   @Test
   void testDefaultFunction() {
     var dispatcher = new StrictCostDispatcher();
-    dispatcher.configure(Map.of(), Optional.empty(), Map.of());
+    dispatcher.configure(Configuration.of(Map.of()));
     Assertions.assertEquals(1, dispatcher.functions.size());
+    Assertions.assertEquals(1, dispatcher.receivers.size());
+  }
+
+  @Test
+  void testCostToScore() {
+    var cost = Map.of(1, 100D, 2, 10D);
+    var score = StrictCostDispatcher.costToScore(cost);
+    Assertions.assertTrue(score.get(2) > score.get(1));
   }
 }