Repository: skiptests/astraea
Pull request #531 (2022-07-31 10:38:58)

Refactor network metrics
as title


Patch:
diff --git a/app/src/main/java/org/astraea/app/metrics/KafkaMetrics.java b/app/src/main/java/org/astraea/app/metrics/KafkaMetrics.java
index a68b5406..0e9aaaf4 100644
--- a/app/src/main/java/org/astraea/app/metrics/KafkaMetrics.java
+++ b/app/src/main/java/org/astraea/app/metrics/KafkaMetrics.java
@@ -20,29 +20,11 @@ import java.util.Arrays;
 import java.util.Collection;
 import java.util.stream.Collectors;
 import org.astraea.app.metrics.broker.HasValue;
-import org.astraea.app.metrics.broker.TotalTimeMs;
 
 public final class KafkaMetrics {
 
   private KafkaMetrics() {}
 
-  public enum Request {
-    Produce,
-    FetchConsumer,
-    FetchFollower;
-
-    public TotalTimeMs totalTimeMs(MBeanClient mBeanClient) {
-      return new TotalTimeMs(
-          mBeanClient.queryBean(
-              BeanQuery.builder()
-                  .domainName("kafka.network")
-                  .property("type", "RequestMetrics")
-                  .property("request", this.name())
-                  .property("name", "TotalTimeMs")
-                  .build()));
-    }
-  }
-
   public enum ReplicaManager {
     AtMinIsrPartitionCount("AtMinIsrPartitionCount"),
     LeaderCount("LeaderCount"),
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
new file mode 100644
index 00000000..23165903
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/metrics/broker/NetworkMetrics.java
@@ -0,0 +1,158 @@
+/*
+ * 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 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;
+
+public class NetworkMetrics {
+
+  public enum Request {
+    PRODUCE("Produce"),
+    FETCH("Fetch"),
+    LIST_OFFSETS("ListOffsets"),
+    METADATA("Metadata"),
+    LEADER_AND_ISR("LeaderAndIsr"),
+    STOP_REPLICA("StopReplica"),
+    UPDATE_METADATA("UpdateMetadata"),
+    CONTROLLED_SHUTDOWN("ControlledShutdown"),
+    OFFSET_COMMIT("OffsetCommit"),
+    OFFSET_FETCH("OffsetFetch"),
+    FIND_COORDINATOR("FindCoordinator"),
+    JOIN_GROUP("JoinGroup"),
+    HEARTBEAT("Heartbeat"),
+    LEAVE_GROUP("LeaveGroup"),
+    SYNC_GROUP("SyncGroup"),
+    DESCRIBE_GROUPS("DescribeGroups"),
+    LIST_GROUPS("ListGroups"),
+    SASL_HANDSHAKE("SaslHandshake"),
+    API_VERSIONS("ApiVersions"),
+    CREATE_TOPICS("CreateTopics"),
+    DELETE_TOPICS("DeleteTopics"),
+    DELETE_RECORDS("DeleteRecords"),
+    INIT_PRODUCER_ID("InitProducerId"),
+    OFFSET_FOR_LEADER_EPOCH("OffsetForLeaderEpoch"),
+    ADD_PARTITIONS_TO_TXN("AddPartitionsToTxn"),
+    ADD_OFFSETS_TO_TXN("AddOffsetsToTxn"),
+    END_TXN("EndTxn"),
+    WRITE_TXN_MARKERS("WriteTxnMarkers"),
+    TXN_OFFSET_COMMIT("TxnOffsetCommit"),
+    DESCRIBE_ACLS("DescribeAcls"),
+    CREATE_ACLS("CreateAcls"),
+    DELETE_ACLS("DeleteAcls"),
+    DESCRIBE_CONFIGS("DescribeConfigs"),
+    ALTER_CONFIGS("AlterConfigs"),
+    ALTER_REPLICA_LOG_DIRS("AlterReplicaLogDirs"),
+    DESCRIBE_LOG_DIRS("DescribeLogDirs"),
+    SASL_AUTHENTICATE("SaslAuthenticate"),
+    CREATE_PARTITIONS("CreatePartitions"),
+    CREATE_DELEGATION_TOKEN("CreateDelegationToken"),
+    RENEW_DELEGATION_TOKEN("RenewDelegationToken"),
+    EXPIRE_DELEGATION_TOKEN("ExpireDelegationToken"),
+    DESCRIBE_DELEGATION_TOKEN("DescribeDelegationToken"),
+    DELETE_GROUPS("DeleteGroups"),
+    ELECT_LEADERS("ElectLeaders"),
+    INCREMENTAL_ALTER_CONFIGS("IncrementalAlterConfigs"),
+    ALTER_PARTITION_REASSIGNMENTS("AlterPartitionReassignments"),
+    LIST_PARTITION_REASSIGNMENTS("ListPartitionReassignments"),
+    OFFSET_DELETE("OffsetDelete"),
+    DESCRIBE_CLIENT_QUOTAS("DescribeClientQuotas"),
+    ALTER_CLIENT_QUOTAS("AlterClientQuotas"),
+    DESCRIBE_USER_SCRAM_CREDENTIALS("DescribeUserScramCredentials"),
+    ALTER_USER_SCRAM_CREDENTIALS("AlterUserScramCredentials"),
+    //    VOTE("Vote"), available for CONTROLLER
+    //    BEGIN_QUORUM_EPOCH("BeginQuorumEpoch"), available for CONTROLLER
+    //    END_QUORUM_EPOCH("EndQuorumEpoch"), available for CONTROLLER
+    //    DESCRIBE_QUORUM("DescribeQuorum"), available for new BROKER and CONTROLLER
+    ALTER_PARTITION("AlterPartition"),
+    UPDATE_FEATURES("UpdateFeatures"),
+    //    ENVELOPE("Envelope"), available for CONTROLLER
+    //    FETCH_SNAPSHOT("FetchSnapshot"), available for CONTROLLER
+    DESCRIBE_CLUSTER("DescribeCluster"),
+    DESCRIBE_PRODUCERS("DescribeProducers"),
+    //    BROKER_REGISTRATION("BrokerRegistration"), available for CONTROLLER
+    //    BROKER_HEARTBEAT("BrokerHeartbeat"), available for CONTROLLER
+    //    UNREGISTER_BROKER("UnregisterBroker"), available for new BROKER and CONTROLLER
+    DESCRIBE_TRANSACTIONS("DescribeTransactions"),
+    LIST_TRANSACTIONS("ListTransactions"),
+    ALLOCATE_PRODUCER_IDS("AllocateProducerIds");
+
+    private final String metricName;
+
+    Request(String metricName) {
+      this.metricName = metricName;
+    }
+
+    public String metricName() {
+      return metricName;
+    }
+
+    public TotalTimeMs totalTimeMs(MBeanClient mBeanClient) {
+      return new TotalTimeMs(
+          mBeanClient.queryBean(
+              BeanQuery.builder()
+                  .domainName("kafka.network")
+                  .property("type", "RequestMetrics")
+                  .property("request", this.metricName())
+                  .property("name", "TotalTimeMs")
+                  .build()));
+    }
+
+    public static class TotalTimeMs implements HasPercentiles, HasCount, HasStatistics {
+
+      private final BeanObject beanObject;
+
+      public TotalTimeMs(BeanObject beanObject) {
+        this.beanObject = beanObject;
+      }
+
+      public Request type() {
+        return Request.of(beanObject.properties().get("request"));
+      }
+
+      @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 + "}";
+      }
+
+      @Override
+      public BeanObject beanObject() {
+        return beanObject;
+      }
+    }
+
+    public static Request of(String metricName) {
+      return Arrays.stream(Request.values())
+          .filter(metric -> metric.metricName().equalsIgnoreCase(metricName))
+          .findFirst()
+          .orElseThrow(() -> new IllegalArgumentException("No such metric: " + metricName));
+    }
+  }
+
+  private NetworkMetrics() {}
+}
diff --git a/app/src/main/java/org/astraea/app/metrics/broker/TotalTimeMs.java b/app/src/main/java/org/astraea/app/metrics/broker/TotalTimeMs.java
deleted file mode 100644
index 91d3d9f7..00000000
--- a/app/src/main/java/org/astraea/app/metrics/broker/TotalTimeMs.java
+++ /dev/null
@@ -1,47 +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.broker;
-
-import java.util.Map;
-import org.astraea.app.metrics.BeanObject;
-
-public class TotalTimeMs implements HasPercentiles, HasCount, HasStatistics {
-
-  private final BeanObject beanObject;
-
-  public TotalTimeMs(BeanObject beanObject) {
-    this.beanObject = beanObject;
-  }
-
-  @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 + "}";
-  }
-
-  @Override
-  public BeanObject beanObject() {
-    return beanObject;
-  }
-}


Test patch:
diff --git a/app/src/test/java/org/astraea/app/metrics/KafkaMetricsTest.java b/app/src/test/java/org/astraea/app/metrics/KafkaMetricsTest.java
index 726a5ba4..32ca19c5 100644
--- a/app/src/test/java/org/astraea/app/metrics/KafkaMetricsTest.java
+++ b/app/src/test/java/org/astraea/app/metrics/KafkaMetricsTest.java
@@ -16,7 +16,6 @@
  */
 package org.astraea.app.metrics;
 
-import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
 
 import java.io.IOException;
@@ -29,7 +28,6 @@ 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.metrics.broker.TotalTimeMs;
 import org.astraea.app.service.RequireBrokerCluster;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
@@ -61,26 +59,6 @@ class KafkaMetricsTest extends RequireBrokerCluster {
     mBeanClient.close();
   }
 
-  @ParameterizedTest()
-  @EnumSource(value = KafkaMetrics.Request.class)
-  void testRequestTotalTimeMs(KafkaMetrics.Request request) {
-    // act
-    TotalTimeMs totalTimeMs = request.totalTimeMs(mBeanClient);
-
-    // assert type casting correct and field exists
-    assertDoesNotThrow(totalTimeMs::percentile50);
-    assertDoesNotThrow(totalTimeMs::percentile75);
-    assertDoesNotThrow(totalTimeMs::percentile95);
-    assertDoesNotThrow(totalTimeMs::percentile98);
-    assertDoesNotThrow(totalTimeMs::percentile99);
-    assertDoesNotThrow(totalTimeMs::percentile999);
-    assertDoesNotThrow(totalTimeMs::count);
-    assertDoesNotThrow(totalTimeMs::max);
-    assertDoesNotThrow(totalTimeMs::mean);
-    assertDoesNotThrow(totalTimeMs::min);
-    assertDoesNotThrow(totalTimeMs::stdDev);
-  }
-
   @ParameterizedTest()
   @EnumSource(value = LogMetrics.Log.class)
   void testTopicPartitionMetrics(LogMetrics.Log request) {
diff --git a/app/src/test/java/org/astraea/app/metrics/broker/NetworkMetricsTest.java b/app/src/test/java/org/astraea/app/metrics/broker/NetworkMetricsTest.java
new file mode 100644
index 00000000..ad5202a0
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/metrics/broker/NetworkMetricsTest.java
@@ -0,0 +1,47 @@
+/*
+ * 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 NetworkMetricsTest extends RequireSingleBrokerCluster {
+
+  @ParameterizedTest()
+  @EnumSource(value = NetworkMetrics.Request.class)
+  void testRequestTotalTimeMs(NetworkMetrics.Request request) {
+    // act
+    var totalTimeMs = request.totalTimeMs(MBeanClient.local());
+
+    // assert type casting correct and field exists
+    Assertions.assertDoesNotThrow(totalTimeMs::percentile50);
+    Assertions.assertDoesNotThrow(totalTimeMs::percentile75);
+    Assertions.assertDoesNotThrow(totalTimeMs::percentile95);
+    Assertions.assertDoesNotThrow(totalTimeMs::percentile98);
+    Assertions.assertDoesNotThrow(totalTimeMs::percentile99);
+    Assertions.assertDoesNotThrow(totalTimeMs::percentile999);
+    Assertions.assertDoesNotThrow(totalTimeMs::count);
+    Assertions.assertDoesNotThrow(totalTimeMs::max);
+    Assertions.assertDoesNotThrow(totalTimeMs::mean);
+    Assertions.assertDoesNotThrow(totalTimeMs::min);
+    Assertions.assertDoesNotThrow(totalTimeMs::stdDev);
+    Assertions.assertEquals(request, totalTimeMs.type());
+  }
+}