Repository: skiptests/astraea
Pull request #454 (2022-07-04 08:18:41)

StrictCostDispatcher should fail immediately if jmx ports are undefined
1. 當有cost functions需要JMX，而JMX又沒有被設置的時候，`StrictCostDispatcher`會直接拋錯
2. 移除`StrictCostDispatcher`多餘的建構式，改用`configure`來測試

chia7712: > StrictCostDispatcher#partition最後回傳的是partition的編號，不是broker id。

不好意思，你是指哪一個路徑回傳了broker id嗎？
chia7712: @chinghongfang 阿阿 我知道你講什麼了，我開另外一隻來處理

Patch:
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 bdb4b6f5..a246316b 100644
--- a/app/src/main/java/org/astraea/app/partitioner/StrictCostDispatcher.java
+++ b/app/src/main/java/org/astraea/app/partitioner/StrictCostDispatcher.java
@@ -17,9 +17,7 @@
 package org.astraea.app.partitioner;
 
 import java.time.Duration;
-import java.util.Collection;
 import java.util.HashMap;
-import java.util.List;
 import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.Objects;
@@ -58,8 +56,11 @@ public class StrictCostDispatcher implements Dispatcher {
   private final BeanCollector beanCollector =
       BeanCollector.builder().interval(Duration.ofSeconds(4)).build();
 
-  /* The cost-functions we consider and the weight of them. It is visible for test.*/
-  Map<CostFunction, Double> functions;
+  // The cost-functions we consider and the weight of them. It is visible for test
+  Map<CostFunction, Double> functions = Map.of(CostFunction.throughput(), 1D);
+
+  // all-in-one fetcher referenced to cost functions
+  Optional<Fetcher> fetcher;
 
   Function<Integer, Integer> jmxPortGetter =
       (id) -> {
@@ -71,18 +72,6 @@ public class StrictCostDispatcher implements Dispatcher {
   volatile RoundRobin<Integer> roundRobin;
   volatile long timeToUpdateRoundRobin = -1;
 
-  public StrictCostDispatcher() {
-    this(List.of(CostFunction.throughput()));
-  }
-
-  // visible for testing
-  StrictCostDispatcher(Collection<CostFunction> functions) {
-    this.functions =
-        functions.stream()
-            .map(f -> Map.entry(f, 1.0))
-            .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
-  }
-
   // visible for testing
   Receiver receiver(String host, int port, Fetcher fetcher) {
     return beanCollector.register().host(host).port(port).fetcher(fetcher).build();
@@ -99,7 +88,7 @@ public class StrictCostDispatcher implements Dispatcher {
 
     // add new receivers for new brokers
     receivers.putAll(
-        Fetcher.of(functions.keySet())
+        fetcher
             .map(
                 fetcher ->
                     partitionLeaders.stream()
@@ -165,16 +154,34 @@ public class StrictCostDispatcher implements Dispatcher {
 
   @Override
   public void configure(Configuration config) {
-    var jmxPortDefault = config.integer(JMX_PORT);
-    var customJmxPort = PartitionerUtils.parseIdJMXPort(config);
-    jmxPortGetter =
+    configure(
+        parseCostFunctionWeight(config),
+        config.integer(JMX_PORT),
+        PartitionerUtils.parseIdJMXPort(config));
+  }
+
+  /**
+   * configure this StrictCostDispatcher. This method is extracted for testing.
+   *
+   * @param functions cost functions used by this dispatcher.
+   * @param jmxPortDefault jmx port by default
+   * @param customJmxPort jmx port for each node
+   */
+  void configure(
+      Map<CostFunction, Double> functions,
+      Optional<Integer> jmxPortDefault,
+      Map<Integer, Integer> customJmxPort) {
+    this.functions = functions;
+    this.fetcher = Fetcher.of(functions.keySet());
+    this.jmxPortGetter =
         id ->
             Optional.ofNullable(customJmxPort.get(id))
                 .or(() -> jmxPortDefault)
                 .orElseThrow(
                     () -> new NoSuchElementException("broker: " + id + " does not have jmx port"));
-
-    functions = parseCostFunctionWeight(config);
+    if (fetcher.isPresent() && jmxPortDefault.isEmpty() && customJmxPort.isEmpty())
+      throw new IllegalArgumentException(
+          "JMX port is empty but the cost functions need metrics from JMX server");
   }
 
   /**


Test patch:
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 4d8d5dfc..1146d497 100644
--- a/app/src/test/java/org/astraea/app/partitioner/StrictCostDispatcherTest.java
+++ b/app/src/test/java/org/astraea/app/partitioner/StrictCostDispatcherTest.java
@@ -68,7 +68,13 @@ public class StrictCostDispatcherTest {
       // Test for cost functions configuring
       dispatcher.configure(
           Configuration.of(
-              Map.of(ThroughputCost.class.getName(), "0.1", BrokerInputCost.class.getName(), "2")));
+              Map.of(
+                  ThroughputCost.class.getName(),
+                  "0.1",
+                  BrokerInputCost.class.getName(),
+                  "2",
+                  "jmx.port",
+                  "1111")));
       Assertions.assertEquals(2, dispatcher.functions.size());
     }
   }
@@ -78,7 +84,13 @@ public class StrictCostDispatcherTest {
     try (var dispatcher = new StrictCostDispatcher()) {
       dispatcher.configure(
           Configuration.of(
-              Map.of(ThroughputCost.class.getName(), "0.1", BrokerInputCost.class.getName(), "2")));
+              Map.of(
+                  ThroughputCost.class.getName(),
+                  "0.1",
+                  BrokerInputCost.class.getName(),
+                  "2",
+                  "jmx.port",
+                  "1111")));
       Assertions.assertEquals(2, dispatcher.functions.size());
     }
   }
@@ -87,7 +99,8 @@ public class StrictCostDispatcherTest {
   void testNoAvailableBrokers() {
     var clusterInfo = Mockito.mock(ClusterInfo.class);
     Mockito.when(clusterInfo.availableReplicaLeaders(Mockito.anyString())).thenReturn(List.of());
-    try (var dispatcher = new StrictCostDispatcher(List.of())) {
+    try (var dispatcher = new StrictCostDispatcher()) {
+      dispatcher.configure(Map.of(), Optional.empty(), Map.of());
       Assertions.assertEquals(
           0, dispatcher.partition("topic", new byte[0], new byte[0], clusterInfo));
     }
@@ -100,7 +113,8 @@ public class StrictCostDispatcherTest {
     var clusterInfo = Mockito.mock(ClusterInfo.class);
     Mockito.when(clusterInfo.availableReplicaLeaders(Mockito.anyString()))
         .thenReturn(List.of(replicaInfo));
-    try (var dispatcher = new StrictCostDispatcher(List.of())) {
+    try (var dispatcher = new StrictCostDispatcher()) {
+      dispatcher.configure(Map.of(), Optional.empty(), Map.of());
       Assertions.assertEquals(
           10, dispatcher.partition("topic", new byte[0], new byte[0], clusterInfo));
     }
@@ -154,7 +168,8 @@ public class StrictCostDispatcherTest {
     Mockito.when(clusterInfo.availableReplicaLeaders(Mockito.anyString()))
         .thenReturn(List.of(replicaInfo0, replicaInfo1));
     Mockito.when(clusterInfo.clusterBean()).thenReturn(ClusterBean.of(Map.of()));
-    try (var dispatcher = new StrictCostDispatcher(List.of(costFunction))) {
+    try (var dispatcher = new StrictCostDispatcher()) {
+      dispatcher.configure(Map.of(costFunction, 1D), Optional.empty(), Map.of());
       dispatcher.partition("topic", new byte[0], new byte[0], clusterInfo);
       Assertions.assertNotNull(dispatcher.roundRobin);
       Assertions.assertEquals(0, dispatcher.receivers.size());
@@ -173,7 +188,7 @@ public class StrictCostDispatcherTest {
     var jmxPort = 12345;
     var count = new AtomicInteger();
     var dispatcher =
-        new StrictCostDispatcher(List.of(costFunction)) {
+        new StrictCostDispatcher() {
           @Override
           Receiver receiver(String host, int port, Fetcher fetcher) {
             Assertions.assertEquals(jmxPort, port);
@@ -181,7 +196,7 @@ public class StrictCostDispatcherTest {
             return Mockito.mock(Receiver.class);
           }
         };
-    dispatcher.jmxPortGetter = ignored -> jmxPort;
+    dispatcher.configure(Map.of(costFunction, 1D), Optional.of(jmxPort), Map.of());
     var replicaInfo0 = ReplicaInfo.of("topic", 0, NodeInfo.of(10, "host", 11111), true, true, true);
     var replicaInfo1 = ReplicaInfo.of("topic", 1, NodeInfo.of(10, "host", 11111), true, true, true);
     var replicaInfo2 =
@@ -203,4 +218,20 @@ public class StrictCostDispatcherTest {
     Assertions.assertEquals(2, dispatcher.receivers.size());
     Assertions.assertEquals(2, count.get());
   }
+
+  @Test
+  void testEmptyJmxPort() {
+    var dispatcher = new StrictCostDispatcher();
+    Assertions.assertThrows(
+        IllegalArgumentException.class,
+        () ->
+            dispatcher.configure(
+                Map.of(CostFunction.throughput(), 1D), Optional.empty(), Map.of()));
+
+    // pass due to default port
+    dispatcher.configure(Map.of(CostFunction.throughput(), 1D), Optional.of(111), Map.of());
+
+    // pass due to specify port
+    dispatcher.configure(Map.of(CostFunction.throughput(), 1D), Optional.empty(), Map.of(222, 111));
+  }
 }