Repository: skiptests/astraea
Pull request #506 (2022-07-25 11:45:15)

Simplify node size cost
1. rename `ReplicaSizeCost.java` to `NodeTopicSizeCost`
2. 移除`HasPartitionCost` from `NodeTopicSizeCost`
3. 移除normalization from `NodeTopicSizeCost`


Patch:
diff --git a/app/src/main/java/org/astraea/app/cost/NodeTopicSizeCost.java b/app/src/main/java/org/astraea/app/cost/NodeTopicSizeCost.java
new file mode 100644
index 00000000..dab447f5
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/cost/NodeTopicSizeCost.java
@@ -0,0 +1,51 @@
+/*
+ * 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.cost;
+
+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.broker.LogMetrics;
+import org.astraea.app.metrics.collector.Fetcher;
+
+public class NodeTopicSizeCost implements HasBrokerCost {
+
+  @Override
+  public Optional<Fetcher> fetcher() {
+    return Optional.of(LogMetrics.Log.SIZE::fetch);
+  }
+
+  /**
+   * @param clusterInfo the clusterInfo that offers the metrics related to topic/partition size
+   * @return a BrokerCost contains the used space for each broker
+   */
+  @Override
+  public BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
+    var result =
+        clusterBean.all().entrySet().stream()
+            .collect(
+                Collectors.toMap(
+                    Map.Entry::getKey,
+                    e ->
+                        LogMetrics.Log.meters(e.getValue(), LogMetrics.Log.SIZE).stream()
+                            .mapToDouble(LogMetrics.Log.Meter::value)
+                            .sum()));
+    return () -> result;
+  }
+}
diff --git a/app/src/main/java/org/astraea/app/cost/ReplicaSizeCost.java b/app/src/main/java/org/astraea/app/cost/ReplicaSizeCost.java
deleted file mode 100644
index 58f391c5..00000000
--- a/app/src/main/java/org/astraea/app/cost/ReplicaSizeCost.java
+++ /dev/null
@@ -1,183 +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.cost;
-
-import java.util.Comparator;
-import java.util.Map;
-import java.util.Optional;
-import java.util.TreeMap;
-import java.util.stream.Collectors;
-import org.astraea.app.admin.ClusterBean;
-import org.astraea.app.admin.ClusterInfo;
-import org.astraea.app.admin.NodeInfo;
-import org.astraea.app.admin.ReplicaInfo;
-import org.astraea.app.admin.TopicPartition;
-import org.astraea.app.admin.TopicPartitionReplica;
-import org.astraea.app.metrics.broker.HasValue;
-import org.astraea.app.metrics.broker.LogMetrics;
-import org.astraea.app.metrics.collector.Fetcher;
-
-/**
- * The result is computed by "Size.Value". "Size.Value" responds to the replica log size of brokers.
- * The calculation method of the score is the replica log usage space divided by the available space
- * on the hard disk
- */
-public class ReplicaSizeCost implements HasBrokerCost, HasPartitionCost {
-  Map<Integer, Integer> totalBrokerCapacity;
-
-  public ReplicaSizeCost(Map<Integer, Integer> totalBrokerCapacity) {
-    this.totalBrokerCapacity = totalBrokerCapacity;
-  }
-
-  @Override
-  public Optional<Fetcher> fetcher() {
-    return Optional.of(LogMetrics.Log.SIZE::fetch);
-  }
-
-  /**
-   * @param clusterInfo the clusterInfo that offers the metrics related to topic/partition size
-   * @return a BrokerCost contains the ratio of the used space to the free space of each broker
-   */
-  @Override
-  public BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
-    var sizeOfReplica = getReplicaSize(clusterBean);
-    var totalReplicaSizeInBroker =
-        clusterInfo.nodes().stream()
-            .collect(
-                Collectors.toMap(
-                    NodeInfo::id,
-                    y ->
-                        sizeOfReplica.entrySet().stream()
-                            .filter(tpr -> tpr.getKey().brokerId() == y.id())
-                            .mapToLong(Map.Entry::getValue)
-                            .sum()));
-    var brokerSizeScore =
-        totalReplicaSizeInBroker.entrySet().stream()
-            .collect(
-                Collectors.toMap(
-                    Map.Entry::getKey,
-                    y ->
-                        Double.valueOf(y.getValue())
-                            / totalBrokerCapacity.get(y.getKey())
-                            / 1048576));
-    return () -> brokerSizeScore;
-  }
-
-  /**
-   * @param clusterInfo the clusterInfo that offers the metrics related to topic/partition size
-   * @return a BrokerCost contains the ratio of the used space to the available space of replicas in
-   *     each broker
-   */
-  @Override
-  public PartitionCost partitionCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
-    final long ONEMEGA = Math.round(Math.pow(2, 20));
-    var sizeOfReplica = getReplicaSize(clusterBean);
-    TreeMap<TopicPartitionReplica, Double> replicaCost =
-        new TreeMap<>(
-            Comparator.comparing(TopicPartitionReplica::brokerId)
-                .thenComparing(TopicPartitionReplica::topic)
-                .thenComparing(TopicPartitionReplica::partition));
-    sizeOfReplica.forEach(
-        (tpr, size) ->
-            replicaCost.put(
-                tpr, (double) size / totalBrokerCapacity.get(tpr.brokerId()) / ONEMEGA));
-
-    var scoreForTopic =
-        clusterInfo.topics().stream()
-            .map(
-                topic ->
-                    Map.entry(
-                        topic,
-                        clusterInfo.replicas(topic).stream()
-                            .filter(ReplicaInfo::isLeader)
-                            .map(
-                                partitionInfo ->
-                                    TopicPartitionReplica.of(
-                                        partitionInfo.topic(),
-                                        partitionInfo.partition(),
-                                        partitionInfo.nodeInfo().id()))
-                            .map(
-                                tpr -> {
-                                  final var score =
-                                      replicaCost.entrySet().stream()
-                                          .filter(
-                                              x ->
-                                                  x.getKey().topic().equals(tpr.topic())
-                                                      && (x.getKey().partition()
-                                                          == tpr.partition()))
-                                          .mapToDouble(Map.Entry::getValue)
-                                          .max()
-                                          .orElseThrow(
-                                              () ->
-                                                  new IllegalStateException(
-                                                      tpr + " topic/partition size not found"));
-                                  return Map.entry(
-                                      TopicPartition.of(tpr.topic(), tpr.partition()), score);
-                                })
-                            .collect(
-                                Collectors.toUnmodifiableMap(
-                                    Map.Entry::getKey, Map.Entry::getValue))))
-            .collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, Map.Entry::getValue));
-
-    var scoreForBroker =
-        clusterInfo.nodes().stream()
-            .map(
-                node ->
-                    Map.entry(
-                        node.id(),
-                        replicaCost.entrySet().stream()
-                            .filter((tprScore) -> tprScore.getKey().brokerId() == node.id())
-                            .collect(
-                                Collectors.toMap(
-                                    x ->
-                                        TopicPartition.of(
-                                            x.getKey().topic(), x.getKey().partition()),
-                                    Map.Entry::getValue))))
-            .collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, Map.Entry::getValue));
-
-    return new PartitionCost() {
-
-      @Override
-      public Map<TopicPartition, Double> value(String topic) {
-        return scoreForTopic.get(topic);
-      }
-
-      @Override
-      public Map<TopicPartition, Double> value(int brokerId) {
-        return scoreForBroker.get(brokerId);
-      }
-    };
-  }
-
-  /**
-   * @param clusterBean offers the metrics related to topic/partition size
-   * @return a map contain the replica log size of each topic/partition
-   */
-  public Map<TopicPartitionReplica, Long> getReplicaSize(ClusterBean clusterBean) {
-    return clusterBean.mapByReplica().entrySet().stream()
-        .flatMap(
-            e ->
-                e.getValue().stream()
-                    .filter(x -> x instanceof HasValue)
-                    .filter(x -> x.beanObject().domainName().equals("kafka.log"))
-                    .filter(x -> x.beanObject().properties().get("type").equals("Log"))
-                    .filter(x -> x.beanObject().properties().get("name").equals("Size"))
-                    .map(x -> (HasValue) x)
-                    .map(x -> Map.entry(e.getKey(), x.value())))
-        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
-  }
-}
diff --git a/app/src/main/java/org/astraea/app/metrics/broker/LogMetrics.java b/app/src/main/java/org/astraea/app/metrics/broker/LogMetrics.java
index ca03157f..3f24a5a3 100644
--- a/app/src/main/java/org/astraea/app/metrics/broker/LogMetrics.java
+++ b/app/src/main/java/org/astraea/app/metrics/broker/LogMetrics.java
@@ -17,8 +17,10 @@
 package org.astraea.app.metrics.broker;
 
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.List;
 import java.util.stream.Collectors;
