Repository: skiptests/astraea
Pull request #456 (2022-07-04 21:13:52)

Introduce ReplicaLeaderCost to be default cost function of StrictCost…
1. 新增`ReplicaLeaderCost`
2. 將`ReplicaLeaderCost`取代`Throughput`做為預設cost function


Patch:
diff --git a/app/src/main/java/org/astraea/app/cost/CostFunction.java b/app/src/main/java/org/astraea/app/cost/CostFunction.java
index 631f15af..91f22fa6 100644
--- a/app/src/main/java/org/astraea/app/cost/CostFunction.java
+++ b/app/src/main/java/org/astraea/app/cost/CostFunction.java
@@ -25,10 +25,6 @@ import org.astraea.app.metrics.collector.Fetcher;
  */
 public interface CostFunction {
 
-  static CostFunction throughput() {
-    return new ThroughputCost();
-  }
-
   /** @return the metrics getters. Those getters are used to fetch mbeans. */
   default Optional<Fetcher> fetcher() {
     return Optional.empty();
diff --git a/app/src/main/java/org/astraea/app/cost/NumberOfLeaderCost.java b/app/src/main/java/org/astraea/app/cost/NumberOfLeaderCost.java
deleted file mode 100644
index 7dd80c1f..00000000
--- a/app/src/main/java/org/astraea/app/cost/NumberOfLeaderCost.java
+++ /dev/null
@@ -1,67 +0,0 @@
-/*
- * 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.Comparator;
-import java.util.Map;
-import java.util.Optional;
-import java.util.stream.Collectors;
-import org.astraea.app.metrics.HasBeanObject;
-import org.astraea.app.metrics.collector.Fetcher;
-import org.astraea.app.metrics.kafka.HasValue;
-import org.astraea.app.metrics.kafka.KafkaMetrics;
-
-/**
- * The result is computed by "LeaderCount.Value". "LeaderCount.Value"" responds to the replica
- * leader number of brokers. The calculation method of the score is the total leader number divided
- * by the total leader number in broker
- */
-public class NumberOfLeaderCost implements HasBrokerCost {
-  @Override
-  public Optional<Fetcher> fetcher() {
-    return Optional.of(KafkaMetrics.ReplicaManager.LeaderCount::fetch);
-  }
-
-  /**
-   * @param clusterInfo cluster information
-   * @return a BrokerCost contain all ratio of leaders that exist on all brokers
-   */
-  @Override
-  public BrokerCost brokerCost(ClusterInfo clusterInfo) {
-    var leaderCount =
-        clusterInfo.clusterBean().all().entrySet().stream()
-            .flatMap(
-                e ->
-                    e.getValue().stream()
-                        .filter(x -> x instanceof HasValue)
-                        .filter(
-                            x -> x.beanObject().getProperties().get("name").equals("LeaderCount"))
-                        .filter(
-                            x ->
-                                x.beanObject().getProperties().get("type").equals("ReplicaManager"))
-                        .sorted(Comparator.comparing(HasBeanObject::createdTimestamp).reversed())
-                        .map(x -> (HasValue) x)
-                        .limit(1)
-                        .map(e2 -> Map.entry(e.getKey(), (int) e2.value())))
-            .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
-    var totalLeader = leaderCount.values().stream().mapToInt(Integer::intValue).sum();
-    var leaderCost =
-        leaderCount.entrySet().stream()
-            .collect(Collectors.toMap(Map.Entry::getKey, e -> (double) e.getValue() / totalLeader));
-    return () -> leaderCost;
-  }
-}
diff --git a/app/src/main/java/org/astraea/app/cost/ReplicaLeaderCost.java b/app/src/main/java/org/astraea/app/cost/ReplicaLeaderCost.java
new file mode 100644
index 00000000..9fe1a952
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/cost/ReplicaLeaderCost.java
@@ -0,0 +1,77 @@
+/*
+ * 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.Comparator;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import org.astraea.app.metrics.HasBeanObject;
+import org.astraea.app.metrics.collector.Fetcher;
+import org.astraea.app.metrics.kafka.HasValue;
+import org.astraea.app.metrics.kafka.KafkaMetrics;
+
+/** more replica leaders -> higher cost */
+public class ReplicaLeaderCost implements HasBrokerCost {
+
+  @Override
+  public BrokerCost brokerCost(ClusterInfo clusterInfo) {
+    var result =
+        leaderCount(clusterInfo).entrySet().stream()
+            .collect(Collectors.toMap(Map.Entry::getKey, e -> (double) e.getValue()));
+    return () -> result;
+  }
+
+  Map<Integer, Integer> leaderCount(ClusterInfo clusterInfo) {
+    return clusterInfo.clusterBean().all().entrySet().stream()
+        .flatMap(
+            e ->
+                e.getValue().stream()
+                    .filter(x -> x instanceof HasValue)
+                    .filter(x -> "LeaderCount".equals(x.beanObject().getProperties().get("name")))
+                    .filter(
+                        x -> "ReplicaManager".equals(x.beanObject().getProperties().get("type")))
+                    .sorted(Comparator.comparing(HasBeanObject::createdTimestamp).reversed())
+                    .map(x -> (HasValue) x)
+                    .limit(1)
+                    .map(e2 -> Map.entry(e.getKey(), (int) e2.value())))
+        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+  }
+
+  @Override
+  public Optional<Fetcher> fetcher() {
+    return Optional.of(KafkaMetrics.ReplicaManager.LeaderCount::fetch);
+  }
+
+  public static class NoMetrics extends ReplicaLeaderCost {
+
+    @Override
+    Map<Integer, Integer> leaderCount(ClusterInfo clusterInfo) {
+      return clusterInfo.topics().stream()
+          .flatMap(t -> clusterInfo.availableReplicaLeaders(t).stream())
+          .collect(Collectors.groupingBy(r -> r.nodeInfo().id()))
+          .entrySet()
+          .stream()
+          .collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, e -> e.getValue().size()));
+    }
+
+    @Override
+    public Optional<Fetcher> fetcher() {
+      return Optional.empty();
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/app/metrics/jmx/MBeanClient.java b/app/src/main/java/org/astraea/app/metrics/jmx/MBeanClient.java
index c46462df..23ad00e1 100644
--- a/app/src/main/java/org/astraea/app/metrics/jmx/MBeanClient.java
+++ b/app/src/main/java/org/astraea/app/metrics/jmx/MBeanClient.java
@@ -54,7 +54,11 @@ public interface MBeanClient extends AutoCloseable {
   }
 
   static MBeanClient of(JMXServiceURL url) {
-    return new MBeanClientImpl(url);
+    return MBeanClientImpl.remote(url);
+  }
+
+  static MBeanClient local() {
+    return MBeanClientImpl.local();
   }
 
   /**
diff --git a/app/src/main/java/org/astraea/app/metrics/jmx/MBeanClientImpl.java b/app/src/main/java/org/astraea/app/metrics/jmx/MBeanClientImpl.java
index 67c44c4a..36260b34 100644
--- a/app/src/main/java/org/astraea/app/metrics/jmx/MBeanClientImpl.java
+++ b/app/src/main/java/org/astraea/app/metrics/jmx/MBeanClientImpl.java
@@ -18,55 +18,80 @@ package org.astraea.app.metrics.jmx;
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
+import java.lang.management.ManagementFactory;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 import java.util.NoSuchElementException;
-import java.util.Objects;
-import java.util.Set;
 import java.util.stream.Collectors;
-import javax.management.Attribute;
 import javax.management.InstanceNotFoundException;
 import javax.management.IntrospectionException;
 import javax.management.MBeanFeatureInfo;
-import javax.management.MBeanInfo;
 import javax.management.MBeanServerConnection;
 import javax.management.ObjectInstance;
 import javax.management.ReflectionException;
-import javax.management.remote.JMXConnector;
 import javax.management.remote.JMXConnectorFactory;
 import javax.management.remote.JMXServiceURL;
 import org.astraea.app.common.Utils;
 
-class MBeanClientImpl implements MBeanClient {
+abstract class MBeanClientImpl implements MBeanClient {
 
-  private final JMXServiceURL jmxServiceURL;
-  private final JMXConnector jmxConnector;
-  private final MBeanServerConnection mBeanServerConnection;
-  private boolean isClosed;
-
-  MBeanClientImpl(JMXServiceURL jmxServiceURL) {
+  static MBeanClientImpl remote(JMXServiceURL jmxServiceURL) {
     try {
-      this.jmxServiceURL = Objects.requireNonNull(jmxServiceURL);
-      this.jmxConnector = JMXConnectorFactory.connect(jmxServiceURL);
-      this.isClosed = false;
-      this.mBeanServerConnection = jmxConnector.getMBeanServerConnection();
+      var jmxConnector = JMXConnectorFactory.connect(jmxServiceURL);
+      return new MBeanClientImpl(jmxConnector.getMBeanServerConnection()) {
+        @Override
+        public String host() {
+          return jmxServiceURL.getHost();
+        }
+
+        @Override
+        public int port() {
+          return jmxServiceURL.getPort();
+        }
+
+        @Override
+        public void close() {
+          Utils.packException(jmxConnector::close);
+        }
+      };
     } catch (IOException e) {
       throw new UncheckedIOException(e);
     }
   }
 
+  static MBeanClientImpl local() {
+    return new MBeanClientImpl(ManagementFactory.getPlatformMBeanServer()) {
+      @Override
+      public String host() {
+        return Utils.hostname();
+      }
+
+      @Override
+      public int port() {
+        return -1;
+      }
+
+      @Override
+      public void close() {}
+    };
+  }
+
+  private final MBeanServerConnection connection;
+
+  MBeanClientImpl(MBeanServerConnection connection) {
+    this.connection = connection;
+  }
+
   @Override
   public BeanObject queryBean(BeanQuery beanQuery) {
-    ensureConnected();
     try {
       // ask for MBeanInfo
-      MBeanInfo mBeanInfo = mBeanServerConnection.getMBeanInfo(beanQuery.objectName());
+      var mBeanInfo = connection.getMBeanInfo(beanQuery.objectName());
 
       // create a list builder all available attributes name
-      List<String> attributeName =
+      var attributeName =
           Arrays.stream(mBeanInfo.getAttributes())
               .map(MBeanFeatureInfo::getName)
               .collect(Collectors.toList());
@@ -84,32 +109,30 @@ class MBeanClientImpl implements MBeanClient {
 
   @Override
   public BeanObject queryBean(BeanQuery beanQuery, Collection<String> attributeNameCollection) {
-    ensureConnected();
     try {
 
       // fetch attribute value from mbean server
-      String[] attributeNameArray = attributeNameCollection.toArray(new String[0]);
-      List<Attribute> attributeList =
-          mBeanServerConnection.getAttributes(beanQuery.objectName(), attributeNameArray).asList();
+      var attributeNameArray = attributeNameCollection.toArray(new String[0]);
+      var attributeList =
+          connection.getAttributes(beanQuery.objectName(), attributeNameArray).asList();
 
       // collect attribute name & value into a map
-      Map<String, Object> attributes = new HashMap<>();
-      for (Attribute attribute : attributeList) {
-        attributes.put(attribute.getName(), attribute.getValue());
-      }
+      var attributes = new HashMap<String, Object>();
+      attributeList.forEach(attribute -> attributes.put(attribute.getName(), attribute.getValue()));
 
       // according to the javadoc of MBeanServerConnection#getAttributes, the API will
       // ignore any
       // error occurring during the fetch process (for example, attribute not exists). Below code
       // check for such condition and try to figure out what exactly the error is. put it into
       // attributes return result.
-      Set<String> notResolvedAttributes =
+      var notResolvedAttributes =
           Arrays.stream(attributeNameArray)
               .filter(str -> !attributes.containsKey(str))
               .collect(Collectors.toSet());
-      for (String attributeName : notResolvedAttributes) {
-        attributes.put(attributeName, fetchAttributeObjectOrException(beanQuery, attributeName));
-      }
+      notResolvedAttributes.forEach(
+          attributeName ->
+              attributes.put(
+                  attributeName, fetchAttributeObjectOrException(beanQuery, attributeName)));
 
       // collect result, and build a new BeanObject as return result
       return new BeanObject(beanQuery.domainName(), beanQuery.properties(), attributes);
@@ -135,7 +158,7 @@ class MBeanClientImpl implements MBeanClient {
     // exception
     // into their result.
     try {
-      return mBeanServerConnection.getAttribute(beanQuery.objectName(), attributeName);
+      return connection.getAttribute(beanQuery.objectName(), attributeName);
     } catch (IOException e) {
       throw new UncheckedIOException(e);
     } catch (Exception e) {
@@ -145,9 +168,8 @@ class MBeanClientImpl implements MBeanClient {
 
   @Override
   public Collection<BeanObject> queryBeans(BeanQuery beanQuery) {
-    ensureConnected();
     try {
-      return mBeanServerConnection.queryMBeans(beanQuery.objectName(), null).stream()
+      return connection.queryMBeans(beanQuery.objectName(), null).stream()
           .map(ObjectInstance::getObjectName)
           .map(BeanQuery::fromObjectName)
           .map(this::queryBean)
@@ -161,40 +183,9 @@ class MBeanClientImpl implements MBeanClient {
   @Override
   public List<String> listDomains() {
     try {
-      return Arrays.asList(mBeanServerConnection.getDomains());
+      return Arrays.asList(connection.getDomains());
     } catch (IOException e) {
       throw new UncheckedIOException(e);
     }
   }
-
-  /**
-   * The JMX URL of current MBeanClient instance
-   *
-   * @return the origin JMX URL used to initiate MBeanClient
-   */
-  public JMXServiceURL getAddress() {
-    return jmxServiceURL;
-  }
-
-  @Override
-  public String host() {
-    return jmxServiceURL.getHost();
-  }
-
-  @Override
-  public int port() {
-    return jmxServiceURL.getPort();
-  }
-
-  private void ensureConnected() {
-    if (isClosed) throw new IllegalStateException("MBean client is closed");
-  }
-
-  @Override
-  public void close() {
-    if (!isClosed) {
-      isClosed = true;
-      Utils.packException(jmxConnector::close);
-    }
-  }
 }
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 878731c0..003f458d 100644
--- a/app/src/main/java/org/astraea/app/partitioner/StrictCostDispatcher.java
+++ b/app/src/main/java/org/astraea/app/partitioner/StrictCostDispatcher.java
@@ -31,14 +31,15 @@ import org.astraea.app.cost.CostFunction;
 import org.astraea.app.cost.HasBrokerCost;
 import org.astraea.app.cost.NodeInfo;
 import org.astraea.app.cost.ReplicaInfo;
+import org.astraea.app.cost.ReplicaLeaderCost;
 import org.astraea.app.metrics.collector.BeanCollector;
 import org.astraea.app.metrics.collector.Fetcher;
 import org.astraea.app.metrics.collector.Receiver;
 
 /**
  * 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.
+ * 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.
  * Normally, all brokers use the same JMX port, so you can just define the `jmx.port=12345`. If one
@@ -57,7 +58,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(CostFunction.throughput(), 1D);
+  Map<CostFunction, Double> functions = Map.of(new ReplicaLeaderCost.NoMetrics(), 1D);
 
   // all-in-one fetcher referenced to cost functions
   Optional<Fetcher> fetcher;
@@ -178,7 +179,7 @@ public class StrictCostDispatcher implements Dispatcher {
       Map<CostFunction, Double> functions,
       Optional<Integer> jmxPortDefault,
       Map<Integer, Integer> customJmxPort) {
-    this.functions = functions;
+    if (!functions.isEmpty()) this.functions = functions;
     this.fetcher = Fetcher.of(functions.keySet());
     this.jmxPortGetter =
         id ->
diff --git a/docker/start_broker.sh b/docker/start_broker.sh
index bbbde237..600032b1 100755
--- a/docker/start_broker.sh
+++ b/docker/start_broker.sh
@@ -337,6 +337,8 @@ setPropertyIfEmpty "num.partitions" "8"
 setPropertyIfEmpty "transaction.state.log.replication.factor" "1"
 setPropertyIfEmpty "offsets.topic.replication.factor" "1"
 setPropertyIfEmpty "transaction.state.log.min.isr" "1"
+setPropertyIfEmpty "min.insync.replicas" "1"
+
 if [[ "$CONFLUENT_BROKER" = "true" ]]; then
     rejectProperty "metric.reporters"
     setPropertyIfEmpty "confluent.metrics.reporter.zookeeper.connect" "$ZOOKEEPER_CONNECT"


Test patch:
diff --git a/app/src/test/java/org/astraea/app/cost/NumberOfLeaderCostTest.java b/app/src/test/java/org/astraea/app/cost/NumberOfLeaderCostTest.java
deleted file mode 100644
index 74ccd364..00000000
--- a/app/src/test/java/org/astraea/app/cost/NumberOfLeaderCostTest.java
+++ /dev/null
@@ -1,122 +0,0 @@
-/*
- * 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.Collection;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-import org.astraea.app.admin.Admin;
-import org.astraea.app.admin.ClusterBean;
-import org.astraea.app.metrics.HasBeanObject;
-import org.astraea.app.metrics.collector.BeanCollector;
-import org.astraea.app.metrics.jmx.BeanObject;
-import org.astraea.app.metrics.kafka.HasValue;
-import org.astraea.app.metrics.kafka.KafkaMetrics;
-import org.astraea.app.service.RequireBrokerCluster;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Test;
-import org.mockito.Mockito;
-
-class NumberOfLeaderCostTest extends RequireBrokerCluster {
-
-  @Test
-  void testBrokerCost() {
-    var topicName = List.of("testLeaderCost-1", "testLeaderCost-2", "testLeaderCost-3");
-    try (var admin = Admin.of(bootstrapServers())) {
-      var host = "localhost";
-      for (var topic : topicName)
-        admin.creator().topic(topic).numberOfPartitions(4).numberOfReplicas((short) 1).create();
-      // wait for topic creation
-      TimeUnit.SECONDS.sleep(5);
-      admin.migrator().partition(topicName.get(0), 0).moveTo(List.of(0));
-      admin.migrator().partition(topicName.get(0), 1).moveTo(List.of(0));
-      admin.migrator().partition(topicName.get(0), 2).moveTo(List.of(0));
-      admin.migrator().partition(topicName.get(0), 3).moveTo(List.of(1));
-      // wait for replica migrate
-      TimeUnit.SECONDS.sleep(1);
-      admin.migrator().topic(topicName.get(1)).moveTo(List.of(1));
-      // wait for replica migrate
-      TimeUnit.SECONDS.sleep(1);
-      admin.migrator().topic(topicName.get(2)).moveTo(List.of(2));
-
-      var allBeans = new HashMap<Integer, Collection<HasBeanObject>>();
-      var jmxAddress = Map.of(1001, jmxServiceURL().getPort());
-
-      NumberOfLeaderCost costFunction = new NumberOfLeaderCost();
-      jmxAddress.forEach(
-          (b, port) -> {
-            var firstBeanObjects =
-                BeanCollector.builder()
-                    .interval(Duration.ofSeconds(4))
-                    .build()
-                    .register()
-                    .host(host)
-                    .port(port)
-                    .fetcher(costFunction.fetcher().get())
-                    .build()
-                    .current();
-            allBeans.put(
-                b,
-                allBeans.containsKey(b)
-                    ? Stream.concat(allBeans.get(b).stream(), firstBeanObjects.stream())
-                        .collect(Collectors.toList())
-                    : firstBeanObjects);
-          });
-      var loadCostFunction = new NumberOfLeaderCost();
-      var load = loadCostFunction.brokerCost(exampleClusterInfo());
-
-      Assertions.assertEquals(3.0 / (3 + 4 + 5), load.value().get(1));
-      Assertions.assertEquals(4.0 / (3 + 4 + 5), load.value().get(2));
-      Assertions.assertEquals(5.0 / (3 + 4 + 5), load.value().get(3));
-    } catch (InterruptedException e) {
-      throw new RuntimeException(e);
-    }
-  }
-
-  private ClusterInfo exampleClusterInfo() {
-    var LeaderCount1 =
-        mockResult("ReplicaManager", KafkaMetrics.ReplicaManager.LeaderCount.metricName(), 3);
-    var LeaderCount2 =
-        mockResult("ReplicaManager", KafkaMetrics.ReplicaManager.LeaderCount.metricName(), 4);
-    var LeaderCount3 =
-        mockResult("ReplicaManager", KafkaMetrics.ReplicaManager.LeaderCount.metricName(), 5);
-
-    Collection<HasBeanObject> broker1 = List.of(LeaderCount1);
-    Collection<HasBeanObject> broker2 = List.of(LeaderCount2);
-    Collection<HasBeanObject> broker3 = List.of(LeaderCount3);
-    return new FakeClusterInfo() {
-      @Override
-      public ClusterBean clusterBean() {
-        return ClusterBean.of(Map.of(1, broker1, 2, broker2, 3, broker3));
-      }
-    };
-  }
-
-  private HasValue mockResult(String type, String name, long count) {
-    var result = Mockito.mock(HasValue.class);
-    var bean = Mockito.mock(BeanObject.class);
-    Mockito.when(result.beanObject()).thenReturn(bean);
-    Mockito.when(bean.getProperties()).thenReturn(Map.of("name", name, "type", type));
-    Mockito.when(result.value()).thenReturn(count);
-    return result;
-  }
-}
diff --git a/app/src/test/java/org/astraea/app/cost/ReplicaLeaderCostTest.java b/app/src/test/java/org/astraea/app/cost/ReplicaLeaderCostTest.java
new file mode 100644
index 00000000..3ce35a60
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/cost/ReplicaLeaderCostTest.java
@@ -0,0 +1,82 @@
+/*
+ * 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.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.astraea.app.admin.ClusterBean;
+import org.astraea.app.metrics.HasBeanObject;
+import org.astraea.app.metrics.jmx.BeanObject;
+import org.astraea.app.metrics.kafka.HasValue;
+import org.astraea.app.metrics.kafka.KafkaMetrics;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+public class ReplicaLeaderCostTest {
+
+  @Test
+  void testNoMetrics() {
+    var replicas =
+        List.of(
+            ReplicaInfo.of("topic", 0, NodeInfo.of(10, "broker0", 1111), true, true, true),
+            ReplicaInfo.of("topic", 0, NodeInfo.of(10, "broker0", 1111), true, true, true),
+            ReplicaInfo.of("topic", 0, NodeInfo.of(11, "broker1", 1111), true, true, true));
+    var function = new ReplicaLeaderCost.NoMetrics();
+    var clusterInfo = Mockito.mock(ClusterInfo.class);
+    Mockito.when(clusterInfo.topics()).thenReturn(Set.of("topic"));
+    Mockito.when(clusterInfo.availableReplicaLeaders(Mockito.anyString())).thenReturn(replicas);
+    var cost = function.leaderCount(clusterInfo);
+    Assertions.assertTrue(cost.containsKey(10));
+    Assertions.assertTrue(cost.containsKey(11));
+    Assertions.assertEquals(2, cost.size());
+    Assertions.assertTrue(cost.get(10) > cost.get(11));
+  }
+
+  @Test
+  void testWithMetrics() {
+    var topicName = List.of("testLeaderCost-1", "testLeaderCost-2", "testLeaderCost-3");
+    var costFunction = new ReplicaLeaderCost();
+    var LeaderCount1 = mockResult(KafkaMetrics.ReplicaManager.LeaderCount.metricName(), 3);
+    var LeaderCount2 = mockResult(KafkaMetrics.ReplicaManager.LeaderCount.metricName(), 4);
+    var LeaderCount3 = mockResult(KafkaMetrics.ReplicaManager.LeaderCount.metricName(), 5);
+
+    Collection<HasBeanObject> broker1 = List.of(LeaderCount1);
+    Collection<HasBeanObject> broker2 = List.of(LeaderCount2);
+    Collection<HasBeanObject> broker3 = List.of(LeaderCount3);
+    var clusterBean = ClusterBean.of(Map.of(1, broker1, 2, broker2, 3, broker3));
+    var clusterInfo = Mockito.mock(ClusterInfo.class);
+    Mockito.when(clusterInfo.clusterBean()).thenReturn(clusterBean);
+    var load = costFunction.brokerCost(clusterInfo);
+
+    Assertions.assertEquals(3, load.value().size());
+    Assertions.assertEquals(3.0, load.value().get(1));
+    Assertions.assertEquals(4.0, load.value().get(2));
+    Assertions.assertEquals(5.0, load.value().get(3));
+  }
+
+  private HasValue mockResult(String name, long count) {
+    var result = Mockito.mock(HasValue.class);
+    var bean = Mockito.mock(BeanObject.class);
+    Mockito.when(result.beanObject()).thenReturn(bean);
+    Mockito.when(bean.getProperties()).thenReturn(Map.of("name", name, "type", "ReplicaManager"));
+    Mockito.when(result.value()).thenReturn(count);
+    return result;
+  }
+}
diff --git a/app/src/test/java/org/astraea/app/metrics/jmx/MBeanClientTest.java b/app/src/test/java/org/astraea/app/metrics/jmx/MBeanClientTest.java
index 1a636e73..b9ffa744 100644
--- a/app/src/test/java/org/astraea/app/metrics/jmx/MBeanClientTest.java
+++ b/app/src/test/java/org/astraea/app/metrics/jmx/MBeanClientTest.java
@@ -45,6 +45,7 @@ import javax.management.remote.JMXConnectorServer;
 import javax.management.remote.JMXConnectorServerFactory;
 import javax.management.remote.JMXServiceURL;
 import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
@@ -233,23 +234,6 @@ class MBeanClientTest {
     }
   }
 
-  @Test
-  void testUseClosedClientWillThrowError() {
-    // arrange
-    var client = MBeanClient.of(jmxServer.getAddress());
-    BeanQuery query =
-        BeanQuery.builder().domainName("java.lang").property("type", "Memory").build();
-
-    // act
-    client.close();
-
-    // assert
-    assertThrows(IllegalStateException.class, () -> client.queryBean(query));
-    assertThrows(
-        IllegalStateException.class, () -> client.queryBean(query, Collections.emptyList()));
-    assertThrows(IllegalStateException.class, () -> client.queryBeans(query));
-  }
-
   @Test
   void testCloseOnceMore() {
     // arrange
@@ -309,9 +293,18 @@ class MBeanClientTest {
   }
 
   @Test
-  void testUsePropertyListPattern() {
+  void testUsePropertyListPatternForRemote() {
+    testUsePropertyListPattern(MBeanClient.of(jmxServer.getAddress()));
+  }
+
+  @Test
+  void testUsePropertyListPatternForLocal() {
+    testUsePropertyListPattern(MBeanClient.local());
+  }
+
+  private void testUsePropertyListPattern(MBeanClient client) {
     // arrange
-    try (var client = MBeanClient.of(jmxServer.getAddress())) {
+    try (client) {
       BeanQuery patternQuery =
           BeanQuery.builder()
               .domainName("java.lang")
@@ -457,4 +450,10 @@ class MBeanClientTest {
       assertEquals(IntStream.range(0, 100).boxed().collect(toSet()), attrValues);
     }
   }
+
+  @Test
+  void testLocal() {
+    var client = MBeanClient.local();
+    Assertions.assertNotEquals(0, client.queryBeans(BeanQuery.all()).size());
+  }
 }
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 6b1bd454..fcfbc2c3 100644
--- a/app/src/test/java/org/astraea/app/partitioner/StrictCostDispatcherTest.java
+++ b/app/src/test/java/org/astraea/app/partitioner/StrictCostDispatcherTest.java
@@ -29,6 +29,7 @@ import org.astraea.app.cost.CostFunction;
 import org.astraea.app.cost.HasBrokerCost;
 import org.astraea.app.cost.NodeInfo;
 import org.astraea.app.cost.ReplicaInfo;
+import org.astraea.app.cost.ReplicaLeaderCost;
 import org.astraea.app.cost.ThroughputCost;
 import org.astraea.app.metrics.collector.Fetcher;
 import org.astraea.app.metrics.collector.Receiver;
@@ -65,13 +66,14 @@ public class StrictCostDispatcherTest {
       Assertions.assertThrows(
           IllegalArgumentException.class,
           () ->
-              dispatcher.configure(Configuration.of(Map.of(ThroughputCost.class.getName(), "-1"))));
+              dispatcher.configure(
+                  Configuration.of(Map.of(ReplicaLeaderCost.NoMetrics.class.getName(), "-1"))));
 
       // Test for cost functions configuring
       dispatcher.configure(
           Configuration.of(
               Map.of(
-                  ThroughputCost.class.getName(),
+                  ReplicaLeaderCost.NoMetrics.class.getName(),
                   "0.1",
                   BrokerInputCost.class.getName(),
                   "2",
@@ -87,7 +89,7 @@ public class StrictCostDispatcherTest {
       dispatcher.configure(
           Configuration.of(
               Map.of(
-                  ThroughputCost.class.getName(),
+                  ReplicaLeaderCost.NoMetrics.class.getName(),
                   "0.1",
                   BrokerInputCost.class.getName(),
                   "2",
@@ -226,15 +228,13 @@ public class StrictCostDispatcherTest {
     var dispatcher = new StrictCostDispatcher();
     Assertions.assertThrows(
         IllegalArgumentException.class,
-        () ->
-            dispatcher.configure(
-                Map.of(CostFunction.throughput(), 1D), Optional.empty(), Map.of()));
+        () -> dispatcher.configure(Map.of(new ThroughputCost(), 1D), Optional.empty(), Map.of()));
 
     // pass due to default port
-    dispatcher.configure(Map.of(CostFunction.throughput(), 1D), Optional.of(111), Map.of());
+    dispatcher.configure(Map.of(new ThroughputCost(), 1D), Optional.of(111), Map.of());
 
     // pass due to specify port
-    dispatcher.configure(Map.of(CostFunction.throughput(), 1D), Optional.empty(), Map.of(222, 111));
+    dispatcher.configure(Map.of(new ThroughputCost(), 1D), Optional.empty(), Map.of(222, 111));
   }
 
   @Test
@@ -264,4 +264,11 @@ public class StrictCostDispatcherTest {
     Assertions.assertEquals(
         partitionId, dispatcher.partition("topic", new byte[0], new byte[0], clusterInfo));
   }
+
+  @Test
+  void testDefaultFunction() {
+    var dispatcher = new StrictCostDispatcher();
+    dispatcher.configure(Map.of(), Optional.empty(), Map.of());
+    Assertions.assertEquals(1, dispatcher.functions.size());
+  }
 }