Repository: skiptests/astraea
Pull request #530 (2022-07-31 08:16:46)

Refactor producer metrics
as title


Patch:
diff --git a/app/src/main/java/org/astraea/app/cost/NodeLatencyCost.java b/app/src/main/java/org/astraea/app/cost/NodeLatencyCost.java
index e3f7842b..976e0071 100644
--- a/app/src/main/java/org/astraea/app/cost/NodeLatencyCost.java
+++ b/app/src/main/java/org/astraea/app/cost/NodeLatencyCost.java
@@ -23,9 +23,9 @@ import java.util.Optional;
 import java.util.stream.Collectors;
 import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
-import org.astraea.app.metrics.KafkaMetrics;
 import org.astraea.app.metrics.collector.Fetcher;
 import org.astraea.app.metrics.producer.HasProducerNodeMetrics;
+import org.astraea.app.metrics.producer.ProducerMetrics;
 
 public class NodeLatencyCost implements HasBrokerCost {
 
@@ -56,6 +56,6 @@ public class NodeLatencyCost implements HasBrokerCost {
 
   @Override
   public Optional<Fetcher> fetcher() {
-    return Optional.of(KafkaMetrics.Producer::nodes);
+    return Optional.of(ProducerMetrics::nodes);
   }
 }
diff --git a/app/src/main/java/org/astraea/app/cost/NodeThroughputCost.java b/app/src/main/java/org/astraea/app/cost/NodeThroughputCost.java
index 0881900b..6561d33f 100644
--- a/app/src/main/java/org/astraea/app/cost/NodeThroughputCost.java
+++ b/app/src/main/java/org/astraea/app/cost/NodeThroughputCost.java
@@ -23,9 +23,9 @@ import java.util.Optional;
 import java.util.stream.Collectors;
 import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
-import org.astraea.app.metrics.KafkaMetrics;
 import org.astraea.app.metrics.collector.Fetcher;
 import org.astraea.app.metrics.producer.HasProducerNodeMetrics;
+import org.astraea.app.metrics.producer.ProducerMetrics;
 
 public class NodeThroughputCost implements HasBrokerCost {
 
@@ -56,6 +56,6 @@ public class NodeThroughputCost implements HasBrokerCost {
 
   @Override
   public Optional<Fetcher> fetcher() {
-    return Optional.of(KafkaMetrics.Producer::nodes);
+    return Optional.of(ProducerMetrics::nodes);
   }
 }
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 81e36646..a68b5406 100644
--- a/app/src/main/java/org/astraea/app/metrics/KafkaMetrics.java
+++ b/app/src/main/java/org/astraea/app/metrics/KafkaMetrics.java
@@ -18,12 +18,9 @@ package org.astraea.app.metrics;
 
 import java.util.Arrays;
 import java.util.Collection;
-import java.util.Map;
 import java.util.stream.Collectors;
 import org.astraea.app.metrics.broker.HasValue;
 import org.astraea.app.metrics.broker.TotalTimeMs;
-import org.astraea.app.metrics.producer.HasProducerNodeMetrics;
-import org.astraea.app.metrics.producer.HasProducerTopicMetrics;
 
 public final class KafkaMetrics {
 
@@ -84,78 +81,4 @@ public final class KafkaMetrics {
           .collect(Collectors.toUnmodifiableList());
     }
   }
-
-  public static final class Producer {
-
-    private static int brokerId(String node) {
-      return Integer.parseInt(node.substring(node.indexOf("-") + 1));
-    }
-
-    private Producer() {}
-
-    /**
-     * node metrics traced by producer
-     *
-     * @param mBeanClient to query beans
-     * @param brokerId broker ids
-     * @return key is client id used by producer, and value is node metrics traced by each producer
-     */
-    public static Map<String, HasProducerNodeMetrics> node(MBeanClient mBeanClient, int brokerId) {
-      return mBeanClient
-          .queryBeans(
-              BeanQuery.builder()
-                  .domainName("kafka.producer")
-                  .property("type", "producer-node-metrics")
-                  .property("node-id", "node-" + brokerId)
-                  .property("client-id", "*")
-                  .build())
-          .stream()
-          .collect(
-              Collectors.toUnmodifiableMap(
-                  b -> b.properties().get("client-id"),
-                  b -> HasProducerNodeMetrics.of(b, brokerId(b.properties().get("node-id")))));
-    }
-
-    /**
-     * collect HasProducerNodeMetrics from all producers.
-     *
-     * @param mBeanClient to query metrics
-     * @return key is broker id, and value is associated to broker metrics recorded by all producers
-     */
-    public static Collection<HasProducerNodeMetrics> nodes(MBeanClient mBeanClient) {
-      return mBeanClient
-          .queryBeans(
-              BeanQuery.builder()
-                  .domainName("kafka.producer")
-                  .property("type", "producer-node-metrics")
-                  .property("node-id", "*")
-                  .property("client-id", "*")
-                  .build())
-          .stream()
-          .map(b -> HasProducerNodeMetrics.of(b, brokerId(b.properties().get("node-id"))))
-          .collect(Collectors.toUnmodifiableList());
-    }
-
-    /**
-     * topic metrics traced by producer
-     *
-     * @param mBeanClient to query beans
-     * @param topic topic name
-     * @return key is client id used by producer, and value is topic metrics traced by each producer
-     */
-    public static Map<String, HasProducerTopicMetrics> topic(
-        MBeanClient mBeanClient, String topic) {
-      return mBeanClient
-          .queryBeans(
-              BeanQuery.builder()
-                  .domainName("kafka.producer")
-                  .property("type", "producer-topic-metrics")
-                  .property("client-id", "*")
-                  .property("topic", topic)
-                  .build())
-          .stream()
-          .collect(
-              Collectors.toUnmodifiableMap(b -> b.properties().get("client-id"), b -> () -> b));
-    }
-  }
 }
diff --git a/app/src/main/java/org/astraea/app/metrics/producer/ProducerMetrics.java b/app/src/main/java/org/astraea/app/metrics/producer/ProducerMetrics.java
new file mode 100644
index 00000000..c8ada5a6
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/metrics/producer/ProducerMetrics.java
@@ -0,0 +1,95 @@
+/*
+ * 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.producer;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.stream.Collectors;
+import org.astraea.app.metrics.BeanQuery;
+import org.astraea.app.metrics.MBeanClient;
+
+public final class ProducerMetrics {
+
+  private static int brokerId(String node) {
+    return Integer.parseInt(node.substring(node.indexOf("-") + 1));
+  }
+
+  /**
+   * node metrics traced by producer
+   *
+   * @param mBeanClient to query beans
+   * @param brokerId broker ids
+   * @return key is client id used by producer, and value is node metrics traced by each producer
+   */
+  public static Map<String, HasProducerNodeMetrics> node(MBeanClient mBeanClient, int brokerId) {
+    return mBeanClient
+        .queryBeans(
+            BeanQuery.builder()
+                .domainName("kafka.producer")
+                .property("type", "producer-node-metrics")
+                .property("node-id", "node-" + brokerId)
+                .property("client-id", "*")
+                .build())
+        .stream()
+        .collect(
+            Collectors.toUnmodifiableMap(
+                b -> b.properties().get("client-id"),
+                b -> HasProducerNodeMetrics.of(b, brokerId(b.properties().get("node-id")))));
+  }
+
+  /**
+   * collect HasProducerNodeMetrics from all producers.
+   *
+   * @param mBeanClient to query metrics
+   * @return key is broker id, and value is associated to broker metrics recorded by all producers
+   */
+  public static Collection<HasProducerNodeMetrics> nodes(MBeanClient mBeanClient) {
+    return mBeanClient
+        .queryBeans(
+            BeanQuery.builder()
+                .domainName("kafka.producer")
+                .property("type", "producer-node-metrics")
+                .property("node-id", "*")
+                .property("client-id", "*")
+                .build())
+        .stream()
+        .map(b -> HasProducerNodeMetrics.of(b, brokerId(b.properties().get("node-id"))))
+        .collect(Collectors.toUnmodifiableList());
+  }
+
+  /**
+   * topic metrics traced by producer
+   *
+   * @param mBeanClient to query beans
+   * @param topic topic name
+   * @return key is client id used by producer, and value is topic metrics traced by each producer
+   */
+  public static Map<String, HasProducerTopicMetrics> topic(MBeanClient mBeanClient, String topic) {
+    return mBeanClient
+        .queryBeans(
+            BeanQuery.builder()
+                .domainName("kafka.producer")
+                .property("type", "producer-topic-metrics")
+                .property("client-id", "*")
+                .property("topic", topic)
+                .build())
+        .stream()
+        .collect(Collectors.toUnmodifiableMap(b -> b.properties().get("client-id"), b -> () -> b));
+  }
+
+  private ProducerMetrics() {}
+}


