Repository: skiptests/astraea
Pull request #411 (2022-06-15 10:44:01)

new interface for read beanObjects
add a new interface for read beanObjects


Patch:
diff --git a/README.md b/README.md
index 9724a9db..f20e1df8 100644
--- a/README.md
+++ b/README.md
@@ -1,5 +1,4 @@
-# Astraea
-a collection of tools used to balance Kafka data
+![alt text](https://github.com/skiptests/astraea/blob/main/logo/logo_with_background.png?raw=true)
 
 # Authors
 - Chia-Ping Tsai <chia7712@gmail.com>
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 e09c6778..d0f50b02 100644
--- a/app/src/main/java/org/astraea/app/admin/Builder.java
+++ b/app/src/main/java/org/astraea/app/admin/Builder.java
@@ -54,7 +54,6 @@ import org.astraea.app.common.Utils;
 import org.astraea.app.cost.ClusterInfo;
 import org.astraea.app.cost.NodeInfo;
 import org.astraea.app.cost.ReplicaInfo;
-import org.astraea.app.metrics.HasBeanObject;
 
 public class Builder {
 
@@ -524,14 +523,8 @@ public class Builder {
         }
 
         @Override
-        public Collection<HasBeanObject> beans(int brokerId) {
-          return List.of();
-        }
-
-        @Override
-        public Map<Integer, Collection<HasBeanObject>> allBeans() {
-          return nodeInfo.stream()
-              .collect(Collectors.toUnmodifiableMap(NodeInfo::id, ignore -> List.of()));
+        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
new file mode 100644
index 00000000..a61a5774
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/admin/ClusterBean.java
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.admin;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.astraea.app.metrics.HasBeanObject;
+
+/** Used to get beanObject using a variety of different keys . */
+public interface ClusterBean {
+  private static Collection<HasBeanObject> compareBeanObject(
+      Collection<HasBeanObject> x1, Collection<HasBeanObject> x2) {
+    var beanObject2 = x2.iterator().next().beanObject();
+    if (x1.stream()
+        .noneMatch(
+            hasBeanObject ->
+                hasBeanObject.beanObject().getProperties().equals(beanObject2.getProperties())
+                    && hasBeanObject.beanObject().domainName().equals(beanObject2.domainName())))
+      return Stream.concat(x1.stream(), x2.stream()).collect(Collectors.toList());
+    return x1;
+  }
+
+  static ClusterBean of(Map<Integer, Collection<HasBeanObject>> allBeans) {
+    Map<TopicPartition, Collection<HasBeanObject>> beanObjectByPartition =
+        allBeans.entrySet().stream()
+            .flatMap(
+                entry ->
+                    entry.getValue().stream()
+                        .filter(
+                            x ->
+                                x.beanObject() != null
+                                    && x.beanObject().getProperties().containsKey("topic")
+                                    && x.beanObject().getProperties().containsKey("partition"))
+                        .filter(
+                            hasBeanObject ->
+                                hasBeanObject.beanObject().getProperties().containsKey("topic")
+                                    && hasBeanObject
+                                        .beanObject()
+                                        .getProperties()
+                                        .containsKey("partition"))
+                        .map(
+                            hasBeanObject -> {
+                              var properties = hasBeanObject.beanObject().getProperties();
+                              var topic = properties.get("topic");
+                              var partition = properties.get("partition");
+                              return Map.entry(
+                                  TopicPartition.of(topic, partition), List.of(hasBeanObject));
+                            }))
+            .collect(
+                Collectors.toMap(
+                    Map.Entry::getKey, Map.Entry::getValue, ClusterBean::compareBeanObject));
+    Map<TopicPartitionReplica, Collection<HasBeanObject>> beanObjectByReplica =
+        allBeans.entrySet().stream()
+            .flatMap(
+                entry ->
+                    entry.getValue().stream()
+                        .filter(
+                            x ->
+                                x.beanObject() != null
+                                    && x.beanObject().getProperties().containsKey("topic")
+                                    && x.beanObject().getProperties().containsKey("partition"))
+                        .map(
+                            hasBeanObject -> {
+                              var properties = hasBeanObject.beanObject().getProperties();
+                              var topic = properties.get("topic");
+                              var partition = Integer.parseInt(properties.get("partition"));
+                              return Map.entry(
+                                  new TopicPartitionReplica(topic, partition, entry.getKey()),
+                                  List.of(hasBeanObject));
+                            }))
+            .collect(
+                Collectors.toMap(
+                    Map.Entry::getKey, Map.Entry::getValue, ClusterBean::compareBeanObject));
+    return new ClusterBean() {
+      @Override
+      public Map<Integer, Collection<HasBeanObject>> all() {
+        return Collections.unmodifiableMap(allBeans);
+      }
+
+      @Override
+      public Map<TopicPartition, Collection<HasBeanObject>> mapByPartition() {
+        return beanObjectByPartition;
+      }
+
+      @Override
+      public Map<TopicPartitionReplica, Collection<HasBeanObject>> mapByReplica() {
+        return beanObjectByReplica;
+      }
+    };
+  }
+
+  /**
+   * @return a {@link Map} collection that contains broker as key and Collection of {@link
+   *     HasBeanObject} as value.
+   */
+  Map<Integer, Collection<HasBeanObject>> all();
+
+  /**
+   * @return a {@link Map} collection that contains {@link TopicPartition} as key and a {@link
+   *     HasBeanObject} as value,note that this can only be used to get partition-related
+   *     beanObjects.
+   */
+  Map<TopicPartition, Collection<HasBeanObject>> mapByPartition();
+
+  /**
+   * @return a {@link Map} collection that contains {@link TopicPartitionReplica} as key and a
+   *     {@link HasBeanObject} as value,note that this can only be used to get partition-related
+   *     beanObjects.
+   */
+  Map<TopicPartitionReplica, Collection<HasBeanObject>> mapByReplica();
+}
diff --git a/app/src/main/java/org/astraea/app/admin/TopicPartitionReplica.java b/app/src/main/java/org/astraea/app/admin/TopicPartitionReplica.java
new file mode 100644
index 00000000..5a074ce5
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/admin/TopicPartitionReplica.java
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.admin;
+
+import java.util.Objects;
+
+public class TopicPartitionReplica implements Comparable<TopicPartitionReplica> {
+  public static TopicPartitionReplica from(org.apache.kafka.common.TopicPartitionReplica tpr) {
+    return new TopicPartitionReplica(tpr.topic(), tpr.partition(), tpr.brokerId());
+  }
+
+  public static org.apache.kafka.common.TopicPartitionReplica to(TopicPartitionReplica tpr) {
+    return new org.apache.kafka.common.TopicPartitionReplica(
+        tpr.topic, tpr.partition, tpr.brokerId());
+  }
+
+  public static TopicPartitionReplica of(String topic, int partition, int brokerId) {
+    return new TopicPartitionReplica(topic, partition, brokerId);
+  }
+
+  private final int brokerId;
+  private final int partition;
+  private final String topic;
+
+  public TopicPartitionReplica(String topic, int partition, int brokerId) {
+    this.partition = partition;
+    this.topic = topic;
+    this.brokerId = brokerId;
+  }
+
+  @Override
+  public int compareTo(TopicPartitionReplica o) {
+    var b = Integer.compare(brokerId, o.brokerId);
+    if (b != 0) return b;
+    var t = topic.compareTo(o.topic);
+    if (t != 0) return t;
+    return Integer.compare(partition, o.partition);
+  }
+
+  public int brokerId() {
+    return brokerId;
+  }
+
+  public int partition() {
+    return partition;
+  }
+
+  public String topic() {
+    return topic;
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(brokerId, topic, partition);
+  }
+
+  @Override
+  public String toString() {
+    return String.format("%d-%s-%d", brokerId, topic, partition);
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) return true;
+    if (o == null || getClass() != o.getClass()) return false;
+    TopicPartitionReplica that = (TopicPartitionReplica) o;
+    return brokerId == that.brokerId
+        && partition == that.partition
+        && Objects.equals(topic, that.topic);
+  }
+}
diff --git a/app/src/main/java/org/astraea/app/cost/ClusterInfo.java b/app/src/main/java/org/astraea/app/cost/ClusterInfo.java
index d2781be1..8aa822c3 100644
--- a/app/src/main/java/org/astraea/app/cost/ClusterInfo.java
+++ b/app/src/main/java/org/astraea/app/cost/ClusterInfo.java
@@ -25,6 +25,7 @@ import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.Set;
 import java.util.stream.Collectors;
+import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.metrics.HasBeanObject;
 
 public interface ClusterInfo {
@@ -88,13 +89,8 @@ public interface ClusterInfo {
       }
 
       @Override
-      public Collection<HasBeanObject> beans(int brokerId) {
-        return List.of();
-      }
-
-      @Override
-      public Map<Integer, Collection<HasBeanObject>> allBeans() {
-        return Map.of();
+      public ClusterBean clusterBean() {
+        return ClusterBean.of(Map.of());
       }
     };
   }
@@ -109,7 +105,8 @@ public interface ClusterInfo {
   static ClusterInfo of(ClusterInfo cluster, Map<Integer, Collection<HasBeanObject>> beans) {
     var all = new HashMap<Integer, List<HasBeanObject>>();
     cluster
-        .allBeans()
+        .clusterBean()
+        .all()
         .forEach((key, value) -> all.computeIfAbsent(key, k -> new ArrayList<>()).addAll(value));
     beans.forEach((key, value) -> all.computeIfAbsent(key, k -> new ArrayList<>()).addAll(value));
     return new ClusterInfo() {
@@ -145,13 +142,8 @@ public interface ClusterInfo {
       }
 
       @Override
-      public Collection<HasBeanObject> beans(int brokerId) {
-        return all.getOrDefault(brokerId, List.of());
-      }
-
-      @Override
-      public Map<Integer, Collection<HasBeanObject>> allBeans() {
-        return Collections.unmodifiableMap(all);
+      public ClusterBean clusterBean() {
+        return ClusterBean.of(Collections.unmodifiableMap(all));
       }
     };
   }
@@ -230,13 +222,6 @@ public interface ClusterInfo {
    */
   List<ReplicaInfo> replicas(String topic);
 
-  /**
-   * @param brokerId broker id
-   * @return return the metrics of broker. It returns empty collection if the broker id is
-   *     nonexistent
-   */
-  Collection<HasBeanObject> beans(int brokerId);
-
-  /** @return all beans of all brokers */
-  Map<Integer, Collection<HasBeanObject>> allBeans();
+  /** @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/LoadCost.java b/app/src/main/java/org/astraea/app/cost/LoadCost.java
index a584dab7..61652e2d 100644
--- a/app/src/main/java/org/astraea/app/cost/LoadCost.java
+++ b/app/src/main/java/org/astraea/app/cost/LoadCost.java
@@ -40,7 +40,7 @@ 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.allBeans());
+    var load = computeLoad(clusterInfo.clusterBean().all());
 
     // Poisson calculation (-> Poisson -> throughputAbility -> to double)
     var brokerScore =
diff --git a/app/src/main/java/org/astraea/app/cost/ThroughputCost.java b/app/src/main/java/org/astraea/app/cost/ThroughputCost.java
index bcc7e1d9..2ace776f 100644
--- a/app/src/main/java/org/astraea/app/cost/ThroughputCost.java
+++ b/app/src/main/java/org/astraea/app/cost/ThroughputCost.java
@@ -29,7 +29,7 @@ public class ThroughputCost implements HasBrokerCost {
 
   @Override
   public BrokerCost brokerCost(ClusterInfo clusterInfo) {
-    var score = score(clusterInfo.allBeans());
+    var score = score(clusterInfo.clusterBean().all());
 
     var max = score.values().stream().mapToDouble(v -> v).max().orElse(1);
 
diff --git a/app/src/main/java/org/astraea/app/cost/broker/BrokerInputCost.java b/app/src/main/java/org/astraea/app/cost/broker/BrokerInputCost.java
index 9140a46d..6d120ad3 100644
--- a/app/src/main/java/org/astraea/app/cost/broker/BrokerInputCost.java
+++ b/app/src/main/java/org/astraea/app/cost/broker/BrokerInputCost.java
@@ -45,7 +45,7 @@ public class BrokerInputCost implements HasBrokerCost {
   @Override
   public BrokerCost brokerCost(ClusterInfo clusterInfo) {
     var costMetrics =
-        clusterInfo.allBeans().entrySet().stream()
+        clusterInfo.clusterBean().all().entrySet().stream()
             .collect(
                 Collectors.toMap(
                     Map.Entry::getKey,
diff --git a/app/src/main/java/org/astraea/app/cost/broker/BrokerOutputCost.java b/app/src/main/java/org/astraea/app/cost/broker/BrokerOutputCost.java
index 73e5c82f..a379a292 100644
--- a/app/src/main/java/org/astraea/app/cost/broker/BrokerOutputCost.java
+++ b/app/src/main/java/org/astraea/app/cost/broker/BrokerOutputCost.java
@@ -45,7 +45,7 @@ public class BrokerOutputCost implements HasBrokerCost {
   @Override
   public BrokerCost brokerCost(ClusterInfo clusterInfo) {
     var costMetrics =
-        clusterInfo.allBeans().entrySet().stream()
+        clusterInfo.clusterBean().all().entrySet().stream()
             .collect(
                 Collectors.toMap(
                     Map.Entry::getKey,
diff --git a/app/src/main/java/org/astraea/app/cost/broker/CpuCost.java b/app/src/main/java/org/astraea/app/cost/broker/CpuCost.java
index 123d7c20..de509ca5 100644
--- a/app/src/main/java/org/astraea/app/cost/broker/CpuCost.java
+++ b/app/src/main/java/org/astraea/app/cost/broker/CpuCost.java
@@ -45,7 +45,7 @@ public class CpuCost implements HasBrokerCost {
   @Override
   public BrokerCost brokerCost(ClusterInfo clusterInfo) {
     var costMetrics =
-        clusterInfo.allBeans().entrySet().stream()
+        clusterInfo.clusterBean().all().entrySet().stream()
             .collect(
                 Collectors.toMap(
                     Map.Entry::getKey,
diff --git a/app/src/main/java/org/astraea/app/cost/broker/MemoryCost.java b/app/src/main/java/org/astraea/app/cost/broker/MemoryCost.java
index 557659ce..90e530ff 100644
--- a/app/src/main/java/org/astraea/app/cost/broker/MemoryCost.java
+++ b/app/src/main/java/org/astraea/app/cost/broker/MemoryCost.java
@@ -46,7 +46,7 @@ public class MemoryCost extends Periodic<Map<Integer, Double>> implements HasBro
   @Override
   public BrokerCost brokerCost(ClusterInfo clusterInfo) {
     var costMetrics =
-        clusterInfo.allBeans().entrySet().stream()
+        clusterInfo.clusterBean().all().entrySet().stream()
             .collect(
                 Collectors.toMap(
                     Map.Entry::getKey,
diff --git a/app/src/main/java/org/astraea/app/cost/broker/NeutralIntegratedCost.java b/app/src/main/java/org/astraea/app/cost/broker/NeutralIntegratedCost.java
index ba972f49..eb851f88 100644
--- a/app/src/main/java/org/astraea/app/cost/broker/NeutralIntegratedCost.java
+++ b/app/src/main/java/org/astraea/app/cost/broker/NeutralIntegratedCost.java
@@ -46,7 +46,7 @@ public class NeutralIntegratedCost implements HasBrokerCost {
   @Override
   public BrokerCost brokerCost(ClusterInfo clusterInfo) {
     var costMetrics =
-        clusterInfo.allBeans().entrySet().stream()
+        clusterInfo.clusterBean().all().entrySet().stream()
             .collect(Collectors.toMap(Map.Entry::getKey, entry -> 0.0));
     costMetrics.forEach(
         (key, value) -> {


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
new file mode 100644
index 00000000..7265f779
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/admin/ClusterBeanTest.java
@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.app.admin;
+
+import java.util.List;
+import java.util.Map;
+import org.astraea.app.cost.ClusterInfo;
+import org.astraea.app.metrics.jmx.BeanObject;
+import org.astraea.app.metrics.kafka.HasValue;
+import org.astraea.app.metrics.kafka.KafkaMetrics;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+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);
+    // BeanObject1 and BeanObject2 is same partition in different broker
+    BeanObject testBeanObjectWithPartition1 =
+        new BeanObject(
+            "kafka.log",
+            Map.of(
+                "name",
+                KafkaMetrics.TopicPartition.Size.metricName(),
+                "type",
+                "Log",
+                "topic",
+                "testBeans",
+                "partition",
+                "0"),
+            Map.of("Value", 100));
+    BeanObject testBeanObjectWithPartition2 =
+        new BeanObject(
+            "kafka.log",
+            Map.of(
+                "name",
+                KafkaMetrics.TopicPartition.Size.metricName(),
+                "type",
+                "Log",
+                "topic",
+                "testBeans",
+                "partition",
+                "0"),
+            Map.of("Value", 100));
+    BeanObject testBeanObjectWithPartition3 =
+        new BeanObject(
+            "kafka.log",
+            Map.of(
+                "name",
+                KafkaMetrics.TopicPartition.LogEndOffset.name(),
+                "type",
+                "Log",
+                "topic",
+                "testBeans",
+                "partition",
+                "0"),
+            Map.of("Value", 100));
+    BeanObject testBeanObjectWithoutPartition =
+        new BeanObject(
+            "kafka.log",
+            Map.of(
+                "name",
+                KafkaMetrics.ReplicaManager.LeaderCount.metricName(),
+                "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)))));
+    // 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());
+
+    // 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());
+    // test get beanObject by replica
+    Assertions.assertEquals(2, clusterInfo.clusterBean().mapByReplica().size());
+    Assertions.assertEquals(
+        2,
+        clusterInfo
+            .clusterBean()
+            .mapByReplica()
+            .get(TopicPartitionReplica.of("testBeans", 0, 2))
+            .size());
+  }
+}
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 8380183b..e16730dc 100644
--- a/app/src/test/java/org/astraea/app/cost/ClusterInfoProvider.java
+++ b/app/src/test/java/org/astraea/app/cost/ClusterInfoProvider.java
@@ -16,16 +16,14 @@
  */
 package org.astraea.app.cost;
 
-import java.util.Collection;
 import java.util.List;
-import java.util.Map;
 import java.util.Random;
 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.TopicPartition;
-import org.astraea.app.metrics.HasBeanObject;
 
 public class ClusterInfoProvider {
 
@@ -127,12 +125,7 @@ public class ClusterInfoProvider {
       }
 
       @Override
-      public Collection<HasBeanObject> beans(int brokerId) {
-        throw new UnsupportedOperationException();
-      }
-
-      @Override
-      public Map<Integer, Collection<HasBeanObject>> allBeans() {
+      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 102cb206..741c6139 100644
--- a/app/src/test/java/org/astraea/app/cost/ClusterInfoTest.java
+++ b/app/src/test/java/org/astraea/app/cost/ClusterInfoTest.java
@@ -16,12 +16,9 @@
  */
 package org.astraea.app.cost;
 
-import java.util.Collection;
 import java.util.List;
-import java.util.Map;
 import java.util.NoSuchElementException;
 import org.apache.kafka.common.Cluster;
-import org.astraea.app.metrics.HasBeanObject;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
@@ -54,24 +51,6 @@ public class ClusterInfoTest {
         NodeInfo.of(node), clusterInfo.replicas(partition.topic()).get(0).nodeInfo());
   }
 
-  @Test
-  void testEmptyBeans() {
-    var clusterInfo = ClusterInfo.of(Mockito.mock(org.apache.kafka.common.Cluster.class));
-    Assertions.assertEquals(0, clusterInfo.allBeans().size());
-    Assertions.assertEquals(0, clusterInfo.beans(19).size());
-  }
-
-  @Test
-  void testBeans() {
-    var beans = Map.of(1, (Collection<HasBeanObject>) List.of(Mockito.mock(HasBeanObject.class)));
-    var origin = Mockito.mock(ClusterInfo.class);
-    Mockito.when(origin.allBeans()).thenReturn(Map.of());
-    var clusterInfo = ClusterInfo.of(origin, beans);
-    Assertions.assertEquals(1, clusterInfo.allBeans().size());
-    Assertions.assertEquals(0, clusterInfo.beans(19).size());
-    Assertions.assertEquals(1, clusterInfo.beans(1).size());
-  }
-
   @Test
   void testIllegalQuery() {
     var clusterInfo = ClusterInfo.of(Cluster.empty());
diff --git a/app/src/test/java/org/astraea/app/cost/FakeClusterInfo.java b/app/src/test/java/org/astraea/app/cost/FakeClusterInfo.java
index ea41dc37..e2adf30a 100644
--- a/app/src/test/java/org/astraea/app/cost/FakeClusterInfo.java
+++ b/app/src/test/java/org/astraea/app/cost/FakeClusterInfo.java
@@ -16,11 +16,10 @@
  */
 package org.astraea.app.cost;
 
-import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import org.astraea.app.metrics.HasBeanObject;
+import org.astraea.app.admin.ClusterBean;
 
 public class FakeClusterInfo implements ClusterInfo {
   @Override
@@ -54,12 +53,7 @@ public class FakeClusterInfo implements ClusterInfo {
   }
 
   @Override
-  public Collection<HasBeanObject> beans(int brokerId) {
-    return List.of();
-  }
-
-  @Override
-  public Map<Integer, Collection<HasBeanObject>> allBeans() {
-    return Map.of();
+  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 9ee6a923..923fba9c 100644
--- a/app/src/test/java/org/astraea/app/cost/LoadCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/LoadCostTest.java
@@ -19,6 +19,7 @@ package org.astraea.app.cost;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
+import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.jmx.BeanObject;
 import org.astraea.app.metrics.kafka.BrokerTopicMetricsResult;
@@ -31,7 +32,7 @@ public class LoadCostTest {
   @Test
   void testComputeLoad() {
     var loadCostFunction = new LoadCost();
-    var allBeans = exampleClusterInfo().allBeans();
+    var allBeans = exampleClusterInfo().clusterBean().all();
     var load = loadCostFunction.computeLoad(allBeans);
 
     Assertions.assertEquals(2, load.get(1));
@@ -59,8 +60,8 @@ public class LoadCostTest {
     Collection<HasBeanObject> broker3 = List.of(BytesInPerSec3, BytesOutPerSec3);
     return new FakeClusterInfo() {
       @Override
-      public Map<Integer, Collection<HasBeanObject>> allBeans() {
-        return Map.of(1, broker1, 2, broker2, 3, broker3);
+      public ClusterBean clusterBean() {
+        return ClusterBean.of(Map.of(1, broker1, 2, broker2, 3, broker3));
       }
     };
   }
diff --git a/app/src/test/java/org/astraea/app/cost/ThroughputCostTest.java b/app/src/test/java/org/astraea/app/cost/ThroughputCostTest.java
index 42263ebc..129336d6 100644
--- a/app/src/test/java/org/astraea/app/cost/ThroughputCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/ThroughputCostTest.java
@@ -19,6 +19,7 @@ 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.metrics.kafka.BrokerTopicMetricsResult;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
@@ -49,7 +50,7 @@ public class ThroughputCostTest {
 
     var cluster = Mockito.mock(ClusterInfo.class);
     Mockito.when(cluster.nodes()).thenReturn(List.of(node));
-    Mockito.when(cluster.allBeans()).thenReturn(Map.of());
+    Mockito.when(cluster.clusterBean()).thenReturn(ClusterBean.of(Map.of()));
     Mockito.when(cluster.topics()).thenReturn(Set.of("t"));
     Mockito.when(cluster.availableReplicas("t"))
         .thenReturn(List.of(ReplicaInfo.of("t", 0, node, true, true, false)));
diff --git a/app/src/test/java/org/astraea/app/cost/broker/BrokerInputCostTest.java b/app/src/test/java/org/astraea/app/cost/broker/BrokerInputCostTest.java
index 7304dca1..d2658964 100644
--- a/app/src/test/java/org/astraea/app/cost/broker/BrokerInputCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/broker/BrokerInputCostTest.java
@@ -19,6 +19,7 @@ package org.astraea.app.cost.broker;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
+import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.cost.ClusterInfo;
 import org.astraea.app.cost.FakeClusterInfo;
 import org.astraea.app.cost.Normalizer;
@@ -71,8 +72,8 @@ public class BrokerInputCostTest {
     Collection<HasBeanObject> broker3 = List.of(BytesInPerSec3);
     return new FakeClusterInfo() {
       @Override
-      public Map<Integer, Collection<HasBeanObject>> allBeans() {
-        return Map.of(1, broker1, 2, broker2, 3, broker3);
+      public ClusterBean clusterBean() {
+        return ClusterBean.of(Map.of(1, broker1, 2, broker2, 3, broker3));
       }
     };
   }
diff --git a/app/src/test/java/org/astraea/app/cost/broker/BrokerOutPutCostTest.java b/app/src/test/java/org/astraea/app/cost/broker/BrokerOutPutCostTest.java
index ef53ffaf..ebebae7c 100644
--- a/app/src/test/java/org/astraea/app/cost/broker/BrokerOutPutCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/broker/BrokerOutPutCostTest.java
@@ -19,6 +19,7 @@ package org.astraea.app.cost.broker;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
+import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.cost.ClusterInfo;
 import org.astraea.app.cost.FakeClusterInfo;
 import org.astraea.app.cost.Normalizer;
@@ -58,8 +59,8 @@ public class BrokerOutPutCostTest {
     Collection<HasBeanObject> broker3 = List.of(BytesInPerSec3);
     return new FakeClusterInfo() {
       @Override
-      public Map<Integer, Collection<HasBeanObject>> allBeans() {
-        return Map.of(1, broker1, 2, broker2, 3, broker3);
+      public ClusterBean clusterBean() {
+        return ClusterBean.of(Map.of(1, broker1, 2, broker2, 3, broker3));
       }
     };
   }
diff --git a/app/src/test/java/org/astraea/app/cost/broker/CpuCostTest.java b/app/src/test/java/org/astraea/app/cost/broker/CpuCostTest.java
index c8edd31e..4da05b53 100644
--- a/app/src/test/java/org/astraea/app/cost/broker/CpuCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/broker/CpuCostTest.java
@@ -20,6 +20,7 @@ import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.cost.ClusterInfo;
 import org.astraea.app.cost.FakeClusterInfo;
 import org.astraea.app.cost.NodeInfo;
@@ -44,8 +45,8 @@ public class CpuCostTest {
     ClusterInfo clusterInfo =
         new FakeClusterInfo() {
           @Override
-          public Map<Integer, Collection<HasBeanObject>> allBeans() {
-            return Map.of(1, broker1, 2, broker2, 3, broker3);
+          public ClusterBean clusterBean() {
+            return ClusterBean.of(Map.of(1, broker1, 2, broker2, 3, broker3));
           }
 
           @Override
@@ -79,8 +80,8 @@ public class CpuCostTest {
     ClusterInfo clusterInfo2 =
         new FakeClusterInfo() {
           @Override
-          public Map<Integer, Collection<HasBeanObject>> allBeans() {
-            return Map.of(1, broker12, 2, broker22, 3, broker32);
+          public ClusterBean clusterBean() {
+            return ClusterBean.of(Map.of(1, broker12, 2, broker22, 3, broker32));
           }
 
           @Override
diff --git a/app/src/test/java/org/astraea/app/cost/broker/MemoryCostTest.java b/app/src/test/java/org/astraea/app/cost/broker/MemoryCostTest.java
index 0521718c..a79148b0 100644
--- a/app/src/test/java/org/astraea/app/cost/broker/MemoryCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/broker/MemoryCostTest.java
@@ -21,6 +21,7 @@ import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.cost.ClusterInfo;
 import org.astraea.app.cost.FakeClusterInfo;
 import org.astraea.app.cost.NodeInfo;
@@ -45,8 +46,8 @@ public class MemoryCostTest {
     ClusterInfo clusterInfo =
         new FakeClusterInfo() {
           @Override
-          public Map<Integer, Collection<HasBeanObject>> allBeans() {
-            return Map.of(1, broker1, 2, broker2, 3, broker3);
+          public ClusterBean clusterBean() {
+            return ClusterBean.of(Map.of(1, broker1, 2, broker2, 3, broker3));
           }
 
           @Override
@@ -80,8 +81,8 @@ public class MemoryCostTest {
     ClusterInfo clusterInfo2 =
         new FakeClusterInfo() {
           @Override
-          public Map<Integer, Collection<HasBeanObject>> allBeans() {
-            return Map.of(1, broker12, 2, broker22, 3, broker32);
+          public ClusterBean clusterBean() {
+            return ClusterBean.of(Map.of(1, broker12, 2, broker22, 3, broker32));
           }
 
           @Override
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 e7e930ef..4f2730e2 100644
--- a/app/src/test/java/org/astraea/app/partitioner/StrictCostDispatcherTest.java
+++ b/app/src/test/java/org/astraea/app/partitioner/StrictCostDispatcherTest.java
@@ -22,6 +22,7 @@ import java.util.NoSuchElementException;
 import java.util.Set;
 import java.util.function.Function;
 import java.util.stream.Collectors;
+import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.cost.BrokerCost;
 import org.astraea.app.cost.ClusterInfo;
 import org.astraea.app.cost.HasBrokerCost;
@@ -108,7 +109,7 @@ public class StrictCostDispatcherTest {
           @Override
           public BrokerCost brokerCost(ClusterInfo clusterInfo) {
             var brokerCost =
-                clusterInfo.allBeans().keySet().stream()
+                clusterInfo.clusterBean().all().keySet().stream()
                     .collect(
                         Collectors.toMap(
                             Function.identity(), id -> id.equals(n0.id()) ? 0.9D : 0.5D));
@@ -125,7 +126,7 @@ public class StrictCostDispatcherTest {
           @Override
           public BrokerCost brokerCost(ClusterInfo clusterInfo) {
             var brokerCost =
-                clusterInfo.allBeans().keySet().stream()
+                clusterInfo.clusterBean().all().keySet().stream()
                     .collect(
                         Collectors.toMap(
                             Function.identity(), id -> id.equals(n0.id()) ? 0.6D : 0.8D));
@@ -155,6 +156,7 @@ public class StrictCostDispatcherTest {
       // there is no available partition
       Mockito.when(clusterInfo.availableReplicaLeaders("aa")).thenReturn(List.of());
       Mockito.when(clusterInfo.topics()).thenReturn(Set.of("aa"));
+      Mockito.when(clusterInfo.clusterBean()).thenReturn(ClusterBean.of(Map.of()));
       Assertions.assertEquals(0, dispatcher.partition("aa", new byte[0], new byte[0], clusterInfo));
 
       // there is only one available partition