+import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.jmx.BeanObject;
 import org.astraea.app.metrics.jmx.BeanQuery;
 import org.astraea.app.metrics.jmx.MBeanClient;
@@ -47,6 +49,14 @@ public final class LogMetrics {
           .orElseThrow(() -> new IllegalArgumentException("No such metric: " + metricName));
     }
 
+    public static Collection<Meter> meters(Collection<HasBeanObject> beans, Log type) {
+      return beans.stream()
+          .filter(m -> m instanceof Meter)
+          .map(m -> (Meter) m)
+          .filter(m -> m.type() == type)
+          .collect(Collectors.toUnmodifiableList());
+    }
+
     public List<Meter> fetch(MBeanClient mBeanClient) {
       return mBeanClient
           .queryBeans(
@@ -62,10 +72,10 @@ public final class LogMetrics {
           .collect(Collectors.toUnmodifiableList());
     }
 
-    static class Meter implements HasValue {
+    public static class Meter implements HasValue {
       private final BeanObject beanObject;
 
-      private Meter(BeanObject beanObject) {
+      public Meter(BeanObject beanObject) {
         this.beanObject = beanObject;
       }
 


Test patch:
diff --git a/app/src/test/java/org/astraea/app/cost/NodeTopicSizeCostTest.java b/app/src/test/java/org/astraea/app/cost/NodeTopicSizeCostTest.java
new file mode 100644
index 00000000..f24b81ba
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/cost/NodeTopicSizeCostTest.java
@@ -0,0 +1,45 @@
+/*
+ * 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.cost;
+
+import java.util.List;
+import java.util.Map;
+import org.astraea.app.admin.ClusterBean;
+import org.astraea.app.admin.ClusterInfo;
+import org.astraea.app.metrics.broker.LogMetrics;
+import org.astraea.app.metrics.jmx.BeanObject;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+class NodeTopicSizeCostTest {
+
+  @Test
+  void testCost() {
+    var bean =
+        new BeanObject(
+            "domain",
+            Map.of("topic", "t", "partition", "10", "name", "SIZE"),
+            Map.of("Value", 777));
+    var meter = new LogMetrics.Log.Meter(bean);
+    var cost = new NodeTopicSizeCost();
+    var result =
+        cost.brokerCost(Mockito.mock(ClusterInfo.class), ClusterBean.of(Map.of(1, List.of(meter))));
+    Assertions.assertEquals(1, result.value().size());
+    Assertions.assertEquals(777, result.value().entrySet().iterator().next().getValue());
+  }
+}
diff --git a/app/src/test/java/org/astraea/app/cost/ReplicaSizeCostTest.java b/app/src/test/java/org/astraea/app/cost/ReplicaSizeCostTest.java
deleted file mode 100644
index 34e0c117..00000000
--- a/app/src/test/java/org/astraea/app/cost/ReplicaSizeCostTest.java
+++ /dev/null
@@ -1,162 +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.cost;
-
-import java.time.Duration;
-import java.util.Collection;
-import java.util.Comparator;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.ExecutionException;
-import java.util.stream.Collectors;
-import org.astraea.app.admin.Admin;
-import org.astraea.app.admin.ClusterBean;
-import org.astraea.app.admin.ClusterInfo;
-import org.astraea.app.admin.NodeInfo;
-import org.astraea.app.admin.ReplicaInfo;
-import org.astraea.app.admin.TopicPartition;
-import org.astraea.app.common.Utils;
-import org.astraea.app.metrics.HasBeanObject;
-import org.astraea.app.metrics.broker.HasValue;
-import org.astraea.app.metrics.broker.LogMetrics;
-import org.astraea.app.metrics.collector.BeanCollector;
-import org.astraea.app.metrics.jmx.BeanObject;
-import org.astraea.app.producer.Producer;
-import org.astraea.app.service.RequireSingleBrokerCluster;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Test;
-import org.mockito.Mockito;
-
-class ReplicaSizeCostTest extends RequireSingleBrokerCluster {
-  private static final HasValue SIZE_TP1_0 =
-      fakeBeanObject("Log", LogMetrics.Log.SIZE.metricName(), "test-1", "0", 891289600);
-  private static final HasValue SIZE_TP1_1 =
-      fakeBeanObject("Log", LogMetrics.Log.SIZE.metricName(), "test-1", "1", 471859200);
-  private static final HasValue SIZE_TP2_0 =
-      fakeBeanObject("Log", LogMetrics.Log.SIZE.metricName(), "test-2", "0", 0);
-  private static final HasValue SIZE_TP2_1 =
-      fakeBeanObject("Log", LogMetrics.Log.SIZE.metricName(), "test-2", "1", 367001600);
-  private static final Collection<HasBeanObject> BROKER_1 =
-      List.of(SIZE_TP1_0, SIZE_TP1_1, SIZE_TP2_1);
-  private static final Collection<HasBeanObject> BROKER_2 = List.of(SIZE_TP1_1, SIZE_TP2_0);
-  private static final Collection<HasBeanObject> BROKER_3 =
-      List.of(SIZE_TP2_1, SIZE_TP1_0, SIZE_TP2_0);
-
-  @Test
-  void testGetMetrics() throws ExecutionException, InterruptedException {
-    var brokerDiskSize = Map.of(1, 1000, 2, 1000, 3, 1000);
-    var topicName = "testGetMetrics-1";
-    try (var admin = Admin.of(bootstrapServers())) {
-      var host = "localhost";
-      admin.creator().topic(topicName).numberOfPartitions(1).numberOfReplicas((short) 1).create();
-      // wait for topic creation
-      Utils.sleep(Duration.ofSeconds(5));
-      var producer = Producer.builder().bootstrapServers(bootstrapServers()).build();
-      producer.sender().topic(topicName).key(new byte[10000]).run().toCompletableFuture().get();
-      ReplicaSizeCost costFunction = new ReplicaSizeCost(brokerDiskSize);
-      var beanObjects =
-          BeanCollector.builder()
-              .interval(Duration.ofSeconds(4))
-              .build()
-              .register()
-              .host(host)
-              .port(jmxServiceURL().getPort())
-              .fetcher(costFunction.fetcher().get())
-              .build()
-              .current();
-      var replicaSize =
-          beanObjects.stream()
-              .filter(x -> x instanceof HasValue)
-              .filter(x -> x.beanObject().properties().get("name").equals("Size"))
-              .filter(x -> x.beanObject().properties().get("type").equals("Log"))
-              .sorted(Comparator.comparing(HasBeanObject::createdTimestamp).reversed())
-              .map(x -> (HasValue) x)
-              .limit(1)
-              .map(e2 -> (int) e2.value())
-              .collect(Collectors.toList());
-      Assertions.assertTrue(replicaSize.get(0) >= 10000);
-    }
-  }
-
-  @Test
-  void partitionCost() {
-    var brokerDiskSize = Map.of(1, 1000, 2, 1000, 3, 1000);
-    var loadCostFunction = new ReplicaSizeCost(brokerDiskSize);
-    var broker1ReplicaLoad = loadCostFunction.partitionCost(clusterInfo(), clusterBean()).value(1);
-    var BROKER_2ReplicaLoad = loadCostFunction.partitionCost(clusterInfo(), clusterBean()).value(2);
-    var BROKER_3ReplicaLoad = loadCostFunction.partitionCost(clusterInfo(), clusterBean()).value(3);
-    // broker1
-    Assertions.assertEquals(0.85, broker1ReplicaLoad.get(TopicPartition.of("test-1", 0)));
-    Assertions.assertEquals(0.45, broker1ReplicaLoad.get(TopicPartition.of("test-1", 1)));
-    Assertions.assertEquals(0.35, broker1ReplicaLoad.get(TopicPartition.of("test-2", 1)));
-    // BROKER_2
-    Assertions.assertEquals(0.45, BROKER_2ReplicaLoad.get(TopicPartition.of("test-1", 1)));
-    Assertions.assertEquals(0, BROKER_2ReplicaLoad.get(TopicPartition.of("test-2", 0)));
-    // BROKER_3
-    Assertions.assertEquals(0.85, BROKER_3ReplicaLoad.get(TopicPartition.of("test-1", 0)));
-    Assertions.assertEquals(0, BROKER_3ReplicaLoad.get(TopicPartition.of("test-2", 0)));
-    Assertions.assertEquals(0.35, BROKER_3ReplicaLoad.get(TopicPartition.of("test-2", 1)));
-  }
-
-  @Test
-  void brokerCost() {
-    var brokerDiskSize = Map.of(1, 1000, 2, 1000, 3, 1000);
-    var loadCostFunction = new ReplicaSizeCost(brokerDiskSize);
-    var brokerReplicaLoad = loadCostFunction.brokerCost(clusterInfo(), clusterBean()).value();
-    Assertions.assertEquals(brokerReplicaLoad.get(1), 0.85 + 0.45 + 0.35);
-    Assertions.assertEquals(brokerReplicaLoad.get(2), 0.45);
-    Assertions.assertEquals(brokerReplicaLoad.get(3), 0.85 + 0.35);
-  }
-
-  private static ClusterInfo clusterInfo() {
-    ClusterInfo clusterInfo = Mockito.mock(ClusterInfo.class);
-    Mockito.when(clusterInfo.nodes())
-        .thenReturn(
-            List.of(NodeInfo.of(1, "", -1), NodeInfo.of(2, "", -1), NodeInfo.of(3, "", -1)));
-    Mockito.when(clusterInfo.topics()).thenReturn(Set.of("test-1", "test-2"));
-    Mockito.when(clusterInfo.replicas(Mockito.anyString()))
-        .thenAnswer(
-            topic ->
-                topic.getArgument(0).equals("test-1")
-                    ? List.of(
-                        ReplicaInfo.of("test-1", 0, NodeInfo.of(1, "", -1), true, true, false),
-                        ReplicaInfo.of("test-1", 0, NodeInfo.of(3, "", -1), false, true, false),
-                        ReplicaInfo.of("test-1", 1, NodeInfo.of(1, "", -1), false, true, false),
-                        ReplicaInfo.of("test-1", 1, NodeInfo.of(2, "", -1), true, true, false))
-                    : List.of(
-                        ReplicaInfo.of("test-2", 0, NodeInfo.of(2, "", -1), false, true, false),
-                        ReplicaInfo.of("test-2", 0, NodeInfo.of(3, "", -1), true, true, false),
-                        ReplicaInfo.of("test-2", 1, NodeInfo.of(1, "", -1), false, true, false),
-                        ReplicaInfo.of("test-2", 1, NodeInfo.of(3, "", -1), true, true, false)));
-    return clusterInfo;
-  }
-
-  private static ClusterBean clusterBean() {
-    return ClusterBean.of(Map.of(1, BROKER_1, 2, BROKER_2, 3, BROKER_3));
-  }
-
-  private static HasValue fakeBeanObject(
-      String type, String name, String topic, String partition, long size) {
-    BeanObject beanObject =
-        new BeanObject(
-            "kafka.log",
-            Map.of("name", name, "type", type, "topic", topic, "partition", partition),
-            Map.of("Value", size));
-    return HasValue.of(beanObject);
-  }
-}
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 cfdbbc90..54e381ba 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
@@ -17,14 +17,18 @@
 package org.astraea.app.metrics.broker;
 
 import java.time.Duration;
+import java.util.List;
 import java.util.stream.Collectors;
 import org.astraea.app.admin.Admin;
 import org.astraea.app.common.Utils;
+import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.jmx.MBeanClient;
 import org.astraea.app.service.RequireSingleBrokerCluster;
 import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.EnumSource;
+import org.mockito.Mockito;
 
 public class LogMetricsTest extends RequireSingleBrokerCluster {
 
@@ -53,4 +57,16 @@ public class LogMetricsTest extends RequireSingleBrokerCluster {
     log.fetch(MBeanClient.local()).forEach(m -> Assertions.assertTrue(m.value() >= 0));
     log.fetch(MBeanClient.local()).forEach(m -> Assertions.assertEquals(m.type(), log));
   }
+
+  @Test
+  void testMeters() {
+    var other = Mockito.mock(HasBeanObject.class);
+    var target = Mockito.mock(LogMetrics.Log.Meter.class);
+    Mockito.when(target.type()).thenReturn(LogMetrics.Log.LOG_END_OFFSET);
+    var result = LogMetrics.Log.meters(List.of(other, target), LogMetrics.Log.LOG_END_OFFSET);
+    Assertions.assertEquals(1, result.size());
+    Assertions.assertEquals(target, result.iterator().next());
+    Assertions.assertEquals(
+        0, LogMetrics.Log.meters(List.of(other, target), LogMetrics.Log.SIZE).size());
+  }
 }