Add Consumer Node Metrics

app/src/main/java/org/astraea/app/cost/NodeLatencyCost.java

@@ -23,9 +23,9 @@
 import java.util.stream.Collectors;
 import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
+import org.astraea.app.metrics.client.HasNodeMetrics;
+import org.astraea.app.metrics.client.producer.ProducerMetrics;
 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 {
 
@@ -34,10 +34,10 @@ public BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
     var result =
         clusterBean.all().values().stream()
             .flatMap(Collection::stream)
-            .filter(b -> b instanceof HasProducerNodeMetrics)
-            .map(b -> (HasProducerNodeMetrics) b)
+            .filter(b -> b instanceof HasNodeMetrics)
+            .map(b -> (HasNodeMetrics) b)
             .filter(b -> !Double.isNaN(b.requestLatencyAvg()))
-            .collect(Collectors.groupingBy(HasProducerNodeMetrics::brokerId))
+            .collect(Collectors.groupingBy(HasNodeMetrics::brokerId))
             .entrySet()
             .stream()
             .collect(
@@ -46,10 +46,9 @@ public BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
                     e ->
                         e.getValue().stream()
                             .sorted(
-                                Comparator.comparing(HasProducerNodeMetrics::createdTimestamp)
-                                    .reversed())
+                                Comparator.comparing(HasNodeMetrics::createdTimestamp).reversed())
                             .limit(1)
-                            .mapToDouble(HasProducerNodeMetrics::requestLatencyAvg)
+                            .mapToDouble(HasNodeMetrics::requestLatencyAvg)
                             .sum()));
     return () -> result;
   }

app/src/main/java/org/astraea/app/cost/NodeThroughputCost.java

@@ -23,9 +23,9 @@
 import java.util.stream.Collectors;
 import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
+import org.astraea.app.metrics.client.HasNodeMetrics;
+import org.astraea.app.metrics.client.producer.ProducerMetrics;
 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 {
 
@@ -34,10 +34,10 @@ public BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
     var result =
         clusterBean.all().values().stream()
             .flatMap(Collection::stream)
-            .filter(b -> b instanceof HasProducerNodeMetrics)
-            .map(b -> (HasProducerNodeMetrics) b)
+            .filter(b -> b instanceof HasNodeMetrics)
+            .map(b -> (HasNodeMetrics) b)
             .filter(b -> !Double.isNaN(b.incomingByteRate()) && !Double.isNaN(b.outgoingByteRate()))
-            .collect(Collectors.groupingBy(HasProducerNodeMetrics::brokerId))
+            .collect(Collectors.groupingBy(HasNodeMetrics::brokerId))
             .entrySet()
             .stream()
             .collect(
@@ -46,8 +46,7 @@ public BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
                     e ->
                         e.getValue().stream()
                             .sorted(
-                                Comparator.comparing(HasProducerNodeMetrics::createdTimestamp)
-                                    .reversed())
+                                Comparator.comparing(HasNodeMetrics::createdTimestamp).reversed())
                             .limit(1)
                             .mapToDouble(m -> m.incomingByteRate() + m.outgoingByteRate())
                             .sum()));

app/src/main/java/org/astraea/app/metrics/producer/HasProducerNodeMetrics.java → app/src/main/java/org/astraea/app/metrics/client/HasNodeMetrics.java

@@ -14,15 +14,15 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.astraea.app.metrics.producer;
+package org.astraea.app.metrics.client;
 
 import org.astraea.app.metrics.BeanObject;
 import org.astraea.app.metrics.HasBeanObject;
 
