Repository: skiptests/astraea
Pull request #532 (2022-07-31 11:34:17)

Refactor ReplicaManager metrics
as title

chia7712: failed test is traced by #533

Patch:
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 97a23666..f87b2b99 100644
--- a/app/src/main/java/org/astraea/app/cost/ReplicaLeaderCost.java
+++ b/app/src/main/java/org/astraea/app/cost/ReplicaLeaderCost.java
@@ -17,14 +17,14 @@
 package org.astraea.app.cost;
 
 import java.util.Comparator;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.stream.Collectors;
 import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.metrics.HasBeanObject;
-import org.astraea.app.metrics.KafkaMetrics;
-import org.astraea.app.metrics.broker.HasValue;
+import org.astraea.app.metrics.broker.ServerMetrics;
 import org.astraea.app.metrics.collector.Fetcher;
 
 /** more replica leaders -> higher cost */
@@ -40,22 +40,22 @@ public class ReplicaLeaderCost implements HasBrokerCost {
 
   Map<Integer, Integer> leaderCount(ClusterInfo ignored, ClusterBean clusterBean) {
     return clusterBean.all().entrySet().stream()
-        .flatMap(
-            e ->
-                e.getValue().stream()
-                    .filter(x -> x instanceof HasValue)
-                    .filter(x -> "LeaderCount".equals(x.beanObject().properties().get("name")))
-                    .filter(x -> "ReplicaManager".equals(x.beanObject().properties().get("type")))
-                    .sorted(Comparator.comparing(HasBeanObject::createdTimestamp).reversed())
-                    .map(x -> (HasValue) x)
-                    .limit(1)
-                    .map(e2 -> Map.entry(e.getKey(), (int) e2.value())))
-        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+        .collect(
+            Collectors.toMap(
+                Map.Entry::getKey,
+                e ->
+                    e.getValue().stream()
+                        .filter(x -> x instanceof ServerMetrics.ReplicaManager.Meter)
+                        .map(x -> (ServerMetrics.ReplicaManager.Meter) x)
+                        .sorted(Comparator.comparing(HasBeanObject::createdTimestamp).reversed())
+                        .limit(1)
+                        .mapToInt(v -> (int) v.value())
+                        .sum()));
   }
 
   @Override
   public Optional<Fetcher> fetcher() {
-    return Optional.of(KafkaMetrics.ReplicaManager.LeaderCount::fetch);
+    return Optional.of(c -> List.of(ServerMetrics.ReplicaManager.LEADER_COUNT.fetch(c)));
   }
 
   public static class NoMetrics extends ReplicaLeaderCost {
diff --git a/app/src/main/java/org/astraea/app/metrics/KafkaMetrics.java b/app/src/main/java/org/astraea/app/metrics/KafkaMetrics.java
deleted file mode 100644
index 0e9aaaf4..00000000
--- a/app/src/main/java/org/astraea/app/metrics/KafkaMetrics.java
+++ /dev/null
@@ -1,66 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.astraea.app.metrics;
-
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.stream.Collectors;
-import org.astraea.app.metrics.broker.HasValue;
-
-public final class KafkaMetrics {
-
-  private KafkaMetrics() {}
-
-  public enum ReplicaManager {
-    AtMinIsrPartitionCount("AtMinIsrPartitionCount"),
-    LeaderCount("LeaderCount"),
-    OfflineReplicaCount("OfflineReplicaCount"),
-    PartitionCount("PartitionCount"),
-    ReassigningPartitions("ReassigningPartitions"),
-    UnderMinIsrPartitionCount("UnderMinIsrPartitionCount"),
-    UnderReplicatedPartitions("UnderReplicatedPartition");
-    private final String metricName;
-
-    ReplicaManager(String name) {
-      this.metricName = name;
-    }
-
-    public String metricName() {
-      return metricName;
-    }
-
-    public static ReplicaManager of(String metricName) {
-      return Arrays.stream(ReplicaManager.values())
-          .filter(metric -> metric.metricName().equalsIgnoreCase(metricName))
-          .findFirst()
-          .orElseThrow(() -> new IllegalArgumentException("No such metric: " + metricName));
-    }
-
-    public Collection<HasBeanObject> fetch(MBeanClient mBeanClient) {
-      return mBeanClient
-          .queryBeans(
-              BeanQuery.builder()
-                  .domainName("kafka.server")
-                  .property("type", "ReplicaManager")
-                  .property("name", metricName)
-                  .build())
-          .stream()
-          .map(HasValue::of)
-          .collect(Collectors.toUnmodifiableList());
-    }
-  }
-}
diff --git a/app/src/main/java/org/astraea/app/metrics/broker/NetworkMetrics.java b/app/src/main/java/org/astraea/app/metrics/broker/NetworkMetrics.java
index 23165903..27980499 100644
--- a/app/src/main/java/org/astraea/app/metrics/broker/NetworkMetrics.java
+++ b/app/src/main/java/org/astraea/app/metrics/broker/NetworkMetrics.java
@@ -17,7 +17,6 @@
 package org.astraea.app.metrics.broker;
 
 import java.util.Arrays;
-import java.util.Map;
 import org.astraea.app.metrics.BeanObject;
 import org.astraea.app.metrics.BeanQuery;
 import org.astraea.app.metrics.MBeanClient;
@@ -129,15 +128,7 @@ public class NetworkMetrics {
 
       @Override
       public String toString() {
-        StringBuilder sb = new StringBuilder();
-        for (Map.Entry<String, Object> e : beanObject().attributes().entrySet()) {
-          sb.append(System.lineSeparator())
-              .append("  ")
-              .append(e.getKey())
-              .append("=")
-              .append(e.getValue());
-        }
-        return beanObject.properties().get("request") + " TotalTimeMs {" + sb + "}";
+        return beanObject().toString();
       }
 
       @Override
diff --git a/app/src/main/java/org/astraea/app/metrics/broker/ServerMetrics.java b/app/src/main/java/org/astraea/app/metrics/broker/ServerMetrics.java
index 305726d5..8c7b9279 100644
--- a/app/src/main/java/org/astraea/app/metrics/broker/ServerMetrics.java
+++ b/app/src/main/java/org/astraea/app/metrics/broker/ServerMetrics.java
@@ -18,7 +18,6 @@ package org.astraea.app.metrics.broker;
 
 import java.util.Arrays;
 import java.util.Collection;
-import java.util.Map;
 import java.util.Objects;
 import java.util.stream.Collectors;
 import org.astraea.app.metrics.BeanObject;
@@ -204,15 +203,70 @@ public final class ServerMetrics {
 
       @Override
       public String toString() {
-        StringBuilder sb = new StringBuilder();
-        for (Map.Entry<String, Object> e : beanObject().attributes().entrySet()) {
-          sb.append(System.lineSeparator())
-              .append("  ")
-              .append(e.getKey())
-              .append("=")
-              .append(e.getValue());
-        }
-        return beanObject().properties().get("name") + "{" + sb + "}";
+        return beanObject().toString();
+      }
+
+      @Override
+      public BeanObject beanObject() {
+        return beanObject;
+      }
+    }
+  }
+
+  public enum ReplicaManager {
+    AT_MIN_ISR_PARTITION_COUNT("AtMinIsrPartitionCount"),
+    LEADER_COUNT("LeaderCount"),
+    OFFLINE_REPLICA_COUNT("OfflineReplicaCount"),
+    PARTITION_COUNT("PartitionCount"),
+    REASSIGNING_PARTITIONS("ReassigningPartitions"),
+    UNDER_MIN_ISR_PARTITION_COUNT("UnderMinIsrPartitionCount"),
+    UNDER_REPLICATED_PARTITIONS("UnderReplicatedPartitions");
+    private final String metricName;
+
+    ReplicaManager(String name) {
+      this.metricName = name;
+    }
+
+    public String metricName() {
+      return metricName;
+    }
+
+    public static ReplicaManager of(String metricName) {
+      return Arrays.stream(ReplicaManager.values())
+          .filter(metric -> metric.metricName().equalsIgnoreCase(metricName))
+          .findFirst()
+          .orElseThrow(() -> new IllegalArgumentException("No such metric: " + metricName));
+    }
+
+    public Meter fetch(MBeanClient mBeanClient) {
+      return new Meter(
+          mBeanClient.queryBean(
+              BeanQuery.builder()
+                  .domainName("kafka.server")
+                  .property("type", "ReplicaManager")
+                  .property("name", metricName)
+                  .build()));
+    }
+
+    public static class Meter implements HasValue {
+
+      private final BeanObject beanObject;
+
+      public Meter(BeanObject beanObject) {
+        this.beanObject = Objects.requireNonNull(beanObject);
+      }
+
+      public String metricsName() {
+        return beanObject().properties().get("name");
+      }
+
+      public ReplicaManager type() {
+        return ReplicaManager.of(metricsName());
+      }
+
+      @Override
+      public String toString() {
+        return beanObject().toString();
       }
 
       @Override
diff --git a/app/src/main/java/org/astraea/app/metrics/platform/OperatingSystemInfo.java b/app/src/main/java/org/astraea/app/metrics/platform/OperatingSystemInfo.java
index 4e8848c9..554a72b9 100644
--- a/app/src/main/java/org/astraea/app/metrics/platform/OperatingSystemInfo.java
+++ b/app/src/main/java/org/astraea/app/metrics/platform/OperatingSystemInfo.java
@@ -16,7 +16,6 @@
  */
 package org.astraea.app.metrics.platform;
 
-import java.util.Map;
 import org.astraea.app.metrics.BeanObject;
 
 public class OperatingSystemInfo implements HasOperatingSystemInfo {
@@ -34,14 +33,6 @@ public class OperatingSystemInfo implements HasOperatingSystemInfo {
 
   @Override
   public String toString() {
-    StringBuilder sb = new StringBuilder();
-    for (Map.Entry<String, Object> e : beanObject().attributes().entrySet()) {
-      sb.append(System.lineSeparator())
-          .append("  ")
-          .append(e.getKey())
-          .append("=")
-          .append(e.getValue());
-    }
-    return "Operating System Info {" + sb + "}";
+    return beanObject().toString();
   }
 }


Test patch:
diff --git a/app/src/test/java/org/astraea/app/admin/ClusterBeanTest.java b/app/src/test/java/org/astraea/app/admin/ClusterBeanTest.java
index 0419b7e7..8477fb0c 100644
--- a/app/src/test/java/org/astraea/app/admin/ClusterBeanTest.java
+++ b/app/src/test/java/org/astraea/app/admin/ClusterBeanTest.java
@@ -19,9 +19,9 @@ package org.astraea.app.admin;
 import java.util.List;
 import java.util.Map;
 import org.astraea.app.metrics.BeanObject;
-import org.astraea.app.metrics.KafkaMetrics;
 import org.astraea.app.metrics.broker.HasValue;
 import org.astraea.app.metrics.broker.LogMetrics;
+import org.astraea.app.metrics.broker.ServerMetrics;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
@@ -76,7 +76,7 @@ class ClusterBeanTest {
             "kafka.log",
             Map.of(
                 "name",
-                KafkaMetrics.ReplicaManager.LeaderCount.metricName(),
+                ServerMetrics.ReplicaManager.LEADER_COUNT.metricName(),
                 "type",
                 "ReplicaManager"),
             Map.of("Value", 300));
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 c8930ec3..43bf6801 100644
--- a/app/src/test/java/org/astraea/app/cost/ReplicaLeaderCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/ReplicaLeaderCostTest.java
@@ -16,7 +16,6 @@
  */
 package org.astraea.app.cost;
 
-import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -26,8 +25,7 @@ import org.astraea.app.admin.NodeInfo;
 import org.astraea.app.admin.ReplicaInfo;
 import org.astraea.app.metrics.BeanObject;
 import org.astraea.app.metrics.HasBeanObject;
-import org.astraea.app.metrics.KafkaMetrics;
-import org.astraea.app.metrics.broker.HasValue;
+import org.astraea.app.metrics.broker.ServerMetrics;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
@@ -55,13 +53,13 @@ public class ReplicaLeaderCostTest {
   @Test
   void testWithMetrics() {
     var costFunction = new ReplicaLeaderCost();
-    var LeaderCount1 = mockResult(KafkaMetrics.ReplicaManager.LeaderCount.metricName(), 3);
-    var LeaderCount2 = mockResult(KafkaMetrics.ReplicaManager.LeaderCount.metricName(), 4);
-    var LeaderCount3 = mockResult(KafkaMetrics.ReplicaManager.LeaderCount.metricName(), 5);
+    var LeaderCount1 = mockResult(ServerMetrics.ReplicaManager.LEADER_COUNT.metricName(), 3);
+    var LeaderCount2 = mockResult(ServerMetrics.ReplicaManager.LEADER_COUNT.metricName(), 4);
+    var LeaderCount3 = mockResult(ServerMetrics.ReplicaManager.LEADER_COUNT.metricName(), 5);
 
-    Collection<HasBeanObject> broker1 = List.of(LeaderCount1);
-    Collection<HasBeanObject> broker2 = List.of(LeaderCount2);
-    Collection<HasBeanObject> broker3 = List.of(LeaderCount3);
+    var broker1 = List.of((HasBeanObject) LeaderCount1);
+    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 load = costFunction.brokerCost(ClusterInfo.EMPTY, clusterBean);
 
@@ -71,8 +69,8 @@ public class ReplicaLeaderCostTest {
     Assertions.assertEquals(5.0, load.value().get(3));
   }
 
-  private HasValue mockResult(String name, long count) {
-    var result = Mockito.mock(HasValue.class);
+  private ServerMetrics.ReplicaManager.Meter mockResult(String name, long count) {
+    var result = Mockito.mock(ServerMetrics.ReplicaManager.Meter.class);
     var bean = Mockito.mock(BeanObject.class);
     Mockito.when(result.beanObject()).thenReturn(bean);
     Mockito.when(bean.properties()).thenReturn(Map.of("name", name, "type", "ReplicaManager"));
diff --git a/app/src/test/java/org/astraea/app/metrics/KafkaMetricsTest.java b/app/src/test/java/org/astraea/app/metrics/KafkaMetricsTest.java
deleted file mode 100644
index 32ca19c5..00000000
--- a/app/src/test/java/org/astraea/app/metrics/KafkaMetricsTest.java
+++ /dev/null
@@ -1,76 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.astraea.app.metrics;
-
-import static org.junit.jupiter.api.Assertions.assertNotEquals;
-
-import java.io.IOException;
-import java.lang.management.ManagementFactory;
-import java.time.Duration;
-import javax.management.MBeanServer;
-import javax.management.remote.JMXConnectorServer;
-import javax.management.remote.JMXConnectorServerFactory;
-import javax.management.remote.JMXServiceURL;
-import org.astraea.app.admin.Admin;
-import org.astraea.app.common.Utils;
-import org.astraea.app.metrics.broker.LogMetrics;
-import org.astraea.app.service.RequireBrokerCluster;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.EnumSource;
-
-class KafkaMetricsTest extends RequireBrokerCluster {
-
-  private MBeanServer mBeanServer;
-  private JMXConnectorServer jmxServer;
-  private MBeanClient mBeanClient;
-
-  @BeforeEach
-  void setUp() throws IOException {
-    JMXServiceURL serviceURL = new JMXServiceURL("service:jmx:rmi://127.0.0.1");
-
-    mBeanServer = ManagementFactory.getPlatformMBeanServer();
-
-    jmxServer = JMXConnectorServerFactory.newJMXConnectorServer(serviceURL, null, mBeanServer);
-    jmxServer.start();
-
-    mBeanClient = MBeanClient.of(jmxServer.getAddress());
-  }
-
-  @AfterEach
-  void tearDown() throws Exception {
-    jmxServer.stop();
-    mBeanServer = null;
-    mBeanClient.close();
-  }
-
-  @ParameterizedTest()
-  @EnumSource(value = LogMetrics.Log.class)
-  void testTopicPartitionMetrics(LogMetrics.Log request) {
-    try (var admin = Admin.of(bootstrapServers())) {
-      // there are only 3 brokers, so 10 partitions can make each broker has some partitions
-      admin.creator().topic(Utils.randomString(5)).numberOfPartitions(10).create();
-    }
-
-    // wait for topic creation
-    Utils.sleep(Duration.ofSeconds(2));
-
-    var beans = request.fetch(mBeanClient);
-    assertNotEquals(0, beans.size());
-  }
-}
diff --git a/app/src/test/java/org/astraea/app/metrics/broker/LogMetricsTest.java b/app/src/test/java/org/astraea/app/metrics/broker/LogMetricsTest.java
index a5288fd6..e1263a3b 100644
--- a/app/src/test/java/org/astraea/app/metrics/broker/LogMetricsTest.java
+++ b/app/src/test/java/org/astraea/app/metrics/broker/LogMetricsTest.java
@@ -16,6 +16,8 @@
  */
 package org.astraea.app.metrics.broker;
 
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+
 import java.time.Duration;
 import java.util.List;
 import java.util.stream.Collectors;
@@ -69,4 +71,19 @@ public class LogMetricsTest extends RequireSingleBrokerCluster {
     Assertions.assertEquals(
         0, LogMetrics.Log.meters(List.of(other, target), LogMetrics.Log.SIZE).size());
   }
+
+  @ParameterizedTest()
+  @EnumSource(value = LogMetrics.Log.class)
+  void testTopicPartitionMetrics(LogMetrics.Log request) {
+    try (var admin = Admin.of(bootstrapServers())) {
+      // there are only 3 brokers, so 10 partitions can make each broker has some partitions
+      admin.creator().topic(Utils.randomString(5)).numberOfPartitions(10).create();
+    }
+
+    // wait for topic creation
+    Utils.sleep(Duration.ofSeconds(2));
+
+    var beans = request.fetch(MBeanClient.local());
+    assertNotEquals(0, beans.size());
+  }
 }
diff --git a/app/src/test/java/org/astraea/app/metrics/broker/ReplicaManagerMetricsTest.java b/app/src/test/java/org/astraea/app/metrics/broker/ReplicaManagerMetricsTest.java
new file mode 100644
index 00000000..acfa4714
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/metrics/broker/ReplicaManagerMetricsTest.java
@@ -0,0 +1,34 @@
+/*
+ * 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.metrics.broker;
+
+import org.astraea.app.metrics.MBeanClient;
+import org.astraea.app.service.RequireSingleBrokerCluster;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+public class ReplicaManagerMetricsTest extends RequireSingleBrokerCluster {
+
+  @ParameterizedTest
+  @EnumSource(ServerMetrics.ReplicaManager.class)
+  void testBrokerTopic(ServerMetrics.ReplicaManager rm) {
+    var meter = rm.fetch(MBeanClient.local());
+    Assertions.assertTrue(meter.value() >= 0);
+    Assertions.assertEquals(rm, meter.type());
+  }
+}