Repository: skiptests/astraea
Pull request #312 (2022-04-29 14:11:30)

Add new cost functions
新增三個balancer會用到的CostFunction

garyparrot: @qoo332001 

https://github.com/skiptests/astraea/issues/289#issuecomment-1092674657

先前提到的第一版中，我們有這些目標

> 考量 Leader 分佈

看起來是實作在 `NumberOfLeaderCost.java`

> 平衡 Produce 流量

這個應該是 `ReplicaDiskInCost.java`

> 平衡 Consume 流量
> 考量 Topic Traffic (避免多數 partition 分佈到同一個節點的情況)

這兩個部分的 Cost Function 有想法嗎



Patch:
diff --git a/app/src/main/java/org/astraea/app/admin/ClusterBean.java b/app/src/main/java/org/astraea/app/admin/ClusterBean.java
index a61a5774..63c5d667 100644
--- a/app/src/main/java/org/astraea/app/admin/ClusterBean.java
+++ b/app/src/main/java/org/astraea/app/admin/ClusterBean.java
@@ -29,11 +29,7 @@ public interface ClusterBean {
   private static Collection<HasBeanObject> compareBeanObject(
       Collection<HasBeanObject> x1, Collection<HasBeanObject> x2) {
     var beanObject2 = x2.iterator().next().beanObject();
-    if (x1.stream()
-        .noneMatch(
-            hasBeanObject ->
-                hasBeanObject.beanObject().getProperties().equals(beanObject2.getProperties())
-                    && hasBeanObject.beanObject().domainName().equals(beanObject2.domainName())))
+    if (x1.stream().noneMatch(hasBeanObject -> hasBeanObject.beanObject().equals(beanObject2)))
       return Stream.concat(x1.stream(), x2.stream()).collect(Collectors.toList());
     return x1;
   }
diff --git a/app/src/main/java/org/astraea/app/cost/broker/BrokerInputCost.java b/app/src/main/java/org/astraea/app/cost/BrokerInputCost.java
similarity index 96%
rename from app/src/main/java/org/astraea/app/cost/broker/BrokerInputCost.java
rename to app/src/main/java/org/astraea/app/cost/BrokerInputCost.java
index 6d120ad3..c8438795 100644
--- a/app/src/main/java/org/astraea/app/cost/broker/BrokerInputCost.java
+++ b/app/src/main/java/org/astraea/app/cost/BrokerInputCost.java
@@ -14,16 +14,13 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.astraea.app.cost.broker;
+package org.astraea.app.cost;
 
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
-import org.astraea.app.cost.BrokerCost;
-import org.astraea.app.cost.ClusterInfo;
-import org.astraea.app.cost.HasBrokerCost;
 import org.astraea.app.metrics.collector.Fetcher;
 import org.astraea.app.metrics.kafka.BrokerTopicMetricsResult;
 import org.astraea.app.metrics.kafka.KafkaMetrics;
diff --git a/app/src/main/java/org/astraea/app/cost/broker/BrokerOutputCost.java b/app/src/main/java/org/astraea/app/cost/BrokerOutputCost.java
similarity index 96%
rename from app/src/main/java/org/astraea/app/cost/broker/BrokerOutputCost.java
rename to app/src/main/java/org/astraea/app/cost/BrokerOutputCost.java
index a379a292..30f76b3d 100644
--- a/app/src/main/java/org/astraea/app/cost/broker/BrokerOutputCost.java
+++ b/app/src/main/java/org/astraea/app/cost/BrokerOutputCost.java
@@ -14,16 +14,13 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.astraea.app.cost.broker;
+package org.astraea.app.cost;
 
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
-import org.astraea.app.cost.BrokerCost;
-import org.astraea.app.cost.ClusterInfo;
-import org.astraea.app.cost.HasBrokerCost;
 import org.astraea.app.metrics.collector.Fetcher;
 import org.astraea.app.metrics.kafka.BrokerTopicMetricsResult;
 import org.astraea.app.metrics.kafka.KafkaMetrics;
diff --git a/app/src/main/java/org/astraea/app/cost/broker/CostUtils.java b/app/src/main/java/org/astraea/app/cost/CostUtils.java
similarity index 98%
rename from app/src/main/java/org/astraea/app/cost/broker/CostUtils.java
rename to app/src/main/java/org/astraea/app/cost/CostUtils.java
index 320a900d..44eb4834 100644
--- a/app/src/main/java/org/astraea/app/cost/broker/CostUtils.java
+++ b/app/src/main/java/org/astraea/app/cost/CostUtils.java
@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.astraea.app.cost.broker;
+package org.astraea.app.cost;
 
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicReference;
diff --git a/app/src/main/java/org/astraea/app/cost/broker/CpuCost.java b/app/src/main/java/org/astraea/app/cost/CpuCost.java
similarity index 96%
rename from app/src/main/java/org/astraea/app/cost/broker/CpuCost.java
rename to app/src/main/java/org/astraea/app/cost/CpuCost.java
index de509ca5..904b8888 100644
--- a/app/src/main/java/org/astraea/app/cost/broker/CpuCost.java
+++ b/app/src/main/java/org/astraea/app/cost/CpuCost.java
@@ -14,16 +14,13 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.astraea.app.cost.broker;
+package org.astraea.app.cost;
 
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
-import org.astraea.app.cost.BrokerCost;
-import org.astraea.app.cost.ClusterInfo;
-import org.astraea.app.cost.HasBrokerCost;
 import org.astraea.app.metrics.collector.Fetcher;
 import org.astraea.app.metrics.java.OperatingSystemInfo;
 import org.astraea.app.metrics.kafka.KafkaMetrics;
diff --git a/app/src/main/java/org/astraea/app/cost/broker/MemoryCost.java b/app/src/main/java/org/astraea/app/cost/MemoryCost.java
similarity index 95%
rename from app/src/main/java/org/astraea/app/cost/broker/MemoryCost.java
rename to app/src/main/java/org/astraea/app/cost/MemoryCost.java
index 90e530ff..1aed9d6a 100644
--- a/app/src/main/java/org/astraea/app/cost/broker/MemoryCost.java
+++ b/app/src/main/java/org/astraea/app/cost/MemoryCost.java
@@ -14,17 +14,13 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.astraea.app.cost.broker;
+package org.astraea.app.cost;
 
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
-import org.astraea.app.cost.BrokerCost;
-import org.astraea.app.cost.ClusterInfo;
-import org.astraea.app.cost.HasBrokerCost;
-import org.astraea.app.cost.Periodic;
 import org.astraea.app.metrics.collector.Fetcher;
 import org.astraea.app.metrics.java.HasJvmMemory;
 import org.astraea.app.metrics.kafka.KafkaMetrics;
diff --git a/app/src/main/java/org/astraea/app/cost/broker/NeutralIntegratedCost.java b/app/src/main/java/org/astraea/app/cost/NeutralIntegratedCost.java
similarity index 96%
rename from app/src/main/java/org/astraea/app/cost/broker/NeutralIntegratedCost.java
rename to app/src/main/java/org/astraea/app/cost/NeutralIntegratedCost.java
index eb851f88..eb82ef5f 100644
--- a/app/src/main/java/org/astraea/app/cost/broker/NeutralIntegratedCost.java
+++ b/app/src/main/java/org/astraea/app/cost/NeutralIntegratedCost.java
@@ -14,19 +14,13 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.astraea.app.cost.broker;
+package org.astraea.app.cost;
 
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
-import org.astraea.app.cost.BrokerCost;
-import org.astraea.app.cost.ClusterInfo;
-import org.astraea.app.cost.CostFunction;
-import org.astraea.app.cost.HasBrokerCost;
-import org.astraea.app.cost.Normalizer;
-import org.astraea.app.cost.WeightProvider;
 import org.astraea.app.metrics.collector.Fetcher;
 
 /**
diff --git a/app/src/main/java/org/astraea/app/cost/NumberOfLeaderCost.java b/app/src/main/java/org/astraea/app/cost/NumberOfLeaderCost.java
new file mode 100644
index 00000000..c8be27f3
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/cost/NumberOfLeaderCost.java
@@ -0,0 +1,66 @@
+/*
+ * 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.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 Fetcher fetcher() {
+    return 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/ReplicaDiskInCost.java b/app/src/main/java/org/astraea/app/cost/ReplicaDiskInCost.java
new file mode 100644
index 00000000..0c15dc04
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/cost/ReplicaDiskInCost.java
@@ -0,0 +1,228 @@
+/*
+ * 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.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.stream.Collectors;
+import org.astraea.app.admin.TopicPartition;
+import org.astraea.app.admin.TopicPartitionReplica;
+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 "Size.Value" ,and createdTimestamp in the metrics. "Size.Value"
+ * responds to the replica log size of brokers. The calculation method of the score is the rate of
+ * increase of log size per unit time divided by the upper limit of broker bandwidth.
+ */
+public class ReplicaDiskInCost implements HasBrokerCost, HasPartitionCost {
+  Map<Integer, Integer> brokerBandwidthCap;
+
+  public ReplicaDiskInCost(Map<Integer, Integer> brokerBandwidthCap) {
+    this.brokerBandwidthCap = brokerBandwidthCap;
+  }
+
+  @Override
+  public BrokerCost brokerCost(ClusterInfo clusterInfo) {
+    final Map<Integer, List<TopicPartitionReplica>> topicPartitionOfEachBroker =
+        clusterInfo.topics().stream()
+            .flatMap(topic -> clusterInfo.replicas(topic).stream())
+            .map(
+                replica ->
+                    new TopicPartitionReplica(
+                        replica.topic(), replica.partition(), replica.nodeInfo().id()))
+            .collect(Collectors.groupingBy(TopicPartitionReplica::brokerId));
+    final var actual =
+        clusterInfo.nodes().stream()
+            .collect(
+                Collectors.toUnmodifiableMap(
+                    NodeInfo::id,
+                    node -> topicPartitionOfEachBroker.getOrDefault(node.id(), List.of())));
+
+    final var topicPartitionDataRate = topicPartitionDataRate(clusterInfo, Duration.ofSeconds(3));
+
+    final var brokerLoad =
+        actual.entrySet().stream()
+            .map(
+                entry ->
+                    Map.entry(
+                        entry.getKey(),
+                        entry.getValue().stream()
+                            .mapToDouble(
+                                x ->
+                                    topicPartitionDataRate.get(
+                                        new TopicPartition(x.topic(), x.partition())))
+                            .sum()))
+            .map(
+                entry ->
+                    Map.entry(
+                        entry.getKey(),
+                        entry.getValue() / brokerBandwidthCap.get(entry.getKey()) / 1024 / 1024))
+            .map(entry -> Map.entry(entry.getKey(), Math.min(entry.getValue(), 1)))
+            .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+
+    return () -> brokerLoad;
+  }
+
+  @Override
+  public PartitionCost partitionCost(ClusterInfo clusterInfo) {
+    var replicaIn = replicaDataRate(clusterInfo, Duration.ofSeconds(2));
+    var replicaScore =
+        new TreeMap<TopicPartitionReplica, Double>(
+            Comparator.comparing(TopicPartitionReplica::brokerId)
+                .thenComparing(TopicPartitionReplica::topic)
+                .thenComparing(TopicPartitionReplica::partition));
+    replicaIn.forEach(
+        (tpr, rate) -> {
+          var score = rate / brokerBandwidthCap.get(tpr.brokerId()) / 1024.0 / 1024.0;
+          if (score >= 1) score = 1;
+          replicaScore.put(tpr, score);
+        });
+    var scoreForTopic =
+        clusterInfo.topics().stream()
+            .map(
+                topic ->
+                    Map.entry(
+                        topic,
+                        replicaScore.entrySet().stream()
+                            .filter(x -> x.getKey().topic().equals(topic))
+                            .collect(
+                                Collectors.groupingBy(
+                                    x ->
+                                        new TopicPartition(
+                                            x.getKey().topic(), x.getKey().partition())))
+                            .entrySet()
+                            .stream()
+                            .map(
+                                entry ->
+                                    Map.entry(
+                                        entry.getKey(),
+                                        entry.getValue().stream()
+                                            .mapToDouble(Map.Entry::getValue)
+                                            .max()
+                                            .orElseThrow()))
+                            .collect(
+                                Collectors.toUnmodifiableMap(
+                                    Map.Entry::getKey, Map.Entry::getValue))))
+            .collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, Map.Entry::getValue));
+
+    var scoreForBroker =
+        clusterInfo.nodes().stream()
+            .map(
+                node ->
+                    Map.entry(
+                        node.id(),
+                        replicaScore.entrySet().stream()
+                            .filter(x -> x.getKey().brokerId() == node.id())
+                            .collect(
+                                Collectors.groupingBy(
+                                    x ->
+                                        new TopicPartition(
+                                            x.getKey().topic(), x.getKey().partition())))
+                            .entrySet()
+                            .stream()
+                            .map(
+                                entry ->
+                                    Map.entry(
+                                        entry.getKey(),
+                                        entry.getValue().stream()
+                                            .mapToDouble(Map.Entry::getValue)
+                                            .max()
+                                            .orElseThrow()))
+                            .collect(
+                                Collectors.toUnmodifiableMap(
+                                    Map.Entry::getKey, Map.Entry::getValue))))
+            .collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, Map.Entry::getValue));
+
+    return new PartitionCost() {
+      @Override
+      public Map<TopicPartition, Double> value(String topic) {
+        return scoreForTopic.get(topic);
+      }
+
+      @Override
+      public Map<TopicPartition, Double> value(int brokerId) {
+        return scoreForBroker.get(brokerId);
+      }
+    };
+  }
+
+  /** @return the metrics getters. Those getters are used to fetch mbeans. */
+  @Override
+  public Fetcher fetcher() {
+    return KafkaMetrics.TopicPartition.Size::fetch;
+  }
+
+  /**
+   * Calculate the maximum increase rate of each topic/partition, across the whole cluster.
+   *
+   * @param clusterInfo the clusterInfo that offers the metrics related to topic/partition size
+   * @param sampleWindow the time interval for calculating the data rate, noted that if the metrics
+   *     doesn't have the sufficient old metric then an exception will likely be thrown.
+   * @return a map contain the maximum increase rate of each topic/partition log
+   */
+  public static Map<TopicPartition, Double> topicPartitionDataRate(
+      ClusterInfo clusterInfo, Duration sampleWindow) {
+    return replicaDataRate(clusterInfo, sampleWindow).entrySet().stream()
+        .map(
+            x -> {
+              var tpr = x.getKey();
+              return Map.entry(new TopicPartition(tpr.topic(), tpr.partition()), x.getValue());
+            })
+        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue, (x1, x2) -> x1));
+  }
+
+  public static Map<TopicPartitionReplica, Double> replicaDataRate(
+      ClusterInfo clusterInfo, Duration sampleWindow) {
+    return clusterInfo.clusterBean().mapByReplica().entrySet().parallelStream()
+        .map(
+            metrics -> {
+              // calculate the increase rate over a specific window of time
+              var sizeTimeSeries =
+                  metrics.getValue().stream()
+                      .filter(bean -> bean instanceof HasValue)
+                      .filter(bean -> bean.beanObject().getProperties().get("type").equals("Log"))
+                      .filter(bean -> bean.beanObject().getProperties().get("name").equals("Size"))
+                      .map(bean -> (HasValue) bean)
+                      .sorted(Comparator.comparingLong(HasBeanObject::createdTimestamp).reversed())
+                      .collect(Collectors.toUnmodifiableList());
+              var latestSize = sizeTimeSeries.stream().findFirst().orElseThrow();
+              var windowSize =
+                  sizeTimeSeries.stream()
+                      .dropWhile(
+                          bean ->
+                              bean.createdTimestamp()
+                                  > latestSize.createdTimestamp() - sampleWindow.toMillis())
+                      .findFirst()
+                      .orElseThrow(
+                          () ->
+                              new IllegalStateException(
+                                  "No sufficient info to determine data rate, try later."));
+              var dataRate =
+                  ((double) (latestSize.value() - windowSize.value()))
+                      / ((double) (latestSize.createdTimestamp() - windowSize.createdTimestamp())
+                          / 1000);
+              return Map.entry(metrics.getKey(), dataRate);
+            })
+        .collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, Map.Entry::getValue));
+  }
+}
diff --git a/app/src/main/java/org/astraea/app/cost/ReplicaSizeCost.java b/app/src/main/java/org/astraea/app/cost/ReplicaSizeCost.java
new file mode 100644
index 00000000..7f373527
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/cost/ReplicaSizeCost.java
@@ -0,0 +1,178 @@
+/*
+ * 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.TreeMap;
+import java.util.stream.Collectors;
+import org.astraea.app.admin.TopicPartition;
+import org.astraea.app.admin.TopicPartitionReplica;
+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 "Size.Value". "Size.Value" responds to the replica log size of brokers.
+ * The calculation method of the score is the replica log usage space divided by the available space
+ * on the hard disk
+ */
+public class ReplicaSizeCost implements HasBrokerCost, HasPartitionCost {
+  Map<Integer, Integer> totalBrokerCapacity;
+
+  public ReplicaSizeCost(Map<Integer, Integer> totalBrokerCapacity) {
+    this.totalBrokerCapacity = totalBrokerCapacity;
+  }
+
+  @Override
+  public Fetcher fetcher() {
+    return KafkaMetrics.TopicPartition.Size::fetch;
+  }
+
+  /**
+   * @param clusterInfo the clusterInfo that offers the metrics related to topic/partition size
+   * @return a BrokerCost contains the ratio of the used space to the free space of each broker
+   */
+  @Override
+  public BrokerCost brokerCost(ClusterInfo clusterInfo) {
+    var sizeOfReplica = getReplicaSize(clusterInfo);
+    var totalReplicaSizeInBroker =
+        clusterInfo.nodes().stream()
+            .collect(
+                Collectors.toMap(
+                    NodeInfo::id,
+                    y ->
+                        sizeOfReplica.entrySet().stream()
+                            .filter(tpr -> tpr.getKey().brokerId() == y.id())
+                            .mapToLong(Map.Entry::getValue)
+                            .sum()));
+    var brokerSizeScore =
+        totalReplicaSizeInBroker.entrySet().stream()
+            .collect(
+                Collectors.toMap(
+                    Map.Entry::getKey,
+                    y ->
+                        Double.valueOf(y.getValue())
+                            / totalBrokerCapacity.get(y.getKey())
+                            / 1048576));
+    return () -> brokerSizeScore;
+  }
+
+  /**
+   * @param clusterInfo the clusterInfo that offers the metrics related to topic/partition size
+   * @return a BrokerCost contains the ratio of the used space to the available space of replicas in
+   *     each broker
+   */
+  @Override
+  public PartitionCost partitionCost(ClusterInfo clusterInfo) {
+    final long ONEMEGA = Math.round(Math.pow(2, 20));
+    var sizeOfReplica = getReplicaSize(clusterInfo);
+    TreeMap<TopicPartitionReplica, Double> replicaCost =
+        new TreeMap<>(
+            Comparator.comparing(TopicPartitionReplica::brokerId)
+                .thenComparing(TopicPartitionReplica::topic)
+                .thenComparing(TopicPartitionReplica::partition));
+    sizeOfReplica.forEach(
+        (tpr, size) ->
+            replicaCost.put(
+                tpr, (double) size / totalBrokerCapacity.get(tpr.brokerId()) / ONEMEGA));
+
+    var scoreForTopic =
+        clusterInfo.topics().stream()
+            .map(
+                topic ->
+                    Map.entry(
+                        topic,
+                        clusterInfo.replicas(topic).stream()
+                            .filter(ReplicaInfo::isLeader)
+                            .map(
+                                partitionInfo ->
+                                    new TopicPartitionReplica(
+                                        partitionInfo.topic(),
+                                        partitionInfo.partition(),
+                                        partitionInfo.nodeInfo().id()))
+                            .map(
+                                tpr -> {
+                                  final var score =
+                                      replicaCost.entrySet().stream()
+                                          .filter(
+                                              x ->
+                                                  x.getKey().topic().equals(tpr.topic())
+                                                      && (x.getKey().partition()
+                                                          == tpr.partition()))
+                                          .mapToDouble(Map.Entry::getValue)
+                                          .max()
+                                          .orElseThrow(
+                                              () ->
+                                                  new IllegalStateException(
+                                                      tpr + " topic/partition size not found"));
+                                  return Map.entry(
+                                      new TopicPartition(tpr.topic(), tpr.partition()), score);
+                                })
+                            .collect(
+                                Collectors.toUnmodifiableMap(
+                                    Map.Entry::getKey, Map.Entry::getValue))))
+            .collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, Map.Entry::getValue));
+
+    var scoreForBroker =
+        clusterInfo.nodes().stream()
+            .map(
+                node ->
+                    Map.entry(
+                        node.id(),
+                        replicaCost.entrySet().stream()
+                            .filter((tprScore) -> tprScore.getKey().brokerId() == node.id())
+                            .collect(
+                                Collectors.toMap(
+                                    x ->
+                                        new TopicPartition(
+                                            x.getKey().topic(), x.getKey().partition()),
+                                    Map.Entry::getValue))))
+            .collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, Map.Entry::getValue));
+
+    return new PartitionCost() {
+
+      @Override
+      public Map<TopicPartition, Double> value(String topic) {
+        return scoreForTopic.get(topic);
+      }
+
+      @Override
+      public Map<TopicPartition, Double> value(int brokerId) {
+        return scoreForBroker.get(brokerId);
+      }
+    };
+  }
+
+  /**
+   * @param clusterInfo the clusterInfo that offers the metrics related to topic/partition size
+   * @return a map contain the replica log size of each topic/partition
+   */
+  public Map<TopicPartitionReplica, Long> getReplicaSize(ClusterInfo clusterInfo) {
+    return clusterInfo.clusterBean().mapByReplica().entrySet().stream()
+        .flatMap(
+            e ->
+                e.getValue().stream()
+                    .filter(x -> x instanceof HasValue)
+                    .filter(x -> x.beanObject().domainName().equals("kafka.log"))
+                    .filter(x -> x.beanObject().getProperties().get("type").equals("Log"))
+                    .filter(x -> x.beanObject().getProperties().get("name").equals("Size"))
+                    .map(x -> (HasValue) x)
+                    .map(x -> Map.entry(e.getKey(), x.value())))
+        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+  }
+}
diff --git a/app/src/main/java/org/astraea/app/metrics/jmx/BeanObject.java b/app/src/main/java/org/astraea/app/metrics/jmx/BeanObject.java
index 34f7557d..4f95c363 100644
--- a/app/src/main/java/org/astraea/app/metrics/jmx/BeanObject.java
+++ b/app/src/main/java/org/astraea/app/metrics/jmx/BeanObject.java
@@ -28,7 +28,7 @@ public class BeanObject {
   private final String domainName;
   private final Map<String, String> properties;
   private final Map<String, Object> attributes;
-  private final long createdTimestamp = System.currentTimeMillis();
+  private final long createdTimestamp;
 
   /**
    * construct a {@link BeanObject}
@@ -42,6 +42,14 @@ public class BeanObject {
    */
   public BeanObject(
       String domainName, Map<String, String> properties, Map<String, Object> attributes) {
+    this(domainName, properties, attributes, System.currentTimeMillis());
+  }
+
+  public BeanObject(
+      String domainName,
+      Map<String, String> properties,
+      Map<String, Object> attributes,
+      long createdTimestamp) {
     this.domainName = Objects.requireNonNull(domainName);
 
     // copy properties, and remove null key or null value
@@ -59,6 +67,7 @@ public class BeanObject {
             .filter(entry -> entry.getKey() != null && entry.getValue() != null)
             .collect(Collectors.toMap(Entry::getKey, Entry::getValue));
     this.attributes = Collections.unmodifiableMap(attributeMap);
+    this.createdTimestamp = createdTimestamp;
   }
 
   public String domainName() {
diff --git a/app/src/main/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobin.java b/app/src/main/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobin.java
index f78a1b77..e0a0f1b2 100644
--- a/app/src/main/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobin.java
+++ b/app/src/main/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobin.java
@@ -23,8 +23,8 @@ import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
 import org.astraea.app.cost.ClusterInfo;
+import org.astraea.app.cost.CostUtils;
 import org.astraea.app.cost.Periodic;
-import org.astraea.app.cost.broker.CostUtils;
 
 /**
  * Given initial key-score pair, it will output a preferred key with the highest current weight. The
diff --git a/app/src/main/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobinDispatcher.java b/app/src/main/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobinDispatcher.java
index 6bfbb393..330a2408 100644
--- a/app/src/main/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobinDispatcher.java
+++ b/app/src/main/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobinDispatcher.java
@@ -32,10 +32,10 @@ import java.util.stream.Collectors;
 import org.apache.kafka.common.Cluster;
 import org.astraea.app.common.Utils;
 import org.astraea.app.cost.ClusterInfo;
+import org.astraea.app.cost.NeutralIntegratedCost;
 import org.astraea.app.cost.NodeInfo;
 import org.astraea.app.cost.Periodic;
 import org.astraea.app.cost.ReplicaInfo;
-import org.astraea.app.cost.broker.NeutralIntegratedCost;
 import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.collector.BeanCollector;
 import org.astraea.app.metrics.collector.Receiver;


Test patch:
diff --git a/app/src/test/java/org/astraea/app/admin/AdminTest.java b/app/src/test/java/org/astraea/app/admin/AdminTest.java
index dd8da894..3d97c418 100644
--- a/app/src/test/java/org/astraea/app/admin/AdminTest.java
+++ b/app/src/test/java/org/astraea/app/admin/AdminTest.java
@@ -27,7 +27,6 @@ import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Supplier;
@@ -43,7 +42,6 @@ import org.astraea.app.concurrent.ThreadPool;
 import org.astraea.app.consumer.Consumer;
 import org.astraea.app.consumer.Deserializer;
 import org.astraea.app.cost.NodeInfo;
-import org.astraea.app.cost.ReplicaInfo;
 import org.astraea.app.producer.Producer;
 import org.astraea.app.producer.Serializer;
 import org.astraea.app.service.RequireBrokerCluster;
@@ -652,37 +650,6 @@ public class AdminTest extends RequireBrokerCluster {
     }
   }
 
-  @Test
-  void somePartitionsOffline() {
-    String topicName1 = "testOfflineTopic-1";
-    String topicName2 = "testOfflineTopic-2";
-    try (var admin = Admin.of(bootstrapServers())) {
-      admin.creator().topic(topicName1).numberOfPartitions(4).numberOfReplicas((short) 1).create();
-      admin.creator().topic(topicName2).numberOfPartitions(4).numberOfReplicas((short) 1).create();
-      // wait for topic creation
-      TimeUnit.SECONDS.sleep(10);
-      var replicaOnBroker0 =
-          admin.replicas(admin.topicNames()).entrySet().stream()
-              .filter(replica -> replica.getValue().get(0).broker() == 0)
-              .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
-      replicaOnBroker0.forEach((tp, replica) -> Assertions.assertFalse(replica.get(0).isOffline()));
-      closeBroker(0);
-      Assertions.assertNull(logFolders().get(0));
-      Assertions.assertNotNull(logFolders().get(1));
-      Assertions.assertNotNull(logFolders().get(2));
-      var offlineReplicaOnBroker0 =
-          admin.replicas(admin.topicNames()).entrySet().stream()
-              .filter(replica -> replica.getValue().get(0).broker() == 0)
-              .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
-      offlineReplicaOnBroker0.forEach(
-          (tp, replica) -> Assertions.assertTrue(replica.get(0).isOffline()));
-
-    } catch (InterruptedException e) {
-      throw new RuntimeException(e);
-    }
-    restartCluster();
-  }
-
   @Test
   void testNodes() {
     try (var admin = Admin.of(bootstrapServers())) {
@@ -753,63 +720,6 @@ public class AdminTest extends RequireBrokerCluster {
     }
   }
 
-  @Test
-  void testClusterInfoWithOfflineNode() throws InterruptedException {
-    try (Admin admin = Admin.of(bootstrapServers())) {
-      var topicName = "ClusterInfo_Offline_" + Utils.randomString();
-      var partitionCount = 30;
-      var replicaCount = (short) 3;
-      admin
-          .creator()
-          .topic(topicName)
-          .numberOfPartitions(partitionCount)
-          .numberOfReplicas(replicaCount)
-          .create();
-      TimeUnit.SECONDS.sleep(3);
-
-      // before node offline
-      var before = admin.clusterInfo(Set.of(topicName));
-      Assertions.assertEquals(
-          partitionCount * replicaCount,
-          before.replicas(topicName).stream().filter(x -> !x.isOfflineReplica()).count());
-      Assertions.assertEquals(
-          partitionCount * replicaCount, before.availableReplicas(topicName).size());
-      Assertions.assertEquals(partitionCount, before.availableReplicaLeaders(topicName).size());
-
-      // act
-      int brokerToClose = ThreadLocalRandom.current().nextInt(0, 3);
-      closeBroker(brokerToClose);
-      TimeUnit.SECONDS.sleep(1);
-
-      // after node offline
-      var after = admin.clusterInfo(Set.of(topicName));
-      Assertions.assertEquals(
-          partitionCount * (replicaCount - 1),
-          after.replicas(topicName).stream().filter(x -> !x.isOfflineReplica()).count());
-      Assertions.assertEquals(
-          partitionCount * (replicaCount - 1), after.availableReplicas(topicName).size());
-      Assertions.assertEquals(
-          partitionCount,
-          after.availableReplicaLeaders(topicName).size(),
-          "One of the rest replicas should take over the leadership");
-      Assertions.assertTrue(
-          after.availableReplicas(topicName).stream()
-              .allMatch(x -> x.nodeInfo().id() != brokerToClose));
-      Assertions.assertTrue(
-          after.availableReplicaLeaders(topicName).stream()
-              .allMatch(x -> x.nodeInfo().id() != brokerToClose));
-      Assertions.assertTrue(
-          after.replicas(topicName).stream()
-              .filter(ReplicaInfo::isOfflineReplica)
-              .allMatch(x -> x.nodeInfo().id() == brokerToClose));
-      Assertions.assertTrue(
-          after.replicas(topicName).stream()
-              .filter(x -> !x.isOfflineReplica())
-              .allMatch(x -> x.nodeInfo().id() != brokerToClose));
-    }
-    restartCluster();
-  }
-
   @ParameterizedTest
   @ValueSource(shorts = {1, 2, 3})
   void preferredLeaderElection(short replicaSize) throws InterruptedException {
diff --git a/app/src/test/java/org/astraea/app/admin/ClusterInfoWithOfflineNodeTest.java b/app/src/test/java/org/astraea/app/admin/ClusterInfoWithOfflineNodeTest.java
new file mode 100644
index 00000000..ccf9e8b2
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/admin/ClusterInfoWithOfflineNodeTest.java
@@ -0,0 +1,84 @@
+/*
+ * 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.admin;
+
+import java.util.Set;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import org.astraea.app.common.Utils;
+import org.astraea.app.cost.ReplicaInfo;
+import org.astraea.app.service.RequireBrokerCluster;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class ClusterInfoWithOfflineNodeTest extends RequireBrokerCluster {
+  @Test
+  void testClusterInfoWithOfflineNode() throws InterruptedException {
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var topicName = "ClusterInfo_Offline_" + Utils.randomString();
+      var partitionCount = 30;
+      var replicaCount = (short) 3;
+      admin
+          .creator()
+          .topic(topicName)
+          .numberOfPartitions(partitionCount)
+          .numberOfReplicas(replicaCount)
+          .create();
+      TimeUnit.SECONDS.sleep(3);
+
+      // before node offline
+      var before = admin.clusterInfo(Set.of(topicName));
+      Assertions.assertEquals(
+          partitionCount * replicaCount,
+          before.replicas(topicName).stream().filter(x -> !x.isOfflineReplica()).count());
+      Assertions.assertEquals(
+          partitionCount * replicaCount, before.availableReplicas(topicName).size());
+      Assertions.assertEquals(partitionCount, before.availableReplicaLeaders(topicName).size());
+
+      // act
+      int brokerToClose = ThreadLocalRandom.current().nextInt(0, 3);
+      closeBroker(brokerToClose);
+      TimeUnit.SECONDS.sleep(1);
+
+      // after node offline
+      var after = admin.clusterInfo(Set.of(topicName));
+      Assertions.assertEquals(
+          partitionCount * (replicaCount - 1),
+          after.replicas(topicName).stream().filter(x -> !x.isOfflineReplica()).count());
+      Assertions.assertEquals(
+          partitionCount * (replicaCount - 1), after.availableReplicas(topicName).size());
+      Assertions.assertEquals(
+          partitionCount,
+          after.availableReplicaLeaders(topicName).size(),
+          "One of the rest replicas should take over the leadership");
+      Assertions.assertTrue(
+          after.availableReplicas(topicName).stream()
+              .allMatch(x -> x.nodeInfo().id() != brokerToClose));
+      Assertions.assertTrue(
+          after.availableReplicaLeaders(topicName).stream()
+              .allMatch(x -> x.nodeInfo().id() != brokerToClose));
+      Assertions.assertTrue(
+          after.replicas(topicName).stream()
+              .filter(ReplicaInfo::isOfflineReplica)
+              .allMatch(x -> x.nodeInfo().id() == brokerToClose));
+      Assertions.assertTrue(
+          after.replicas(topicName).stream()
+              .filter(x -> !x.isOfflineReplica())
+              .allMatch(x -> x.nodeInfo().id() != brokerToClose));
+    }
+  }
+}
diff --git a/app/src/test/java/org/astraea/app/admin/SomePartitionOfflineTest.java b/app/src/test/java/org/astraea/app/admin/SomePartitionOfflineTest.java
new file mode 100644
index 00000000..bec53857
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/admin/SomePartitionOfflineTest.java
@@ -0,0 +1,56 @@
+/*
+ * 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.admin;
+
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import org.astraea.app.service.RequireBrokerCluster;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class SomePartitionOfflineTest extends RequireBrokerCluster {
+  @Test
+  void somePartitionsOffline() {
+    String topicName1 = "testOfflineTopic-1";
+    String topicName2 = "testOfflineTopic-2";
+    try (var admin = Admin.of(bootstrapServers())) {
+      admin.creator().topic(topicName1).numberOfPartitions(4).numberOfReplicas((short) 1).create();
+      admin.creator().topic(topicName2).numberOfPartitions(4).numberOfReplicas((short) 1).create();
+      // wait for topic creation
+      TimeUnit.SECONDS.sleep(10);
+      var replicaOnBroker0 =
+          admin.replicas(admin.topicNames()).entrySet().stream()
+              .filter(replica -> replica.getValue().get(0).broker() == 0)
+              .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+      replicaOnBroker0.forEach((tp, replica) -> Assertions.assertFalse(replica.get(0).isOffline()));
+      closeBroker(0);
+      Assertions.assertNull(logFolders().get(0));
+      Assertions.assertNotNull(logFolders().get(1));
+      Assertions.assertNotNull(logFolders().get(2));
+      var offlineReplicaOnBroker0 =
+          admin.replicas(admin.topicNames()).entrySet().stream()
+              .filter(replica -> replica.getValue().get(0).broker() == 0)
+              .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+      offlineReplicaOnBroker0.forEach(
+          (tp, replica) -> Assertions.assertTrue(replica.get(0).isOffline()));
+
+    } catch (InterruptedException e) {
+      throw new RuntimeException(e);
+    }
+  }
+}
diff --git a/app/src/test/java/org/astraea/app/cost/broker/BrokerInputCostTest.java b/app/src/test/java/org/astraea/app/cost/BrokerInputCostTest.java
similarity index 96%
rename from app/src/test/java/org/astraea/app/cost/broker/BrokerInputCostTest.java
rename to app/src/test/java/org/astraea/app/cost/BrokerInputCostTest.java
index b635ae68..b1c43fed 100644
--- a/app/src/test/java/org/astraea/app/cost/broker/BrokerInputCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/BrokerInputCostTest.java
@@ -14,15 +14,12 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.astraea.app.cost.broker;
+package org.astraea.app.cost;
 
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import org.astraea.app.admin.ClusterBean;
-import org.astraea.app.cost.ClusterInfo;
-import org.astraea.app.cost.FakeClusterInfo;
-import org.astraea.app.cost.Normalizer;
 import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.collector.BeanCollector;
 import org.astraea.app.metrics.collector.Receiver;
diff --git a/app/src/test/java/org/astraea/app/cost/broker/BrokerOutPutCostTest.java b/app/src/test/java/org/astraea/app/cost/BrokerOutPutCostTest.java
similarity index 96%
rename from app/src/test/java/org/astraea/app/cost/broker/BrokerOutPutCostTest.java
rename to app/src/test/java/org/astraea/app/cost/BrokerOutPutCostTest.java
index 7afe49ce..ed72efa9 100644
--- a/app/src/test/java/org/astraea/app/cost/broker/BrokerOutPutCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/BrokerOutPutCostTest.java
@@ -14,15 +14,12 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.astraea.app.cost.broker;
+package org.astraea.app.cost;
 
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import org.astraea.app.admin.ClusterBean;
-import org.astraea.app.cost.ClusterInfo;
-import org.astraea.app.cost.FakeClusterInfo;
-import org.astraea.app.cost.Normalizer;
 import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.collector.BeanCollector;
 import org.astraea.app.metrics.collector.Receiver;
diff --git a/app/src/test/java/org/astraea/app/cost/broker/CpuCostTest.java b/app/src/test/java/org/astraea/app/cost/CpuCostTest.java
similarity index 95%
rename from app/src/test/java/org/astraea/app/cost/broker/CpuCostTest.java
rename to app/src/test/java/org/astraea/app/cost/CpuCostTest.java
index ec4faece..88f62b3e 100644
--- a/app/src/test/java/org/astraea/app/cost/broker/CpuCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/CpuCostTest.java
@@ -14,18 +14,13 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.astraea.app.cost.broker;
+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.cost.ClusterInfo;
-import org.astraea.app.cost.FakeClusterInfo;
-import org.astraea.app.cost.NodeInfo;
-import org.astraea.app.cost.Normalizer;
-import org.astraea.app.cost.ReplicaInfo;
 import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.collector.BeanCollector;
 import org.astraea.app.metrics.collector.Receiver;
diff --git a/app/src/test/java/org/astraea/app/cost/broker/MemoryCostTest.java b/app/src/test/java/org/astraea/app/cost/MemoryCostTest.java
similarity index 95%
rename from app/src/test/java/org/astraea/app/cost/broker/MemoryCostTest.java
rename to app/src/test/java/org/astraea/app/cost/MemoryCostTest.java
index 51995e64..4364b009 100644
--- a/app/src/test/java/org/astraea/app/cost/broker/MemoryCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/MemoryCostTest.java
@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.astraea.app.cost.broker;
+package org.astraea.app.cost;
 
 import java.lang.management.MemoryUsage;
 import java.util.Collection;
@@ -22,11 +22,6 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import org.astraea.app.admin.ClusterBean;
-import org.astraea.app.cost.ClusterInfo;
-import org.astraea.app.cost.FakeClusterInfo;
-import org.astraea.app.cost.NodeInfo;
-import org.astraea.app.cost.Normalizer;
-import org.astraea.app.cost.ReplicaInfo;
 import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.collector.BeanCollector;
 import org.astraea.app.metrics.collector.Receiver;
diff --git a/app/src/test/java/org/astraea/app/cost/NumberOfLeaderCostTest.java b/app/src/test/java/org/astraea/app/cost/NumberOfLeaderCostTest.java
new file mode 100644
index 00000000..f24bf8fe
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/cost/NumberOfLeaderCostTest.java
@@ -0,0 +1,124 @@
+/*
+ * 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.Set;
+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.collector.Fetcher;
+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(Fetcher.of(Set.of(costFunction.fetcher())))
+                    .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/ReplicaDiskInCostTest.java b/app/src/test/java/org/astraea/app/cost/ReplicaDiskInCostTest.java
new file mode 100644
index 00000000..16e2c42c
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/cost/ReplicaDiskInCostTest.java
@@ -0,0 +1,146 @@
+/*
+ * 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.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.astraea.app.admin.ClusterBean;
+import org.astraea.app.admin.TopicPartition;
+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.astraea.app.service.RequireBrokerCluster;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class ReplicaDiskInCostTest extends RequireBrokerCluster {
+
+  @Test
+  void testPartitionCost() {
+    Map<Integer, Integer> brokerBandwidth = new HashMap<>();
+    brokerBandwidth.put(1, 50);
+    brokerBandwidth.put(2, 50);
+    brokerBandwidth.put(3, 50);
+    var loadCostFunction = new ReplicaDiskInCost(brokerBandwidth);
+    var broker1ReplicaLoad = loadCostFunction.partitionCost(exampleClusterInfo()).value(1);
+    var broker2ReplicaLoad = loadCostFunction.partitionCost(exampleClusterInfo()).value(2);
+    var broker3ReplicaLoad = loadCostFunction.partitionCost(exampleClusterInfo()).value(3);
+    // broker1
+    Assertions.assertEquals(
+        0.23841381072998047, broker1ReplicaLoad.get(new TopicPartition("test-1", 0)));
+    Assertions.assertEquals(
+        0.1907339096069336, broker1ReplicaLoad.get(new TopicPartition("test-2", 0)));
+    // broker2
+    Assertions.assertEquals(
+        0.23841381072998047, broker2ReplicaLoad.get(new TopicPartition("test-1", 0)));
+    Assertions.assertEquals(
+        0.476834774017334, broker2ReplicaLoad.get(new TopicPartition("test-1", 1)));
+    // broker3
+    Assertions.assertEquals(
+        0.476834774017334, broker3ReplicaLoad.get(new TopicPartition("test-1", 1)));
+    Assertions.assertEquals(
+        0.1907339096069336, broker3ReplicaLoad.get(new TopicPartition("test-2", 0)));
+  }
+
+  @Test
+  void testBrokerCost() {
+    Map<Integer, Integer> properties = new HashMap<>();
+    properties.put(1, 50);
+    properties.put(2, 50);
+    properties.put(3, 50);
+    var loadCostFunction = new ReplicaDiskInCost(properties);
+    var brokerLoad = loadCostFunction.brokerCost(exampleClusterInfo()).value();
+    Assertions.assertEquals(0.23841381072998047 + 0.1907339096069336, brokerLoad.get(1));
+    Assertions.assertEquals(0.23841381072998047 + 0.476834774017334, brokerLoad.get(2));
+    Assertions.assertEquals(0.476834774017334 + 0.1907339096069336, brokerLoad.get(3));
+  }
+
+  private ClusterInfo exampleClusterInfo() {
+    var oldTP1_0 =
+        fakeBeanObject(
+            "Log", KafkaMetrics.TopicPartition.Size.metricName(), "test-1", "0", 1000, 1000L);
+    var newTP1_0 =
+        fakeBeanObject(
+            "Log", KafkaMetrics.TopicPartition.Size.metricName(), "test-1", "0", 50000000, 5000L);
+    var oldTP1_1 =
+        fakeBeanObject(
+            "Log", KafkaMetrics.TopicPartition.Size.metricName(), "test-1", "1", 500, 1000L);
+    var newTP1_1 =
+        fakeBeanObject(
+            "Log", KafkaMetrics.TopicPartition.Size.metricName(), "test-1", "1", 100000000, 5000L);
+    var oldTP2_0 =
+        fakeBeanObject(
+            "Log", KafkaMetrics.TopicPartition.Size.metricName(), "test-2", "0", 200, 1000L);
+    var newTP2_0 =
+        fakeBeanObject(
+            "Log", KafkaMetrics.TopicPartition.Size.metricName(), "test-2", "0", 40000000, 5000L);
+    /*
+    test replica distribution :
+        broker1 : test-1-0,test-2-0
+        broker2 : test-1-0,test-1-1
+        broker1 : test-1-1,test-2-0
+     */
+    Collection<HasBeanObject> broker1 = List.of(oldTP1_0, newTP1_0, oldTP2_0, newTP2_0);
+    Collection<HasBeanObject> broker2 = List.of(oldTP1_0, newTP1_0, oldTP1_1, newTP1_1);
+    Collection<HasBeanObject> broker3 = List.of(oldTP1_1, newTP1_1, oldTP2_0, newTP2_0);
+    return new FakeClusterInfo() {
+      @Override
+      public List<NodeInfo> nodes() {
+        return List.of(NodeInfo.of(1, "", -1), NodeInfo.of(2, "", -1), NodeInfo.of(3, "", -1));
+      }
+
+      @Override
+      public Set<String> topics() {
+        return Set.of("test-1", "test-2");
+      }
+
+      @Override
+      public List<ReplicaInfo> replicas(String topic) {
+        if (topic.equals("test-1"))
+          return List.of(
+              ReplicaInfo.of("test-1", 0, NodeInfo.of(1, "", -1), true, true, false),
+              ReplicaInfo.of("test-1", 0, NodeInfo.of(2, "", -1), false, true, false),
+              ReplicaInfo.of("test-1", 1, NodeInfo.of(2, "", -1), false, true, false),
+              ReplicaInfo.of("test-1", 1, NodeInfo.of(3, "", -1), true, true, false));
+        else
+          return List.of(
+              ReplicaInfo.of("test-2", 0, NodeInfo.of(1, "", -1), false, true, false),
+              ReplicaInfo.of("test-2", 0, NodeInfo.of(3, "", -1), true, true, false));
+      }
+
+      @Override
+      public ClusterBean clusterBean() {
+        return ClusterBean.of(Map.of(1, broker1, 2, broker2, 3, broker3));
+      }
+    };
+  }
+
+  private HasValue fakeBeanObject(
+      String type, String name, String topic, String partition, long size, long time) {
+    BeanObject beanObject =
+        new BeanObject(
+            "",
+            Map.of("name", name, "type", type, "topic", topic, "partition", partition),
+            Map.of("Value", size),
+            time);
+    return HasValue.of(beanObject);
+  }
+}
diff --git a/app/src/test/java/org/astraea/app/cost/ReplicaSizeCostTest.java b/app/src/test/java/org/astraea/app/cost/ReplicaSizeCostTest.java
new file mode 100644
index 00000000..acaa50e7
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/cost/ReplicaSizeCostTest.java
@@ -0,0 +1,169 @@
+/*
+ * 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.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import org.astraea.app.admin.Admin;
+import org.astraea.app.admin.ClusterBean;
+import org.astraea.app.admin.TopicPartition;
+import org.astraea.app.metrics.HasBeanObject;
+import org.astraea.app.metrics.collector.BeanCollector;
+import org.astraea.app.metrics.collector.Fetcher;
+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.producer.Producer;
+import org.astraea.app.service.RequireSingleBrokerCluster;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class ReplicaSizeCostTest extends RequireSingleBrokerCluster {
+  @Test
+  void testGetMetrics() {
+    var brokerDiskSize = Map.of(1, 1000, 2, 1000, 3, 1000);
+    var topicName = "testGetMetrics-1";
+    try (var admin = Admin.of(bootstrapServers())) {
+      var host = "localhost";
+      admin.creator().topic(topicName).numberOfPartitions(1).numberOfReplicas((short) 1).create();
+      // wait for topic creation
+      TimeUnit.SECONDS.sleep(5);
+      var producer = Producer.builder().bootstrapServers(bootstrapServers()).build();
+      producer.sender().topic(topicName).key(new byte[10000]).run().toCompletableFuture().get();
+      ReplicaSizeCost costFunction = new ReplicaSizeCost(brokerDiskSize);
+      var beanObjects =
+          BeanCollector.builder()
+              .interval(Duration.ofSeconds(4))
+              .build()
+              .register()
+              .host(host)
+              .port(jmxServiceURL().getPort())
+              .fetcher(Fetcher.of(Set.of(costFunction.fetcher())))
+              .build()
+              .current();
+      var replicaSize =
+          beanObjects.stream()
+              .filter(x -> x instanceof HasValue)
+              .filter(x -> x.beanObject().getProperties().get("name").equals("Size"))
+              .filter(x -> x.beanObject().getProperties().get("type").equals("Log"))
+              .sorted(Comparator.comparing(HasBeanObject::createdTimestamp).reversed())
+              .map(x -> (HasValue) x)
+              .limit(1)
+              .map(e2 -> (int) e2.value())
+              .collect(Collectors.toList());
+      Assertions.assertTrue(replicaSize.get(0) >= 10000);
+    } catch (InterruptedException | ExecutionException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  @Test
+  void partitionCost() {
+    var brokerDiskSize = Map.of(1, 1000, 2, 1000, 3, 1000);
+    var loadCostFunction = new ReplicaSizeCost(brokerDiskSize);
+    var broker1ReplicaLoad = loadCostFunction.partitionCost(exampleClusterInfo()).value(1);
+    var broker2ReplicaLoad = loadCostFunction.partitionCost(exampleClusterInfo()).value(2);
+    var broker3ReplicaLoad = loadCostFunction.partitionCost(exampleClusterInfo()).value(3);
+    // broker1
+    Assertions.assertEquals(0.85, broker1ReplicaLoad.get(new TopicPartition("test-1", 0)));
+    Assertions.assertEquals(0.45, broker1ReplicaLoad.get(new TopicPartition("test-1", 1)));
+    Assertions.assertEquals(0.35, broker1ReplicaLoad.get(new TopicPartition("test-2", 1)));
+    // broker2
+    Assertions.assertEquals(0.45, broker2ReplicaLoad.get(new TopicPartition("test-1", 1)));
+    Assertions.assertEquals(0, broker2ReplicaLoad.get(new TopicPartition("test-2", 0)));
+    // broker3
+    Assertions.assertEquals(0.85, broker3ReplicaLoad.get(new TopicPartition("test-1", 0)));
+    Assertions.assertEquals(0, broker3ReplicaLoad.get(new TopicPartition("test-2", 0)));
+    Assertions.assertEquals(0.35, broker3ReplicaLoad.get(new TopicPartition("test-2", 1)));
+  }
+
+  @Test
+  void brokerCost() {
+    var brokerDiskSize = Map.of(1, 1000, 2, 1000, 3, 1000);
+    var loadCostFunction = new ReplicaSizeCost(brokerDiskSize);
+    var brokerReplicaLoad = loadCostFunction.brokerCost(exampleClusterInfo()).value();
+    Assertions.assertEquals(brokerReplicaLoad.get(1), 0.85 + 0.45 + 0.35);
+    Assertions.assertEquals(brokerReplicaLoad.get(2), 0.45);
+    Assertions.assertEquals(brokerReplicaLoad.get(3), 0.85 + 0.35);
+  }
+
+  private ClusterInfo exampleClusterInfo() {
+    var sizeTP1_0 =
+        fakeBeanObject(
+            "Log", KafkaMetrics.TopicPartition.Size.metricName(), "test-1", "0", 891289600);
+    var sizeTP1_1 =
+        fakeBeanObject(
+            "Log", KafkaMetrics.TopicPartition.Size.metricName(), "test-1", "1", 471859200);
+    var sizeTP2_0 =
+        fakeBeanObject("Log", KafkaMetrics.TopicPartition.Size.metricName(), "test-2", "0", 0);
+    var sizeTP2_1 =
+        fakeBeanObject(
+            "Log", KafkaMetrics.TopicPartition.Size.metricName(), "test-2", "1", 367001600);
+    Collection<HasBeanObject> broker1 = List.of(sizeTP1_0, sizeTP1_1, sizeTP2_1);
+    Collection<HasBeanObject> broker2 = List.of(sizeTP1_1, sizeTP2_0);
+    Collection<HasBeanObject> broker3 = List.of(sizeTP2_1, sizeTP1_0, sizeTP2_0);
+    return new FakeClusterInfo() {
+      @Override
+      public List<NodeInfo> nodes() {
+        return List.of(NodeInfo.of(1, "", -1), NodeInfo.of(2, "", -1), NodeInfo.of(3, "", -1));
+      }
+
+      @Override
+      public Set<String> topics() {
+        return Set.of("test-1", "test-2");
+      }
+
+      @Override
+      public List<ReplicaInfo> replicas(String topic) {
+        if (topic.equals("test-1"))
+          return List.of(
+              ReplicaInfo.of("test-1", 0, NodeInfo.of(1, "", -1), true, true, false),
+              ReplicaInfo.of("test-1", 0, NodeInfo.of(3, "", -1), false, true, false),
+              ReplicaInfo.of("test-1", 1, NodeInfo.of(1, "", -1), false, true, false),
+              ReplicaInfo.of("test-1", 1, NodeInfo.of(2, "", -1), true, true, false));
+        else
+          return List.of(
+              ReplicaInfo.of("test-2", 0, NodeInfo.of(2, "", -1), false, true, false),
+              ReplicaInfo.of("test-2", 0, NodeInfo.of(3, "", -1), true, true, false),
+              ReplicaInfo.of("test-2", 1, NodeInfo.of(1, "", -1), false, true, false),
+              ReplicaInfo.of("test-2", 1, NodeInfo.of(3, "", -1), true, true, false));
+      }
+
+      @Override
+      public ClusterBean clusterBean() {
+        return ClusterBean.of(Map.of(1, broker1, 2, broker2, 3, broker3));
+      }
+    };
+  }
+
+  private HasValue fakeBeanObject(
+      String type, String name, String topic, String partition, long size) {
+    BeanObject beanObject =
+        new BeanObject(
+            "kafka.log",
+            Map.of("name", name, "type", type, "topic", topic, "partition", partition),
+            Map.of("Value", size));
+    return HasValue.of(beanObject);
+  }
+}
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 4f2730e2..56c407a5 100644
--- a/app/src/test/java/org/astraea/app/partitioner/StrictCostDispatcherTest.java
+++ b/app/src/test/java/org/astraea/app/partitioner/StrictCostDispatcherTest.java
@@ -24,12 +24,12 @@ import java.util.function.Function;
 import java.util.stream.Collectors;
 import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.cost.BrokerCost;
+import org.astraea.app.cost.BrokerInputCost;
 import org.astraea.app.cost.ClusterInfo;
 import org.astraea.app.cost.HasBrokerCost;
 import org.astraea.app.cost.NodeInfo;
 import org.astraea.app.cost.ReplicaInfo;
 import org.astraea.app.cost.ThroughputCost;
-import org.astraea.app.cost.broker.BrokerInputCost;
 import org.astraea.app.metrics.collector.Fetcher;
 import org.astraea.app.metrics.collector.Receiver;
 import org.junit.jupiter.api.Assertions;
@@ -178,24 +178,20 @@ public class StrictCostDispatcherTest {
     var config =
         Configuration.of(
             Map.of(
-                "org.astraea.app.cost.broker.BrokerInputCost",
+                "org.astraea.app.cost.BrokerInputCost",
                 "20",
-                "org.astraea.app.cost.broker.BrokerOutputCost",
+                "org.astraea.app.cost.BrokerOutputCost",
                 "1.25"));
     var ans = StrictCostDispatcher.parseCostFunctionWeight(config);
     Assertions.assertEquals(2, ans.size());
     for (var entry : ans.entrySet()) {
-      if (entry
-          .getKey()
-          .getClass()
-          .getName()
-          .equals("org.astraea.app.cost.broker.BrokerInputCost")) {
+      if (entry.getKey().getClass().getName().equals("org.astraea.app.cost.BrokerInputCost")) {
         Assertions.assertEquals(20.0, entry.getValue());
       } else if (entry
           .getKey()
           .getClass()
           .getName()
-          .equals("org.astraea.app.cost.broker.BrokerOutputCost")) {
+          .equals("org.astraea.app.cost.BrokerOutputCost")) {
         Assertions.assertEquals(1.25, entry.getValue());
       } else {
         Assertions.assertEquals(0.0, entry.getValue());
@@ -206,9 +202,9 @@ public class StrictCostDispatcherTest {
     var config2 =
         Configuration.of(
             Map.of(
-                "org.astraea.app.cost.broker.BrokerInputCost",
+                "org.astraea.app.cost.BrokerInputCost",
                 "-20",
-                "org.astraea.app.cost.broker.BrokerOutputCost",
+                "org.astraea.app.cost.BrokerOutputCost",
                 "1.25"));
     Assertions.assertThrows(
         IllegalArgumentException.class,
diff --git a/app/src/test/java/org/astraea/app/service/RequireBrokerCluster.java b/app/src/test/java/org/astraea/app/service/RequireBrokerCluster.java
index bcabd719..afe6359d 100644
--- a/app/src/test/java/org/astraea/app/service/RequireBrokerCluster.java
+++ b/app/src/test/java/org/astraea/app/service/RequireBrokerCluster.java
@@ -27,8 +27,8 @@ import org.junit.jupiter.api.AfterAll;
  */
 public abstract class RequireBrokerCluster extends RequireJmxServer {
   private static final int NUMBER_OF_BROKERS = 3;
-  private static ZookeeperCluster ZOOKEEPER_CLUSTER = Services.zookeeperCluster();
-  private static BrokerCluster BROKER_CLUSTER =
+  private static final ZookeeperCluster ZOOKEEPER_CLUSTER = Services.zookeeperCluster();
+  private static final BrokerCluster BROKER_CLUSTER =
       Services.brokerCluster(ZOOKEEPER_CLUSTER, NUMBER_OF_BROKERS);
 
   protected static String bootstrapServers() {
@@ -39,12 +39,6 @@ public abstract class RequireBrokerCluster extends RequireJmxServer {
     return BROKER_CLUSTER.logFolders();
   }
 
-  protected static void restartCluster() {
-    shutdownClusters();
-    ZOOKEEPER_CLUSTER = Services.zookeeperCluster();
-    BROKER_CLUSTER = Services.brokerCluster(ZOOKEEPER_CLUSTER, NUMBER_OF_BROKERS);
-  }
-
   protected static void closeBroker(int brokerID) {
     BROKER_CLUSTER.close(brokerID);
   }
diff --git a/app/src/test/java/org/astraea/app/service/RequireSingleBrokerCluster.java b/app/src/test/java/org/astraea/app/service/RequireSingleBrokerCluster.java
new file mode 100644
index 00000000..655677a4
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/service/RequireSingleBrokerCluster.java
@@ -0,0 +1,51 @@
+/*
+ * 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.service;
+
+import java.util.Map;
+import java.util.Set;
+import org.astraea.app.common.Utils;
+import org.junit.jupiter.api.AfterAll;
+
+/**
+ * This class offers a way to have embedded kafka cluster. It is useful to test code which is
+ * depended on true cluster.
+ */
+public abstract class RequireSingleBrokerCluster extends RequireJmxServer {
+  private static final int NUMBER_OF_BROKERS = 1;
+  private static final ZookeeperCluster ZOOKEEPER_CLUSTER = Services.zookeeperCluster();
+  private static final BrokerCluster BROKER_CLUSTER =
+      Services.brokerCluster(ZOOKEEPER_CLUSTER, NUMBER_OF_BROKERS);
+
+  protected static String bootstrapServers() {
+    return BROKER_CLUSTER.bootstrapServers();
+  }
+
+  protected static Map<Integer, Set<String>> logFolders() {
+    return BROKER_CLUSTER.logFolders();
+  }
+
+  protected static Set<Integer> brokerIds() {
+    return logFolders().keySet();
+  }
+
+  @AfterAll
+  static void shutdownClusters() {
+    Utils.swallowException(BROKER_CLUSTER::close);
+    Utils.swallowException(ZOOKEEPER_CLUSTER::close);
+  }
+}