Repository: skiptests/astraea
Pull request #547 (2022-08-04 19:12:25)

simplify sizeTimeSeries
簡化ReplicaDiskInCost中 sizeTimeSeries的計算方法，改用LogMetrics.Log.Meter


Patch:
diff --git a/app/src/main/java/org/astraea/app/cost/CostFunction.java b/app/src/main/java/org/astraea/app/cost/CostFunction.java
index 91f22fa6..c36f45a3 100644
--- a/app/src/main/java/org/astraea/app/cost/CostFunction.java
+++ b/app/src/main/java/org/astraea/app/cost/CostFunction.java
@@ -18,10 +18,14 @@ package org.astraea.app.cost;
 
 import java.util.Optional;
 import org.astraea.app.metrics.collector.Fetcher;
+import org.astraea.app.partitioner.Configuration;
 
 /**
  * It is meaningless to implement this interface. Instead, we should implement interfaces like
- * {@link HasBrokerCost} or {@link HasPartitionCost}.
+ * {@link HasBrokerCost} ,{@link HasPartitionCost} or {@link HasPartitionCost}.
+ *
+ * <p>Constructors in CostFunction can only take no or only one parameter {@link Configuration}, ex.
+ * Constructor({@link Configuration} configuration) or Constructor()
  */
 public interface CostFunction {
 
diff --git a/app/src/main/java/org/astraea/app/cost/ReplicaDiskInCost.java b/app/src/main/java/org/astraea/app/cost/ReplicaDiskInCost.java
index 35a39956..49f8a3f1 100644
--- a/app/src/main/java/org/astraea/app/cost/ReplicaDiskInCost.java
+++ b/app/src/main/java/org/astraea/app/cost/ReplicaDiskInCost.java
@@ -21,7 +21,6 @@ import java.util.Comparator;
 import java.util.List;
 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;
@@ -29,9 +28,9 @@ import org.astraea.app.admin.NodeInfo;
 import org.astraea.app.admin.TopicPartition;
 import org.astraea.app.admin.TopicPartitionReplica;
 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.Fetcher;
+import org.astraea.app.partitioner.Configuration;
 
 /**
  * The result is computed by "Size.Value" ,and createdTimestamp in the metrics. "Size.Value"
@@ -39,10 +38,11 @@ import org.astraea.app.metrics.collector.Fetcher;
  * increase of log size per unit time divided by the upper limit of broker bandwidth.
  */
 public class ReplicaDiskInCost implements HasBrokerCost, HasPartitionCost {
-  Map<Integer, Integer> brokerBandwidthCap;
+  private final Duration duration;
 
-  public ReplicaDiskInCost(Map<Integer, Integer> brokerBandwidthCap) {
-    this.brokerBandwidthCap = brokerBandwidthCap;
+  public ReplicaDiskInCost(Configuration configuration) {
+    duration =
+        Duration.ofSeconds(Integer.parseInt(configuration.string("metrics.duration").orElse("30")));
   }
 
   @Override
@@ -62,7 +62,7 @@ public class ReplicaDiskInCost implements HasBrokerCost, HasPartitionCost {
                     NodeInfo::id,
                     node -> topicPartitionOfEachBroker.getOrDefault(node.id(), List.of())));
 
-    final var topicPartitionDataRate = topicPartitionDataRate(clusterBean, Duration.ofSeconds(3));
+    final var topicPartitionDataRate = topicPartitionDataRate(clusterBean, duration);
 
     final var brokerLoad =
         actual.entrySet().stream()
@@ -76,12 +76,7 @@ public class ReplicaDiskInCost implements HasBrokerCost, HasPartitionCost {
                                     topicPartitionDataRate.get(
                                         TopicPartition.of(x.topic(), x.partition())))
                             .sum()))