-public interface HasProducerNodeMetrics extends HasBeanObject {
+public interface HasNodeMetrics extends HasBeanObject {
 
-  static HasProducerNodeMetrics of(BeanObject beanObject, int brokerId) {
-    return new HasProducerNodeMetrics() {
+  static HasNodeMetrics of(BeanObject beanObject, int brokerId) {
+    return new HasNodeMetrics() {
       @Override
       public int brokerId() {
         return brokerId;

app/src/main/java/org/astraea/app/metrics/client/consumer/ConsumerMetrics.java

@@ -0,0 +1,73 @@
+/*
+ * 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.client.consumer;
+
+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;
+import org.astraea.app.metrics.client.HasNodeMetrics;
+
+public class ConsumerMetrics {
+  private static int brokerId(String node) {
+    return Integer.parseInt(node.substring(node.indexOf("-") + 1));
+  }
+
+  /**
+   * node metrics traced by consumer
+   *
+   * @param mBeanClient to query beans
+   * @param brokerId broker ids
+   * @return key is client id used by consumer, and value is node metrics traced by each consumer
+   */
+  public static Map<String, HasNodeMetrics> node(MBeanClient mBeanClient, int brokerId) {
+    return mBeanClient
+        .queryBeans(
+            BeanQuery.builder()
+                .domainName("kafka.consumer")
+                .property("type", "consumer-node-metrics")
+                .property("node-id", "node-" + brokerId)
+                .property("client-id", "*")
+                .build())
+        .stream()
+        .collect(
+            Collectors.toUnmodifiableMap(
+                b -> b.properties().get("client-id"),
+                b -> HasNodeMetrics.of(b, brokerId(b.properties().get("node-id")))));
+  }
+
+  /**
+   * collect HasNodeMetrics from all consumers.
+   *
+   * @param mBeanClient to query metrics
+   * @return key is broker id, and value is associated to broker metrics recorded by all consumers
+   */
+  public static Collection<HasNodeMetrics> nodes(MBeanClient mBeanClient) {
+    return mBeanClient
+        .queryBeans(
+            BeanQuery.builder()
+                .domainName("kafka.consumer")
+                .property("type", "consumer-node-metrics")
+                .property("node-id", "*")
+                .property("client-id", "*")
+                .build())
+        .stream()
+        .map(b -> HasNodeMetrics.of(b, brokerId(b.properties().get("node-id"))))
+        .collect(Collectors.toUnmodifiableList());
+  }
+}

app/src/main/java/org/astraea/app/metrics/producer/HasProducerTopicMetrics.java → app/src/main/java/org/astraea/app/metrics/client/producer/HasProducerTopicMetrics.java

@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.astraea.app.metrics.producer;
+package org.astraea.app.metrics.client.producer;
 
 import org.astraea.app.metrics.HasBeanObject;
 

app/src/main/java/org/astraea/app/metrics/producer/ProducerMetrics.java → app/src/main/java/org/astraea/app/metrics/client/producer/ProducerMetrics.java

@@ -14,13 +14,14 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.astraea.app.metrics.producer;
+package org.astraea.app.metrics.client.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;
+import org.astraea.app.metrics.client.HasNodeMetrics;
 
 public final class ProducerMetrics {
 
@@ -35,7 +36,7 @@ private static int brokerId(String node) {
    * @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) {
+  public static Map<String, HasNodeMetrics> node(MBeanClient mBeanClient, int brokerId) {
     return mBeanClient
         .queryBeans(
             BeanQuery.builder()
@@ -48,7 +49,7 @@ public static Map<String, HasProducerNodeMetrics> node(MBeanClient mBeanClient,
         .collect(
             Collectors.toUnmodifiableMap(
                 b -> b.properties().get("client-id"),
-                b -> HasProducerNodeMetrics.of(b, brokerId(b.properties().get("node-id")))));
+                b -> HasNodeMetrics.of(b, brokerId(b.properties().get("node-id")))));
   }
 
   /**
@@ -57,7 +58,7 @@ public static Map<String, HasProducerNodeMetrics> node(MBeanClient mBeanClient,
    * @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) {
+  public static Collection<HasNodeMetrics> nodes(MBeanClient mBeanClient) {
     return mBeanClient
         .queryBeans(
             BeanQuery.builder()
@@ -67,7 +68,7 @@ public static Collection<HasProducerNodeMetrics> nodes(MBeanClient mBeanClient)
                 .property("client-id", "*")
                 .build())
         .stream()
-        .map(b -> HasProducerNodeMetrics.of(b, brokerId(b.properties().get("node-id"))))
+        .map(b -> HasNodeMetrics.of(b, brokerId(b.properties().get("node-id"))))
         .collect(Collectors.toUnmodifiableList());
   }
 

app/src/test/java/org/astraea/app/cost/NodeLatencyCostTest.java

@@ -27,8 +27,8 @@
 import org.astraea.app.metrics.BeanObject;
 import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.MBeanClient;
-import org.astraea.app.metrics.producer.HasProducerNodeMetrics;
-import org.astraea.app.metrics.producer.ProducerMetrics;
+import org.astraea.app.metrics.client.HasNodeMetrics;
+import org.astraea.app.metrics.client.producer.ProducerMetrics;
 import org.astraea.app.producer.Producer;
 import org.astraea.app.service.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
@@ -39,7 +39,7 @@ public class NodeLatencyCostTest extends RequireBrokerCluster {
 
   @Test
   void testNan() {
-    var bean = Mockito.mock(HasProducerNodeMetrics.class);
+    var bean = Mockito.mock(HasNodeMetrics.class);
     Mockito.when(bean.brokerId()).thenReturn(1);
     Mockito.when(bean.requestLatencyAvg()).thenReturn(Double.NaN);
     var clusterBean = ClusterBean.of(Map.of(-1, List.of(bean)));
@@ -50,7 +50,7 @@ void testNan() {
 
   @Test
   void testBrokerId() {
-    var bean = Mockito.mock(HasProducerNodeMetrics.class);
+    var bean = Mockito.mock(HasNodeMetrics.class);
     Mockito.when(bean.brokerId()).thenReturn(1);
     Mockito.when(bean.requestLatencyAvg()).thenReturn(10D);
     var clusterBean = ClusterBean.of(Map.of(-1, List.of(bean)));

app/src/test/java/org/astraea/app/cost/NodeThroughputCostTest.java

@@ -22,7 +22,7 @@
 import org.astraea.app.admin.ClusterInfo;
 import org.astraea.app.metrics.BeanObject;
 import org.astraea.app.metrics.MBeanClient;
-import org.astraea.app.metrics.producer.HasProducerNodeMetrics;
+import org.astraea.app.metrics.client.HasNodeMetrics;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
@@ -31,7 +31,7 @@ public class NodeThroughputCostTest {
 
   @Test
   void testNan() {
-    var bean = Mockito.mock(HasProducerNodeMetrics.class);
+    var bean = Mockito.mock(HasNodeMetrics.class);
     Mockito.when(bean.brokerId()).thenReturn(1);
     Mockito.when(bean.incomingByteRate()).thenReturn(Double.NaN);
     Mockito.when(bean.outgoingByteRate()).thenReturn(Double.NaN);
@@ -43,7 +43,7 @@ void testNan() {
 
   @Test
   void testBrokerId() {
-    var bean = Mockito.mock(HasProducerNodeMetrics.class);
+    var bean = Mockito.mock(HasNodeMetrics.class);
     Mockito.when(bean.brokerId()).thenReturn(1);
     Mockito.when(bean.incomingByteRate()).thenReturn(10D);
     Mockito.when(bean.outgoingByteRate()).thenReturn(10D);
@@ -57,11 +57,11 @@ void testBrokerId() {
   @Test
   void testCost() {
     var throughputCost = new NodeThroughputCost();
-    var bean0 = Mockito.mock(HasProducerNodeMetrics.class);
+    var bean0 = Mockito.mock(HasNodeMetrics.class);
     Mockito.when(bean0.incomingByteRate()).thenReturn(10D);
     Mockito.when(bean0.outgoingByteRate()).thenReturn(20D);
     Mockito.when(bean0.brokerId()).thenReturn(10);
-    var bean1 = Mockito.mock(HasProducerNodeMetrics.class);
+    var bean1 = Mockito.mock(HasNodeMetrics.class);
     Mockito.when(bean1.incomingByteRate()).thenReturn(2D);
     Mockito.when(bean1.outgoingByteRate()).thenReturn(3D);
     Mockito.when(bean1.brokerId()).thenReturn(11);

app/src/test/java/org/astraea/app/metrics/client/consumer/ConsumerMetricsTest.java

@@ -0,0 +1,83 @@
+/*
+ * 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.client.consumer;
+
+import java.time.Duration;
+import java.util.Set;
+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.consumer.Consumer;
+import org.astraea.app.metrics.MBeanClient;
+import org.astraea.app.metrics.client.HasNodeMetrics;
+import org.astraea.app.service.RequireBrokerCluster;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class ConsumerMetricsTest extends RequireBrokerCluster {
+  @Test
+  void testSingleBroker() {
+    var topic = Utils.randomString(10);
+    try (var admin = Admin.of(bootstrapServers());
+        var consumer =
+            Consumer.forTopics(Set.of(topic)).bootstrapServers(bootstrapServers()).build()) {
+      admin.creator().topic(topic).numberOfPartitions(1).create();
+      Utils.sleep(Duration.ofSeconds(3));
+      var owner = admin.replicas(Set.of(topic)).get(TopicPartition.of(topic, 0)).get(0).broker();
+      consumer.poll(Duration.ofSeconds(5));
+      var metrics = ConsumerMetrics.node(MBeanClient.local(), owner);
+      Assertions.assertEquals(1, metrics.size());
+      check(metrics.values().stream().findAny().get());
+    }
+  }
+
+  @Test
+  void testMultiBrokers() {
+    var topic = Utils.randomString(10);
+    try (var admin = Admin.of(bootstrapServers());
+        var consumer =
+            Consumer.forTopics(Set.of(topic)).bootstrapServers(bootstrapServers()).build()) {
+      admin.creator().topic(topic).numberOfPartitions(3).create();
+      Utils.sleep(Duration.ofSeconds(3));
+      consumer.poll(Duration.ofSeconds(5));
+      var metrics = ConsumerMetrics.nodes(MBeanClient.local());
+      Assertions.assertNotEquals(1, metrics.size());
+      Assertions.assertTrue(
+          metrics.stream()
+              .map(HasNodeMetrics::brokerId)
+              .collect(Collectors.toUnmodifiableList())
+              .containsAll(brokerIds()));
+      metrics.forEach(ConsumerMetricsTest::check);
+    }
+  }
+
+  private static void check(HasNodeMetrics metrics) {
+    Assertions.assertNotEquals(0D, metrics.incomingByteRate());
+    Assertions.assertNotEquals(0D, metrics.incomingByteTotal());
+    Assertions.assertNotEquals(0D, metrics.outgoingByteRate());
+    Assertions.assertNotEquals(0D, metrics.outgoingByteTotal());
+    Assertions.assertEquals(Double.NaN, metrics.requestLatencyAvg());
+    Assertions.assertEquals(Double.NaN, metrics.requestLatencyMax());
+    Assertions.assertNotEquals(0D, metrics.requestRate());
+    Assertions.assertNotEquals(0D, metrics.requestSizeAvg());
+    Assertions.assertNotEquals(0D, metrics.requestSizeMax());
+    Assertions.assertNotEquals(0D, metrics.requestTotal());
+    Assertions.assertNotEquals(0D, metrics.responseRate());
+    Assertions.assertNotEquals(0D, metrics.responseTotal());
+  }
+}

app/src/test/java/org/astraea/app/metrics/producer/HasProducerTopicMetricsTest.java → app/src/test/java/org/astraea/app/metrics/client/producer/HasProducerTopicMetricsTest.java

@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.astraea.app.metrics.producer;
+package org.astraea.app.metrics.client.producer;
 
 import java.util.concurrent.ExecutionException;
 import org.astraea.app.common.Utils;