Repository: skiptests/astraea
Pull request #451 (2022-07-02 18:37:21)

Add todo comment in RecordHandler#get


Patch:
diff --git a/app/src/main/java/org/astraea/app/cost/BrokerInputCost.java b/app/src/main/java/org/astraea/app/cost/BrokerInputCost.java
index c8438795..f2825623 100644
--- a/app/src/main/java/org/astraea/app/cost/BrokerInputCost.java
+++ b/app/src/main/java/org/astraea/app/cost/BrokerInputCost.java
@@ -19,6 +19,7 @@ package org.astraea.app.cost;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import org.astraea.app.metrics.collector.Fetcher;
@@ -92,8 +93,8 @@ public class BrokerInputCost implements HasBrokerCost {
   }
 
   @Override
-  public Fetcher fetcher() {
-    return client -> List.of(KafkaMetrics.BrokerTopic.BytesInPerSec.fetch(client));
+  public Optional<Fetcher> fetcher() {
+    return Optional.of(client -> List.of(KafkaMetrics.BrokerTopic.BytesInPerSec.fetch(client)));
   }
 
   private static class BrokerMetric {
diff --git a/app/src/main/java/org/astraea/app/cost/BrokerOutputCost.java b/app/src/main/java/org/astraea/app/cost/BrokerOutputCost.java
index 30f76b3d..fee5a31e 100644
--- a/app/src/main/java/org/astraea/app/cost/BrokerOutputCost.java
+++ b/app/src/main/java/org/astraea/app/cost/BrokerOutputCost.java
@@ -19,6 +19,7 @@ package org.astraea.app.cost;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import org.astraea.app.metrics.collector.Fetcher;
@@ -92,8 +93,8 @@ public class BrokerOutputCost implements HasBrokerCost {
   }
 
   @Override
-  public Fetcher fetcher() {
-    return client -> List.of(KafkaMetrics.BrokerTopic.BytesOutPerSec.fetch(client));
+  public Optional<Fetcher> fetcher() {
+    return Optional.of(client -> List.of(KafkaMetrics.BrokerTopic.BytesOutPerSec.fetch(client)));
   }
 
   private static class BrokerMetric {
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 e40f3be4..631f15af 100644
--- a/app/src/main/java/org/astraea/app/cost/CostFunction.java
+++ b/app/src/main/java/org/astraea/app/cost/CostFunction.java
@@ -16,6 +16,7 @@
  */
 package org.astraea.app.cost;
 
+import java.util.Optional;
 import org.astraea.app.metrics.collector.Fetcher;
 
 /**
@@ -29,5 +30,7 @@ public interface CostFunction {
   }
 
   /** @return the metrics getters. Those getters are used to fetch mbeans. */
-  Fetcher fetcher();
+  default Optional<Fetcher> fetcher() {
+    return Optional.empty();
+  }
 }
diff --git a/app/src/main/java/org/astraea/app/cost/CpuCost.java b/app/src/main/java/org/astraea/app/cost/CpuCost.java
index 904b8888..37d90138 100644
--- a/app/src/main/java/org/astraea/app/cost/CpuCost.java
+++ b/app/src/main/java/org/astraea/app/cost/CpuCost.java
@@ -19,6 +19,7 @@ package org.astraea.app.cost;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import org.astraea.app.metrics.collector.Fetcher;
@@ -86,8 +87,8 @@ public class CpuCost implements HasBrokerCost {
   }
 
   @Override
-  public Fetcher fetcher() {
-    return client -> List.of(KafkaMetrics.Host.operatingSystem(client));
+  public Optional<Fetcher> fetcher() {
+    return Optional.of(client -> List.of(KafkaMetrics.Host.operatingSystem(client)));
   }
 
   private static class BrokerMetric {
diff --git a/app/src/main/java/org/astraea/app/cost/LoadCost.java b/app/src/main/java/org/astraea/app/cost/LoadCost.java
index 61652e2d..6e2f9994 100644
--- a/app/src/main/java/org/astraea/app/cost/LoadCost.java
+++ b/app/src/main/java/org/astraea/app/cost/LoadCost.java
@@ -20,6 +20,7 @@ import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import org.astraea.app.metrics.HasBeanObject;
@@ -145,11 +146,12 @@ public class LoadCost implements HasBrokerCost {
 
   /** @return the metrics getters. Those getters are used to fetch mbeans. */
   @Override
-  public Fetcher fetcher() {
-    return client ->
-        List.of(
-            KafkaMetrics.BrokerTopic.BytesInPerSec.fetch(client),
-            KafkaMetrics.BrokerTopic.BytesOutPerSec.fetch(client));
+  public Optional<Fetcher> fetcher() {
+    return Optional.of(
+        client ->
+            List.of(
+                KafkaMetrics.BrokerTopic.BytesInPerSec.fetch(client),
+                KafkaMetrics.BrokerTopic.BytesOutPerSec.fetch(client)));
   }
 
   private static class BrokerMetric {
diff --git a/app/src/main/java/org/astraea/app/cost/MemoryCost.java b/app/src/main/java/org/astraea/app/cost/MemoryCost.java
index 1aed9d6a..5a33a9f8 100644
--- a/app/src/main/java/org/astraea/app/cost/MemoryCost.java
+++ b/app/src/main/java/org/astraea/app/cost/MemoryCost.java
@@ -19,6 +19,7 @@ package org.astraea.app.cost;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import org.astraea.app.metrics.collector.Fetcher;
@@ -86,8 +87,8 @@ public class MemoryCost extends Periodic<Map<Integer, Double>> implements HasBro
   }
 
   @Override
-  public Fetcher fetcher() {
-    return client -> List.of(KafkaMetrics.Host.jvmMemory(client));
+  public Optional<Fetcher> fetcher() {
+    return Optional.of(client -> List.of(KafkaMetrics.Host.jvmMemory(client)));
   }
 
   private static class BrokerMetric {
diff --git a/app/src/main/java/org/astraea/app/cost/NeutralIntegratedCost.java b/app/src/main/java/org/astraea/app/cost/NeutralIntegratedCost.java
index eb82ef5f..46bee83e 100644
--- a/app/src/main/java/org/astraea/app/cost/NeutralIntegratedCost.java
+++ b/app/src/main/java/org/astraea/app/cost/NeutralIntegratedCost.java
@@ -20,6 +20,7 @@ import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.stream.Collectors;
 import org.astraea.app.metrics.collector.Fetcher;
 
@@ -33,7 +34,8 @@ import org.astraea.app.metrics.collector.Fetcher;
 public class NeutralIntegratedCost implements HasBrokerCost {
   private final List<HasBrokerCost> metricsCost =
       List.of(new BrokerInputCost(), new BrokerOutputCost(), new CpuCost(), new MemoryCost());
-  private final Map<Integer, BrokerMetrics> brokersMetric = new HashMap<>();
+  // Visible for test
+  Map<Integer, BrokerMetrics> brokersMetric = new HashMap<>();
   private final AHPEmpowerment ahpEmpowerment = new AHPEmpowerment();
   private final WeightProvider weightProvider = WeightProvider.entropy(Normalizer.minMax(true));
 
@@ -49,58 +51,80 @@ public class NeutralIntegratedCost implements HasBrokerCost {
           }
         });
 
-    metricsCost.forEach(
-        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);
-          }
-        });
+    metricsCost.forEach(hasBrokerCost -> setBrokerMetrics(hasBrokerCost, clusterInfo));
 
+    var entropyEmpowerment = weight(weightProvider, brokersMetric);
+    var entropyEmpowermentSum =
+        entropyEmpowerment.entrySet().stream()
+            .mapToDouble(
+                entry -> entry.getValue() * ahpEmpowerment.empowerment().get(entry.getKey()))
+            .sum();
+    // The weight of each metric is obtained by combining Entropy and AHP.
     var integratedEmpowerment =
-        weight(weightProvider, brokersMetric).entrySet().stream()
+        entropyEmpowerment.entrySet().stream()
             .collect(
                 Collectors.toMap(
                     Map.Entry::getKey,
                     entry ->
-                        entry.getValue() * 0.5
-                            + ahpEmpowerment.empowerment().get(entry.getKey()) * 0.5));
+                        entry.getValue()
+                            * ahpEmpowerment.empowerment().get(entry.getKey())
+                            / entropyEmpowermentSum));
 
-    return () ->
+    var integratedScore =
         brokersMetric.entrySet().stream()
             .collect(
                 Collectors.toMap(
                     Map.Entry::getKey,
                     entry ->
-                        entry.getValue().inputScore
+                        entry.getValue().inputTScore
                                 * integratedEmpowerment.get(Metrics.inputThroughput.metricName)
-                            + entry.getValue().outputScore
+                            + entry.getValue().outputTScore
                                 * integratedEmpowerment.get(Metrics.outputThroughput.metricName)
-                            + entry.getValue().cpuScore
+                            + entry.getValue().cpuTScore
                                 * integratedEmpowerment.get(Metrics.cpu.metricName)
-                            + entry.getValue().memoryScore
+                            + entry.getValue().memoryTScore
                                 * integratedEmpowerment.get(Metrics.memory.metricName)));
+
+    return () -> integratedScore;
+  }
+
+  // Save the original value of each metric and the value of the metric after the TScore
+  // calculation.
+  // TODO Refactor
+  void setBrokerMetrics(HasBrokerCost hasBrokerCost, ClusterInfo clusterInfo) {
+    if (hasBrokerCost instanceof BrokerInputCost) {
+      var inputBrokerCost = hasBrokerCost.brokerCost(clusterInfo);
+      inputBrokerCost
+          .value()
+          .forEach((brokerID, value) -> brokersMetric.get(brokerID).inputScore = value);
+      inputBrokerCost
+          .normalize(Normalizer.TScore())
+          .value()
+          .forEach((brokerID, value) -> brokersMetric.get(brokerID).inputTScore = value);
+    } else if (hasBrokerCost instanceof BrokerOutputCost) {
+      var outPutBrokerCost = hasBrokerCost.brokerCost(clusterInfo);
+      outPutBrokerCost
+          .value()
+          .forEach((brokerID, value) -> brokersMetric.get(brokerID).outputScore = value);
+      outPutBrokerCost
+          .normalize(Normalizer.TScore())
+          .value()
+          .forEach((brokerID, value) -> brokersMetric.get(brokerID).outputTScore = value);
+    } else if (hasBrokerCost instanceof CpuCost) {
+      var CPUBrokerCost = hasBrokerCost.brokerCost(clusterInfo);
+      CPUBrokerCost.value()
+          .forEach((brokerID, value) -> brokersMetric.get(brokerID).cpuScore = value);
+      CPUBrokerCost.normalize(Normalizer.TScore())
+          .value()
+          .forEach((brokerID, value) -> brokersMetric.get(brokerID).cpuTScore = value);
+    } else if (hasBrokerCost instanceof MemoryCost) {
+      var MemoryBrokerCost = hasBrokerCost.brokerCost(clusterInfo);
+      MemoryBrokerCost.value()
+          .forEach((brokerID, value) -> brokersMetric.get(brokerID).memoryScore = value);
+      MemoryBrokerCost.normalize(Normalizer.TScore())
+          .value()
+          .forEach((brokerID, value) -> brokersMetric.get(brokerID).memoryTScore = value);
+    }
   }
 
   static Map<String, Double> weight(
@@ -137,8 +161,8 @@ public class NeutralIntegratedCost implements HasBrokerCost {
   }
 
   @Override
-  public Fetcher fetcher() {
-    return Fetcher.of(metricsCost.stream().map(CostFunction::fetcher).collect(Collectors.toList()));
+  public Optional<Fetcher> fetcher() {
+    return Fetcher.of(metricsCost);
   }
 
   static class BrokerMetrics {
@@ -146,6 +170,10 @@ public class NeutralIntegratedCost implements HasBrokerCost {
     double outputScore = 0.0;
     double cpuScore = 0.0;
     double memoryScore = 0.0;
+    double inputTScore = 0.0;
+    double outputTScore = 0.0;
+    double cpuTScore = 0.0;
+    double memoryTScore = 0.0;
 
     BrokerMetrics() {}
   }
diff --git a/app/src/main/java/org/astraea/app/cost/NumberOfLeaderCost.java b/app/src/main/java/org/astraea/app/cost/NumberOfLeaderCost.java
index c8be27f3..7dd80c1f 100644
--- a/app/src/main/java/org/astraea/app/cost/NumberOfLeaderCost.java
+++ b/app/src/main/java/org/astraea/app/cost/NumberOfLeaderCost.java
@@ -18,6 +18,7 @@ package org.astraea.app.cost;
 
 import java.util.Comparator;
 import java.util.Map;
+import java.util.Optional;
 import java.util.stream.Collectors;
 import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.collector.Fetcher;
@@ -31,8 +32,8 @@ import org.astraea.app.metrics.kafka.KafkaMetrics;
  */
 public class NumberOfLeaderCost implements HasBrokerCost {
   @Override
-  public Fetcher fetcher() {
-    return KafkaMetrics.ReplicaManager.LeaderCount::fetch;
+  public Optional<Fetcher> fetcher() {
+    return Optional.of(KafkaMetrics.ReplicaManager.LeaderCount::fetch);
   }
 
   /**
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 0c15dc04..eb6e25b9 100644
--- a/app/src/main/java/org/astraea/app/cost/ReplicaDiskInCost.java
+++ b/app/src/main/java/org/astraea/app/cost/ReplicaDiskInCost.java
@@ -20,6 +20,7 @@ import java.time.Duration;
 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.TopicPartition;
@@ -168,8 +169,8 @@ public class ReplicaDiskInCost implements HasBrokerCost, HasPartitionCost {
 
   /** @return the metrics getters. Those getters are used to fetch mbeans. */
   @Override
-  public Fetcher fetcher() {
-    return KafkaMetrics.TopicPartition.Size::fetch;
+  public Optional<Fetcher> fetcher() {
+    return Optional.of(KafkaMetrics.TopicPartition.Size::fetch);
   }
 
   /**
diff --git a/app/src/main/java/org/astraea/app/cost/ReplicaSizeCost.java b/app/src/main/java/org/astraea/app/cost/ReplicaSizeCost.java
index 7f373527..7d7e50fa 100644
--- a/app/src/main/java/org/astraea/app/cost/ReplicaSizeCost.java
+++ b/app/src/main/java/org/astraea/app/cost/ReplicaSizeCost.java
@@ -18,6 +18,7 @@ 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.TopicPartition;
@@ -39,8 +40,8 @@ public class ReplicaSizeCost implements HasBrokerCost, HasPartitionCost {
   }
 
   @Override
-  public Fetcher fetcher() {
-    return KafkaMetrics.TopicPartition.Size::fetch;
+  public Optional<Fetcher> fetcher() {
+    return Optional.of(KafkaMetrics.TopicPartition.Size::fetch);
   }
 
   /**
diff --git a/app/src/main/java/org/astraea/app/cost/ThroughputCost.java b/app/src/main/java/org/astraea/app/cost/ThroughputCost.java
index 2ace776f..27d0434f 100644
--- a/app/src/main/java/org/astraea/app/cost/ThroughputCost.java
+++ b/app/src/main/java/org/astraea/app/cost/ThroughputCost.java
@@ -19,6 +19,7 @@ package org.astraea.app.cost;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.stream.Collectors;
 import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.collector.Fetcher;
@@ -55,10 +56,11 @@ public class ThroughputCost implements HasBrokerCost {
   }
 
   @Override
-  public Fetcher fetcher() {
-    return client ->
-        List.of(
-            KafkaMetrics.BrokerTopic.BytesInPerSec.fetch(client),
-            KafkaMetrics.BrokerTopic.BytesOutPerSec.fetch(client));
+  public Optional<Fetcher> fetcher() {
+    return Optional.of(
+        client ->
+            List.of(
+                KafkaMetrics.BrokerTopic.BytesInPerSec.fetch(client),
+                KafkaMetrics.BrokerTopic.BytesOutPerSec.fetch(client)));
   }
 }
diff --git a/app/src/main/java/org/astraea/app/metrics/collector/Fetcher.java b/app/src/main/java/org/astraea/app/metrics/collector/Fetcher.java
index c0a5e3d3..818c227a 100644
--- a/app/src/main/java/org/astraea/app/metrics/collector/Fetcher.java
+++ b/app/src/main/java/org/astraea/app/metrics/collector/Fetcher.java
@@ -17,7 +17,9 @@
 package org.astraea.app.metrics.collector;
 
 import java.util.Collection;
+import java.util.Optional;
 import java.util.stream.Collectors;
+import org.astraea.app.cost.CostFunction;
 import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.jmx.MBeanClient;
 
@@ -25,14 +27,24 @@ import org.astraea.app.metrics.jmx.MBeanClient;
 public interface Fetcher {
 
   /**
-   * merge multiples fetchers to single one.
+   * merge all fetchers into single one
    *
-   * @param fs fetchers
-   * @return single fetcher
+   * @param functions cost function
+   * @return fetcher if there is available fetcher. Otherwise, empty is returned
    */
-  static Fetcher of(Collection<Fetcher> fs) {
-    return client ->
-        fs.stream().flatMap(f -> f.fetch(client).stream()).collect(Collectors.toUnmodifiableList());
+  static Optional<Fetcher> of(Collection<? extends CostFunction> functions) {
+    var fs =
+        functions.stream()
+            .map(CostFunction::fetcher)
+            .filter(Optional::isPresent)
+            .map(Optional::get)
+            .collect(Collectors.toUnmodifiableList());
+    if (fs.isEmpty()) return Optional.empty();
+    return Optional.of(
+        client ->
+            fs.stream()
+                .flatMap(f -> f.fetch(client).stream())
+                .collect(Collectors.toUnmodifiableList()));
   }
 
   /**
diff --git a/app/src/main/java/org/astraea/app/partitioner/StrictCostDispatcher.java b/app/src/main/java/org/astraea/app/partitioner/StrictCostDispatcher.java
index bc087276..50f1c009 100644
--- a/app/src/main/java/org/astraea/app/partitioner/StrictCostDispatcher.java
+++ b/app/src/main/java/org/astraea/app/partitioner/StrictCostDispatcher.java
@@ -149,11 +149,8 @@ public class StrictCostDispatcher implements Dispatcher {
         .register()
         .host(host)
         .port(port)
-        .fetcher(
-            Fetcher.of(
-                functions.keySet().stream()
-                    .map(CostFunction::fetcher)
-                    .collect(Collectors.toUnmodifiableList())))
+        // TODO: return optional receiver
+        .fetcher(Fetcher.of(functions.keySet()).get())
         .build();
   }
 
diff --git a/app/src/main/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobinDispatcher.java b/app/src/main/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobinDispatcher.java
index 330a2408..9846dc07 100644
--- a/app/src/main/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobinDispatcher.java
+++ b/app/src/main/java/org/astraea/app/partitioner/smooth/SmoothWeightRoundRobinDispatcher.java
@@ -150,7 +150,8 @@ public class SmoothWeightRoundRobinDispatcher extends Periodic<Map<Integer, Doub
         .register()
         .host(host)
         .port(port)
-        .fetcher(neutralIntegratedCost.fetcher())
+        // TODO: handle the empty fetcher
+        .fetcher(neutralIntegratedCost.fetcher().get())
         .build();
   }
 
diff --git a/app/src/main/java/org/astraea/app/web/RecordHandler.java b/app/src/main/java/org/astraea/app/web/RecordHandler.java
index c80d51db..024217df 100644
--- a/app/src/main/java/org/astraea/app/web/RecordHandler.java
+++ b/app/src/main/java/org/astraea/app/web/RecordHandler.java
@@ -127,6 +127,8 @@ public class RecordHandler implements Handler {
           consumer
               .poll(
                   maxRecords,
+                  // TODO: Use DurationField format in timeout
+                  // (https://github.com/skiptests/astraea/issues/450)
                   Duration.ofSeconds(Integer.parseInt(queries.getOrDefault(TIMEOUT, "5"))))
               .stream()
               .map(Record::new)


Test patch:
diff --git a/app/src/test/java/org/astraea/app/cost/BrokerInputCostTest.java b/app/src/test/java/org/astraea/app/cost/BrokerInputCostTest.java
index b1c43fed..ef70eac5 100644
--- a/app/src/test/java/org/astraea/app/cost/BrokerInputCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/BrokerInputCostTest.java
@@ -68,7 +68,7 @@ public class BrokerInputCostTest extends RequireBrokerCluster {
             .register()
             .host(jmxServiceURL().getHost())
             .port(jmxServiceURL().getPort())
-            .fetcher(new BrokerInputCost().fetcher())
+            .fetcher(new BrokerInputCost().fetcher().get())
             .build()) {
       Assertions.assertFalse(receiver.current().isEmpty());
 
diff --git a/app/src/test/java/org/astraea/app/cost/BrokerOutPutCostTest.java b/app/src/test/java/org/astraea/app/cost/BrokerOutPutCostTest.java
index ed72efa9..3f258bfc 100644
--- a/app/src/test/java/org/astraea/app/cost/BrokerOutPutCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/BrokerOutPutCostTest.java
@@ -57,7 +57,7 @@ public class BrokerOutPutCostTest extends RequireBrokerCluster {
             .register()
             .host(jmxServiceURL().getHost())
             .port(jmxServiceURL().getPort())
-            .fetcher(new BrokerOutputCost().fetcher())
+            .fetcher(new BrokerOutputCost().fetcher().get())
             .build()) {
       Assertions.assertFalse(receiver.current().isEmpty());
 
diff --git a/app/src/test/java/org/astraea/app/cost/CpuCostTest.java b/app/src/test/java/org/astraea/app/cost/CpuCostTest.java
index 88f62b3e..26959105 100644
--- a/app/src/test/java/org/astraea/app/cost/CpuCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/CpuCostTest.java
@@ -109,7 +109,7 @@ public class CpuCostTest extends RequireBrokerCluster {
             .register()
             .host(jmxServiceURL().getHost())
             .port(jmxServiceURL().getPort())
-            .fetcher(new CpuCost().fetcher())
+            .fetcher(new CpuCost().fetcher().get())
             .build()) {
       Assertions.assertFalse(receiver.current().isEmpty());
       Assertions.assertTrue(
diff --git a/app/src/test/java/org/astraea/app/cost/MemoryCostTest.java b/app/src/test/java/org/astraea/app/cost/MemoryCostTest.java
index 4364b009..ff41c0e2 100644
--- a/app/src/test/java/org/astraea/app/cost/MemoryCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/MemoryCostTest.java
@@ -111,7 +111,7 @@ public class MemoryCostTest extends RequireBrokerCluster {
             .register()
             .host(jmxServiceURL().getHost())
             .port(jmxServiceURL().getPort())
-            .fetcher(new MemoryCost().fetcher())
+            .fetcher(new MemoryCost().fetcher().get())
             .build()) {
       Assertions.assertFalse(receiver.current().isEmpty());
       Assertions.assertTrue(receiver.current().stream().allMatch(o -> o instanceof JvmMemory));
diff --git a/app/src/test/java/org/astraea/app/cost/NeutralIntegratedCostTest.java b/app/src/test/java/org/astraea/app/cost/NeutralIntegratedCostTest.java
new file mode 100644
index 00000000..6820960f
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/cost/NeutralIntegratedCostTest.java
@@ -0,0 +1,77 @@
+/*
+ * 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.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+public class NeutralIntegratedCostTest {
+  @Test
+  void testSetBrokerMetrics() {
+    var neutralIntegratedCost = new NeutralIntegratedCost();
+    var clusterInfo = Mockito.mock(ClusterInfo.class);
+    neutralIntegratedCost.brokersMetric =
+        new HashMap<>(
+            Map.of(
+                0,
+                new NeutralIntegratedCost.BrokerMetrics(),
+                1,
+                new NeutralIntegratedCost.BrokerMetrics(),
+                2,
+                new NeutralIntegratedCost.BrokerMetrics()));
+
+    HasBrokerCost input = Mockito.mock(BrokerInputCost.class);
+    HasBrokerCost output = Mockito.mock(BrokerOutputCost.class);
+    HasBrokerCost cpu = Mockito.mock(CpuCost.class);
+    HasBrokerCost memory = Mockito.mock(MemoryCost.class);
+
+    // Convert Map to BrokerCost, Map represents the score of each node metrics
+    var inputBrokerCost = getBrokerCost(new HashMap<>(Map.of(0, 50.0, 1, 100.0, 2, 150.0)));
+    var outputBrokerCost = getBrokerCost(new HashMap<>(Map.of(0, 500.0, 1, 1000.0, 2, 1500.0)));
+    var cpuBrokerCost = getBrokerCost(new HashMap<>(Map.of(0, 0.05, 1, 0.1, 2, 0.15)));
+    var memoryBrokerCost = getBrokerCost(new HashMap<>(Map.of(0, 0.1, 1, 0.3, 2, 0.5)));
+    Mockito.when(input.brokerCost(clusterInfo)).thenReturn(inputBrokerCost);
+    Mockito.when(output.brokerCost(clusterInfo)).thenReturn(outputBrokerCost);
+    Mockito.when(memory.brokerCost(clusterInfo)).thenReturn(memoryBrokerCost);
+    Mockito.when(cpu.brokerCost(clusterInfo)).thenReturn(cpuBrokerCost);
+    List<HasBrokerCost> metricsCost = List.of(input, output, cpu, memory);
+    metricsCost.forEach(
+        hasBrokerCost -> neutralIntegratedCost.setBrokerMetrics(hasBrokerCost, clusterInfo));
+
+    Assertions.assertEquals(neutralIntegratedCost.brokersMetric.get(0).inputScore, 50.0);
+    Assertions.assertEquals(neutralIntegratedCost.brokersMetric.get(1).inputScore, 100.0);
+    Assertions.assertEquals(neutralIntegratedCost.brokersMetric.get(2).inputScore, 150.0);
+    Assertions.assertEquals(neutralIntegratedCost.brokersMetric.get(0).outputScore, 500.0);
+    Assertions.assertEquals(neutralIntegratedCost.brokersMetric.get(1).outputScore, 1000.0);
+    Assertions.assertEquals(neutralIntegratedCost.brokersMetric.get(2).outputScore, 1500.0);
+
+    Assertions.assertEquals(neutralIntegratedCost.brokersMetric.get(0).cpuTScore, 0.38);
+    Assertions.assertEquals(neutralIntegratedCost.brokersMetric.get(1).cpuTScore, 0.5);
+    Assertions.assertEquals(neutralIntegratedCost.brokersMetric.get(2).cpuTScore, 0.62);
+    Assertions.assertEquals(neutralIntegratedCost.brokersMetric.get(0).memoryTScore, 0.38);
+    Assertions.assertEquals(neutralIntegratedCost.brokersMetric.get(1).memoryTScore, 0.5);
+    Assertions.assertEquals(neutralIntegratedCost.brokersMetric.get(2).memoryTScore, 0.62);
+  }
+
+  BrokerCost getBrokerCost(Map<Integer, Double> map) {
+    return () -> map;
+  }
+}
diff --git a/app/src/test/java/org/astraea/app/cost/NumberOfLeaderCostTest.java b/app/src/test/java/org/astraea/app/cost/NumberOfLeaderCostTest.java
index f24bf8fe..74ccd364 100644
--- a/app/src/test/java/org/astraea/app/cost/NumberOfLeaderCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/NumberOfLeaderCostTest.java
@@ -21,7 +21,6 @@ import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -29,7 +28,6 @@ import org.astraea.app.admin.Admin;
 import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.collector.BeanCollector;
-import org.astraea.app.metrics.collector.Fetcher;
 import org.astraea.app.metrics.jmx.BeanObject;
 import org.astraea.app.metrics.kafka.HasValue;
 import org.astraea.app.metrics.kafka.KafkaMetrics;
@@ -73,7 +71,7 @@ class NumberOfLeaderCostTest extends RequireBrokerCluster {
                     .register()
                     .host(host)
                     .port(port)
-                    .fetcher(Fetcher.of(Set.of(costFunction.fetcher())))
+                    .fetcher(costFunction.fetcher().get())
                     .build()
                     .current();
             allBeans.put(
diff --git a/app/src/test/java/org/astraea/app/cost/ReplicaSizeCostTest.java b/app/src/test/java/org/astraea/app/cost/ReplicaSizeCostTest.java
index acaa50e7..a2c30e3c 100644
--- a/app/src/test/java/org/astraea/app/cost/ReplicaSizeCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/ReplicaSizeCostTest.java
@@ -30,7 +30,6 @@ import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.TopicPartition;
 import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.collector.BeanCollector;
-import org.astraea.app.metrics.collector.Fetcher;
 import org.astraea.app.metrics.jmx.BeanObject;
 import org.astraea.app.metrics.kafka.HasValue;
 import org.astraea.app.metrics.kafka.KafkaMetrics;
@@ -59,7 +58,7 @@ class ReplicaSizeCostTest extends RequireSingleBrokerCluster {
               .register()
               .host(host)
               .port(jmxServiceURL().getPort())
-              .fetcher(Fetcher.of(Set.of(costFunction.fetcher())))
+              .fetcher(costFunction.fetcher().get())
               .build()
               .current();
       var replicaSize =
diff --git a/app/src/test/java/org/astraea/app/cost/WeightProviderTest.java b/app/src/test/java/org/astraea/app/cost/WeightProviderTest.java
index e5f0708f..8a6841b7 100644
--- a/app/src/test/java/org/astraea/app/cost/WeightProviderTest.java
+++ b/app/src/test/java/org/astraea/app/cost/WeightProviderTest.java
@@ -17,16 +17,86 @@
 package org.astraea.app.cost;
 
 import java.util.Comparator;
+import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
 import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
 
 public class WeightProviderTest {
+  @Test
+  void testPositiveEntropy() {
+    Normalizer normalizer = Normalizer.minMax(true);
+    var weightProvider = new WeightProvider.EntropyWeightProvider(normalizer);
+    var confusion =
+        IntStream.range(0, 1)
+            .boxed()
+            .collect(
+                Collectors.toMap(
+                    String::valueOf,
+                    ignored ->
+                        IntStream.range(0, 100)
+                            .mapToObj(i -> Math.random() * i)
+                            .collect(Collectors.toUnmodifiableList())));
+    confusion.put(
+        "1",
+        IntStream.range(0, 100)
+            .mapToObj(i -> 0.0 + i % 10 * 0.1)
+            .collect(Collectors.toUnmodifiableList()));
+    var entropy = weightProvider.weight(confusion);
+    // "0" range of 0 to 100 numbers * Math.random().
+    // "1" range of 0 to 1 numbers.The number is accurate to one decimal places.
+    // "0" is more confusing than "1".
+    Assertions.assertTrue(entropy.get("0") > entropy.get("1"));
+
+    var metrics1 = Math.random();
+    var metrics2 = Math.random();
+    var metrics3 = Math.random();
+    var metrics =
+        Map.of(
+            0,
+            List.of(metrics1, metrics2, metrics3),
+            1,
+            List.of(metrics1 * 100, metrics2 * 100, metrics3 * 100));
+    var sameEntropy = weightProvider.weight(metrics);
+    Assertions.assertEquals(
+        Math.round(sameEntropy.get(0) * 100) / 100, Math.round(sameEntropy.get(1) * 100) / 100);
+
+    var AABAndABB =
+        Map.of(
+            0,
+            List.of(metrics1 * 10, metrics2, metrics2),
+            1,
+            List.of(metrics1 * 10, metrics1 * 10, metrics2));
+    var ABEntropy = weightProvider.weight(AABAndABB);
+    Assertions.assertEquals(ABEntropy.get(0), ABEntropy.get(1));
+
+    var uniformDistribution =
+        IntStream.range(0, 1)
+            .boxed()
+            .collect(
+                Collectors.toMap(
+                    String::valueOf,
+                    ignored ->
+                        IntStream.range(0, 100)
+                            .mapToObj(i -> 1.0 + i % 10 * 0.1)
+                            .collect(Collectors.toUnmodifiableList())));
+    // Since entropy represents the degree of uncertainty, uniform distribution does not represent
+    // certainty in the concept of information entropy. On the contrary, because the uniform
+    // distribution means that it is possible at every point, the degree of uncertainty is greater
+    // and the weight given by the entropy method is also greater.
+    uniformDistribution.put(
+        "1",
+        IntStream.range(0, 100).mapToObj(i -> 0.0 + i).collect(Collectors.toUnmodifiableList()));
+
+    var uniformEntropy = weightProvider.weight(uniformDistribution);
+    Assertions.assertTrue(uniformEntropy.get("0") < uniformEntropy.get("1"));
+  }
 
   @ParameterizedTest
   @MethodSource("normalizers")
@@ -51,7 +121,6 @@ public class WeightProviderTest {
         weightProvider.entropies(raw).entrySet().stream()
             .sorted(Map.Entry.comparingByValue())
             .collect(Collectors.toUnmodifiableList());
-
     entropies.forEach(e -> Assertions.assertTrue(0 <= e.getValue() && e.getValue() <= 1));
 
     var weights =
diff --git a/app/src/test/java/org/astraea/app/metrics/collector/FetcherTest.java b/app/src/test/java/org/astraea/app/metrics/collector/FetcherTest.java
index 0bdbc556..28ec73c6 100644
--- a/app/src/test/java/org/astraea/app/metrics/collector/FetcherTest.java
+++ b/app/src/test/java/org/astraea/app/metrics/collector/FetcherTest.java
@@ -17,6 +17,8 @@
 package org.astraea.app.metrics.collector;
 
 import java.util.List;
+import java.util.Optional;
+import org.astraea.app.cost.CostFunction;
 import org.astraea.app.metrics.HasBeanObject;
 import org.astraea.app.metrics.jmx.MBeanClient;
 import org.junit.jupiter.api.Assertions;
@@ -32,7 +34,22 @@ public class FetcherTest {
     var mbean1 = Mockito.mock(HasBeanObject.class);
     Fetcher fetcher1 = client -> List.of(mbean1);
 
-    var fetcher = Fetcher.of(List.of(fetcher1, fetcher0));
+    var fetcher =
+        Fetcher.of(
+                List.of(
+                    new CostFunction() {
+                      @Override
+                      public Optional<Fetcher> fetcher() {
+                        return Optional.of(fetcher0);
+                      }
+                    },
+                    new CostFunction() {
+                      @Override
+                      public Optional<Fetcher> fetcher() {
+                        return Optional.of(fetcher1);
+                      }
+                    }))
+            .get();
 
     var result = fetcher.fetch(Mockito.mock(MBeanClient.class));
 
@@ -40,4 +57,19 @@ public class FetcherTest {
     Assertions.assertTrue(result.contains(mbean0));
     Assertions.assertTrue(result.contains(mbean1));
   }
+
+  @Test
+  void testEmptyCostFunction() {
+    Assertions.assertEquals(Optional.empty(), Fetcher.of(List.of()));
+    Assertions.assertEquals(
+        Optional.empty(),
+        Fetcher.of(
+            List.of(
+                new CostFunction() {
+                  @Override
+                  public Optional<Fetcher> fetcher() {
+                    return Optional.empty();
+                  }
+                })));
+  }
 }
diff --git a/app/src/test/java/org/astraea/app/partitioner/StrictCostDispatcherTest.java b/app/src/test/java/org/astraea/app/partitioner/StrictCostDispatcherTest.java
index 56c407a5..2ef04315 100644
--- a/app/src/test/java/org/astraea/app/partitioner/StrictCostDispatcherTest.java
+++ b/app/src/test/java/org/astraea/app/partitioner/StrictCostDispatcherTest.java
@@ -30,7 +30,6 @@ import org.astraea.app.cost.HasBrokerCost;
 import org.astraea.app.cost.NodeInfo;
 import org.astraea.app.cost.ReplicaInfo;
 import org.astraea.app.cost.ThroughputCost;
-import org.astraea.app.metrics.collector.Fetcher;
 import org.astraea.app.metrics.collector.Receiver;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
@@ -115,11 +114,6 @@ public class StrictCostDispatcherTest {
                             Function.identity(), id -> id.equals(n0.id()) ? 0.9D : 0.5D));
             return () -> brokerCost;
           }
-
-          @Override
-          public Fetcher fetcher() {
-            return client -> List.of();
-          }
         };
     var costFunction2 =
         new HasBrokerCost() {
@@ -132,11 +126,6 @@ public class StrictCostDispatcherTest {
                             Function.identity(), id -> id.equals(n0.id()) ? 0.6D : 0.8D));
             return () -> brokerCost;
           }
-
-          @Override
-          public Fetcher fetcher() {
-            return client -> List.of();
-          }
         };
 
     try (var dispatcher =