Repository: skiptests/astraea
Pull request #645 (2022-08-28 10:52:13)

Add template to ClusterInfo
https://github.com/skiptests/astraea/pull/641#discussion_r956566515

為了避免上述狀況再度發生，這隻PR還是將template加上了去，這樣之後各個元件可以明確定義需要“哪種replica"



Patch:
diff --git a/app/src/main/java/org/astraea/app/admin/Admin.java b/app/src/main/java/org/astraea/app/admin/Admin.java
index 059e95a9..a57c9dfd 100644
--- a/app/src/main/java/org/astraea/app/admin/Admin.java
+++ b/app/src/main/java/org/astraea/app/admin/Admin.java
@@ -200,7 +200,7 @@ public interface Admin extends Closeable {
   Collection<Quota> quotas();
 
   /** @return a snapshot object of cluster state at the moment */
-  default ClusterInfo clusterInfo() {
+  default ClusterInfo<Replica> clusterInfo() {
     return clusterInfo(topicNames());
   }
 
@@ -208,7 +208,7 @@ public interface Admin extends Closeable {
    * @param topics query only this subset of topics
    * @return a snapshot object of cluster state at the moment
    */
-  ClusterInfo clusterInfo(Set<String> topics);
+  ClusterInfo<Replica> clusterInfo(Set<String> topics);
 
   /** @return all transaction ids */
   Set<String> transactionIds();
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 9a015675..1a2bbe11 100644
--- a/app/src/main/java/org/astraea/app/admin/Builder.java
+++ b/app/src/main/java/org/astraea/app/admin/Builder.java
@@ -446,7 +446,7 @@ public class Builder {
     }
 
     @Override
-    public ClusterInfo clusterInfo(Set<String> topics) {
+    public ClusterInfo<Replica> clusterInfo(Set<String> topics) {
       final var nodeInfo = this.nodes().stream().collect(Collectors.toUnmodifiableList());
 
       var replicas =
@@ -454,17 +454,16 @@ public class Builder {
               () ->
                   replicas(topics).values().stream()
                       .flatMap(Collection::stream)
-                      .map(r -> (ReplicaInfo) r)
                       .collect(Collectors.toUnmodifiableList()));
 
-      return new ClusterInfo() {
+      return new ClusterInfo<>() {
         @Override
         public List<NodeInfo> nodes() {
           return nodeInfo;
         }
 
         @Override
-        public List<ReplicaInfo> replicas() {
+        public List<Replica> replicas() {
           return replicas;
         }
       };
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 4289df4d..3fb513ca 100644
--- a/app/src/main/java/org/astraea/app/admin/ClusterInfo.java
+++ b/app/src/main/java/org/astraea/app/admin/ClusterInfo.java
@@ -23,9 +23,9 @@ import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
 
-public interface ClusterInfo {
-  ClusterInfo EMPTY =
-      new ClusterInfo() {
+public interface ClusterInfo<T extends ReplicaInfo> {
+  ClusterInfo<ReplicaInfo> EMPTY =
+      new ClusterInfo<>() {
 
         @Override
         public List<NodeInfo> nodes() {
@@ -38,6 +38,11 @@ public interface ClusterInfo {
         }
       };
 
+  @SuppressWarnings("unchecked")
+  static <T extends ReplicaInfo> ClusterInfo<T> empty() {
+    return (ClusterInfo<T>) EMPTY;
+  }
+
   /**
    * convert the kafka Cluster to our ClusterInfo. Noted: this method is used by {@link
    * org.astraea.app.cost.HasBrokerCost} normally, so all data structure are converted immediately
@@ -45,7 +50,7 @@ public interface ClusterInfo {
    * @param cluster kafka ClusterInfo
    * @return ClusterInfo
    */
-  static ClusterInfo of(org.apache.kafka.common.Cluster cluster) {
+  static ClusterInfo<ReplicaInfo> of(org.apache.kafka.common.Cluster cluster) {
     return of(
         cluster.nodes().stream().map(NodeInfo::of).collect(Collectors.toUnmodifiableList()),
         cluster.topics().stream()
@@ -54,7 +59,7 @@ public interface ClusterInfo {
             .collect(Collectors.toUnmodifiableList()));
   }
 
-  static ClusterInfo of(List<NodeInfo> nodes, List<ReplicaInfo> replicas) {
+  static <T extends ReplicaInfo> ClusterInfo<T> of(List<NodeInfo> nodes, List<T> replicas) {
     var topics = replicas.stream().map(ReplicaInfo::topic).collect(Collectors.toUnmodifiableSet());
     var replicasForTopic = replicas.stream().collect(Collectors.groupingBy(ReplicaInfo::topic));
     var availableReplicasForTopic =
@@ -73,7 +78,7 @@ public interface ClusterInfo {
             .filter(ReplicaInfo::isLeader)
             .collect(Collectors.groupingBy(r -> Map.entry(r.nodeInfo().id(), r.topic())));
 
-    return new ClusterInfo() {
+    return new ClusterInfo<>() {
       @Override
       public List<NodeInfo> nodes() {
         return nodes;
@@ -84,28 +89,28 @@ public interface ClusterInfo {
       }
 
       @Override
-      public List<ReplicaInfo> availableReplicaLeaders(String topic) {
+      public List<T> availableReplicaLeaders(String topic) {
         return availableReplicaLeadersForTopics.getOrDefault(topic, List.of());
       }
 
       @Override
-      public List<ReplicaInfo> availableReplicaLeaders(int broker, String topic) {
+      public List<T> availableReplicaLeaders(int broker, String topic) {
         return availableLeaderReplicasForBrokersTopics.getOrDefault(
             Map.entry(broker, topic), List.of());
       }
 
       @Override
-      public List<ReplicaInfo> availableReplicas(String topic) {
+      public List<T> availableReplicas(String topic) {
         return availableReplicasForTopic.getOrDefault(topic, List.of());
       }
 
       @Override
-      public List<ReplicaInfo> replicas(String topic) {
+      public List<T> replicas(String topic) {
         return replicasForTopic.getOrDefault(topic, List.of());
       }
 
       @Override
-      public List<ReplicaInfo> replicas() {
+      public List<T> replicas() {
         return replicas;
       }
     };
@@ -131,7 +136,7 @@ public interface ClusterInfo {
    * @param topic The Topic name
    * @return A list of {@link ReplicaInfo}.
    */
-  default List<ReplicaInfo> availableReplicaLeaders(String topic) {
+  default List<T> availableReplicaLeaders(String topic) {
     return replicas(topic).stream()
         .filter(ReplicaInfo::isLeader)
         .collect(Collectors.toUnmodifiableList());
@@ -145,7 +150,7 @@ public interface ClusterInfo {
    * @param topic The Topic name
    * @return A list of {@link ReplicaInfo}.
    */
-  default List<ReplicaInfo> availableReplicaLeaders(int broker, String topic) {
+  default List<T> availableReplicaLeaders(int broker, String topic) {
     return availableReplicaLeaders(topic).stream()
         .filter(r -> r.nodeInfo().id() == broker)
         .collect(Collectors.toUnmodifiableList());
@@ -158,7 +163,7 @@ public interface ClusterInfo {
    * @param topic The topic name
    * @return A list of {@link ReplicaInfo}.
    */
-  default List<ReplicaInfo> availableReplicas(String topic) {
+  default List<T> availableReplicas(String topic) {
     return replicas(topic).stream()
         .filter(ReplicaInfo::isOnline)
         .collect(Collectors.toUnmodifiableList());
@@ -182,7 +187,7 @@ public interface ClusterInfo {
    * @param topic The topic name
    * @return A list of {@link ReplicaInfo}.
    */
-  default List<ReplicaInfo> replicas(String topic) {
+  default List<T> replicas(String topic) {
     return replicas().stream()
         .filter(r -> r.topic().equals(topic))
         .collect(Collectors.toUnmodifiableList());
@@ -192,10 +197,10 @@ public interface ClusterInfo {
    * @param replica to search
    * @return the replica matched to input replica
    */
-  default Optional<ReplicaInfo> replica(TopicPartitionReplica replica) {
+  default Optional<T> replica(TopicPartitionReplica replica) {
     return replicas().stream().filter(r -> r.topicPartitionReplica().equals(replica)).findFirst();
   }
 
   /** @return all replicas cached by this cluster info. */
-  List<ReplicaInfo> replicas();
+  List<T> replicas();
 }
diff --git a/app/src/main/java/org/astraea/app/balancer/BalancerUtils.java b/app/src/main/java/org/astraea/app/balancer/BalancerUtils.java
index 7f8ecdd6..4d7450f6 100644
--- a/app/src/main/java/org/astraea/app/balancer/BalancerUtils.java
+++ b/app/src/main/java/org/astraea/app/balancer/BalancerUtils.java
@@ -30,7 +30,6 @@ import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.admin.NodeInfo;
 import org.astraea.app.admin.Replica;
-import org.astraea.app.admin.ReplicaInfo;
 import org.astraea.app.admin.TopicPartition;
 import org.astraea.app.balancer.log.ClusterLogAllocation;
 import org.astraea.app.cost.HasClusterCost;
@@ -47,7 +46,8 @@ public class BalancerUtils {
    * @param allocation offers new host and data folder
    * @return new cluster info
    */
-  public static ClusterInfo update(ClusterInfo clusterInfo, ClusterLogAllocation allocation) {
+  public static ClusterInfo<Replica> update(
+      ClusterInfo<Replica> clusterInfo, ClusterLogAllocation allocation) {
     var newReplicas =
         clusterInfo.replicas().stream()
             .collect(Collectors.groupingBy(r -> TopicPartition.of(r.topic(), r.partition())))
@@ -56,10 +56,7 @@ public class BalancerUtils {
             .flatMap(
                 entry -> {
                   var lps = allocation.logPlacements(entry.getKey());
-                  var replicas =
-                      entry.getValue().stream()
-                          .map(r -> (Replica) r)
-                          .collect(Collectors.toUnmodifiableList());
+                  var replicas = entry.getValue();
                   return IntStream.range(0, replicas.size())
                       .mapToObj(
                           index -> {
@@ -67,19 +64,18 @@ public class BalancerUtils {
                             // return previous replica due to no new information
                             if (index >= lps.size()) return previous;
                             var lp = lps.get(index);
-                            return (ReplicaInfo)
-                                Replica.of(
-                                    previous.topic(),
-                                    previous.partition(),
-                                    clusterInfo.node(lp.broker()),
-                                    previous.lag(),
-                                    previous.size(),
-                                    index == 0,
-                                    previous.inSync(),
-                                    previous.isFuture(),
-                                    previous.isOffline(),
-                                    previous.isPreferredLeader(),
-                                    lp.dataFolder());
+                            return Replica.of(
+                                previous.topic(),
+                                previous.partition(),
+                                clusterInfo.node(lp.broker()),
+                                previous.lag(),
+                                previous.size(),
+                                index == 0,
+                                previous.inSync(),
+                                previous.isFuture(),
+                                previous.isOffline(),
+                                previous.isPreferredLeader(),
+                                lp.dataFolder());
                           });
                 })
             .collect(Collectors.toUnmodifiableList());
@@ -100,12 +96,13 @@ public class BalancerUtils {
    *     locked.
    * @return a {@link ClusterInfo} with its log placement replaced.
    */
-  public static ClusterInfo merge(ClusterInfo clusterInfo, ClusterLogAllocation allocation) {
-    return new ClusterInfo() {
+  public static ClusterInfo<Replica> merge(
+      ClusterInfo<Replica> clusterInfo, ClusterLogAllocation allocation) {
+    return new ClusterInfo<>() {
       // TODO: maybe add a field to tell if this cluster info is mocked.
       private final Map<Integer, NodeInfo> nodeIdMap =
           nodes().stream().collect(Collectors.toUnmodifiableMap(NodeInfo::id, Function.identity()));
-      private final List<ReplicaInfo> replicas =
+      private final List<Replica> replicas =
           allocation.topicPartitions().stream()
               .map(tp -> Map.entry(tp, allocation.logPlacements(tp)))
               .flatMap(
@@ -139,7 +136,7 @@ public class BalancerUtils {
       }
 
       @Override
-      public List<ReplicaInfo> replicas() {
+      public List<Replica> replicas() {
         return replicas;
       }
     };
@@ -187,7 +184,7 @@ public class BalancerUtils {
   }
 
   static double evaluateCost(
-      ClusterInfo clusterInfo,
+      ClusterInfo<Replica> clusterInfo,
       Map<HasClusterCost, Map<Integer, Collection<HasBeanObject>>> metrics) {
     var scores =
         metrics.keySet().stream()
diff --git a/app/src/main/java/org/astraea/app/balancer/executor/RebalanceAdmin.java b/app/src/main/java/org/astraea/app/balancer/executor/RebalanceAdmin.java
index 69ca4d77..5107a2d2 100644
--- a/app/src/main/java/org/astraea/app/balancer/executor/RebalanceAdmin.java
+++ b/app/src/main/java/org/astraea/app/balancer/executor/RebalanceAdmin.java
@@ -23,6 +23,7 @@ import java.util.concurrent.CompletableFuture;
 import java.util.function.Predicate;
 import org.astraea.app.admin.Admin;
 import org.astraea.app.admin.ClusterInfo;
+import org.astraea.app.admin.Replica;
 import org.astraea.app.admin.TopicPartition;
 import org.astraea.app.admin.TopicPartitionReplica;
 import org.astraea.app.balancer.log.LogPlacement;
@@ -87,7 +88,7 @@ public interface RebalanceAdmin {
    */
   LeaderElectionTask leaderElection(TopicPartition topicPartition);
 
-  ClusterInfo clusterInfo();
+  ClusterInfo<Replica> clusterInfo();
 
   /**
    * @return a {@link Predicate<String>} indicate which topic name is allowed to operate by this
diff --git a/app/src/main/java/org/astraea/app/balancer/executor/RebalanceAdminImpl.java b/app/src/main/java/org/astraea/app/balancer/executor/RebalanceAdminImpl.java
index 61980a11..c85fe9d5 100644
--- a/app/src/main/java/org/astraea/app/balancer/executor/RebalanceAdminImpl.java
+++ b/app/src/main/java/org/astraea/app/balancer/executor/RebalanceAdminImpl.java
@@ -226,7 +226,7 @@ class RebalanceAdminImpl implements RebalanceAdmin {
   }
 
   @Override
-  public ClusterInfo clusterInfo() {
+  public ClusterInfo<Replica> clusterInfo() {
     return admin.clusterInfo(
         admin.topicNames().stream().filter(topicFilter).collect(Collectors.toUnmodifiableSet()));
   }
diff --git a/app/src/main/java/org/astraea/app/balancer/log/ClusterLogAllocation.java b/app/src/main/java/org/astraea/app/balancer/log/ClusterLogAllocation.java
index e4c649a1..b12ac3f7 100644
--- a/app/src/main/java/org/astraea/app/balancer/log/ClusterLogAllocation.java
+++ b/app/src/main/java/org/astraea/app/balancer/log/ClusterLogAllocation.java
@@ -37,11 +37,9 @@ import org.astraea.app.admin.TopicPartition;
  */
 public interface ClusterLogAllocation {
 
-  static ClusterLogAllocation of(ClusterInfo clusterInfo) {
+  static ClusterLogAllocation of(ClusterInfo<Replica> clusterInfo) {
     return of(
         clusterInfo.replicas().stream()
-            .filter(r -> r instanceof Replica)
-            .map(r -> (Replica) r)
             .collect(Collectors.groupingBy(r -> TopicPartition.of(r.topic(), r.partition())))
             .entrySet()
             .stream()
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 fd8aa7e4..d9273d0c 100644
--- a/app/src/main/java/org/astraea/app/cost/BrokerInputCost.java
+++ b/app/src/main/java/org/astraea/app/cost/BrokerInputCost.java
@@ -22,12 +22,14 @@ 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.admin.ReplicaInfo;
 import org.astraea.app.metrics.broker.ServerMetrics;
 import org.astraea.app.metrics.collector.Fetcher;
 
 public class BrokerInputCost implements HasBrokerCost {
   @Override
-  public BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
+  public BrokerCost brokerCost(
+      ClusterInfo<? extends ReplicaInfo> clusterInfo, ClusterBean clusterBean) {
     var brokerCost =
         clusterBean.all().entrySet().stream()
             .collect(
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 14acdd85..ce040e31 100644
--- a/app/src/main/java/org/astraea/app/cost/BrokerOutputCost.java
+++ b/app/src/main/java/org/astraea/app/cost/BrokerOutputCost.java
@@ -22,13 +22,15 @@ 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.admin.ReplicaInfo;
 import org.astraea.app.metrics.broker.ServerMetrics;
 import org.astraea.app.metrics.collector.Fetcher;
 
 public class BrokerOutputCost implements HasBrokerCost {
 
   @Override
-  public BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
+  public BrokerCost brokerCost(
+      ClusterInfo<? extends ReplicaInfo> clusterInfo, ClusterBean clusterBean) {
     var brokerCost =
         clusterBean.all().entrySet().stream()
             .collect(
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 dc2e1aa9..70c97d27 100644
--- a/app/src/main/java/org/astraea/app/cost/CpuCost.java
+++ b/app/src/main/java/org/astraea/app/cost/CpuCost.java
@@ -23,6 +23,7 @@ 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.admin.ReplicaInfo;
 import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.collector.Fetcher;
 import org.astraea.app.metrics.platform.HostMetrics;
@@ -35,7 +36,8 @@ import org.astraea.app.metrics.platform.OperatingSystemInfo;
 public class CpuCost implements HasBrokerCost {
 
   @Override
-  public BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
+  public BrokerCost brokerCost(
+      ClusterInfo<? extends ReplicaInfo> clusterInfo, ClusterBean clusterBean) {
     var cpuCosts =
         clusterBean.all().entrySet().stream()
             .collect(
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 63e3b3be..e54e8d9e 100644
--- a/app/src/main/java/org/astraea/app/cost/HasBrokerCost.java
+++ b/app/src/main/java/org/astraea/app/cost/HasBrokerCost.java
@@ -18,6 +18,7 @@ package org.astraea.app.cost;
 
 import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
+import org.astraea.app.admin.ReplicaInfo;
 
 public interface HasBrokerCost extends CostFunction {
   /**
@@ -27,5 +28,5 @@ public interface HasBrokerCost extends CostFunction {
    * @param clusterBean cluster metrics
    * @return the score of each broker.
    */
-  BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean);
+  BrokerCost brokerCost(ClusterInfo<? extends ReplicaInfo> clusterInfo, ClusterBean clusterBean);
 }
diff --git a/app/src/main/java/org/astraea/app/cost/HasClusterCost.java b/app/src/main/java/org/astraea/app/cost/HasClusterCost.java
index 44bb0c64..cbaa2e86 100644
--- a/app/src/main/java/org/astraea/app/cost/HasClusterCost.java
+++ b/app/src/main/java/org/astraea/app/cost/HasClusterCost.java
@@ -18,6 +18,7 @@ package org.astraea.app.cost;
 
 import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
+import org.astraea.app.admin.Replica;
 
 public interface HasClusterCost extends CostFunction {
   /**
@@ -27,5 +28,5 @@ public interface HasClusterCost extends CostFunction {
    * @param clusterBean cluster metrics
    * @return the score of cluster.
    */
-  ClusterCost clusterCost(ClusterInfo clusterInfo, ClusterBean clusterBean);
+  ClusterCost clusterCost(ClusterInfo<Replica> 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 ed44e03d..7e83b4e4 100644
--- a/app/src/main/java/org/astraea/app/cost/LoadCost.java
+++ b/app/src/main/java/org/astraea/app/cost/LoadCost.java
@@ -25,6 +25,7 @@ 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.ReplicaInfo;
 import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.broker.ServerMetrics;
 import org.astraea.app.metrics.collector.Fetcher;
@@ -41,7 +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, ClusterBean clusterBean) {
+  public BrokerCost brokerCost(
+      ClusterInfo<? extends ReplicaInfo> clusterInfo, ClusterBean clusterBean) {
     var load = computeLoad(clusterBean.all());
 
     // Poisson calculation (-> Poisson -> throughputAbility -> to double)
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 d4d1df2c..3db2950b 100644
--- a/app/src/main/java/org/astraea/app/cost/MemoryCost.java
+++ b/app/src/main/java/org/astraea/app/cost/MemoryCost.java
@@ -23,6 +23,7 @@ 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.admin.ReplicaInfo;
 import org.astraea.app.metrics.collector.Fetcher;
 import org.astraea.app.metrics.platform.HasJvmMemory;
 import org.astraea.app.metrics.platform.HostMetrics;
@@ -34,7 +35,8 @@ public class MemoryCost extends Periodic<Map<Integer, Double>> implements HasBro
    * to the memory usage of brokers.
    */
   @Override
-  public BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
+  public BrokerCost brokerCost(
+      ClusterInfo<? extends ReplicaInfo> clusterInfo, ClusterBean clusterBean) {
     var memoryCosts =
         clusterBean.all().entrySet().stream()
             .collect(
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 90e6eb45..2c570bd4 100644
--- a/app/src/main/java/org/astraea/app/cost/NeutralIntegratedCost.java
+++ b/app/src/main/java/org/astraea/app/cost/NeutralIntegratedCost.java
@@ -24,6 +24,7 @@ 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.admin.ReplicaInfo;
 import org.astraea.app.metrics.collector.Fetcher;
 
 /**
@@ -42,7 +43,8 @@ public class NeutralIntegratedCost implements HasBrokerCost {
   private final WeightProvider weightProvider = WeightProvider.entropy(Normalizer.minMax(true));
 
   @Override
-  public BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
+  public BrokerCost brokerCost(
+      ClusterInfo<? extends ReplicaInfo> clusterInfo, ClusterBean clusterBean) {
     var costMetrics =
         clusterBean.all().entrySet().stream()
             .collect(Collectors.toMap(Map.Entry::getKey, entry -> 0.0));
@@ -94,7 +96,9 @@ public class NeutralIntegratedCost implements HasBrokerCost {
   // calculation.
   // TODO Refactor
   void setBrokerMetrics(
-      HasBrokerCost hasBrokerCost, ClusterInfo clusterInfo, ClusterBean clusterBean) {
+      HasBrokerCost hasBrokerCost,
+      ClusterInfo<? extends ReplicaInfo> clusterInfo,
+      ClusterBean clusterBean) {
     if (hasBrokerCost instanceof BrokerInputCost) {
       var inputBrokerCost = hasBrokerCost.brokerCost(clusterInfo, clusterBean);
       inputBrokerCost
diff --git a/app/src/main/java/org/astraea/app/cost/NodeMetricsCost.java b/app/src/main/java/org/astraea/app/cost/NodeMetricsCost.java
index 1baa53c1..8cd4c534 100644
--- a/app/src/main/java/org/astraea/app/cost/NodeMetricsCost.java
+++ b/app/src/main/java/org/astraea/app/cost/NodeMetricsCost.java
@@ -23,6 +23,7 @@ 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.admin.ReplicaInfo;
 import org.astraea.app.metrics.client.HasNodeMetrics;
 import org.astraea.app.metrics.client.producer.ProducerMetrics;
 import org.astraea.app.metrics.collector.Fetcher;
@@ -30,7 +31,8 @@ import org.astraea.app.metrics.collector.Fetcher;
 /** Calculate the cost by client-node-metrics. */
 public abstract class NodeMetricsCost implements HasBrokerCost {
   @Override
-  public BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
+  public BrokerCost brokerCost(
+      ClusterInfo<? extends ReplicaInfo> clusterInfo, ClusterBean clusterBean) {
     var result =
         clusterBean.all().values().stream()
             .flatMap(Collection::stream)
diff --git a/app/src/main/java/org/astraea/app/cost/NodeTopicSizeCost.java b/app/src/main/java/org/astraea/app/cost/NodeTopicSizeCost.java
index 0db27e09..754a21fa 100644
--- a/app/src/main/java/org/astraea/app/cost/NodeTopicSizeCost.java
+++ b/app/src/main/java/org/astraea/app/cost/NodeTopicSizeCost.java
@@ -23,6 +23,7 @@ import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.admin.NodeInfo;
 import org.astraea.app.admin.Replica;
+import org.astraea.app.admin.ReplicaInfo;
 import org.astraea.app.metrics.broker.LogMetrics;
 import org.astraea.app.metrics.collector.Fetcher;
 
@@ -39,7 +40,8 @@ public class NodeTopicSizeCost implements HasBrokerCost, HasClusterCost {
    * @return a BrokerCost contains the used space for each broker
    */
   @Override
-  public BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
+  public BrokerCost brokerCost(
+      ClusterInfo<? extends ReplicaInfo> clusterInfo, ClusterBean clusterBean) {
     var result =
         clusterBean.all().entrySet().stream()
             .collect(
@@ -53,7 +55,7 @@ public class NodeTopicSizeCost implements HasBrokerCost, HasClusterCost {
   }
 
   @Override
-  public ClusterCost clusterCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
+  public ClusterCost clusterCost(ClusterInfo<Replica> clusterInfo, ClusterBean clusterBean) {
     var brokerCost =
         clusterInfo.nodes().stream()
             .collect(
@@ -62,7 +64,7 @@ public class NodeTopicSizeCost implements HasBrokerCost, HasClusterCost {
                     nodeInfo ->
                         clusterInfo.replicas().stream()
                             .filter(r -> r.nodeInfo().id() == nodeInfo.id())
-                            .mapToLong(r -> r instanceof Replica ? ((Replica) r).size() : 0)
+                            .mapToLong(Replica::size)
                             .sum()));
     return () ->
         dispersion.calculate(
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 eb772f1b..81c78198 100644
--- a/app/src/main/java/org/astraea/app/cost/ReplicaDiskInCost.java
+++ b/app/src/main/java/org/astraea/app/cost/ReplicaDiskInCost.java
@@ -24,6 +24,8 @@ 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.Replica;
+import org.astraea.app.admin.ReplicaInfo;
 import org.astraea.app.admin.TopicPartition;
 import org.astraea.app.admin.TopicPartitionReplica;
 import org.astraea.app.metrics.HasBeanObject;
@@ -47,7 +49,7 @@ public class ReplicaDiskInCost implements HasClusterCost, HasBrokerCost {
   }
 
   @Override
-  public ClusterCost clusterCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
+  public ClusterCost clusterCost(ClusterInfo<Replica> clusterInfo, ClusterBean clusterBean) {
     var brokerCost = brokerCost(clusterInfo, clusterBean).value();
     // when retention occur, brokerCost will be set to -1 , and return a big score to reject this
     // plan.
@@ -56,7 +58,8 @@ public class ReplicaDiskInCost implements HasClusterCost, HasBrokerCost {
   }
 
   @Override
-  public BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
+  public BrokerCost brokerCost(
+      ClusterInfo<? extends ReplicaInfo> clusterInfo, ClusterBean clusterBean) {
     var partitionCost = partitionCost(clusterInfo, clusterBean);
     var brokerLoad =
         clusterInfo.nodes().stream()
@@ -72,7 +75,7 @@ public class ReplicaDiskInCost implements HasClusterCost, HasBrokerCost {
   }
 
   private Function<Integer, Map<TopicPartition, Double>> partitionCost(
-      ClusterInfo clusterInfo, ClusterBean clusterBean) {
+      ClusterInfo<? extends ReplicaInfo> clusterInfo, ClusterBean clusterBean) {
     var replicaIn = replicaDataRate(clusterBean, duration);
     var scoreForBroker =
         clusterInfo.nodes().stream()
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 2dc19629..7ccbeea7 100644
--- a/app/src/main/java/org/astraea/app/cost/ReplicaLeaderCost.java
+++ b/app/src/main/java/org/astraea/app/cost/ReplicaLeaderCost.java
@@ -23,6 +23,8 @@ 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.admin.Replica;
+import org.astraea.app.admin.ReplicaInfo;
 import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.broker.ServerMetrics;
 import org.astraea.app.metrics.collector.Fetcher;
@@ -32,7 +34,8 @@ public class ReplicaLeaderCost implements HasBrokerCost, HasClusterCost {
   private final Dispersion dispersion = Dispersion.correlationCoefficient();
 
   @Override
-  public BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
+  public BrokerCost brokerCost(
+      ClusterInfo<? extends ReplicaInfo> clusterInfo, ClusterBean clusterBean) {
     var result =
         leaderCount(clusterInfo, clusterBean).entrySet().stream()
             .collect(Collectors.toMap(Map.Entry::getKey, e -> (double) e.getValue()));
@@ -40,13 +43,13 @@ public class ReplicaLeaderCost implements HasBrokerCost, HasClusterCost {
   }
 
   @Override
-  public ClusterCost clusterCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
+  public ClusterCost clusterCost(ClusterInfo<Replica> clusterInfo, ClusterBean clusterBean) {
     var brokerScore = brokerCost(clusterInfo, clusterBean).value();
     return () -> dispersion.calculate(brokerScore.values());
   }
 
   private static Map<Integer, Integer> leaderCount(
-      ClusterInfo clusterInfo, ClusterBean clusterBean) {
+      ClusterInfo<? extends ReplicaInfo> clusterInfo, ClusterBean clusterBean) {
     var leaderCount = leaderCount(clusterBean);
     // if there is no available metrics, we re-count the leaders based on cluster information
     if (leaderCount.values().stream().mapToInt(i -> i).sum() == 0) return leaderCount(clusterInfo);
@@ -68,7 +71,7 @@ public class ReplicaLeaderCost implements HasBrokerCost, HasClusterCost {
                         .sum()));
   }
 
-  static Map<Integer, Integer> leaderCount(ClusterInfo clusterInfo) {
+  static Map<Integer, Integer> leaderCount(ClusterInfo<? extends ReplicaInfo> clusterInfo) {
     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/partitioner/Dispatcher.java b/app/src/main/java/org/astraea/app/partitioner/Dispatcher.java
index 85310bab..2f35d308 100644
--- a/app/src/main/java/org/astraea/app/partitioner/Dispatcher.java
+++ b/app/src/main/java/org/astraea/app/partitioner/Dispatcher.java
@@ -22,6 +22,7 @@ import java.util.stream.Collectors;
 import org.apache.kafka.clients.producer.Partitioner;
 import org.apache.kafka.common.Cluster;
 import org.astraea.app.admin.ClusterInfo;
+import org.astraea.app.admin.ReplicaInfo;
 
 public interface Dispatcher extends Partitioner {
   /**
@@ -38,7 +39,7 @@ public interface Dispatcher extends Partitioner {
    * @param value The value to partition
    * @param clusterInfo The current cluster metadata
    */
-  int partition(String topic, byte[] key, byte[] value, ClusterInfo clusterInfo);
+  int partition(String topic, byte[] key, byte[] value, ClusterInfo<ReplicaInfo> clusterInfo);
 
   /**
    * configure this dispatcher. This method is called only once.
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 aa66b66e..eae00657 100644
--- a/app/src/main/java/org/astraea/app/partitioner/StrictCostDispatcher.java
+++ b/app/src/main/java/org/astraea/app/partitioner/StrictCostDispatcher.java
@@ -29,6 +29,7 @@ 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;
 import org.astraea.app.argument.DurationField;
 import org.astraea.app.common.Utils;
 import org.astraea.app.cost.CostFunction;
@@ -85,7 +86,7 @@ public class StrictCostDispatcher implements Dispatcher {
     return beanCollector.register().host(host).port(port).fetcher(fetcher).build();
   }
 
-  void tryToUpdateFetcher(ClusterInfo clusterInfo) {
+  void tryToUpdateFetcher(ClusterInfo<ReplicaInfo> clusterInfo) {
     // add new receivers for new brokers
     fetcher.ifPresent(
         fetcher ->
@@ -102,7 +103,8 @@ public class StrictCostDispatcher implements Dispatcher {
   }
 
   @Override
-  public int partition(String topic, byte[] key, byte[] value, ClusterInfo clusterInfo) {
+  public int partition(
+      String topic, byte[] key, byte[] value, ClusterInfo<ReplicaInfo> clusterInfo) {
     var partitionLeaders = clusterInfo.availableReplicaLeaders(topic);
     // just return first partition if there is no available partitions
     if (partitionLeaders.isEmpty()) return 0;
@@ -125,7 +127,7 @@ public class StrictCostDispatcher implements Dispatcher {
     return candidate.get((int) (Math.random() * candidate.size())).partition();
   }
 
-  synchronized void tryToUpdateRoundRobin(ClusterInfo clusterInfo) {
+  synchronized void tryToUpdateRoundRobin(ClusterInfo<ReplicaInfo> clusterInfo) {
     if (System.currentTimeMillis() >= timeToUpdateRoundRobin) {
       var roundRobin =
           newRoundRobin(
@@ -167,7 +169,9 @@ public class StrictCostDispatcher implements Dispatcher {
    * @return SmoothWeightedRoundRobin
    */
   static RoundRobin<Integer> newRoundRobin(
-      Map<HasBrokerCost, Double> costFunctions, ClusterInfo clusterInfo, ClusterBean clusterBean) {
+      Map<HasBrokerCost, Double> costFunctions,
+      ClusterInfo<ReplicaInfo> clusterInfo,
+      ClusterBean clusterBean) {
     var weightedCost =
         costFunctions.entrySet().stream()
             .flatMap(
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 c4699306..01b4fcba 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
@@ -24,6 +24,7 @@ import java.util.Map;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 import org.astraea.app.admin.ClusterInfo;
+import org.astraea.app.admin.ReplicaInfo;
 import org.astraea.app.cost.Periodic;
 
 /**
@@ -113,7 +114,7 @@ public final class SmoothWeightRoundRobin
    *
    * @return the preferred ID
    */
-  public synchronized int getAndChoose(String topic, ClusterInfo clusterInfo) {
+  public synchronized int getAndChoose(String topic, ClusterInfo<ReplicaInfo> clusterInfo) {
     // TODO Update brokerID with ClusterInfo frequency.
     var brokerID =
         brokersIDofTopic.computeIfAbsent(
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 93d648db..f78972bd 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
@@ -71,7 +71,8 @@ public class SmoothWeightRoundRobinDispatcher extends Periodic<Map<Integer, Doub
   public static final String JMX_PORT = "jmx.port";
 
   @Override
-  public int partition(String topic, byte[] key, byte[] value, ClusterInfo clusterInfo) {
+  public int partition(
+      String topic, byte[] key, byte[] value, ClusterInfo<ReplicaInfo> clusterInfo) {
     var targetPartition = unusedPartitions.poll();
     tryUpdateAfterOneSecond(
         () -> {
@@ -156,7 +157,7 @@ public class SmoothWeightRoundRobinDispatcher extends Periodic<Map<Integer, Doub
         .build();
   }
 
-  private int nextValue(String topic, ClusterInfo clusterInfo, int targetBroker) {
+  private int nextValue(String topic, ClusterInfo<ReplicaInfo> clusterInfo, int targetBroker) {
     return topicCounter
         .computeIfAbsent(topic, k -> new BrokerNextCounter(clusterInfo))
         .brokerCounter
@@ -164,7 +165,7 @@ public class SmoothWeightRoundRobinDispatcher extends Periodic<Map<Integer, Doub
         .getAndIncrement();
   }
 
-  private void refreshPartitionMetaData(ClusterInfo clusterInfo, String topic) {
+  private void refreshPartitionMetaData(ClusterInfo<ReplicaInfo> clusterInfo, String topic) {
     partitions = clusterInfo.availableReplicas(topic);
     partitions.forEach(
         p ->
@@ -183,7 +184,7 @@ public class SmoothWeightRoundRobinDispatcher extends Periodic<Map<Integer, Doub
   private static class BrokerNextCounter {
     private final Map<Integer, AtomicInteger> brokerCounter;
 
-    BrokerNextCounter(ClusterInfo clusterInfo) {
+    BrokerNextCounter(ClusterInfo<ReplicaInfo> clusterInfo) {
       brokerCounter =
           clusterInfo.nodes().stream()
               .collect(Collectors.toMap(NodeInfo::id, node -> new AtomicInteger(0)));
diff --git a/app/src/main/java/org/astraea/app/web/BalancerHandler.java b/app/src/main/java/org/astraea/app/web/BalancerHandler.java
index fe8d699f..7f2ff9c1 100644
--- a/app/src/main/java/org/astraea/app/web/BalancerHandler.java
+++ b/app/src/main/java/org/astraea/app/web/BalancerHandler.java
@@ -108,7 +108,7 @@ class BalancerHandler implements Handler {
                                                 .replica(
                                                     TopicPartitionReplica.of(
                                                         tp.topic(), tp.partition(), l.broker()))
-                                                .map(r -> ((Replica) r).size())
+                                                .map(Replica::size)
                                                 .orElse(null)),
                                     placements(entry.getKey().logPlacements(tp), ignored -> null)))
                         .collect(Collectors.toUnmodifiableList()))


Test patch:
diff --git a/app/src/test/java/org/astraea/app/balancer/BalancerUtilsIntegratedTest.java b/app/src/test/java/org/astraea/app/balancer/BalancerUtilsIntegratedTest.java
index a67384e6..6bee86a8 100644
--- a/app/src/test/java/org/astraea/app/balancer/BalancerUtilsIntegratedTest.java
+++ b/app/src/test/java/org/astraea/app/balancer/BalancerUtilsIntegratedTest.java
@@ -22,7 +22,6 @@ import java.util.Map;
 import java.util.Set;
 import java.util.stream.IntStream;
 import org.astraea.app.admin.Admin;
-import org.astraea.app.admin.Replica;
 import org.astraea.app.admin.TopicPartition;
 import org.astraea.app.balancer.log.ClusterLogAllocation;
 import org.astraea.app.balancer.log.LogPlacement;
@@ -47,7 +46,7 @@ public class BalancerUtilsIntegratedTest extends RequireBrokerCluster {
       }
 
       var clusterInfo = admin.clusterInfo(Set.of(topicName));
-      clusterInfo.replicas().forEach(r -> Assertions.assertTrue(((Replica) r).size() > 0));
+      clusterInfo.replicas().forEach(r -> Assertions.assertTrue(r.size() > 0));
 
       var replica = clusterInfo.replicas().iterator().next();
       var newBrokerId =
@@ -60,11 +59,11 @@ public class BalancerUtilsIntegratedTest extends RequireBrokerCluster {
                   Map.of(
                       TopicPartition.of(topicName, 0),
                       // change the broker
-                      List.of(LogPlacement.of(newBrokerId, ((Replica) replica).dataFolder())))));
+                      List.of(LogPlacement.of(newBrokerId, replica.dataFolder())))));
 
       Assertions.assertEquals(clusterInfo.replicas().size(), merged.replicas().size());
       Assertions.assertEquals(clusterInfo.topics().size(), merged.topics().size());
-      merged.replicas().forEach(r -> Assertions.assertTrue(((Replica) r).size() > 0));
+      merged.replicas().forEach(r -> Assertions.assertTrue(r.size() > 0));
       Assertions.assertEquals(1, merged.replicas(topicName).size());
       Assertions.assertEquals(newBrokerId, merged.replicas(topicName).get(0).nodeInfo().id());
     }
diff --git a/app/src/test/java/org/astraea/app/balancer/BalancerUtilsTest.java b/app/src/test/java/org/astraea/app/balancer/BalancerUtilsTest.java
index 3091acfe..c89085cc 100644
--- a/app/src/test/java/org/astraea/app/balancer/BalancerUtilsTest.java
+++ b/app/src/test/java/org/astraea/app/balancer/BalancerUtilsTest.java
@@ -19,11 +19,9 @@ package org.astraea.app.balancer;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
-import org.apache.kafka.common.Cluster;
-import org.apache.kafka.common.Node;
-import org.apache.kafka.common.PartitionInfo;
 import org.astraea.app.admin.ClusterInfo;
+import org.astraea.app.admin.NodeInfo;
+import org.astraea.app.admin.Replica;
 import org.astraea.app.admin.TopicPartition;
 import org.astraea.app.balancer.log.ClusterLogAllocation;
 import org.astraea.app.balancer.log.LogPlacement;
@@ -68,22 +66,25 @@ class BalancerUtilsTest {
     var tp2 = TopicPartition.of("testMockCluster", 0);
     var logPlacement1 = List.of(LogPlacement.of(0, "/data"), LogPlacement.of(1, "/data"));
     var logPlacement2 = List.of(LogPlacement.of(1, "/data1"), LogPlacement.of(2, "/data1"));
-    var nodes =
-        new Node[] {
-          new Node(0, "localhost", 9092),
-          new Node(1, "localhost", 9092),
-          new Node(2, "localhost", 9092)
-        };
-    var partitionInfo = new PartitionInfo("test-1", 1, nodes[0], nodes, nodes);
     var clusterInfo =
         ClusterInfo.of(
-            new Cluster(
-                "",
-                List.of(nodes[0], nodes[1], nodes[2]),
-                List.of(partitionInfo),
-                Set.of(),
-                Set.of(),
-                Node.noNode()));
+            List.of(
+                NodeInfo.of(0, "localhost", 9092),
+                NodeInfo.of(1, "localhost", 9092),
+                NodeInfo.of(2, "localhost", 9092)),
+            List.of(
+                Replica.of(
+                    "test-1",
+                    1,
+                    NodeInfo.of(0, "localhost", 9092),
+                    0,
+                    100,
+                    true,
+                    true,
+                    false,
+                    false,
+                    true,
+                    "/tmp/aa")));
     var cla = ClusterLogAllocation.of(Map.of(tp1, logPlacement1, tp2, logPlacement2));
     var mockClusterInfo = BalancerUtils.merge(clusterInfo, cla);
     Assertions.assertEquals(mockClusterInfo.replicas("testMockCluster").size(), 4);
@@ -94,22 +95,14 @@ class BalancerUtilsTest {
 
   @Test
   void testEvaluateCost() {
-    var node1 = new Node(0, "localhost", 9092);
-    var node2 = new Node(1, "localhost", 9092);
-    var partitionInfo1 =
-        new PartitionInfo("test-1", 0, node1, new Node[] {node1}, new Node[] {node1});
-    var partitionInfo2 =
-        new PartitionInfo("test-1", 1, node2, new Node[] {node2}, new Node[] {node2});
+    var node1 = NodeInfo.of(0, "localhost", 9092);
+    var node2 = NodeInfo.of(1, "localhost", 9092);
     var clusterInfo =
         ClusterInfo.of(
-            new Cluster(
-                "",
-                List.of(node1, node2),
-                List.of(partitionInfo1, partitionInfo2),
-                Set.of(),
-                Set.of(),
-                Node.noNode()));
-
+            List.of(node1, node2),
+            List.of(
+                Replica.of("test-1", 0, node1, 0, 100, true, true, false, false, true, "/tmp/aa"),
+                Replica.of("test-1", 1, node2, 0, 100, true, true, false, false, true, "/tmp/aa")));
     var cf1 = new ReplicaLeaderCost();
     var cf2 = new ReplicaDiskInCost(Configuration.of(Map.of("metrics.duration", "5")));
     var cost = BalancerUtils.evaluateCost(clusterInfo, Map.of(cf1, beanObjects, cf2, beanObjects));
diff --git a/app/src/test/java/org/astraea/app/balancer/FakeClusterInfo.java b/app/src/test/java/org/astraea/app/balancer/FakeClusterInfo.java
index e926e753..7d9a3037 100644
--- a/app/src/test/java/org/astraea/app/balancer/FakeClusterInfo.java
+++ b/app/src/test/java/org/astraea/app/balancer/FakeClusterInfo.java
@@ -27,10 +27,9 @@ import java.util.stream.IntStream;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.admin.NodeInfo;
 import org.astraea.app.admin.Replica;
-import org.astraea.app.admin.ReplicaInfo;
 import org.astraea.app.admin.TopicPartition;
 
-public class FakeClusterInfo implements ClusterInfo {
+public class FakeClusterInfo implements ClusterInfo<Replica> {
 
   public static FakeClusterInfo of(
       int nodeCount, int topicCount, int partitionCount, int replicaCount) {
@@ -85,20 +84,19 @@ public class FakeClusterInfo implements ClusterInfo {
                     IntStream.range(0, replicaCount)
                         .mapToObj(
                             r ->
-                                (ReplicaInfo)
-                                    Replica.of(
-                                        tp.topic(),
-                                        tp.partition(),
-                                        nodes.get(r),
-                                        0,
-                                        -1,
-                                        r == 0,
-                                        true,
-                                        false,
-                                        false,
-                                        false,
-                                        dataDirectoryList.get(
-                                            tp.partition() % dataDirectories.size()))))
+                                Replica.of(
+                                    tp.topic(),
+                                    tp.partition(),
+                                    nodes.get(r),
+                                    0,
+                                    -1,
+                                    r == 0,
+                                    true,
+                                    false,
+                                    false,
+                                    false,
+                                    dataDirectoryList.get(
+                                        tp.partition() % dataDirectories.size()))))
             .collect(Collectors.toUnmodifiableList());
 
     return new FakeClusterInfo(
@@ -110,10 +108,10 @@ public class FakeClusterInfo implements ClusterInfo {
 
   private final List<NodeInfo> nodes;
   private final Map<Integer, Set<String>> dataDirectories;
-  private final List<ReplicaInfo> replicas;
+  private final List<Replica> replicas;
 
   FakeClusterInfo(
-      List<NodeInfo> nodes, Map<Integer, Set<String>> dataDirectories, List<ReplicaInfo> replicas) {
+      List<NodeInfo> nodes, Map<Integer, Set<String>> dataDirectories, List<Replica> replicas) {
     this.nodes = nodes;
     this.dataDirectories = dataDirectories;
     this.replicas = replicas;
@@ -129,7 +127,7 @@ public class FakeClusterInfo implements ClusterInfo {
   }
 
   @Override
-  public List<ReplicaInfo> replicas() {
+  public List<Replica> replicas() {
     return replicas;
   }
 }
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 eb085537..03b2e905 100644
--- a/app/src/test/java/org/astraea/app/cost/BrokerInputCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/BrokerInputCostTest.java
@@ -36,7 +36,7 @@ public class BrokerInputCostTest extends RequireBrokerCluster {
     var scores =
         brokerInputCost
             .brokerCost(
-                ClusterInfo.EMPTY,
+                ClusterInfo.empty(),
                 ClusterBean.of(
                     Map.of(
                         1,
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 092aaa52..28844478 100644
--- a/app/src/test/java/org/astraea/app/cost/BrokerOutputCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/BrokerOutputCostTest.java
@@ -36,7 +36,7 @@ public class BrokerOutputCostTest extends RequireBrokerCluster {
     var scores =
         brokerOutputCost
             .brokerCost(
-                ClusterInfo.EMPTY,
+                ClusterInfo.empty(),
                 ClusterBean.of(
                     Map.of(
                         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 e03b9a0d..1365d508 100644
--- a/app/src/test/java/org/astraea/app/cost/ReplicaDiskInCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/ReplicaDiskInCostTest.java
@@ -23,7 +23,7 @@ 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;
+import org.astraea.app.admin.Replica;
 import org.astraea.app.metrics.BeanObject;
 import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.broker.HasGauge;
@@ -79,8 +79,8 @@ class ReplicaDiskInCostTest extends RequireBrokerCluster {
     Assertions.assertEquals(0.20721255412897746, brokerLoad);
   }
 
-  private ClusterInfo clusterInfo() {
-    ClusterInfo clusterInfo = Mockito.mock(ClusterInfo.class);
+  private ClusterInfo<Replica> clusterInfo() {
+    var clusterInfo = (ClusterInfo<Replica>) Mockito.mock(ClusterInfo.class);
     Mockito.when(clusterInfo.nodes())
         .thenReturn(
             List.of(NodeInfo.of(1, "", -1), NodeInfo.of(2, "", -1), NodeInfo.of(3, "", -1)));
@@ -90,13 +90,79 @@ class ReplicaDiskInCostTest extends RequireBrokerCluster {
             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))
+                        Replica.of(
+                            "test-1",
+                            0,
+                            NodeInfo.of(1, "", -1),
+                            0,
+                            100,
+                            true,
+                            true,
+                            false,
+                            false,
+                            true,
+                            "/tmp/aa"),
+                        Replica.of(
+                            "test-1",
+                            0,
+                            NodeInfo.of(2, "", -1),
+                            0,
+                            100,
+                            false,
+                            true,
+                            false,
+                            false,
+                            false,
+                            "/tmp/aa"),
+                        Replica.of(
+                            "test-1",
+                            1,
+                            NodeInfo.of(2, "", -1),
+                            0,
+                            100,
+                            false,
+                            true,
+                            false,
+                            false,
+                            false,
+                            "/tmp/aa"),
+                        Replica.of(
+                            "test-1",
+                            1,
+                            NodeInfo.of(3, "", -1),
+                            0,
+                            100,
+                            true,
+                            true,
+                            false,
+                            false,
+                            true,
+                            "/tmp/aa"))
                     : 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)));
+                        Replica.of(
+                            "test-2",
+                            0,
+                            NodeInfo.of(1, "", -1),
+                            0,
+                            100,
+                            false,
+                            true,
+                            false,
+                            false,
+                            false,
+                            "/tmp/aa"),
+                        Replica.of(
+                            "test-2",
+                            0,
+                            NodeInfo.of(3, "", -1),
+                            0,
+                            100,
+                            true,
+                            true,
+                            false,
+                            false,
+                            true,
+                            "/tmp/aa")));
     return clusterInfo;
   }
 
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 a3bd9685..9711f0b6 100644
--- a/app/src/test/java/org/astraea/app/cost/ReplicaLeaderCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/ReplicaLeaderCostTest.java
@@ -39,7 +39,8 @@ public class ReplicaLeaderCostTest {
             ReplicaInfo.of("topic", 0, NodeInfo.of(10, "broker0", 1111), true, true, true),
             ReplicaInfo.of("topic", 0, NodeInfo.of(10, "broker0", 1111), true, true, true),
             ReplicaInfo.of("topic", 0, NodeInfo.of(11, "broker1", 1111), true, true, true));
-    var clusterInfo = Mockito.mock(ClusterInfo.class);
+    @SuppressWarnings("unchecked")
+    ClusterInfo<ReplicaInfo> clusterInfo = Mockito.mock(ClusterInfo.class);
     Mockito.when(clusterInfo.topics()).thenReturn(Set.of("topic"));
     Mockito.when(clusterInfo.availableReplicaLeaders(Mockito.anyString())).thenReturn(replicas);
     var cost = ReplicaLeaderCost.leaderCount(clusterInfo);
@@ -60,8 +61,8 @@ public class ReplicaLeaderCostTest {
     var broker2 = List.of((HasBeanObject) LeaderCount2);
     var broker3 = List.of((HasBeanObject) LeaderCount3);
     var clusterBean = ClusterBean.of(Map.of(1, broker1, 2, broker2, 3, broker3));
-    var brokerLoad = costFunction.brokerCost(ClusterInfo.EMPTY, clusterBean);
-    var clusterLoad = costFunction.clusterCost(ClusterInfo.EMPTY, clusterBean);
+    var brokerLoad = costFunction.brokerCost(ClusterInfo.empty(), clusterBean);
+    var clusterLoad = costFunction.clusterCost(ClusterInfo.empty(), clusterBean);
 
     Assertions.assertEquals(3, brokerLoad.value().size());
     Assertions.assertEquals(3.0, brokerLoad.value().get(1));
diff --git a/app/src/test/java/org/astraea/app/partitioner/DispatcherTest.java b/app/src/test/java/org/astraea/app/partitioner/DispatcherTest.java
index bf5629d5..125d1436 100644
--- a/app/src/test/java/org/astraea/app/partitioner/DispatcherTest.java
+++ b/app/src/test/java/org/astraea/app/partitioner/DispatcherTest.java
@@ -23,6 +23,7 @@ import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.kafka.common.Cluster;
 import org.astraea.app.admin.ClusterInfo;
+import org.astraea.app.admin.ReplicaInfo;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
@@ -34,7 +35,8 @@ public class DispatcherTest {
     var dispatcher =
         new Dispatcher() {
           @Override
-          public int partition(String topic, byte[] key, byte[] value, ClusterInfo clusterInfo) {
+          public int partition(
+              String topic, byte[] key, byte[] value, ClusterInfo<ReplicaInfo> clusterInfo) {
             Assertions.assertEquals(0, Objects.requireNonNull(key).length);
             Assertions.assertEquals(0, Objects.requireNonNull(value).length);
             count.incrementAndGet();
@@ -59,7 +61,8 @@ public class DispatcherTest {
     var dispatcher =
         new Dispatcher() {
           @Override
-          public int partition(String topic, byte[] key, byte[] value, ClusterInfo clusterInfo) {
+          public int partition(
+              String topic, byte[] key, byte[] value, ClusterInfo<ReplicaInfo> clusterInfo) {
             return 0;
           }
         };
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 af49d3d8..b98392f2 100644
--- a/app/src/test/java/org/astraea/app/partitioner/StrictCostDispatcherTest.java
+++ b/app/src/test/java/org/astraea/app/partitioner/StrictCostDispatcherTest.java
@@ -187,7 +187,8 @@ public class StrictCostDispatcherTest {
     var costFunction =
         new HasBrokerCost() {
           @Override
-          public BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
+          public BrokerCost brokerCost(
+              ClusterInfo<? extends ReplicaInfo> clusterInfo, ClusterBean clusterBean) {
             return Mockito.mock(BrokerCost.class);
           }
 
@@ -263,7 +264,8 @@ public class StrictCostDispatcherTest {
       var costFunction =
           new HasBrokerCost() {
             @Override
-            public BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
+            public BrokerCost brokerCost(
+                ClusterInfo<? extends ReplicaInfo> clusterInfo, ClusterBean clusterBean) {
               return () -> Map.of(brokerId, 10D);
             }
           };