Repository: skiptests/astraea
Pull request #479 (2022-07-13 16:50:50)

Move ClusterBean out of ClusterInfo
這隻PR主要是將`ClusterBean`獨立出來，原因是原本`beans`只是個簡單的資料結構，但經過#411調整後，`ClusterBean`成為一個獨立且具有豐富功能的物件，因此已經不適合再包裹在`ClusterInfo`底下


Patch:
diff --git a/app/src/main/java/org/astraea/app/admin/Builder.java b/app/src/main/java/org/astraea/app/admin/Builder.java
index cd157f37..b8e92763 100644
--- a/app/src/main/java/org/astraea/app/admin/Builder.java
+++ b/app/src/main/java/org/astraea/app/admin/Builder.java
@@ -520,11 +520,6 @@ public class Builder {
                 "This ClusterInfo have no information about topic \"" + topic + "\"");
           else return replicaInfos;
         }
-
-        @Override
-        public ClusterBean clusterBean() {
-          return ClusterBean.of(Map.of());
-        }
       };
     }
 
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 788b7654..74d54697 100644
--- a/app/src/main/java/org/astraea/app/admin/ClusterBean.java
+++ b/app/src/main/java/org/astraea/app/admin/ClusterBean.java
@@ -26,6 +26,8 @@ import org.astraea.app.metrics.HasBeanObject;
 
 /** Used to get beanObject using a variety of different keys . */
 public interface ClusterBean {
+  ClusterBean EMPTY = ClusterBean.of(Map.of());
+
   private static Collection<HasBeanObject> compareBeanObject(
       Collection<HasBeanObject> x1, Collection<HasBeanObject> x2) {
     var beanObject2 = x2.iterator().next().beanObject();
diff --git a/app/src/main/java/org/astraea/app/admin/ClusterInfo.java b/app/src/main/java/org/astraea/app/admin/ClusterInfo.java
index 764c1f72..83076704 100644
--- a/app/src/main/java/org/astraea/app/admin/ClusterInfo.java
+++ b/app/src/main/java/org/astraea/app/admin/ClusterInfo.java
@@ -16,16 +16,46 @@
  */
 package org.astraea.app.admin;
 
-import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.Set;
 import java.util.stream.Collectors;
-import java.util.stream.Stream;
-import org.astraea.app.metrics.HasBeanObject;
 
 public interface ClusterInfo {
+  ClusterInfo EMPTY =
+      new ClusterInfo() {
+
+        @Override
+        public List<NodeInfo> nodes() {
+          return List.of();
+        }
+
+        @Override
+        public Set<String> dataDirectories(int brokerId) {
+          return Set.of();
+        }
+
+        @Override
+        public List<ReplicaInfo> availableReplicaLeaders(String topic) {
+          return List.of();
+        }
+
+        @Override
+        public List<ReplicaInfo> availableReplicas(String topic) {
+          return List.of();
+        }
+
+        @Override
+        public Set<String> topics() {
+          return Set.of();
+        }
+
+        @Override
+        public List<ReplicaInfo> replicas(String topic) {
+          return List.of();
+        }
+      };
 
   /**
    * convert the kafka Cluster to our ClusterInfo. All data structure are converted immediately, so
@@ -82,72 +112,6 @@ public interface ClusterInfo {
       public List<ReplicaInfo> replicas(String topic) {
         return replicas;
       }
-
-      @Override
-      public ClusterBean clusterBean() {
-        return ClusterBean.of(Map.of());
-      }
-    };
-  }
-
-  /**
-   * merge the beans into cluster information
-   *
-   * @param clusterInfo cluster information
-   * @param beans extra beans
-   * @return a new cluster information with extra beans
-   */
-  static ClusterInfo of(
-      ClusterInfo clusterInfo, Map<Integer, ? extends Collection<HasBeanObject>> beans) {
-    var clusterBean =
-        ClusterBean.of(
-            Stream.concat(
-                    clusterInfo.clusterBean().all().entrySet().stream(), beans.entrySet().stream())
-                .collect(
-                    Collectors.toMap(
-                        Map.Entry::getKey,
-                        Map.Entry::getValue,
-                        // Normally, the cluster info used to merge with beans has no beans. Hence,
-                        // the cost of merge is acceptable
-                        (l, r) ->
-                            Stream.concat(l.stream(), r.stream())
-                                .collect(Collectors.toUnmodifiableList()))));
-    return new ClusterInfo() {
-
-      @Override
-      public List<NodeInfo> nodes() {
-        return clusterInfo.nodes();
-      }
-
-      @Override
-      public Set<String> dataDirectories(int brokerId) {
-        return clusterInfo.dataDirectories(brokerId);
-      }
-
-      @Override
-      public List<ReplicaInfo> availableReplicaLeaders(String topic) {
-        return clusterInfo.availableReplicaLeaders(topic);
-      }
-
-      @Override
-      public List<ReplicaInfo> availableReplicas(String topic) {
-        return clusterInfo.availableReplicas(topic);
-      }
-
-      @Override
-      public Set<String> topics() {
-        return clusterInfo.topics();
-      }
-
-      @Override
-      public List<ReplicaInfo> replicas(String topic) {
-        return clusterInfo.replicas(topic);
-      }
-
-      @Override
-      public ClusterBean clusterBean() {
-        return clusterBean;
-      }
     };
   }
 
@@ -224,7 +188,4 @@ public interface ClusterInfo {
    *     the given topic is unknown to this ClusterInfo
    */
   List<ReplicaInfo> replicas(String topic);
-
-  /** @return a {@link ClusterBean} used to get beanObject using a variety of different keys. */
-  ClusterBean clusterBean();
 }
diff --git a/app/src/main/java/org/astraea/app/cost/BrokerInputCost.java b/app/src/main/java/org/astraea/app/cost/BrokerInputCost.java
index f3bf6c75..9b5f9de5 100644
--- a/app/src/main/java/org/astraea/app/cost/BrokerInputCost.java
+++ b/app/src/main/java/org/astraea/app/cost/BrokerInputCost.java
@@ -22,6 +22,7 @@ import java.util.Map;
 import java.util.Optional;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
+import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.metrics.KafkaMetrics;
 import org.astraea.app.metrics.broker.BrokerTopicMetricsResult;
@@ -42,9 +43,9 @@ public class BrokerInputCost implements HasBrokerCost {
   private final Map<Integer, BrokerMetric> brokersMetric = new HashMap<>();
 
   @Override
-  public BrokerCost brokerCost(ClusterInfo clusterInfo) {
+  public BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
     var costMetrics =
-        clusterInfo.clusterBean().all().entrySet().stream()
+        clusterBean.all().entrySet().stream()
             .collect(
                 Collectors.toMap(
                     Map.Entry::getKey,
diff --git a/app/src/main/java/org/astraea/app/cost/BrokerOutputCost.java b/app/src/main/java/org/astraea/app/cost/BrokerOutputCost.java
index bba706c8..56ed1ae4 100644
--- a/app/src/main/java/org/astraea/app/cost/BrokerOutputCost.java
+++ b/app/src/main/java/org/astraea/app/cost/BrokerOutputCost.java
@@ -22,6 +22,7 @@ import java.util.Map;
 import java.util.Optional;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
+import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.metrics.KafkaMetrics;
 import org.astraea.app.metrics.broker.BrokerTopicMetricsResult;
@@ -42,9 +43,9 @@ public class BrokerOutputCost implements HasBrokerCost {
   private final Map<Integer, BrokerMetric> brokersMetric = new HashMap<>();
 
   @Override
-  public BrokerCost brokerCost(ClusterInfo clusterInfo) {
+  public BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
     var costMetrics =
-        clusterInfo.clusterBean().all().entrySet().stream()
+        clusterBean.all().entrySet().stream()
             .collect(
                 Collectors.toMap(
                     Map.Entry::getKey,
diff --git a/app/src/main/java/org/astraea/app/cost/CpuCost.java b/app/src/main/java/org/astraea/app/cost/CpuCost.java
index 8b2772df..e8883e2c 100644
--- a/app/src/main/java/org/astraea/app/cost/CpuCost.java
+++ b/app/src/main/java/org/astraea/app/cost/CpuCost.java
@@ -22,6 +22,7 @@ import java.util.Map;
 import java.util.Optional;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
+import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.metrics.KafkaMetrics;
 import org.astraea.app.metrics.collector.Fetcher;
@@ -42,9 +43,9 @@ public class CpuCost implements HasBrokerCost {
   private final Map<Integer, BrokerMetric> brokersMetric = new HashMap<>();
 
   @Override
-  public BrokerCost brokerCost(ClusterInfo clusterInfo) {
+  public BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
     var costMetrics =
-        clusterInfo.clusterBean().all().entrySet().stream()
+        clusterBean.all().entrySet().stream()
             .collect(
                 Collectors.toMap(
                     Map.Entry::getKey,
diff --git a/app/src/main/java/org/astraea/app/cost/HasBrokerCost.java b/app/src/main/java/org/astraea/app/cost/HasBrokerCost.java
index 5d6efd7b..63e3b3be 100644
--- a/app/src/main/java/org/astraea/app/cost/HasBrokerCost.java
+++ b/app/src/main/java/org/astraea/app/cost/HasBrokerCost.java
@@ -16,6 +16,7 @@
  */
 package org.astraea.app.cost;
 
+import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
 
 public interface HasBrokerCost extends CostFunction {
@@ -23,7 +24,8 @@ public interface HasBrokerCost extends CostFunction {
    * score all nodes for a particular metrics according to passed beans and cluster information.
    *
    * @param clusterInfo cluster information
+   * @param clusterBean cluster metrics
    * @return the score of each broker.
    */
-  BrokerCost brokerCost(ClusterInfo clusterInfo);
+  BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean);
 }
diff --git a/app/src/main/java/org/astraea/app/cost/HasPartitionCost.java b/app/src/main/java/org/astraea/app/cost/HasPartitionCost.java
index d3609744..bba1c06a 100644
--- a/app/src/main/java/org/astraea/app/cost/HasPartitionCost.java
+++ b/app/src/main/java/org/astraea/app/cost/HasPartitionCost.java
@@ -16,6 +16,7 @@
  */
 package org.astraea.app.cost;
 
+import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
 
 public interface HasPartitionCost extends CostFunction {
@@ -23,7 +24,8 @@ public interface HasPartitionCost extends CostFunction {
    * score all nodes according to passed beans and cluster information.
    *
    * @param clusterInfo cluster information
-   * @return the score of each partition. The score ranges in [0 - 1].
+   * @param clusterBean cluster metrics
+   * @return the score of each partition
    */
-  PartitionCost partitionCost(ClusterInfo clusterInfo);
+  PartitionCost partitionCost(ClusterInfo clusterInfo, ClusterBean clusterBean);
 }
diff --git a/app/src/main/java/org/astraea/app/cost/LoadCost.java b/app/src/main/java/org/astraea/app/cost/LoadCost.java
index 1a59c45f..56aa4c7c 100644
--- a/app/src/main/java/org/astraea/app/cost/LoadCost.java
+++ b/app/src/main/java/org/astraea/app/cost/LoadCost.java
@@ -23,6 +23,7 @@ import java.util.Map;
 import java.util.Optional;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
+import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.KafkaMetrics;
@@ -41,8 +42,8 @@ public class LoadCost implements HasBrokerCost {
 
   /** Do "Poisson" and "weightPoisson" calculation on "load". And change output to double. */
   @Override
-  public BrokerCost brokerCost(ClusterInfo clusterInfo) {
-    var load = computeLoad(clusterInfo.clusterBean().all());
+  public BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
+    var load = computeLoad(clusterBean.all());
 
     // Poisson calculation (-> Poisson -> throughputAbility -> to double)
     var brokerScore =
diff --git a/app/src/main/java/org/astraea/app/cost/MemoryCost.java b/app/src/main/java/org/astraea/app/cost/MemoryCost.java
index 1461c56d..654d59da 100644
--- a/app/src/main/java/org/astraea/app/cost/MemoryCost.java
+++ b/app/src/main/java/org/astraea/app/cost/MemoryCost.java
@@ -22,6 +22,7 @@ import java.util.Map;
 import java.util.Optional;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
+import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.metrics.KafkaMetrics;
 import org.astraea.app.metrics.collector.Fetcher;
@@ -42,9 +43,9 @@ public class MemoryCost extends Periodic<Map<Integer, Double>> implements HasBro
    * </ol>
    */
   @Override
-  public BrokerCost brokerCost(ClusterInfo clusterInfo) {
+  public BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
     var costMetrics =
-        clusterInfo.clusterBean().all().entrySet().stream()
+        clusterBean.all().entrySet().stream()
             .collect(
                 Collectors.toMap(
                     Map.Entry::getKey,
diff --git a/app/src/main/java/org/astraea/app/cost/NeutralIntegratedCost.java b/app/src/main/java/org/astraea/app/cost/NeutralIntegratedCost.java
index e43c0624..802bc32f 100644
--- a/app/src/main/java/org/astraea/app/cost/NeutralIntegratedCost.java
+++ b/app/src/main/java/org/astraea/app/cost/NeutralIntegratedCost.java
@@ -22,6 +22,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.stream.Collectors;
+import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.metrics.collector.Fetcher;
 
@@ -41,9 +42,9 @@ public class NeutralIntegratedCost implements HasBrokerCost {
   private final WeightProvider weightProvider = WeightProvider.entropy(Normalizer.minMax(true));
 
   @Override
-  public BrokerCost brokerCost(ClusterInfo clusterInfo) {
+  public BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
     var costMetrics =
-        clusterInfo.clusterBean().all().entrySet().stream()
+        clusterBean.all().entrySet().stream()
             .collect(Collectors.toMap(Map.Entry::getKey, entry -> 0.0));
     costMetrics.forEach(
         (key, value) -> {
@@ -52,7 +53,7 @@ public class NeutralIntegratedCost implements HasBrokerCost {
           }
         });
 
-    metricsCost.forEach(hasBrokerCost -> setBrokerMetrics(hasBrokerCost, clusterInfo));
+    metricsCost.forEach(hasBrokerCost -> setBrokerMetrics(hasBrokerCost, clusterInfo, clusterBean));
 
     var entropyEmpowerment = weight(weightProvider, brokersMetric);
     var entropyEmpowermentSum =
@@ -92,9 +93,10 @@ public class NeutralIntegratedCost implements HasBrokerCost {
   // Save the original value of each metric and the value of the metric after the TScore
   // calculation.
   // TODO Refactor
-  void setBrokerMetrics(HasBrokerCost hasBrokerCost, ClusterInfo clusterInfo) {
+  void setBrokerMetrics(
+      HasBrokerCost hasBrokerCost, ClusterInfo clusterInfo, ClusterBean clusterBean) {
     if (hasBrokerCost instanceof BrokerInputCost) {
-      var inputBrokerCost = hasBrokerCost.brokerCost(clusterInfo);
+      var inputBrokerCost = hasBrokerCost.brokerCost(clusterInfo, clusterBean);
       inputBrokerCost
           .value()
           .forEach((brokerID, value) -> brokersMetric.get(brokerID).inputScore = value);
@@ -103,7 +105,7 @@ public class NeutralIntegratedCost implements HasBrokerCost {
           .value()
           .forEach((brokerID, value) -> brokersMetric.get(brokerID).inputTScore = value);
     } else if (hasBrokerCost instanceof BrokerOutputCost) {
-      var outPutBrokerCost = hasBrokerCost.brokerCost(clusterInfo);
+      var outPutBrokerCost = hasBrokerCost.brokerCost(clusterInfo, clusterBean);
       outPutBrokerCost
           .value()
           .forEach((brokerID, value) -> brokersMetric.get(brokerID).outputScore = value);
@@ -112,14 +114,14 @@ public class NeutralIntegratedCost implements HasBrokerCost {
           .value()
           .forEach((brokerID, value) -> brokersMetric.get(brokerID).outputTScore = value);
     } else if (hasBrokerCost instanceof CpuCost) {
-      var CPUBrokerCost = hasBrokerCost.brokerCost(clusterInfo);
+      var CPUBrokerCost = hasBrokerCost.brokerCost(clusterInfo, clusterBean);
       CPUBrokerCost.value()
           .forEach((brokerID, value) -> brokersMetric.get(brokerID).cpuScore = value);
       CPUBrokerCost.normalize(Normalizer.TScore())
           .value()
           .forEach((brokerID, value) -> brokersMetric.get(brokerID).cpuTScore = value);
     } else if (hasBrokerCost instanceof MemoryCost) {
-      var MemoryBrokerCost = hasBrokerCost.brokerCost(clusterInfo);
+      var MemoryBrokerCost = hasBrokerCost.brokerCost(clusterInfo, clusterBean);
       MemoryBrokerCost.value()
           .forEach((brokerID, value) -> brokersMetric.get(brokerID).memoryScore = value);
       MemoryBrokerCost.normalize(Normalizer.TScore())
diff --git a/app/src/main/java/org/astraea/app/cost/NodeLatencyCost.java b/app/src/main/java/org/astraea/app/cost/NodeLatencyCost.java
index 908830b0..a2392d06 100644
--- a/app/src/main/java/org/astraea/app/cost/NodeLatencyCost.java
+++ b/app/src/main/java/org/astraea/app/cost/NodeLatencyCost.java
@@ -20,6 +20,7 @@ import java.util.Collection;
 import java.util.Map;
 import java.util.Optional;
 import java.util.stream.Collectors;
+import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.KafkaMetrics;
@@ -29,9 +30,9 @@ import org.astraea.app.metrics.producer.HasProducerNodeMetrics;
 public class NodeLatencyCost implements HasBrokerCost {
 
   @Override
-  public BrokerCost brokerCost(ClusterInfo clusterInfo) {
+  public BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
     var result =
-        clusterInfo.clusterBean().all().entrySet().stream()
+        clusterBean.all().entrySet().stream()
             .collect(
                 Collectors.toMap(
                     Map.Entry::getKey,
diff --git a/app/src/main/java/org/astraea/app/cost/NodeThroughputCost.java b/app/src/main/java/org/astraea/app/cost/NodeThroughputCost.java
index 43401875..ecfd6c8d 100644
--- a/app/src/main/java/org/astraea/app/cost/NodeThroughputCost.java
+++ b/app/src/main/java/org/astraea/app/cost/NodeThroughputCost.java
@@ -20,6 +20,7 @@ import java.util.Collection;
 import java.util.Map;
 import java.util.Optional;
 import java.util.stream.Collectors;
+import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.metrics.KafkaMetrics;
 import org.astraea.app.metrics.collector.Fetcher;
@@ -28,9 +29,9 @@ import org.astraea.app.metrics.producer.HasProducerNodeMetrics;
 public class NodeThroughputCost implements HasBrokerCost {
 
   @Override
-  public BrokerCost brokerCost(ClusterInfo clusterInfo) {
+  public BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
     var result =
-        clusterInfo.clusterBean().all().entrySet().stream()
+        clusterBean.all().entrySet().stream()
             .collect(
                 Collectors.toMap(
                     Map.Entry::getKey,
diff --git a/app/src/main/java/org/astraea/app/cost/ReplicaDiskInCost.java b/app/src/main/java/org/astraea/app/cost/ReplicaDiskInCost.java
index 06259f83..1fa60d9a 100644
--- a/app/src/main/java/org/astraea/app/cost/ReplicaDiskInCost.java
+++ b/app/src/main/java/org/astraea/app/cost/ReplicaDiskInCost.java
@@ -23,6 +23,7 @@ import java.util.Map;
 import java.util.Optional;
 import java.util.TreeMap;
 import java.util.stream.Collectors;
+import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.admin.NodeInfo;
 import org.astraea.app.admin.TopicPartition;
@@ -45,7 +46,7 @@ public class ReplicaDiskInCost implements HasBrokerCost, HasPartitionCost {
   }
 
   @Override
-  public BrokerCost brokerCost(ClusterInfo clusterInfo) {
+  public BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
     final Map<Integer, List<TopicPartitionReplica>> topicPartitionOfEachBroker =
         clusterInfo.topics().stream()
             .flatMap(topic -> clusterInfo.replicas(topic).stream())
@@ -61,7 +62,7 @@ public class ReplicaDiskInCost implements HasBrokerCost, HasPartitionCost {
                     NodeInfo::id,
                     node -> topicPartitionOfEachBroker.getOrDefault(node.id(), List.of())));
 
-    final var topicPartitionDataRate = topicPartitionDataRate(clusterInfo, Duration.ofSeconds(3));
+    final var topicPartitionDataRate = topicPartitionDataRate(clusterBean, Duration.ofSeconds(3));
 
     final var brokerLoad =
         actual.entrySet().stream()
@@ -87,8 +88,8 @@ public class ReplicaDiskInCost implements HasBrokerCost, HasPartitionCost {
   }
 
   @Override
-  public PartitionCost partitionCost(ClusterInfo clusterInfo) {
-    var replicaIn = replicaDataRate(clusterInfo, Duration.ofSeconds(2));
+  public PartitionCost partitionCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
+    var replicaIn = replicaDataRate(clusterBean, Duration.ofSeconds(2));
     var replicaScore =
         new TreeMap<TopicPartitionReplica, Double>(
             Comparator.comparing(TopicPartitionReplica::brokerId)
@@ -178,14 +179,14 @@ public class ReplicaDiskInCost implements HasBrokerCost, HasPartitionCost {
   /**
    * 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 clusterBean 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()
+      ClusterBean clusterBean, Duration sampleWindow) {
+    return replicaDataRate(clusterBean, sampleWindow).entrySet().stream()
         .map(
             x -> {
               var tpr = x.getKey();
@@ -195,8 +196,8 @@ public class ReplicaDiskInCost implements HasBrokerCost, HasPartitionCost {
   }
 
   public static Map<TopicPartitionReplica, Double> replicaDataRate(
-      ClusterInfo clusterInfo, Duration sampleWindow) {
-    return clusterInfo.clusterBean().mapByReplica().entrySet().parallelStream()
+      ClusterBean clusterBean, Duration sampleWindow) {
+    return clusterBean.mapByReplica().entrySet().parallelStream()
         .map(
             metrics -> {
               // calculate the increase rate over a specific window of time
diff --git a/app/src/main/java/org/astraea/app/cost/ReplicaLeaderCost.java b/app/src/main/java/org/astraea/app/cost/ReplicaLeaderCost.java
index 3011dd3f..97a23666 100644
--- a/app/src/main/java/org/astraea/app/cost/ReplicaLeaderCost.java
+++ b/app/src/main/java/org/astraea/app/cost/ReplicaLeaderCost.java
@@ -20,6 +20,7 @@ import java.util.Comparator;
 import java.util.Map;
 import java.util.Optional;
 import java.util.stream.Collectors;
+import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.KafkaMetrics;
@@ -30,15 +31,15 @@ import org.astraea.app.metrics.collector.Fetcher;
 public class ReplicaLeaderCost implements HasBrokerCost {
 
   @Override
-  public BrokerCost brokerCost(ClusterInfo clusterInfo) {
+  public BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
     var result =
-        leaderCount(clusterInfo).entrySet().stream()
+        leaderCount(clusterInfo, clusterBean).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()
+  Map<Integer, Integer> leaderCount(ClusterInfo ignored, ClusterBean clusterBean) {
+    return clusterBean.all().entrySet().stream()
         .flatMap(
             e ->
                 e.getValue().stream()
@@ -60,7 +61,7 @@ public class ReplicaLeaderCost implements HasBrokerCost {
   public static class NoMetrics extends ReplicaLeaderCost {
 
     @Override
-    Map<Integer, Integer> leaderCount(ClusterInfo clusterInfo) {
+    Map<Integer, Integer> leaderCount(ClusterInfo clusterInfo, ClusterBean ignored) {
       return clusterInfo.topics().stream()
           .flatMap(t -> clusterInfo.availableReplicaLeaders(t).stream())
           .collect(Collectors.groupingBy(r -> r.nodeInfo().id()))
diff --git a/app/src/main/java/org/astraea/app/cost/ReplicaSizeCost.java b/app/src/main/java/org/astraea/app/cost/ReplicaSizeCost.java
index ba7736fa..98669408 100644
--- a/app/src/main/java/org/astraea/app/cost/ReplicaSizeCost.java
+++ b/app/src/main/java/org/astraea/app/cost/ReplicaSizeCost.java
@@ -21,6 +21,7 @@ import java.util.Map;
 import java.util.Optional;
 import java.util.TreeMap;
 import java.util.stream.Collectors;
+import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.admin.NodeInfo;
 import org.astraea.app.admin.ReplicaInfo;
@@ -52,8 +53,8 @@ public class ReplicaSizeCost implements HasBrokerCost, HasPartitionCost {
    * @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);
+  public BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
+    var sizeOfReplica = getReplicaSize(clusterBean);
     var totalReplicaSizeInBroker =
         clusterInfo.nodes().stream()
             .collect(
@@ -82,9 +83,9 @@ public class ReplicaSizeCost implements HasBrokerCost, HasPartitionCost {
    *     each broker
    */
   @Override
-  public PartitionCost partitionCost(ClusterInfo clusterInfo) {
+  public PartitionCost partitionCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
     final long ONEMEGA = Math.round(Math.pow(2, 20));
-    var sizeOfReplica = getReplicaSize(clusterInfo);
+    var sizeOfReplica = getReplicaSize(clusterBean);
     TreeMap<TopicPartitionReplica, Double> replicaCost =
         new TreeMap<>(
             Comparator.comparing(TopicPartitionReplica::brokerId)
@@ -163,11 +164,11 @@ public class ReplicaSizeCost implements HasBrokerCost, HasPartitionCost {
   }
 
   /**
-   * @param clusterInfo the clusterInfo that offers the metrics related to topic/partition size
+   * @param clusterBean 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()
+  public Map<TopicPartitionReplica, Long> getReplicaSize(ClusterBean clusterBean) {
+    return clusterBean.mapByReplica().entrySet().stream()
         .flatMap(
             e ->
                 e.getValue().stream()
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 79fbdd1b..5aec6d23 100644
--- a/app/src/main/java/org/astraea/app/partitioner/StrictCostDispatcher.java
+++ b/app/src/main/java/org/astraea/app/partitioner/StrictCostDispatcher.java
@@ -24,6 +24,7 @@ import java.util.Optional;
 import java.util.TreeMap;
 import java.util.function.Function;
 import java.util.stream.Collectors;
+import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.admin.NodeInfo;
 import org.astraea.app.admin.ReplicaInfo;
@@ -126,8 +127,8 @@ public class StrictCostDispatcher implements Dispatcher {
       roundRobin =
           newRoundRobin(
               functions,
-              ClusterInfo.of(
-                  clusterInfo,
+              clusterInfo,
+              ClusterBean.of(
                   receivers.entrySet().stream()
                       .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().current()))));
       timeToUpdateRoundRobin = System.currentTimeMillis() + roundRobinLease.toMillis();
@@ -158,14 +159,14 @@ public class StrictCostDispatcher implements Dispatcher {
    * @return SmoothWeightedRoundRobin
    */
   static RoundRobin<Integer> newRoundRobin(
-      Map<CostFunction, Double> costFunctions, ClusterInfo clusterInfo) {
+      Map<CostFunction, Double> costFunctions, ClusterInfo clusterInfo, ClusterBean clusterBean) {
     var weightedCost =
         costFunctions.entrySet().stream()
             .filter(e -> e.getKey() instanceof HasBrokerCost)
             .flatMap(
                 functionWeight ->
                     ((HasBrokerCost) functionWeight.getKey())
-                        .brokerCost(clusterInfo).value().entrySet().stream()
+                        .brokerCost(clusterInfo, clusterBean).value().entrySet().stream()
                             .map(
                                 idAndCost ->
                                     Map.entry(
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 94b2691b..0447d0e9 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
@@ -30,6 +30,7 @@ import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 import org.apache.kafka.common.Cluster;
+import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.admin.NodeInfo;
 import org.astraea.app.admin.ReplicaInfo;
@@ -81,7 +82,7 @@ public class SmoothWeightRoundRobinDispatcher extends Periodic<Map<Integer, Doub
                   .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().current()));
 
           var compoundScore =
-              neutralIntegratedCost.brokerCost(ClusterInfo.of(clusterInfo, beans)).value();
+              neutralIntegratedCost.brokerCost(clusterInfo, ClusterBean.of(beans)).value();
 
           if (smoothWeightRoundRobinCal == null) {
             smoothWeightRoundRobinCal = new SmoothWeightRoundRobin(compoundScore);


Test patch:
diff --git a/app/src/test/java/org/astraea/app/admin/ClusterBeanTest.java b/app/src/test/java/org/astraea/app/admin/ClusterBeanTest.java
index 914993ae..0dabb63b 100644
--- a/app/src/test/java/org/astraea/app/admin/ClusterBeanTest.java
+++ b/app/src/test/java/org/astraea/app/admin/ClusterBeanTest.java
@@ -27,14 +27,6 @@ import org.mockito.Mockito;
 
 class ClusterBeanTest {
 
-  @Test
-  void testEmptyBeans() {
-    var clusterInfo = ClusterInfo.of(Mockito.mock(org.apache.kafka.common.Cluster.class));
-    Assertions.assertEquals(0, clusterInfo.clusterBean().all().size());
-    Assertions.assertEquals(0, clusterInfo.clusterBean().mapByPartition().size());
-    Assertions.assertEquals(0, clusterInfo.clusterBean().mapByReplica().size());
-  }
-
   @Test
   void testBeans() {
     var clusterInfo = Mockito.mock(ClusterInfo.class);
@@ -87,37 +79,30 @@ class ClusterBeanTest {
                 "type",
                 "ReplicaManager"),
             Map.of("Value", 300));
-    Mockito.when(clusterInfo.clusterBean())
-        .thenReturn(
-            ClusterBean.of(
-                Map.of(
-                    1,
-                    List.of(HasValue.of(testBeanObjectWithPartition1)),
-                    2,
-                    List.of(
-                        HasValue.of(testBeanObjectWithoutPartition),
-                        HasValue.of(testBeanObjectWithPartition2),
-                        HasValue.of(testBeanObjectWithPartition3)))));
+    var clusterBean =
+        ClusterBean.of(
+            Map.of(
+                1,
+                List.of(HasValue.of(testBeanObjectWithPartition1)),
+                2,
+                List.of(
+                    HasValue.of(testBeanObjectWithoutPartition),
+                    HasValue.of(testBeanObjectWithPartition2),
+                    HasValue.of(testBeanObjectWithPartition3))));
     // test all
-    clusterInfo.clusterBean().mapByPartition();
-    Assertions.assertEquals(2, clusterInfo.clusterBean().all().size());
-    Assertions.assertEquals(1, clusterInfo.clusterBean().all().get(1).size());
-    Assertions.assertEquals(3, clusterInfo.clusterBean().all().get(2).size());
+    clusterBean.mapByPartition();
+    Assertions.assertEquals(2, clusterBean.all().size());
+    Assertions.assertEquals(1, clusterBean.all().get(1).size());
+    Assertions.assertEquals(3, clusterBean.all().get(2).size());
 
     // test get beanObject by partition
     // when call beanObjectByPartition() will return a map and the key is TopicPartition it's will
     // ignore replicas and get the metrics of first replicas
     Assertions.assertEquals(
-        2,
-        clusterInfo.clusterBean().mapByPartition().get(TopicPartition.of("testBeans", "0")).size());
+        2, clusterBean.mapByPartition().get(TopicPartition.of("testBeans", "0")).size());
     // test get beanObject by replica
-    Assertions.assertEquals(2, clusterInfo.clusterBean().mapByReplica().size());
+    Assertions.assertEquals(2, clusterBean.mapByReplica().size());
     Assertions.assertEquals(
-        2,
-        clusterInfo
-            .clusterBean()
-            .mapByReplica()
-            .get(TopicPartitionReplica.of("testBeans", 0, 2))
-            .size());
+        2, clusterBean.mapByReplica().get(TopicPartitionReplica.of("testBeans", 0, 2)).size());
   }
 }
diff --git a/app/src/test/java/org/astraea/app/cost/BrokerInputCostTest.java b/app/src/test/java/org/astraea/app/cost/BrokerInputCostTest.java
index 10c7ed5b..5eef5bc9 100644
--- a/app/src/test/java/org/astraea/app/cost/BrokerInputCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/BrokerInputCostTest.java
@@ -35,16 +35,21 @@ import org.mockito.Mockito;
 public class BrokerInputCostTest extends RequireBrokerCluster {
   @Test
   void testTScoreCost() {
-    ClusterInfo clusterInfo = exampleClusterInfo(50000L, 20000L, 5000L);
-
     var brokerInputCost = new BrokerInputCost();
-    var scores = brokerInputCost.brokerCost(clusterInfo).normalize(Normalizer.TScore()).value();
+    var scores =
+        brokerInputCost
+            .brokerCost(ClusterInfo.EMPTY, clusterBean(50000L, 20000L, 5000L))
+            .normalize(Normalizer.TScore())
+            .value();
     Assertions.assertEquals(0.63, scores.get(1));
     Assertions.assertEquals(0.47, scores.get(2));
     Assertions.assertEquals(0.39, scores.get(3));
 
-    ClusterInfo clusterInfo2 = exampleClusterInfo(55555L, 25352L, 25000L);
-    scores = brokerInputCost.brokerCost(clusterInfo2).normalize(Normalizer.TScore()).value();
+    scores =
+        brokerInputCost
+            .brokerCost(ClusterInfo.EMPTY, clusterBean(55555L, 25352L, 25000L))
+            .normalize(Normalizer.TScore())
+            .value();
     Assertions.assertEquals(0.64, scores.get(1));
     Assertions.assertEquals(0.43, scores.get(2));
     Assertions.assertEquals(0.43, scores.get(3));
@@ -52,10 +57,9 @@ public class BrokerInputCostTest extends RequireBrokerCluster {
 
   @Test
   void testNoNormalize() {
-    ClusterInfo clusterInfo = exampleClusterInfo(10000L, 20000L, 5000L);
-
     var brokerInputCost = new BrokerInputCost();
-    var scores = brokerInputCost.brokerCost(clusterInfo).value();
+    var scores =
+        brokerInputCost.brokerCost(ClusterInfo.EMPTY, clusterBean(10000L, 20000L, 5000L)).value();
     Assertions.assertEquals(10000.0, scores.get(1));
     Assertions.assertEquals(20000.0, scores.get(2));
     Assertions.assertEquals(5000.0, scores.get(3));
@@ -90,7 +94,7 @@ public class BrokerInputCostTest extends RequireBrokerCluster {
     }
   }
 
-  private ClusterInfo exampleClusterInfo(long in1, long in2, long in3) {
+  private ClusterBean clusterBean(long in1, long in2, long in3) {
     var BytesInPerSec1 = mockResult(KafkaMetrics.BrokerTopic.BytesInPerSec.metricName(), in1);
     var BytesInPerSec2 = mockResult(KafkaMetrics.BrokerTopic.BytesInPerSec.metricName(), in2);
     var BytesInPerSec3 = mockResult(KafkaMetrics.BrokerTopic.BytesInPerSec.metricName(), in3);
@@ -98,12 +102,7 @@ public class BrokerInputCostTest extends RequireBrokerCluster {
     Collection<HasBeanObject> broker1 = List.of(BytesInPerSec1);
     Collection<HasBeanObject> broker2 = List.of(BytesInPerSec2);
     Collection<HasBeanObject> broker3 = List.of(BytesInPerSec3);
-    return new FakeClusterInfo() {
-      @Override
-      public ClusterBean clusterBean() {
-        return ClusterBean.of(Map.of(1, broker1, 2, broker2, 3, broker3));
-      }
-    };
+    return ClusterBean.of(Map.of(1, broker1, 2, broker2, 3, broker3));
   }
 
   private BrokerTopicMetricsResult mockResult(String name, long count) {
diff --git a/app/src/test/java/org/astraea/app/cost/BrokerOutPutCostTest.java b/app/src/test/java/org/astraea/app/cost/BrokerOutPutCostTest.java
index 8159df3f..3c6c2822 100644
--- a/app/src/test/java/org/astraea/app/cost/BrokerOutPutCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/BrokerOutPutCostTest.java
@@ -35,16 +35,21 @@ import org.mockito.Mockito;
 public class BrokerOutPutCostTest extends RequireBrokerCluster {
   @Test
   void testCost() {
-    ClusterInfo clusterInfo = exampleClusterInfo(10000L, 20000L, 5000L);
-
     var brokerOutputCost = new BrokerOutputCost();
-    var scores = brokerOutputCost.brokerCost(clusterInfo).normalize(Normalizer.TScore()).value();
+    var scores =
+        brokerOutputCost
+            .brokerCost(ClusterInfo.EMPTY, clusterBean(10000L, 20000L, 5000L))
+            .normalize(Normalizer.TScore())
+            .value();
     Assertions.assertEquals(0.47, scores.get(1));
     Assertions.assertEquals(0.63, scores.get(2));
     Assertions.assertEquals(0.39, scores.get(3));
 
-    ClusterInfo clusterInfo2 = exampleClusterInfo(55555L, 25352L, 25000L);
-    scores = brokerOutputCost.brokerCost(clusterInfo2).normalize(Normalizer.TScore()).value();
+    scores =
+        brokerOutputCost
+            .brokerCost(ClusterInfo.EMPTY, clusterBean(55555L, 25352L, 25000L))
+            .normalize(Normalizer.TScore())
+            .value();
     Assertions.assertEquals(0.64, scores.get(1));
     Assertions.assertEquals(0.43, scores.get(2));
     Assertions.assertEquals(0.43, scores.get(3));
@@ -79,7 +84,7 @@ public class BrokerOutPutCostTest extends RequireBrokerCluster {
     }
   }
 
-  private ClusterInfo exampleClusterInfo(long out1, long out2, long out3) {
+  private ClusterBean clusterBean(long out1, long out2, long out3) {
     var BytesInPerSec1 = mockResult(KafkaMetrics.BrokerTopic.BytesOutPerSec.metricName(), out1);
     var BytesInPerSec2 = mockResult(KafkaMetrics.BrokerTopic.BytesOutPerSec.metricName(), out2);
     var BytesInPerSec3 = mockResult(KafkaMetrics.BrokerTopic.BytesOutPerSec.metricName(), out3);
@@ -87,12 +92,7 @@ public class BrokerOutPutCostTest extends RequireBrokerCluster {
     Collection<HasBeanObject> broker1 = List.of(BytesInPerSec1);
     Collection<HasBeanObject> broker2 = List.of(BytesInPerSec2);
     Collection<HasBeanObject> broker3 = List.of(BytesInPerSec3);
-    return new FakeClusterInfo() {
-      @Override
-      public ClusterBean clusterBean() {
-        return ClusterBean.of(Map.of(1, broker1, 2, broker2, 3, broker3));
-      }
-    };
+    return ClusterBean.of(Map.of(1, broker1, 2, broker2, 3, broker3));
   }
 
   private BrokerTopicMetricsResult mockResult(String name, long count) {
diff --git a/app/src/test/java/org/astraea/app/cost/ClusterInfoProvider.java b/app/src/test/java/org/astraea/app/cost/ClusterInfoProvider.java
index a87a8e41..c689cf40 100644
--- a/app/src/test/java/org/astraea/app/cost/ClusterInfoProvider.java
+++ b/app/src/test/java/org/astraea/app/cost/ClusterInfoProvider.java
@@ -22,7 +22,6 @@ import java.util.Set;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
-import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.admin.NodeInfo;
 import org.astraea.app.admin.ReplicaInfo;
@@ -126,11 +125,6 @@ public class ClusterInfoProvider {
       public List<ReplicaInfo> replicas(String topic) {
         return replicas.get(topic);
       }
-
-      @Override
-      public ClusterBean clusterBean() {
-        throw new UnsupportedOperationException();
-      }
     };
   }
 }
diff --git a/app/src/test/java/org/astraea/app/cost/ClusterInfoTest.java b/app/src/test/java/org/astraea/app/cost/ClusterInfoTest.java
index df94f547..4e325291 100644
--- a/app/src/test/java/org/astraea/app/cost/ClusterInfoTest.java
+++ b/app/src/test/java/org/astraea/app/cost/ClusterInfoTest.java
@@ -17,14 +17,11 @@
 package org.astraea.app.cost;
 
 import java.util.List;
-import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.Set;
 import org.apache.kafka.common.Cluster;
-import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.admin.NodeInfo;
-import org.astraea.app.metrics.HasBeanObject;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
@@ -69,17 +66,4 @@ public class ClusterInfoTest {
     Assertions.assertThrows(
         UnsupportedOperationException.class, () -> clusterInfo.dataDirectories(0));
   }
-
-  @Test
-  void testMergeWithBeans() {
-    var bean0 = Mockito.mock(HasBeanObject.class);
-    var bean1 = Mockito.mock(HasBeanObject.class);
-    var clusterBean = ClusterBean.of(Map.of(10, List.of(bean0)));
-    var mergedBeans = Map.of(10, List.of(bean1));
-    var origin = Mockito.mock(ClusterInfo.class);
-    Mockito.when(origin.clusterBean()).thenReturn(clusterBean);
-    var newClusterInfo = ClusterInfo.of(origin, mergedBeans);
-    Assertions.assertEquals(1, newClusterInfo.clusterBean().all().size());
-    Assertions.assertEquals(2, newClusterInfo.clusterBean().all().get(10).size());
-  }
 }
diff --git a/app/src/test/java/org/astraea/app/cost/CpuCostTest.java b/app/src/test/java/org/astraea/app/cost/CpuCostTest.java
index ffebc1bf..079b27c9 100644
--- a/app/src/test/java/org/astraea/app/cost/CpuCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/CpuCostTest.java
@@ -43,29 +43,21 @@ public class CpuCostTest extends RequireBrokerCluster {
     Collection<HasBeanObject> broker1 = List.of(cpuUsage1);
     Collection<HasBeanObject> broker2 = List.of(cpuUsage2);
     Collection<HasBeanObject> broker3 = List.of(cpuUsage3);
-    ClusterInfo clusterInfo =
-        new FakeClusterInfo() {
-          @Override
-          public ClusterBean clusterBean() {
-            return ClusterBean.of(Map.of(1, broker1, 2, broker2, 3, broker3));
-          }
 
-          @Override
-          public Set<String> topics() {
-            return Set.of("t");
-          }
-
-          @Override
-          public List<ReplicaInfo> availableReplicas(String topic) {
-            return List.of(
+    ClusterInfo clusterInfo = Mockito.mock(ClusterInfo.class);
+    Mockito.when(clusterInfo.topics()).thenReturn(Set.of("t"));
+    Mockito.when(clusterInfo.availableReplicas(Mockito.anyString()))
+        .thenReturn(
+            List.of(
                 ReplicaInfo.of("t", 0, NodeInfo.of(1, "host1", 9092), true, true, false),
                 ReplicaInfo.of("t", 0, NodeInfo.of(2, "host2", 9092), false, true, false),
-                ReplicaInfo.of("t", 0, NodeInfo.of(3, "host3", 9092), false, true, false));
-          }
-        };
-
+                ReplicaInfo.of("t", 0, NodeInfo.of(3, "host3", 9092), false, true, false)));
     var cpuCost = new CpuCost();
-    var scores = cpuCost.brokerCost(clusterInfo).normalize(Normalizer.TScore()).value();
+    var scores =
+        cpuCost
+            .brokerCost(clusterInfo, ClusterBean.of(Map.of(1, broker1, 2, broker2, 3, broker3)))
+            .normalize(Normalizer.TScore())
+            .value();
     Assertions.assertEquals(0.39, scores.get(1));
     Assertions.assertEquals(0.63, scores.get(2));
     Assertions.assertEquals(0.48, scores.get(3));
@@ -78,27 +70,20 @@ public class CpuCostTest extends RequireBrokerCluster {
     Collection<HasBeanObject> broker12 = List.of(cpuUsage1);
     Collection<HasBeanObject> broker22 = List.of(cpuUsage2);
     Collection<HasBeanObject> broker32 = List.of(cpuUsage3);
-    ClusterInfo clusterInfo2 =
-        new FakeClusterInfo() {
-          @Override
-          public ClusterBean clusterBean() {
-            return ClusterBean.of(Map.of(1, broker12, 2, broker22, 3, broker32));
-          }
-
-          @Override
-          public Set<String> topics() {
-            return Set.of("t");
-          }
 
-          @Override
-          public List<ReplicaInfo> availableReplicas(String topic) {
-            return List.of(
+    ClusterInfo clusterInfo2 = Mockito.mock(ClusterInfo.class);
+    Mockito.when(clusterInfo2.topics()).thenReturn(Set.of("t"));
+    Mockito.when(clusterInfo2.availableReplicas(Mockito.anyString()))
+        .thenReturn(
+            List.of(
                 ReplicaInfo.of("t", 0, NodeInfo.of(1, "host1", 9092), true, true, false),
                 ReplicaInfo.of("t", 0, NodeInfo.of(2, "host2", 9092), false, true, false),
-                ReplicaInfo.of("t", 0, NodeInfo.of(3, "host3", 9092), false, true, false));
-          }
-        };
-    scores = cpuCost.brokerCost(clusterInfo2).normalize(Normalizer.TScore()).value();
+                ReplicaInfo.of("t", 0, NodeInfo.of(3, "host3", 9092), false, true, false)));
+    scores =
+        cpuCost
+            .brokerCost(clusterInfo2, ClusterBean.of(Map.of(1, broker12, 2, broker22, 3, broker32)))
+            .normalize(Normalizer.TScore())
+            .value();
     Assertions.assertEquals(0.52, scores.get(1));
     Assertions.assertEquals(0.61, scores.get(2));
     Assertions.assertEquals(0.37, scores.get(3));
diff --git a/app/src/test/java/org/astraea/app/cost/FakeClusterInfo.java b/app/src/test/java/org/astraea/app/cost/FakeClusterInfo.java
deleted file mode 100644
index 649652bd..00000000
--- a/app/src/test/java/org/astraea/app/cost/FakeClusterInfo.java
+++ /dev/null
@@ -1,62 +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.List;
-import java.util.Map;
-import java.util.Set;
-import org.astraea.app.admin.ClusterBean;
-import org.astraea.app.admin.ClusterInfo;
-import org.astraea.app.admin.NodeInfo;
-import org.astraea.app.admin.ReplicaInfo;
-
-public class FakeClusterInfo implements ClusterInfo {
-  @Override
-  public List<NodeInfo> nodes() {
-    return List.of();
-  }
-
-  @Override
-  public Set<String> dataDirectories(int brokerId) {
-    return Set.of();
-  }
-
-  @Override
-  public List<ReplicaInfo> availableReplicaLeaders(String topic) {
-    return List.of();
-  }
-
-  @Override
-  public List<ReplicaInfo> availableReplicas(String topic) {
-    return List.of();
-  }
-
-  @Override
-  public Set<String> topics() {
-    return Set.of();
-  }
-
-  @Override
-  public List<ReplicaInfo> replicas(String topic) {
-    return List.of();
-  }
-
-  @Override
-  public ClusterBean clusterBean() {
-    return ClusterBean.of(Map.of());
-  }
-}
diff --git a/app/src/test/java/org/astraea/app/cost/LoadCostTest.java b/app/src/test/java/org/astraea/app/cost/LoadCostTest.java
index 4e8c3872..113beea4 100644
--- a/app/src/test/java/org/astraea/app/cost/LoadCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/LoadCostTest.java
@@ -20,7 +20,6 @@ import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import org.astraea.app.admin.ClusterBean;
-import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.KafkaMetrics;
 import org.astraea.app.metrics.broker.BrokerTopicMetricsResult;
@@ -33,7 +32,7 @@ public class LoadCostTest {
   @Test
   void testComputeLoad() {
     var loadCostFunction = new LoadCost();
-    var allBeans = exampleClusterInfo().clusterBean().all();
+    var allBeans = clusterBean().all();
     var load = loadCostFunction.computeLoad(allBeans);
 
     Assertions.assertEquals(2, load.get(1));
@@ -48,7 +47,7 @@ public class LoadCostTest {
     Assertions.assertEquals(2, load.get(3));
   }
 
-  private ClusterInfo exampleClusterInfo() {
+  private ClusterBean clusterBean() {
     var BytesInPerSec1 = mockResult(KafkaMetrics.BrokerTopic.BytesInPerSec.metricName(), 50000L);
     var BytesInPerSec2 = mockResult(KafkaMetrics.BrokerTopic.BytesInPerSec.metricName(), 100000L);
     var BytesInPerSec3 = mockResult(KafkaMetrics.BrokerTopic.BytesInPerSec.metricName(), 200000L);
@@ -59,12 +58,7 @@ public class LoadCostTest {
     Collection<HasBeanObject> broker1 = List.of(BytesInPerSec1, BytesOutPerSec1);
     Collection<HasBeanObject> broker2 = List.of(BytesInPerSec2, BytesOutPerSec2);
     Collection<HasBeanObject> broker3 = List.of(BytesInPerSec3, BytesOutPerSec3);
-    return new FakeClusterInfo() {
-      @Override
-      public ClusterBean clusterBean() {
-        return ClusterBean.of(Map.of(1, broker1, 2, broker2, 3, broker3));
-      }
-    };
+    return ClusterBean.of(Map.of(1, broker1, 2, broker2, 3, broker3));
   }
 
   private BrokerTopicMetricsResult mockResult(String name, long count) {
diff --git a/app/src/test/java/org/astraea/app/cost/MemoryCostTest.java b/app/src/test/java/org/astraea/app/cost/MemoryCostTest.java
index de67451e..09b31a51 100644
--- a/app/src/test/java/org/astraea/app/cost/MemoryCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/MemoryCostTest.java
@@ -45,29 +45,20 @@ public class MemoryCostTest extends RequireBrokerCluster {
     Collection<HasBeanObject> broker1 = List.of(jvmMemory1);
     Collection<HasBeanObject> broker2 = List.of(jvmMemory2);
     Collection<HasBeanObject> broker3 = List.of(jvmMemory3);
-    ClusterInfo clusterInfo =
-        new FakeClusterInfo() {
-          @Override
-          public ClusterBean clusterBean() {
-            return ClusterBean.of(Map.of(1, broker1, 2, broker2, 3, broker3));
-          }
-
-          @Override
-          public Set<String> topics() {
-            return Set.of("t");
-          }
-
-          @Override
-          public List<ReplicaInfo> availableReplicas(String topic) {
-            return List.of(
+    ClusterInfo clusterInfo = Mockito.mock(ClusterInfo.class);
+    Mockito.when(clusterInfo.topics()).thenReturn(Set.of("t"));
+    Mockito.when(clusterInfo.availableReplicas(Mockito.anyString()))
+        .thenReturn(
+            List.of(
                 ReplicaInfo.of("t", 0, NodeInfo.of(1, "host1", 9092), true, false, false),
                 ReplicaInfo.of("t", 0, NodeInfo.of(2, "host2", 9092), false, true, false),
-                ReplicaInfo.of("t", 0, NodeInfo.of(3, "host3", 9092), false, true, false));
-          }
-        };
-
+                ReplicaInfo.of("t", 0, NodeInfo.of(3, "host3", 9092), false, true, false)));
     var memoryCost = new MemoryCost();
-    var scores = memoryCost.brokerCost(clusterInfo).normalize(Normalizer.TScore()).value();
+    var scores =
+        memoryCost
+            .brokerCost(clusterInfo, ClusterBean.of(Map.of(1, broker1, 2, broker2, 3, broker3)))
+            .normalize(Normalizer.TScore())
+            .value();
     Assertions.assertEquals(0.39, scores.get(1));
     Assertions.assertEquals(0.63, scores.get(2));
     Assertions.assertEquals(0.47, scores.get(3));
@@ -80,27 +71,20 @@ public class MemoryCostTest extends RequireBrokerCluster {
     Collection<HasBeanObject> broker12 = List.of(jvmMemory1);
     Collection<HasBeanObject> broker22 = List.of(jvmMemory2);
     Collection<HasBeanObject> broker32 = List.of(jvmMemory3);
-    ClusterInfo clusterInfo2 =
-        new FakeClusterInfo() {
-          @Override
-          public ClusterBean clusterBean() {
-            return ClusterBean.of(Map.of(1, broker12, 2, broker22, 3, broker32));
-          }
-
-          @Override
-          public Set<String> topics() {
-            return Set.of("t");
-          }
 
-          @Override
-          public List<ReplicaInfo> availableReplicas(String topic) {
-            return List.of(
+    ClusterInfo clusterInfo2 = Mockito.mock(ClusterInfo.class);
+    Mockito.when(clusterInfo2.topics()).thenReturn(Set.of("t"));
+    Mockito.when(clusterInfo2.availableReplicas(Mockito.anyString()))
+        .thenReturn(
+            List.of(
                 ReplicaInfo.of("t", 0, NodeInfo.of(1, "host1", 9092), true, true, false),
                 ReplicaInfo.of("t", 0, NodeInfo.of(2, "host2", 9092), false, true, false),
-                ReplicaInfo.of("t", 0, NodeInfo.of(3, "host3", 9092), false, true, false));
-          }
-        };
-    scores = memoryCost.brokerCost(clusterInfo2).normalize(Normalizer.TScore()).value();
+                ReplicaInfo.of("t", 0, NodeInfo.of(3, "host3", 9092), false, true, false)));
+    scores =
+        memoryCost
+            .brokerCost(clusterInfo2, ClusterBean.of(Map.of(1, broker12, 2, broker22, 3, broker32)))
+            .normalize(Normalizer.TScore())
+            .value();
     Assertions.assertEquals(0.36, scores.get(1));
     Assertions.assertEquals(0.58, scores.get(2));
     Assertions.assertEquals(0.56, scores.get(3));
diff --git a/app/src/test/java/org/astraea/app/cost/NeutralIntegratedCostTest.java b/app/src/test/java/org/astraea/app/cost/NeutralIntegratedCostTest.java
index 629f6c34..f3bb4762 100644
--- a/app/src/test/java/org/astraea/app/cost/NeutralIntegratedCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/NeutralIntegratedCostTest.java
@@ -19,6 +19,7 @@ package org.astraea.app.cost;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
@@ -49,13 +50,14 @@ public class NeutralIntegratedCostTest {
     var outputBrokerCost = getBrokerCost(new HashMap<>(Map.of(0, 500.0, 1, 1000.0, 2, 1500.0)));
     var cpuBrokerCost = getBrokerCost(new HashMap<>(Map.of(0, 0.05, 1, 0.1, 2, 0.15)));
     var memoryBrokerCost = getBrokerCost(new HashMap<>(Map.of(0, 0.1, 1, 0.3, 2, 0.5)));
-    Mockito.when(input.brokerCost(clusterInfo)).thenReturn(inputBrokerCost);
-    Mockito.when(output.brokerCost(clusterInfo)).thenReturn(outputBrokerCost);
-    Mockito.when(memory.brokerCost(clusterInfo)).thenReturn(memoryBrokerCost);
-    Mockito.when(cpu.brokerCost(clusterInfo)).thenReturn(cpuBrokerCost);
+    Mockito.when(input.brokerCost(clusterInfo, ClusterBean.EMPTY)).thenReturn(inputBrokerCost);
+    Mockito.when(output.brokerCost(clusterInfo, ClusterBean.EMPTY)).thenReturn(outputBrokerCost);
+    Mockito.when(memory.brokerCost(clusterInfo, ClusterBean.EMPTY)).thenReturn(memoryBrokerCost);
+    Mockito.when(cpu.brokerCost(clusterInfo, ClusterBean.EMPTY)).thenReturn(cpuBrokerCost);
     List<HasBrokerCost> metricsCost = List.of(input, output, cpu, memory);
     metricsCost.forEach(
-        hasBrokerCost -> neutralIntegratedCost.setBrokerMetrics(hasBrokerCost, clusterInfo));
+        hasBrokerCost ->
+            neutralIntegratedCost.setBrokerMetrics(hasBrokerCost, clusterInfo, ClusterBean.EMPTY));
 
     Assertions.assertEquals(neutralIntegratedCost.brokersMetric.get(0).inputScore, 50.0);
     Assertions.assertEquals(neutralIntegratedCost.brokersMetric.get(1).inputScore, 100.0);
diff --git a/app/src/test/java/org/astraea/app/cost/NodeLatencyCostTest.java b/app/src/test/java/org/astraea/app/cost/NodeLatencyCostTest.java
index c491e629..cd1de97d 100644
--- a/app/src/test/java/org/astraea/app/cost/NodeLatencyCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/NodeLatencyCostTest.java
@@ -56,10 +56,8 @@ public class NodeLatencyCostTest extends RequireBrokerCluster {
                       .map(b -> (HasBeanObject) b)
                       .collect(Collectors.toUnmodifiableList())));
       var clusterInfo = Mockito.mock(ClusterInfo.class);
-      Mockito.when(clusterInfo.clusterBean()).thenReturn(clusterBean);
-
       var function = new NodeLatencyCost();
-      var cost = function.brokerCost(clusterInfo);
+      var cost = function.brokerCost(clusterInfo, clusterBean);
       Assertions.assertEquals(1, cost.value().size());
       Assertions.assertNotNull(cost.value().get(brokerId));
     }
diff --git a/app/src/test/java/org/astraea/app/cost/NodeThroughputCostTest.java b/app/src/test/java/org/astraea/app/cost/NodeThroughputCostTest.java
index c7878fb8..39fb0f73 100644
--- a/app/src/test/java/org/astraea/app/cost/NodeThroughputCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/NodeThroughputCostTest.java
@@ -40,9 +40,7 @@ public class NodeThroughputCostTest {
     Mockito.when(bean1.outgoingByteRate()).thenReturn(3D);
     var clusterBean = ClusterBean.of(Map.of(0, List.of(bean0), 1, List.of(bean1)));
     var clusterInfo = Mockito.mock(ClusterInfo.class);
-    Mockito.when(clusterInfo.clusterBean()).thenReturn(clusterBean);
-
-    var cost = throughputCost.brokerCost(clusterInfo);
+    var cost = throughputCost.brokerCost(clusterInfo, clusterBean);
     Assertions.assertEquals(30D, cost.value().get(0));
     Assertions.assertEquals(5D, cost.value().get(1));
   }
diff --git a/app/src/test/java/org/astraea/app/cost/ReplicaDiskInCostTest.java b/app/src/test/java/org/astraea/app/cost/ReplicaDiskInCostTest.java
index 38237588..3fe2db91 100644
--- a/app/src/test/java/org/astraea/app/cost/ReplicaDiskInCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/ReplicaDiskInCostTest.java
@@ -33,8 +33,40 @@ import org.astraea.app.metrics.jmx.BeanObject;
 import org.astraea.app.service.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
 
 class ReplicaDiskInCostTest extends RequireBrokerCluster {
+  private static final HasValue OLD_TP1_0 =
+      fakeBeanObject(
+          "Log", KafkaMetrics.TopicPartition.Size.metricName(), "test-1", "0", 1000, 1000L);
+  private static final HasValue NEW_TP1_0 =
+      fakeBeanObject(
+          "Log", KafkaMetrics.TopicPartition.Size.metricName(), "test-1", "0", 50000000, 5000L);
+  private static final HasValue OLD_TP1_1 =
+      fakeBeanObject(
+          "Log", KafkaMetrics.TopicPartition.Size.metricName(), "test-1", "1", 500, 1000L);
+  private static final HasValue NEW_TP1_1 =
+      fakeBeanObject(
+          "Log", KafkaMetrics.TopicPartition.Size.metricName(), "test-1", "1", 100000000, 5000L);
+  private static final HasValue OLD_TP2_0 =
+      fakeBeanObject(
+          "Log", KafkaMetrics.TopicPartition.Size.metricName(), "test-2", "0", 200, 1000L);
+  private static final HasValue NEW_TP2_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
+   */
+  private static final Collection<HasBeanObject> broker1 =
+      List.of(OLD_TP1_0, NEW_TP1_0, OLD_TP2_0, NEW_TP2_0);
+  private static final Collection<HasBeanObject> broker2 =
+      List.of(OLD_TP1_0, NEW_TP1_0, OLD_TP1_1, NEW_TP1_1);
+  private static final Collection<HasBeanObject> broker3 =
+      List.of(OLD_TP1_1, NEW_TP1_1, OLD_TP2_0, NEW_TP2_0);
 
   @Test
   void testPartitionCost() {
@@ -43,9 +75,9 @@ class ReplicaDiskInCostTest extends RequireBrokerCluster {
     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);
+    var broker1ReplicaLoad = loadCostFunction.partitionCost(clusterInfo(), clusterBean()).value(1);
+    var broker2ReplicaLoad = loadCostFunction.partitionCost(clusterInfo(), clusterBean()).value(2);
+    var broker3ReplicaLoad = loadCostFunction.partitionCost(clusterInfo(), clusterBean()).value(3);
     // broker1
     Assertions.assertEquals(
         0.23841381072998047, broker1ReplicaLoad.get(new TopicPartition("test-1", 0)));
@@ -70,73 +102,39 @@ class ReplicaDiskInCostTest extends RequireBrokerCluster {
     properties.put(2, 50);
     properties.put(3, 50);
     var loadCostFunction = new ReplicaDiskInCost(properties);
-    var brokerLoad = loadCostFunction.brokerCost(exampleClusterInfo()).value();
+    var brokerLoad = loadCostFunction.brokerCost(clusterInfo(), clusterBean()).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");
-      }
+  private ClusterInfo clusterInfo() {
 
-      @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));
-      }
+    ClusterInfo clusterInfo = Mockito.mock(ClusterInfo.class);
+    Mockito.when(clusterInfo.nodes())
+        .thenReturn(
+            List.of(NodeInfo.of(1, "", -1), NodeInfo.of(2, "", -1), NodeInfo.of(3, "", -1)));
+    Mockito.when(clusterInfo.topics()).thenReturn(Set.of("test-1", "test-2"));
+    Mockito.when(clusterInfo.replicas(Mockito.anyString()))
+        .thenAnswer(
+            topic ->
+                topic.getArgument(0).equals("test-1")
+                    ? 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))
+                    : 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)));
+    return clusterInfo;
+  }
 
-      @Override
-      public ClusterBean clusterBean() {
-        return ClusterBean.of(Map.of(1, broker1, 2, broker2, 3, broker3));
-      }
-    };
+  private static ClusterBean clusterBean() {
+    return ClusterBean.of(Map.of(1, broker1, 2, broker2, 3, broker3));
   }
 
-  private HasValue fakeBeanObject(
+  private static HasValue fakeBeanObject(
       String type, String name, String topic, String partition, long size, long time) {
     BeanObject beanObject =
         new BeanObject(
diff --git a/app/src/test/java/org/astraea/app/cost/ReplicaLeaderCostTest.java b/app/src/test/java/org/astraea/app/cost/ReplicaLeaderCostTest.java
index 8b45abf2..d5a803d1 100644
--- a/app/src/test/java/org/astraea/app/cost/ReplicaLeaderCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/ReplicaLeaderCostTest.java
@@ -45,7 +45,7 @@ public class ReplicaLeaderCostTest {
     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);
+    var cost = function.leaderCount(clusterInfo, ClusterBean.EMPTY);
     Assertions.assertTrue(cost.containsKey(10));
     Assertions.assertTrue(cost.containsKey(11));
     Assertions.assertEquals(2, cost.size());
@@ -64,9 +64,7 @@ public class ReplicaLeaderCostTest {
     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);
+    var load = costFunction.brokerCost(ClusterInfo.EMPTY, clusterBean);
 
     Assertions.assertEquals(3, load.value().size());
     Assertions.assertEquals(3.0, load.value().get(1));
diff --git a/app/src/test/java/org/astraea/app/cost/ReplicaSizeCostTest.java b/app/src/test/java/org/astraea/app/cost/ReplicaSizeCostTest.java
index 368be8f3..9072258f 100644
--- a/app/src/test/java/org/astraea/app/cost/ReplicaSizeCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/ReplicaSizeCostTest.java
@@ -40,8 +40,26 @@ import org.astraea.app.producer.Producer;
 import org.astraea.app.service.RequireSingleBrokerCluster;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
 
 class ReplicaSizeCostTest extends RequireSingleBrokerCluster {
+  private static final HasValue SIZE_TP1_0 =
+      fakeBeanObject(
+          "Log", KafkaMetrics.TopicPartition.Size.metricName(), "test-1", "0", 891289600);
+  private static final HasValue SIZE_TP1_1 =
+      fakeBeanObject(
+          "Log", KafkaMetrics.TopicPartition.Size.metricName(), "test-1", "1", 471859200);
+  private static final HasValue SIZE_TP2_0 =
+      fakeBeanObject("Log", KafkaMetrics.TopicPartition.Size.metricName(), "test-2", "0", 0);
+  private static final HasValue SIZE_TP2_1 =
+      fakeBeanObject(
+          "Log", KafkaMetrics.TopicPartition.Size.metricName(), "test-2", "1", 367001600);
+  private static final Collection<HasBeanObject> BROKER_1 =
+      List.of(SIZE_TP1_0, SIZE_TP1_1, SIZE_TP2_1);
+  private static final Collection<HasBeanObject> BROKER_2 = List.of(SIZE_TP1_1, SIZE_TP2_0);
+  private static final Collection<HasBeanObject> BROKER_3 =
+      List.of(SIZE_TP2_1, SIZE_TP1_0, SIZE_TP2_0);
+
   @Test
   void testGetMetrics() throws ExecutionException, InterruptedException {
     var brokerDiskSize = Map.of(1, 1000, 2, 1000, 3, 1000);
@@ -82,82 +100,60 @@ class ReplicaSizeCostTest extends RequireSingleBrokerCluster {
   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);
+    var broker1ReplicaLoad = loadCostFunction.partitionCost(clusterInfo(), clusterBean()).value(1);
+    var BROKER_2ReplicaLoad = loadCostFunction.partitionCost(clusterInfo(), clusterBean()).value(2);
+    var BROKER_3ReplicaLoad = loadCostFunction.partitionCost(clusterInfo(), clusterBean()).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)));
+    // BROKER_2
+    Assertions.assertEquals(0.45, BROKER_2ReplicaLoad.get(new TopicPartition("test-1", 1)));
+    Assertions.assertEquals(0, BROKER_2ReplicaLoad.get(new TopicPartition("test-2", 0)));
+    // BROKER_3
+    Assertions.assertEquals(0.85, BROKER_3ReplicaLoad.get(new TopicPartition("test-1", 0)));
+    Assertions.assertEquals(0, BROKER_3ReplicaLoad.get(new TopicPartition("test-2", 0)));
+    Assertions.assertEquals(0.35, BROKER_3ReplicaLoad.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();
+    var brokerReplicaLoad = loadCostFunction.brokerCost(clusterInfo(), clusterBean()).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));
-      }
+  private static ClusterInfo clusterInfo() {
+    ClusterInfo clusterInfo = Mockito.mock(ClusterInfo.class);
+    Mockito.when(clusterInfo.nodes())
+        .thenReturn(
+            List.of(NodeInfo.of(1, "", -1), NodeInfo.of(2, "", -1), NodeInfo.of(3, "", -1)));
+    Mockito.when(clusterInfo.topics()).thenReturn(Set.of("test-1", "test-2"));
+    Mockito.when(clusterInfo.replicas(Mockito.anyString()))
+        .thenAnswer(
+            topic ->
+                topic.getArgument(0).equals("test-1")
+                    ? 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))
+                    : 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)));
+    return clusterInfo;
+  }
 
-      @Override
-      public ClusterBean clusterBean() {
-        return ClusterBean.of(Map.of(1, broker1, 2, broker2, 3, broker3));
-      }
-    };
+  private static ClusterBean clusterBean() {
+    return ClusterBean.of(Map.of(1, BROKER_1, 2, BROKER_2, 3, BROKER_3));
   }
 
-  private HasValue fakeBeanObject(
+  private static HasValue fakeBeanObject(
       String type, String name, String topic, String partition, long size) {
     BeanObject beanObject =
         new 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 3d43770f..d747a17e 100644
--- a/app/src/test/java/org/astraea/app/partitioner/StrictCostDispatcherTest.java
+++ b/app/src/test/java/org/astraea/app/partitioner/StrictCostDispatcherTest.java
@@ -171,7 +171,6 @@ public class StrictCostDispatcherTest {
     var clusterInfo = Mockito.mock(ClusterInfo.class);
     Mockito.when(clusterInfo.availableReplicaLeaders(Mockito.anyString()))
         .thenReturn(List.of(replicaInfo0, replicaInfo1));
-    Mockito.when(clusterInfo.clusterBean()).thenReturn(ClusterBean.of(Map.of()));
     try (var dispatcher = new StrictCostDispatcher()) {
       dispatcher.configure(
           Map.of(costFunction, 1D), Optional.empty(), Map.of(), Duration.ofSeconds(10));
@@ -210,7 +209,6 @@ public class StrictCostDispatcherTest {
     var clusterInfo = Mockito.mock(ClusterInfo.class);
     Mockito.when(clusterInfo.availableReplicaLeaders(Mockito.anyString()))
         .thenReturn(List.of(replicaInfo0, replicaInfo1, replicaInfo2));
-    Mockito.when(clusterInfo.clusterBean()).thenReturn(ClusterBean.of(Map.of()));
     // there is one local receiver by default
     Assertions.assertEquals(1, dispatcher.receivers.size());
     Assertions.assertEquals(-1, dispatcher.receivers.keySet().iterator().next());
@@ -254,7 +252,7 @@ public class StrictCostDispatcherTest {
     var costFunction =
         new HasBrokerCost() {
           @Override
-          public BrokerCost brokerCost(ClusterInfo clusterInfo) {
+          public BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
             return () -> Map.of(brokerId, 10D);
           }
         };
@@ -269,7 +267,6 @@ public class StrictCostDispatcherTest {
     var clusterInfo = Mockito.mock(ClusterInfo.class);
     Mockito.when(clusterInfo.availableReplicaLeaders(Mockito.anyString()))
         .thenReturn(List.of(replicaInfo0, replicaInfo1));
-    Mockito.when(clusterInfo.clusterBean()).thenReturn(ClusterBean.of(Map.of()));
 
     Assertions.assertEquals(
         partitionId, dispatcher.partition("topic", new byte[0], new byte[0], clusterInfo));
@@ -298,7 +295,6 @@ public class StrictCostDispatcherTest {
     Assertions.assertEquals(Duration.ofSeconds(2), dispatcher.roundRobinLease);
 
     var clusterInfo = Mockito.mock(ClusterInfo.class);
-    Mockito.when(clusterInfo.clusterBean()).thenReturn(ClusterBean.of(Map.of()));
     dispatcher.tryToUpdateRoundRobin(clusterInfo);
     var rr = dispatcher.roundRobin;
     Assertions.assertNotNull(rr);
diff --git a/app/src/test/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobinDispatchTest.java b/app/src/test/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobinDispatchTest.java
index bfc84288..05410734 100644
--- a/app/src/test/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobinDispatchTest.java
+++ b/app/src/test/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobinDispatchTest.java
@@ -32,6 +32,7 @@ import java.util.stream.IntStream;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.serialization.ByteArraySerializer;
 import org.astraea.app.admin.Admin;
+import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.admin.NodeInfo;
 import org.astraea.app.admin.ReplicaInfo;
 import org.astraea.app.common.Utils;
@@ -41,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.consumer.Header;
-import org.astraea.app.cost.FakeClusterInfo;
 import org.astraea.app.producer.Producer;
 import org.astraea.app.producer.Serializer;
 import org.astraea.app.service.RequireBrokerCluster;
@@ -299,13 +299,9 @@ public class SmoothWeightRoundRobinDispatchTest extends RequireBrokerCluster {
     var node3 = Mockito.mock(NodeInfo.class);
     Mockito.when(re3.nodeInfo()).thenReturn(node3);
     Mockito.when(node3.id()).thenReturn(3);
-    var testCluster =
-        new FakeClusterInfo() {
-          @Override
-          public List<ReplicaInfo> availableReplicaLeaders(String topic) {
-            return List.of(re1, re2, re3);
-          }
-        };
+    var testCluster = Mockito.mock(ClusterInfo.class);
+    Mockito.when(testCluster.availableReplicaLeaders(Mockito.anyString()))
+        .thenReturn(List.of(re1, re2, re3));
     Assertions.assertEquals(1, smoothWeight.getAndChoose(topic, testCluster));
     Assertions.assertEquals(2, smoothWeight.getAndChoose(topic, testCluster));
     Assertions.assertEquals(3, smoothWeight.getAndChoose(topic, testCluster));
diff --git a/app/src/test/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobinTest.java b/app/src/test/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobinTest.java
index 927068b9..6930bf2f 100644
--- a/app/src/test/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobinTest.java
+++ b/app/src/test/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobinTest.java
@@ -21,7 +21,6 @@ import java.util.Map;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.admin.NodeInfo;
 import org.astraea.app.admin.ReplicaInfo;
-import org.astraea.app.cost.FakeClusterInfo;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
@@ -72,11 +71,9 @@ public class SmoothWeightRoundRobinTest {
     var node3 = Mockito.mock(NodeInfo.class);
     Mockito.when(re3.nodeInfo()).thenReturn(node3);
     Mockito.when(node3.id()).thenReturn(3);
-    return new FakeClusterInfo() {
-      @Override
-      public List<ReplicaInfo> availableReplicaLeaders(String topic) {
-        return List.of(re1, re2, re3);
-      }
-    };
+    var clusterInfo = Mockito.mock(ClusterInfo.class);
+    Mockito.when(clusterInfo.availableReplicaLeaders(Mockito.anyString()))
+        .thenReturn(List.of(re1, re2, re3));
+    return clusterInfo;
   }
 }