Repository: skiptests/astraea
Pull request #389 (2022-06-02 15:15:17)

make brokerCost() flexibly select the normalize policy
#387 在嘗試修改的過程中，我發現了一個問題。
如果傳noNormalizer進去讓brokerCost不進行標準化直接輸出metrics值，我認爲有違背brokerCost的初衷，因爲回傳值不會在0～1之間。
但我仍然認爲加入normalizer是可取的，可以靈活調整計算分數的標準化策略，所以我發了這支pr添加這一新特性。
entropy需要原值的問題我會發另一個pr處理，應該會採用新增一個metrics()方法，直接把值撈出來 。


chia7712: @wycccccc 抱歉#392造成了許多衝突，麻煩你解決一下衝突
wycccccc: 上述修改都已完成，麻煩再review一下。

Patch:
diff --git a/app/build.gradle b/app/build.gradle
index b5ab15ae..c4386d41 100644
--- a/app/build.gradle
+++ b/app/build.gradle
@@ -26,7 +26,7 @@ repositories {
 }
 
 def versions = [
-    kafka: project.properties['kafka.version'] ?: "3.1.1",
+    kafka: project.properties['kafka.version'] ?: "3.2.0",
     junit: project.properties['junit.version'] ?: "5.8.2",
     mockito: project.properties['mockito.version'] ?: "4.2.0",
     jcommander: project.properties['jcommander.version'] ?: "1.81",
diff --git a/app/src/main/java/org/astraea/app/cost/BrokerCost.java b/app/src/main/java/org/astraea/app/cost/BrokerCost.java
index 6b712c2f..78574636 100644
--- a/app/src/main/java/org/astraea/app/cost/BrokerCost.java
+++ b/app/src/main/java/org/astraea/app/cost/BrokerCost.java
@@ -20,6 +20,22 @@ import java.util.Map;
 
 /** Return type of cost function, `HasBrokerCost`. It returns the score of brokers. */
 public interface BrokerCost {
+  /**
+   * A new BrokerCost has been normalized.
+   *
+   * @param normalizer
+   * @return A normalized BrokerCost.
+   */
+  default BrokerCost normalize(Normalizer normalizer) {
+    var map = this.value();
+    var keys = map.keySet().iterator();
+    var normalization = normalizer.normalize(map.values()).iterator();
+    while (normalization.hasNext()) {
+      map.replace(keys.next(), normalization.next());
+    }
+    return () -> map;
+  }
+
   /** @return broker-id and its "cost" */
   Map<Integer, Double> value();
 }
diff --git a/app/src/main/java/org/astraea/app/cost/HasBrokerCost.java b/app/src/main/java/org/astraea/app/cost/HasBrokerCost.java
index 9207d5e6..0eab831c 100644
--- a/app/src/main/java/org/astraea/app/cost/HasBrokerCost.java
+++ b/app/src/main/java/org/astraea/app/cost/HasBrokerCost.java
@@ -18,10 +18,10 @@ package org.astraea.app.cost;
 
 public interface HasBrokerCost extends CostFunction {
   /**
-   * score all nodes according to passed beans and cluster information.
+   * score all nodes for a particular metrics according to passed beans and cluster information.
    *
    * @param clusterInfo cluster information
-   * @return the score of each broker. The score ranges in [0 - 1].
+   * @return the score of each broker.
    */
   BrokerCost brokerCost(ClusterInfo clusterInfo);
 }
diff --git a/app/src/main/java/org/astraea/app/cost/Normalizer.java b/app/src/main/java/org/astraea/app/cost/Normalizer.java
index 2ce2e99c..ff669406 100644
--- a/app/src/main/java/org/astraea/app/cost/Normalizer.java
+++ b/app/src/main/java/org/astraea/app/cost/Normalizer.java
@@ -27,7 +27,11 @@ public interface Normalizer {
 
   /** @return all normalizers */
   static List<Normalizer> all() {
-    return List.of(Normalizer.proportion(), Normalizer.minMax(true), Normalizer.minMax(false));
+    return List.of(
+        Normalizer.proportion(),
+        Normalizer.minMax(true),
+        Normalizer.minMax(false),
+        Normalizer.TScore());
   }
 
   /**
@@ -70,6 +74,35 @@ public interface Normalizer {
     };
   }
 
+  /**
+   * implement the TScore normalization
+   *
+   * <p>(value -avg/standardDeviation)*10+50
+   *
+   * @return TScore normalizer
+   */
+  static Normalizer TScore() {
+    return values -> {
+      var avg = values.stream().mapToDouble(i -> i).sum() / values.size();
+      var standardDeviation =
+          Math.sqrt(values.stream().mapToDouble(i -> (i - avg) * (i - avg)).sum() / values.size());
+
+      return values.stream()
+          .map(i -> (i - avg) / standardDeviation)
+          .map(
+              i -> {
+                var score = (i * 10 + 50) / 100.0;
+                if (score > 1) {
+                  score = 1.0;
+                } else if (score < 0) {
+                  score = 0.0;
+                }
+                return Math.round(score * 100) / 100.0;
+              })
+          .collect(Collectors.toUnmodifiableList());
+    };
+  }
+
   /**
    * rescales the values into a range of [0,1]
    *
diff --git a/app/src/main/java/org/astraea/app/cost/broker/BrokerInputCost.java b/app/src/main/java/org/astraea/app/cost/broker/BrokerInputCost.java
index 53c9d774..9140a46d 100644
--- a/app/src/main/java/org/astraea/app/cost/broker/BrokerInputCost.java
+++ b/app/src/main/java/org/astraea/app/cost/broker/BrokerInputCost.java
@@ -73,8 +73,7 @@ public class BrokerInputCost implements HasBrokerCost {
                       broker.accumulateCount = inBean.count();
                       return count;
                     }));
-
-    CostUtils.TScore(costMetrics).forEach((broker, v) -> brokersMetric.get(broker).updateLoad(v));
+    costMetrics.forEach((broker, v) -> brokersMetric.get(broker).updateLoad(v));
 
     return this::computeLoad;
   }
diff --git a/app/src/main/java/org/astraea/app/cost/broker/BrokerOutputCost.java b/app/src/main/java/org/astraea/app/cost/broker/BrokerOutputCost.java
index c8a268f7..73e5c82f 100644
--- a/app/src/main/java/org/astraea/app/cost/broker/BrokerOutputCost.java
+++ b/app/src/main/java/org/astraea/app/cost/broker/BrokerOutputCost.java
@@ -73,8 +73,7 @@ public class BrokerOutputCost implements HasBrokerCost {
                       broker.accumulateCount = inBean.count();
                       return count;
                     }));
-
-    CostUtils.TScore(costMetrics).forEach((broker, v) -> brokersMetric.get(broker).updateLoad(v));
+    costMetrics.forEach((broker, v) -> brokersMetric.get(broker).updateLoad(v));
 
     return this::computeLoad;
   }
diff --git a/app/src/main/java/org/astraea/app/cost/broker/MemoryCost.java b/app/src/main/java/org/astraea/app/cost/broker/MemoryCost.java
index b7488a0e..557659ce 100644
--- a/app/src/main/java/org/astraea/app/cost/broker/MemoryCost.java
+++ b/app/src/main/java/org/astraea/app/cost/broker/MemoryCost.java
@@ -68,8 +68,7 @@ public class MemoryCost extends Periodic<Map<Integer, Double>> implements HasBro
                       return (jvmBean.heapMemoryUsage().getUsed() + 0.0)
                           / (jvmBean.heapMemoryUsage().getMax() + 1);
                     }));
-
-    CostUtils.TScore(costMetrics).forEach((broker, v) -> brokersMetric.get(broker).updateLoad(v));
+    costMetrics.forEach((broker, v) -> brokersMetric.get(broker).updateLoad(v));
 
     return this::computeLoad;
   }
diff --git a/app/src/main/java/org/astraea/app/cost/broker/NeutralIntegratedCost.java b/app/src/main/java/org/astraea/app/cost/broker/NeutralIntegratedCost.java
index a447d86c..ba972f49 100644
--- a/app/src/main/java/org/astraea/app/cost/broker/NeutralIntegratedCost.java
+++ b/app/src/main/java/org/astraea/app/cost/broker/NeutralIntegratedCost.java
@@ -60,21 +60,25 @@ public class NeutralIntegratedCost implements HasBrokerCost {
           if (hasBrokerCost instanceof BrokerInputCost) {
             hasBrokerCost
                 .brokerCost(clusterInfo)
+                .normalize(Normalizer.TScore())
                 .value()
                 .forEach((brokerID, value) -> brokersMetric.get(brokerID).inputScore = value);
           } else if (hasBrokerCost instanceof BrokerOutputCost) {
             hasBrokerCost
                 .brokerCost(clusterInfo)
+                .normalize(Normalizer.TScore())
                 .value()
                 .forEach((brokerID, value) -> brokersMetric.get(brokerID).outputScore = value);
           } else if (hasBrokerCost instanceof CpuCost) {
             hasBrokerCost
                 .brokerCost(clusterInfo)
+                .normalize(Normalizer.TScore())
                 .value()
                 .forEach((brokerID, value) -> brokersMetric.get(brokerID).cpuScore = value);
           } else if (hasBrokerCost instanceof MemoryCost) {
             hasBrokerCost
                 .brokerCost(clusterInfo)
+                .normalize(Normalizer.TScore())
                 .value()
                 .forEach((brokerID, value) -> brokersMetric.get(brokerID).memoryScore = value);
           }
diff --git a/docker/start_broker.sh b/docker/start_broker.sh
index 7d26a199..bbbde237 100755
--- a/docker/start_broker.sh
+++ b/docker/start_broker.sh
@@ -18,7 +18,7 @@ declare -r DOCKER_FOLDER=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" &>/dev/null
 source $DOCKER_FOLDER/docker_build_common.sh
 
 # ===============================[global variables]===============================
-declare -r VERSION=${REVISION:-${VERSION:-3.1.1}}
+declare -r VERSION=${REVISION:-${VERSION:-3.2.0}}
 declare -r DOCKERFILE=$DOCKER_FOLDER/broker.dockerfile
 declare -r CONFLUENT_BROKER=${CONFLUENT_BROKER:-false}
 declare -r CONFLUENT_VERSION=${CONFLUENT_VERSION:-7.0.1}
@@ -68,7 +68,7 @@ function showHelp() {
   echo "    REPO=astraea/broker                      set the docker repo"
   echo "    HEAP_OPTS=\"-Xmx2G -Xms2G\"                set broker JVM memory"
   echo "    REVISION=trunk                           set revision of kafka source code to build container"
-  echo "    VERSION=3.1.1                            set version of kafka distribution"
+  echo "    VERSION=3.2.0                            set version of kafka distribution"
   echo "    BUILD=false                              set true if you want to build image locally"
   echo "    RUN=false                                set false if you want to build/pull image only"
   echo "    DATA_FOLDERS=/tmp/folder1                set host folders used by broker"


Test patch:
diff --git a/app/src/test/java/org/astraea/app/cost/BrokerCostTest.java b/app/src/test/java/org/astraea/app/cost/BrokerCostTest.java
new file mode 100644
index 00000000..7cfe08f5
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/cost/BrokerCostTest.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.Map;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class BrokerCostTest {
+  @Test
+  void testNormalize() {
+    var testMap =
+        IntStream.range(0, 100).boxed().collect(Collectors.toMap(i -> i, i -> i * 2 + 0.0));
+    var oldKey = -1;
+    var oldValue = -1.0;
+    brokerCost(testMap)
+        .normalize(Normalizer.TScore())
+        .value()
+        .forEach(
+            (key, value) -> {
+              Assertions.assertTrue(key > oldKey);
+              Assertions.assertTrue(value > oldValue);
+            });
+  }
+
+  BrokerCost brokerCost(Map<Integer, Double> map) {
+    return () -> map;
+  }
+}
diff --git a/app/src/test/java/org/astraea/app/cost/broker/BrokerInputCostTest.java b/app/src/test/java/org/astraea/app/cost/broker/BrokerInputCostTest.java
index df25c5d8..7304dca1 100644
--- a/app/src/test/java/org/astraea/app/cost/broker/BrokerInputCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/broker/BrokerInputCostTest.java
@@ -21,6 +21,7 @@ import java.util.List;
 import java.util.Map;
 import org.astraea.app.cost.ClusterInfo;
 import org.astraea.app.cost.FakeClusterInfo;
+import org.astraea.app.cost.Normalizer;
 import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.jmx.BeanObject;
 import org.astraea.app.metrics.kafka.BrokerTopicMetricsResult;
@@ -31,20 +32,33 @@ import org.mockito.Mockito;
 
 public class BrokerInputCostTest {
   @Test
-  void testCost() {
+  void testTScoreCost() {
+    ClusterInfo clusterInfo = exampleClusterInfo(50000L, 20000L, 5000L);
+
+    var brokerInputCost = new BrokerInputCost();
+    var scores = brokerInputCost.brokerCost(clusterInfo).normalize(Normalizer.TScore()).value();
+    Assertions.assertEquals(0.63, scores.get(1));
+    Assertions.assertEquals(0.47, scores.get(2));
+    Assertions.assertEquals(0.39, scores.get(3));
+
+    ClusterInfo clusterInfo2 = exampleClusterInfo(55555L, 25352L, 25000L);
+    scores = brokerInputCost.brokerCost(clusterInfo2).normalize(Normalizer.TScore()).value();
+    Assertions.assertEquals(0.64, scores.get(1));
+    Assertions.assertEquals(0.43, scores.get(2));
+    Assertions.assertEquals(0.43, scores.get(3));
+  }
+
+  @Test
+  void testNoNormalize() {
     ClusterInfo clusterInfo = exampleClusterInfo(10000L, 20000L, 5000L);
 
     var brokerInputCost = new BrokerInputCost();
     var scores = brokerInputCost.brokerCost(clusterInfo).value();
-    Assertions.assertEquals(0.47, scores.get(1));
-    Assertions.assertEquals(0.63, scores.get(2));
-    Assertions.assertEquals(0.39, scores.get(3));
+    Assertions.assertEquals(10000.0, scores.get(1));
+    Assertions.assertEquals(20000.0, scores.get(2));
+    Assertions.assertEquals(5000.0, scores.get(3));
 
     ClusterInfo clusterInfo2 = exampleClusterInfo(55555L, 25352L, 25000L);
-    scores = brokerInputCost.brokerCost(clusterInfo2).value();
-    Assertions.assertEquals(0.55, scores.get(1));
-    Assertions.assertEquals(0.51, scores.get(2));
-    Assertions.assertEquals(0.44, scores.get(3));
   }
 
   private ClusterInfo exampleClusterInfo(long in1, long in2, long in3) {
diff --git a/app/src/test/java/org/astraea/app/cost/broker/BrokerOutPutCostTest.java b/app/src/test/java/org/astraea/app/cost/broker/BrokerOutPutCostTest.java
index 5db1a14f..ef53ffaf 100644
--- a/app/src/test/java/org/astraea/app/cost/broker/BrokerOutPutCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/broker/BrokerOutPutCostTest.java
@@ -21,6 +21,7 @@ import java.util.List;
 import java.util.Map;
 import org.astraea.app.cost.ClusterInfo;
 import org.astraea.app.cost.FakeClusterInfo;
+import org.astraea.app.cost.Normalizer;
 import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.jmx.BeanObject;
 import org.astraea.app.metrics.kafka.BrokerTopicMetricsResult;
@@ -35,16 +36,16 @@ public class BrokerOutPutCostTest {
     ClusterInfo clusterInfo = exampleClusterInfo(10000L, 20000L, 5000L);
 
     var brokerOutputCost = new BrokerOutputCost();
-    var scores = brokerOutputCost.brokerCost(clusterInfo).value();
+    var scores = brokerOutputCost.brokerCost(clusterInfo).normalize(Normalizer.TScore()).value();
     Assertions.assertEquals(0.47, scores.get(1));
     Assertions.assertEquals(0.63, scores.get(2));
     Assertions.assertEquals(0.39, scores.get(3));
 
     ClusterInfo clusterInfo2 = exampleClusterInfo(55555L, 25352L, 25000L);
-    scores = brokerOutputCost.brokerCost(clusterInfo2).value();
-    Assertions.assertEquals(0.55, scores.get(1));
-    Assertions.assertEquals(0.51, scores.get(2));
-    Assertions.assertEquals(0.44, scores.get(3));
+    scores = brokerOutputCost.brokerCost(clusterInfo2).normalize(Normalizer.TScore()).value();
+    Assertions.assertEquals(0.64, scores.get(1));
+    Assertions.assertEquals(0.43, scores.get(2));
+    Assertions.assertEquals(0.43, scores.get(3));
   }
 
   private ClusterInfo exampleClusterInfo(long out1, long out2, long out3) {
diff --git a/app/src/test/java/org/astraea/app/cost/broker/CpuCostTest.java b/app/src/test/java/org/astraea/app/cost/broker/CpuCostTest.java
index 1496a584..c8edd31e 100644
--- a/app/src/test/java/org/astraea/app/cost/broker/CpuCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/broker/CpuCostTest.java
@@ -23,6 +23,7 @@ import java.util.Set;
 import org.astraea.app.cost.ClusterInfo;
 import org.astraea.app.cost.FakeClusterInfo;
 import org.astraea.app.cost.NodeInfo;
+import org.astraea.app.cost.Normalizer;
 import org.astraea.app.cost.ReplicaInfo;
 import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.java.OperatingSystemInfo;
@@ -62,7 +63,7 @@ public class CpuCostTest {
         };
 
     var cpuCost = new CpuCost();
-    var scores = cpuCost.brokerCost(clusterInfo).value();
+    var scores = cpuCost.brokerCost(clusterInfo).normalize(Normalizer.TScore()).value();
     Assertions.assertEquals(0.39, scores.get(1));
     Assertions.assertEquals(0.63, scores.get(2));
     Assertions.assertEquals(0.48, scores.get(3));
@@ -95,10 +96,10 @@ public class CpuCostTest {
                 ReplicaInfo.of("t", 0, NodeInfo.of(3, "host3", 9092), false, true, false));
           }
         };
-    scores = cpuCost.brokerCost(clusterInfo2).value();
-    Assertions.assertEquals(0.51, scores.get(1));
-    Assertions.assertEquals(0.55, scores.get(2));
-    Assertions.assertEquals(0.44, scores.get(3));
+    scores = cpuCost.brokerCost(clusterInfo2).normalize(Normalizer.TScore()).value();
+    Assertions.assertEquals(0.52, scores.get(1));
+    Assertions.assertEquals(0.61, scores.get(2));
+    Assertions.assertEquals(0.37, scores.get(3));
   }
 
   private OperatingSystemInfo mockResult(double usage) {
diff --git a/app/src/test/java/org/astraea/app/cost/broker/MemoryCostTest.java b/app/src/test/java/org/astraea/app/cost/broker/MemoryCostTest.java
index 9c7f3fb6..0521718c 100644
--- a/app/src/test/java/org/astraea/app/cost/broker/MemoryCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/broker/MemoryCostTest.java
@@ -24,6 +24,7 @@ import java.util.Set;
 import org.astraea.app.cost.ClusterInfo;
 import org.astraea.app.cost.FakeClusterInfo;
 import org.astraea.app.cost.NodeInfo;
+import org.astraea.app.cost.Normalizer;
 import org.astraea.app.cost.ReplicaInfo;
 import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.java.HasJvmMemory;
@@ -63,7 +64,7 @@ public class MemoryCostTest {
         };
 
     var memoryCost = new MemoryCost();
-    var scores = memoryCost.brokerCost(clusterInfo).value();
+    var scores = memoryCost.brokerCost(clusterInfo).normalize(Normalizer.TScore()).value();
     Assertions.assertEquals(0.39, scores.get(1));
     Assertions.assertEquals(0.63, scores.get(2));
     Assertions.assertEquals(0.47, scores.get(3));
@@ -96,10 +97,10 @@ public class MemoryCostTest {
                 ReplicaInfo.of("t", 0, NodeInfo.of(3, "host3", 9092), false, true, false));
           }
         };
-    scores = memoryCost.brokerCost(clusterInfo2).value();
-    Assertions.assertEquals(0.42, scores.get(1));
-    Assertions.assertEquals(0.52, scores.get(2));
-    Assertions.assertEquals(0.55, scores.get(3));
+    scores = memoryCost.brokerCost(clusterInfo2).normalize(Normalizer.TScore()).value();
+    Assertions.assertEquals(0.36, scores.get(1));
+    Assertions.assertEquals(0.58, scores.get(2));
+    Assertions.assertEquals(0.56, scores.get(3));
   }
 
   private HasJvmMemory mockResult(long used, long max) {