Test patch:
diff --git a/app/src/test/java/org/astraea/app/cost/NodeLatencyCostTest.java b/app/src/test/java/org/astraea/app/cost/NodeLatencyCostTest.java
index f05b28a8..187598ab 100644
--- a/app/src/test/java/org/astraea/app/cost/NodeLatencyCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/NodeLatencyCostTest.java
@@ -26,9 +26,9 @@ import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.common.Utils;
 import org.astraea.app.metrics.BeanObject;
 import org.astraea.app.metrics.HasBeanObject;
-import org.astraea.app.metrics.KafkaMetrics;
 import org.astraea.app.metrics.MBeanClient;
 import org.astraea.app.metrics.producer.HasProducerNodeMetrics;
+import org.astraea.app.metrics.producer.ProducerMetrics;
 import org.astraea.app.producer.Producer;
 import org.astraea.app.service.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
@@ -71,7 +71,7 @@ public class NodeLatencyCostTest extends RequireBrokerCluster {
       producer.sender().topic(Utils.randomString(10)).value(new byte[100]).run();
       producer.flush();
 
-      var beans = KafkaMetrics.Producer.nodes(MBeanClient.local());
+      var beans = ProducerMetrics.nodes(MBeanClient.local());
       var clusterBean =
           ClusterBean.of(
               Map.of(
diff --git a/app/src/test/java/org/astraea/app/metrics/producer/HasProducerNodeMetricsTest.java b/app/src/test/java/org/astraea/app/metrics/producer/HasProducerNodeMetricsTest.java
index d888f5f6..9869b3df 100644
--- a/app/src/test/java/org/astraea/app/metrics/producer/HasProducerNodeMetricsTest.java
+++ b/app/src/test/java/org/astraea/app/metrics/producer/HasProducerNodeMetricsTest.java
@@ -23,7 +23,6 @@ import java.util.stream.Collectors;
 import org.astraea.app.admin.Admin;
 import org.astraea.app.admin.TopicPartition;
 import org.astraea.app.common.Utils;
-import org.astraea.app.metrics.KafkaMetrics;
 import org.astraea.app.metrics.MBeanClient;
 import org.astraea.app.producer.Producer;
 import org.astraea.app.service.RequireBrokerCluster;
@@ -41,7 +40,7 @@ public class HasProducerNodeMetricsTest extends RequireBrokerCluster {
       Utils.sleep(Duration.ofSeconds(3));
       var owner = admin.replicas(Set.of(topic)).get(TopicPartition.of(topic, 0)).get(0).broker();
       producer.sender().topic(topic).run().toCompletableFuture().get();
-      var metrics = KafkaMetrics.Producer.node(MBeanClient.local(), owner);
+      var metrics = ProducerMetrics.node(MBeanClient.local(), owner);
       Assertions.assertEquals(1, metrics.size());
       check(metrics.get("producer-1"));
     }
@@ -78,7 +77,7 @@ public class HasProducerNodeMetricsTest extends RequireBrokerCluster {
           .run()
           .toCompletableFuture()
           .get();
-      var metrics = KafkaMetrics.Producer.nodes(MBeanClient.local());
+      var metrics = ProducerMetrics.nodes(MBeanClient.local());
       Assertions.assertNotEquals(1, metrics.size());
       Assertions.assertTrue(
           metrics.stream()
diff --git a/app/src/test/java/org/astraea/app/metrics/producer/HasProducerTopicMetricsTest.java b/app/src/test/java/org/astraea/app/metrics/producer/HasProducerTopicMetricsTest.java
index df49dcd0..8e2c905b 100644
--- a/app/src/test/java/org/astraea/app/metrics/producer/HasProducerTopicMetricsTest.java
+++ b/app/src/test/java/org/astraea/app/metrics/producer/HasProducerTopicMetricsTest.java
@@ -18,7 +18,6 @@ package org.astraea.app.metrics.producer;
 
 import java.util.concurrent.ExecutionException;
 import org.astraea.app.common.Utils;
-import org.astraea.app.metrics.KafkaMetrics;
 import org.astraea.app.metrics.MBeanClient;
 import org.astraea.app.producer.Producer;
 import org.astraea.app.service.RequireSingleBrokerCluster;
@@ -32,7 +31,7 @@ public class HasProducerTopicMetricsTest extends RequireSingleBrokerCluster {
     var topic = Utils.randomString(10);
     try (var producer = Producer.of(bootstrapServers())) {
       producer.sender().topic(topic).run().toCompletableFuture().get();
-      var metrics = KafkaMetrics.Producer.topic(MBeanClient.local(), topic);
+      var metrics = ProducerMetrics.topic(MBeanClient.local(), topic);
       Assertions.assertEquals(1, metrics.size());
       var producerTopicMetrics = metrics.get("producer-1");
       Assertions.assertNotEquals(0D, producerTopicMetrics.byteRate());