Repository: skiptests/astraea
Pull request #524 (2022-07-28 19:03:44)

Remove properties in cost
移除`ReplicaDiskInCCost`建構式要求先輸入總量的要求，以及新增可以自訂Duration的建構式

chia7712: @qoo332001 麻煩在 cost function 介面上描述建構式的規範

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..b22cbdf0 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;
@@ -32,6 +31,7 @@ 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 +39,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 +63,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 +77,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 +85,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 +120,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(
@@ -223,6 +208,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/cost/ReplicaDiskInCostTest.java b/app/src/test/java/org/astraea/app/cost/ReplicaDiskInCostTest.java
index fdf653ff..2541617f 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(