-            .map(
-                entry ->
-                    Map.entry(
-                        entry.getKey(),
-                        entry.getValue() / brokerBandwidthCap.get(entry.getKey()) / 1024 / 1024))
-            .map(entry -> Map.entry(entry.getKey(), Math.min(entry.getValue(), 1)))
+            .map(entry -> Map.entry(entry.getKey(), entry.getValue()))
             .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
 
     return () -> brokerLoad;
@@ -89,25 +84,14 @@ public class ReplicaDiskInCost implements HasBrokerCost, HasPartitionCost {
 
   @Override
   public PartitionCost partitionCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
-    var replicaIn = replicaDataRate(clusterBean, Duration.ofSeconds(2));
-    var replicaScore =
-        new TreeMap<TopicPartitionReplica, Double>(
-            Comparator.comparing(TopicPartitionReplica::brokerId)
-                .thenComparing(TopicPartitionReplica::topic)
-                .thenComparing(TopicPartitionReplica::partition));
-    replicaIn.forEach(
-        (tpr, rate) -> {
-          var score = rate / brokerBandwidthCap.get(tpr.brokerId()) / 1024.0 / 1024.0;
-          if (score >= 1) score = 1;
-          replicaScore.put(tpr, score);
-        });
+    var replicaIn = replicaDataRate(clusterBean, duration);
     var scoreForTopic =
         clusterInfo.topics().stream()
             .map(
                 topic ->
                     Map.entry(
                         topic,
-                        replicaScore.entrySet().stream()
+                        replicaIn.entrySet().stream()
                             .filter(x -> x.getKey().topic().equals(topic))
                             .collect(
                                 Collectors.groupingBy(
@@ -135,7 +119,7 @@ public class ReplicaDiskInCost implements HasBrokerCost, HasPartitionCost {
                 node ->
                     Map.entry(
                         node.id(),
-                        replicaScore.entrySet().stream()
+                        replicaIn.entrySet().stream()
                             .filter(x -> x.getKey().brokerId() == node.id())
                             .collect(
                                 Collectors.groupingBy(
@@ -202,11 +186,7 @@ public class ReplicaDiskInCost implements HasBrokerCost, HasPartitionCost {
             metrics -> {
               // calculate the increase rate over a specific window of time
               var sizeTimeSeries =
-                  metrics.getValue().stream()
-                      .filter(bean -> bean instanceof HasValue)
-                      .filter(bean -> bean.beanObject().properties().get("type").equals("Log"))
-                      .filter(bean -> bean.beanObject().properties().get("name").equals("Size"))
-                      .map(bean -> (HasValue) bean)
+                  LogMetrics.Log.meters(metrics.getValue(), LogMetrics.Log.SIZE).stream()
                       .sorted(Comparator.comparingLong(HasBeanObject::createdTimestamp).reversed())
                       .collect(Collectors.toUnmodifiableList());
               var latestSize = sizeTimeSeries.stream().findFirst().orElseThrow();
@@ -223,6 +203,8 @@ public class ReplicaDiskInCost implements HasBrokerCost, HasPartitionCost {
                                   "No sufficient info to determine data rate, try later."));
               var dataRate =
                   ((double) (latestSize.value() - windowSize.value()))
+                      / 1024.0
+                      / 1024.0
                       / ((double) (latestSize.createdTimestamp() - windowSize.createdTimestamp())
                           / 1000);
               return Map.entry(metrics.getKey(), dataRate);


Test patch:
diff --git a/app/src/test/java/org/astraea/app/admin/AdminTest.java b/app/src/test/java/org/astraea/app/admin/AdminTest.java
index 2e8fab32..9aeb6c0c 100644
--- a/app/src/test/java/org/astraea/app/admin/AdminTest.java
+++ b/app/src/test/java/org/astraea/app/admin/AdminTest.java
@@ -1165,8 +1165,11 @@ public class AdminTest extends RequireBrokerCluster {
     try (var admin = Admin.of(bootstrapServers())) {
       topicNames.forEach(
           x -> admin.creator().topic(x).numberOfPartitions(3).numberOfReplicas((short) 3).create());
+      Utils.sleep(Duration.ofSeconds(2));
 
       admin.deleteTopics(Set.of(topicNames.get(0), topicNames.get(1)));
+      Utils.sleep(Duration.ofSeconds(2));
+
       var latestTopicNames = admin.topicNames();
       Assertions.assertFalse(latestTopicNames.contains(topicNames.get(0)));
       Assertions.assertFalse(latestTopicNames.contains(topicNames.get(1)));
@@ -1174,6 +1177,8 @@ public class AdminTest extends RequireBrokerCluster {
       Assertions.assertTrue(latestTopicNames.contains(topicNames.get(3)));
 
       admin.deleteTopics(Set.of(topicNames.get(3)));
+      Utils.sleep(Duration.ofSeconds(2));
+
       latestTopicNames = admin.topicNames();
       Assertions.assertFalse(latestTopicNames.contains(topicNames.get(3)));
       Assertions.assertTrue(latestTopicNames.contains(topicNames.get(2)));
diff --git a/app/src/test/java/org/astraea/app/cost/ReplicaDiskInCostTest.java b/app/src/test/java/org/astraea/app/cost/ReplicaDiskInCostTest.java
index fdf653ff..bb8ed0f8 100644
--- a/app/src/test/java/org/astraea/app/cost/ReplicaDiskInCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/ReplicaDiskInCostTest.java
@@ -17,7 +17,6 @@
 package org.astraea.app.cost;
 
 import java.util.Collection;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -30,6 +29,7 @@ import org.astraea.app.metrics.BeanObject;
 import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.broker.HasValue;
 import org.astraea.app.metrics.broker.LogMetrics;
+import org.astraea.app.partitioner.Configuration;
 import org.astraea.app.service.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
@@ -64,46 +64,39 @@ class ReplicaDiskInCostTest extends RequireBrokerCluster {
 
   @Test
   void testPartitionCost() {
-    Map<Integer, Integer> brokerBandwidth = new HashMap<>();
-    brokerBandwidth.put(1, 50);
-    brokerBandwidth.put(2, 50);
-    brokerBandwidth.put(3, 50);
-    var loadCostFunction = new ReplicaDiskInCost(brokerBandwidth);
+    var configuration = Configuration.of(Map.of("metrics.duration", "3"));
+    var loadCostFunction = new ReplicaDiskInCost(configuration);
     var broker1ReplicaLoad = loadCostFunction.partitionCost(clusterInfo(), clusterBean()).value(1);
     var broker2ReplicaLoad = loadCostFunction.partitionCost(clusterInfo(), clusterBean()).value(2);
     var broker3ReplicaLoad = loadCostFunction.partitionCost(clusterInfo(), clusterBean()).value(3);
     // broker1
     Assertions.assertEquals(
-        0.23841381072998047, broker1ReplicaLoad.get(TopicPartition.of("test-1", 0)));
+        11.920690536499023, broker1ReplicaLoad.get(TopicPartition.of("test-1", 0)));
     Assertions.assertEquals(
-        0.1907339096069336, broker1ReplicaLoad.get(TopicPartition.of("test-2", 0)));
+        9.53669548034668, broker1ReplicaLoad.get(TopicPartition.of("test-2", 0)));
     // broker2
     Assertions.assertEquals(
-        0.23841381072998047, broker2ReplicaLoad.get(TopicPartition.of("test-1", 0)));
+        11.920690536499023, broker2ReplicaLoad.get(TopicPartition.of("test-1", 0)));
     Assertions.assertEquals(
-        0.476834774017334, broker2ReplicaLoad.get(TopicPartition.of("test-1", 1)));
+        23.8417387008667, broker2ReplicaLoad.get(TopicPartition.of("test-1", 1)));
     // broker3
     Assertions.assertEquals(
-        0.476834774017334, broker3ReplicaLoad.get(TopicPartition.of("test-1", 1)));
+        23.8417387008667, broker3ReplicaLoad.get(TopicPartition.of("test-1", 1)));
     Assertions.assertEquals(
-        0.1907339096069336, broker3ReplicaLoad.get(TopicPartition.of("test-2", 0)));
+        9.53669548034668, broker3ReplicaLoad.get(TopicPartition.of("test-2", 0)));
   }
 
   @Test
   void testBrokerCost() {
-    Map<Integer, Integer> properties = new HashMap<>();
-    properties.put(1, 50);
-    properties.put(2, 50);
-    properties.put(3, 50);
-    var loadCostFunction = new ReplicaDiskInCost(properties);
+    var configuration = Configuration.of(Map.of("metrics.duration", "3"));
+    var loadCostFunction = new ReplicaDiskInCost(configuration);
     var brokerLoad = loadCostFunction.brokerCost(clusterInfo(), clusterBean()).value();
-    Assertions.assertEquals(0.23841381072998047 + 0.1907339096069336, brokerLoad.get(1));
-    Assertions.assertEquals(0.23841381072998047 + 0.476834774017334, brokerLoad.get(2));
-    Assertions.assertEquals(0.476834774017334 + 0.1907339096069336, brokerLoad.get(3));
+    Assertions.assertEquals(21.457386016845703, brokerLoad.get(1));
+    Assertions.assertEquals(35.76242923736572, brokerLoad.get(2));
+    Assertions.assertEquals(33.37843418121338, brokerLoad.get(3));
   }
 
   private ClusterInfo clusterInfo() {
-
     ClusterInfo clusterInfo = Mockito.mock(ClusterInfo.class);
     Mockito.when(clusterInfo.nodes())
         .thenReturn(
@@ -128,14 +121,13 @@ class ReplicaDiskInCostTest extends RequireBrokerCluster {
     return ClusterBean.of(Map.of(1, broker1, 2, broker2, 3, broker3));
   }
 
-  private static HasValue fakeBeanObject(
+  private static LogMetrics.Log.Meter fakeBeanObject(
       String type, String name, String topic, String partition, long size, long time) {
-    BeanObject beanObject =
+    return new LogMetrics.Log.Meter(
         new BeanObject(
-            "",
+            "kafka.log",
             Map.of("name", name, "type", type, "topic", topic, "partition", partition),
             Map.of("Value", size),
-            time);
-    return HasValue.of(beanObject);
+            time));
   }
 }
diff --git a/app/src/test/java/org/astraea/app/web/TopicHandlerTest.java b/app/src/test/java/org/astraea/app/web/TopicHandlerTest.java
index d3b051fb..c4411251 100644
--- a/app/src/test/java/org/astraea/app/web/TopicHandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/TopicHandlerTest.java
@@ -204,14 +204,19 @@ public class TopicHandlerTest extends RequireBrokerCluster {
       var handler = new TopicHandler(admin);
       topicNames.forEach(
           x -> admin.creator().topic(x).numberOfPartitions(3).numberOfReplicas((short) 3).create());
+      Utils.sleep(Duration.ofSeconds(2));
 
       handler.delete(topicNames.get(0), Map.of());
+      Utils.sleep(Duration.ofSeconds(2));
+
       var latestTopicNames = admin.topicNames();
       Assertions.assertFalse(latestTopicNames.contains(topicNames.get(0)));
       Assertions.assertTrue(latestTopicNames.contains(topicNames.get(1)));
       Assertions.assertTrue(latestTopicNames.contains(topicNames.get(2)));
 
       handler.delete(topicNames.get(2), Map.of());
+      Utils.sleep(Duration.ofSeconds(2));
+
       latestTopicNames = admin.topicNames();
       Assertions.assertFalse(latestTopicNames.contains(topicNames.get(2)));
       Assertions.assertTrue(latestTopicNames.contains(topicNames.get(1)));