Repository: skiptests/astraea
Pull request #538 (2022-08-01 17:42:10)

Apply Utils.constructCostFunction
as title

chia7712: @chinghongfang @garyparrot 你們的意見我已經處理，再請看一下

Patch:
diff --git a/app/src/main/java/org/astraea/app/App.java b/app/src/main/java/org/astraea/app/App.java
index c1177822..bb64f1b5 100644
--- a/app/src/main/java/org/astraea/app/App.java
+++ b/app/src/main/java/org/astraea/app/App.java
@@ -23,6 +23,7 @@ import java.util.List;
 import java.util.Map;
 import org.astraea.app.admin.ReplicaSyncingMonitor;
 import org.astraea.app.automation.Automation;
+import org.astraea.app.balancer.BalanceProcessDemo;
 import org.astraea.app.metrics.MetricExplorer;
 import org.astraea.app.performance.Performance;
 import org.astraea.app.web.WebService;
@@ -34,7 +35,8 @@ public class App {
           "performance", Performance.class,
           "monitor", ReplicaSyncingMonitor.class,
           "automation", Automation.class,
-          "web", WebService.class);
+          "web", WebService.class,
+          "balance-demo", BalanceProcessDemo.class);
 
   static void execute(Map<String, Class<?>> mains, List<String> args) throws Throwable {
 
diff --git a/app/src/main/java/org/astraea/app/balancer/BalanceProcessDemo.java b/app/src/main/java/org/astraea/app/balancer/BalanceProcessDemo.java
new file mode 100644
index 00000000..d83c962a
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/balancer/BalanceProcessDemo.java
@@ -0,0 +1,72 @@
+/*
+ * 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.balancer;
+
+import com.beust.jcommander.Parameter;
+import java.util.Set;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+import org.astraea.app.admin.Admin;
+import org.astraea.app.balancer.executor.RebalanceAdmin;
+import org.astraea.app.balancer.executor.StraightPlanExecutor;
+import org.astraea.app.balancer.generator.ShufflePlanGenerator;
+import org.astraea.app.balancer.log.ClusterLogAllocation;
+
+/**
+ * A simple demo for Balancer, it does the following:
+ *
+ * <ol>
+ *   <li>Generate a random plan for the target cluster.
+ *   <li>Print the information of the generated plan.
+ *   <li>Execute the generated plan against the target cluster.
+ * </ol>
+ */
+public class BalanceProcessDemo {
+
+  public static void main(String[] args) {
+    var argument = org.astraea.app.argument.Argument.parse(new Argument(), args);
+    var rebalancePlanGenerator = new ShufflePlanGenerator(1, 5);
+    var rebalancePlanExecutor = new StraightPlanExecutor();
+    try (var admin = Admin.of(argument.configs())) {
+      Predicate<String> topicFilter = (topic) -> !argument.ignoredTopics.contains(topic);
+      var topics =
+          admin.topicNames().stream().filter(topicFilter).collect(Collectors.toUnmodifiableSet());
+      var clusterInfo = admin.clusterInfo(topics);
+
+      // TODO: implement one interface to select the best plan from many plan ,see #544
+      var rebalancePlan = rebalancePlanGenerator.generate(clusterInfo).findFirst().orElseThrow();
+      System.out.println(rebalancePlan);
+      var targetAllocation =
+          rebalancePlan
+              .rebalancePlan()
+              .orElseThrow(() -> new IllegalStateException("No suitable plan found"));
+
+      System.out.println("[Target Allocation]");
+      System.out.println(ClusterLogAllocation.toString(targetAllocation));
+
+      var rebalanceAdmin = RebalanceAdmin.of(admin, topicFilter);
+      rebalancePlanExecutor.run(rebalanceAdmin, targetAllocation);
+    }
+  }
+
+  public static class Argument extends org.astraea.app.argument.Argument {
+
+    @Parameter(names = {"--ignored.topics"})
+    public Set<String> ignoredTopics =
+        Set.of("__consumer_offsets", "__transaction_state", "__cluster_metadata");
+  }
+}
diff --git a/app/src/main/java/org/astraea/app/balancer/log/ClusterLogAllocation.java b/app/src/main/java/org/astraea/app/balancer/log/ClusterLogAllocation.java
index 482418df..fe96cd54 100644
--- a/app/src/main/java/org/astraea/app/balancer/log/ClusterLogAllocation.java
+++ b/app/src/main/java/org/astraea/app/balancer/log/ClusterLogAllocation.java
@@ -150,7 +150,10 @@ public interface ClusterLogAllocation {
                   .forEach(
                       log ->
                           stringBuilder.append(
-                              String.format("%s(%s) ", log.broker(), log.logDirectory())));
+                              String.format(
+                                  "(%s, %s) ",
+                                  log.broker(),
+                                  log.logDirectory().orElse("log dir not specified"))));
 
               stringBuilder.append(System.lineSeparator());
             });
diff --git a/app/src/main/java/org/astraea/app/common/Utils.java b/app/src/main/java/org/astraea/app/common/Utils.java
index bc770546..7a140361 100644
--- a/app/src/main/java/org/astraea/app/common/Utils.java
+++ b/app/src/main/java/org/astraea/app/common/Utils.java
@@ -17,7 +17,6 @@
 package org.astraea.app.common;
 
 import java.io.File;
-import java.lang.reflect.Constructor;
 import java.net.InetAddress;
 import java.net.ServerSocket;
 import java.nio.file.Files;
@@ -256,22 +255,16 @@ public final class Utils {
         .thenApply(f -> futures.stream().map(CompletableFuture::join).collect(Collectors.toList()));
   }
 
-  public static CostFunction constructCostFunction(
-      Class<? extends CostFunction> costClass, Configuration configuration) {
-    for (Constructor<?> constructor : costClass.getConstructors()) {
-      Class<?>[] types = constructor.getParameterTypes();
-      if (types.length == 1 && types[0].isAssignableFrom(Configuration.class)) {
-        return (CostFunction) Utils.packException(() -> constructor.newInstance(configuration));
-      }
-    }
-    for (Constructor<?> constructor : costClass.getConstructors()) {
-      Class<?>[] types = constructor.getParameterTypes();
-      if (types.length == 0) {
-        return (CostFunction) Utils.packException(() -> constructor.newInstance());
-      }
+  public static <T extends CostFunction> T constructCostFunction(
+      Class<T> costClass, Configuration configuration) {
+    try {
+      // case 0: create cost function by configuration
+      var constructor = costClass.getConstructor(Configuration.class);
+      return Utils.packException(() -> constructor.newInstance(configuration));
+    } catch (NoSuchMethodException e) {
+      // case 1: create cost function by empty constructor
+      return Utils.packException(() -> costClass.getConstructor().newInstance());
     }
-    throw new IllegalArgumentException(
-        "No suitable constructor found for class " + costClass.getName());
   }
 
   private Utils() {}
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 1fc30053..ccc6ace6 100644
--- a/app/src/main/java/org/astraea/app/partitioner/StrictCostDispatcher.java
+++ b/app/src/main/java/org/astraea/app/partitioner/StrictCostDispatcher.java
@@ -62,7 +62,7 @@ public class StrictCostDispatcher implements Dispatcher {
   Duration roundRobinLease;
 
   // The cost-functions we consider and the weight of them. It is visible for test
-  Map<CostFunction, Double> functions = Map.of();
+  Map<HasBrokerCost, Double> functions = Map.of();
 
   // all-in-one fetcher referenced to cost functions
   Optional<Fetcher> fetcher;
@@ -159,10 +159,9 @@ public class StrictCostDispatcher implements Dispatcher {
    * @return SmoothWeightedRoundRobin
    */
   static RoundRobin<Integer> newRoundRobin(
-      Map<CostFunction, Double> costFunctions, ClusterInfo clusterInfo, ClusterBean clusterBean) {
+      Map<HasBrokerCost, Double> costFunctions, ClusterInfo clusterInfo, ClusterBean clusterBean) {
     var weightedCost =
         costFunctions.entrySet().stream()
-            .filter(e -> e.getKey() instanceof HasBrokerCost)
             .flatMap(
                 functionWeight ->
                     ((HasBrokerCost) functionWeight.getKey())
@@ -202,7 +201,7 @@ public class StrictCostDispatcher implements Dispatcher {
    * @param customJmxPort jmx port for each node
    */
   void configure(
-      Map<CostFunction, Double> functions,
+      Map<HasBrokerCost, Double> functions,
       Optional<Integer> jmxPortDefault,
       Map<Integer, Integer> customJmxPort,
       Duration roundRobinLease) {
@@ -234,36 +233,29 @@ public class StrictCostDispatcher implements Dispatcher {
    * @param config that contains cost-function names and its corresponding weight
    * @return pairs of cost-function object and its corresponding weight
    */
-  public static Map<CostFunction, Double> parseCostFunctionWeight(Configuration config) {
+  @SuppressWarnings("unchecked")
+  public static Map<HasBrokerCost, Double> parseCostFunctionWeight(Configuration config) {
     return config.entrySet().stream()
         .map(
             nameAndWeight -> {
-              Class<?> name;
-              double weight;
+              Class<?> clz;
               try {
-                name = Class.forName(nameAndWeight.getKey());
-                weight = Double.parseDouble(nameAndWeight.getValue());
-                if (weight < 0.0)
-                  throw new IllegalArgumentException("Cost-function weight should not be negative");
+                clz = Class.forName(nameAndWeight.getKey());
               } catch (ClassNotFoundException ignore) {
-                /* To delete all config option that is not for configuring cost-function. */
+                // this config is not cost function, so we just skip it.
                 return null;
               }
-              return Map.entry(name, weight);
+              var weight = Double.parseDouble(nameAndWeight.getValue());
+              if (weight < 0.0)
+                throw new IllegalArgumentException("Cost-function weight should not be negative");
+              return Map.entry(clz, weight);
             })
         .filter(Objects::nonNull)
-        .filter(e -> CostFunction.class.isAssignableFrom(e.getKey()))
-        .map(
-            e -> {
-              try {
-                return Map.entry(
-                    (CostFunction) e.getKey().getConstructor().newInstance(), e.getValue());
-              } catch (Exception ex) {
-                ex.printStackTrace();
-                throw new IllegalArgumentException(ex);
-              }
-            })
-        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+        .filter(e -> HasBrokerCost.class.isAssignableFrom(e.getKey()))
+        .collect(
+            Collectors.toMap(
+                e -> Utils.constructCostFunction((Class<HasBrokerCost>) e.getKey(), config),
+                Map.Entry::getValue));
   }
 
   @Override
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 92e74fee..d43ee1a5 100644
--- a/app/src/main/java/org/astraea/app/web/RecordHandler.java
+++ b/app/src/main/java/org/astraea/app/web/RecordHandler.java
@@ -152,12 +152,7 @@ public class RecordHandler implements Handler {
 
     try (var consumer = consumerBuilder.build()) {
       var limit = Integer.parseInt(queries.getOrDefault(LIMIT, "1"));
-      return new Records(
-          consumer.poll(limit, timeout).stream()
-              .map(Record::new)
-              // TODO: remove limit here (https://github.com/skiptests/astraea/issues/441)
-              .limit(limit)
-              .collect(toList()));
+      return new Records(consumer.poll(limit, timeout).stream().map(Record::new).collect(toList()));
     }
   }
 
diff --git a/docs/web_server/web_api_records_chinese.md b/docs/web_server/web_api_records_chinese.md
index c8554b44..cddf76e7 100644
--- a/docs/web_server/web_api_records_chinese.md
+++ b/docs/web_server/web_api_records_chinese.md
@@ -113,7 +113,7 @@ GET /records/{topic}
 | partition             | (選填) 指定要讀取之 partition                                                                                        | 無      |
 | keyDeserializer       | (選填) key deserializer                                                                                        | string |
 | valueDeserializer     | (選填) value deserializer                                                                                      | string |
-| limit                 | (選填) 回傳資料筆數上限                                                                                                | 1      |
+| limit                 | (選填) 回傳資料筆數上限，注意此僅為一建議值，您仍有可能取得大於此值之資料筆數                                                                     | 1      |
 | timeout               | (選填) 請求資料最大時限，逾時便中斷本次操作。格式為 `數值` + `單位`，單位可填寫 `days`, `day`, `h`, `m`, `s`, `ms`, `us`, `ns`。範例：1h，5s，1000ms | 5s     |
 | distanceFromLatest    | (選填) 距離最新 offset 往前多少位移量開始拉取資料                                                                               | 無      |
 | distanceFromBeginning | (選填) 距離起始 offset 往後多少位移量開始拉取資料                                                                               | 無      |


Test patch:
diff --git a/app/src/test/java/org/astraea/app/balancer/BalanceProcessDemoTest.java b/app/src/test/java/org/astraea/app/balancer/BalanceProcessDemoTest.java
new file mode 100644
index 00000000..47b79a12
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/balancer/BalanceProcessDemoTest.java
@@ -0,0 +1,65 @@
+/*
+ * 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.balancer;
+
+import java.time.Duration;
+import org.astraea.app.admin.Admin;
+import org.astraea.app.balancer.log.ClusterLogAllocation;
+import org.astraea.app.common.Utils;
+import org.astraea.app.service.RequireBrokerCluster;
+import org.junit.jupiter.api.Test;
+
+public class BalanceProcessDemoTest extends RequireBrokerCluster {
+
+  @Test
+  void run() {
+    // prepare topics
+    createTopics();
+
+    // before operation allocation
+    System.out.println("[Before]");
+    describeCurrentAllocation();
+
+    // run
+    System.out.println("[Execute]");
+    BalanceProcessDemo.main(new String[] {"--bootstrap.servers", bootstrapServers()});
+
+    // after operation allocation
+    System.out.println("[After]");
+    describeCurrentAllocation();
+  }
+
+  void describeCurrentAllocation() {
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      var cla = ClusterLogAllocation.of(admin.clusterInfo());
+      System.out.println(ClusterLogAllocation.toString(cla));
+      System.out.println();
+    }
+  }
+
+  void createTopics() {
+    try (Admin admin = Admin.of(bootstrapServers())) {
+      admin
+          .creator()
+          .topic("TestingTopic")
+          .numberOfPartitions(10)
+          .numberOfReplicas((short) 2)
+          .create();
+    }
+    Utils.sleep(Duration.ofSeconds(1));
+  }
+}
diff --git a/app/src/test/java/org/astraea/app/common/UtilsTest.java b/app/src/test/java/org/astraea/app/common/UtilsTest.java
index 47db461d..0a000fa1 100644
--- a/app/src/test/java/org/astraea/app/common/UtilsTest.java
+++ b/app/src/test/java/org/astraea/app/common/UtilsTest.java
@@ -121,6 +121,6 @@ public class UtilsTest {
 
     // act, assert
     Assertions.assertThrows(
-        IllegalArgumentException.class, () -> Utils.constructCostFunction(aClass, config));
+        RuntimeException.class, () -> Utils.constructCostFunction(aClass, config));
   }
 }
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 d747a17e..7be347f0 100644
--- a/app/src/test/java/org/astraea/app/partitioner/StrictCostDispatcherTest.java
+++ b/app/src/test/java/org/astraea/app/partitioner/StrictCostDispatcherTest.java
@@ -28,7 +28,6 @@ import org.astraea.app.admin.ReplicaInfo;
 import org.astraea.app.common.Utils;
 import org.astraea.app.cost.BrokerCost;
 import org.astraea.app.cost.BrokerInputCost;
-import org.astraea.app.cost.CostFunction;
 import org.astraea.app.cost.HasBrokerCost;
 import org.astraea.app.cost.NodeThroughputCost;
 import org.astraea.app.cost.ReplicaLeaderCost;
@@ -164,7 +163,7 @@ public class StrictCostDispatcherTest {
 
   @Test
   void testCostFunctionWithoutFetcher() {
-    var costFunction = new CostFunction() {};
+    HasBrokerCost costFunction = (clusterInfo, bean) -> Mockito.mock(BrokerCost.class);
     var replicaInfo0 = ReplicaInfo.of("topic", 0, NodeInfo.of(10, "host", 11111), true, true, true);
     var replicaInfo1 =
         ReplicaInfo.of("topic", 1, NodeInfo.of(12, "host2", 11111), true, true, true);
@@ -183,7 +182,12 @@ public class StrictCostDispatcherTest {
   @Test
   void testReceivers() {
     var costFunction =
-        new CostFunction() {
+        new HasBrokerCost() {
+          @Override
+          public BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
+            return Mockito.mock(BrokerCost.class);
+          }
+
           @Override
           public Optional<Fetcher> fetcher() {
             return Optional.of(Mockito.mock(Fetcher.class));
diff --git a/app/src/test/java/org/astraea/app/web/RecordHandlerTest.java b/app/src/test/java/org/astraea/app/web/RecordHandlerTest.java
index de7b1c1c..cefc8ba1 100644
--- a/app/src/test/java/org/astraea/app/web/RecordHandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/RecordHandlerTest.java
@@ -303,9 +303,9 @@ public class RecordHandlerTest extends RequireBrokerCluster {
                 Optional.of(topic),
                 Map.of(DISTANCE_FROM_LATEST, "2", VALUE_DESERIALIZER, "integer")));
 
-    Assertions.assertEquals(1, response.data.size());
+    Assertions.assertEquals(2, response.data.size());
     Assertions.assertEquals(
-        List.of(8), response.data.stream().map(record -> record.value).collect(toList()));
+        List.of(8, 9), response.data.stream().map(record -> record.value).collect(toList()));
   }
 
   @Test
@@ -321,9 +321,9 @@ public class RecordHandlerTest extends RequireBrokerCluster {
                 Optional.of(topic),
                 Map.of(DISTANCE_FROM_BEGINNING, "8", VALUE_DESERIALIZER, "integer")));
 
-    Assertions.assertEquals(1, response.data.size());
+    Assertions.assertEquals(2, response.data.size());
     Assertions.assertEquals(
-        List.of(8), response.data.stream().map(record -> record.value).collect(toList()));
+        List.of(8, 9), response.data.stream().map(record -> record.value).collect(toList()));
   }
 
   @Test
@@ -337,9 +337,9 @@ public class RecordHandlerTest extends RequireBrokerCluster {
             RecordHandler.Records.class,
             handler.get(Optional.of(topic), Map.of(SEEK_TO, "3", VALUE_DESERIALIZER, "integer")));
 
-    Assertions.assertEquals(1, response.data.size());
+    Assertions.assertEquals(2, response.data.size());
     Assertions.assertEquals(
-        List.of(3), response.data.stream().map(record -> record.value).collect(toList()));
+        List.of(3, 4), response.data.stream().map(record -> record.value).collect(toList()));
   }
 
   @Test
@@ -370,8 +370,8 @@ public class RecordHandlerTest extends RequireBrokerCluster {
             RecordHandler.Records.class,
             handler.get(Optional.of(topic), Map.of(DISTANCE_FROM_BEGINNING, "1", PARTITION, "1")));
 
-    Assertions.assertEquals(
-        List.of(1), response.data.stream().map(r -> r.partition).collect(toList()));
+    Assertions.assertTrue(
+        response.data.stream().map(r -> r.partition).filter(p -> p != 1).findAny().isEmpty());
   }
 
   @Test
@@ -387,9 +387,11 @@ public class RecordHandlerTest extends RequireBrokerCluster {
                 Optional.of(topic),
                 Map.of(DISTANCE_FROM_BEGINNING, "2", LIMIT, "3", VALUE_DESERIALIZER, "integer")));
 
-    Assertions.assertEquals(3, response.data.size());
+    // limit is just a recommended size here, we might get more records than limit
+    Assertions.assertEquals(8, response.data.size());
     Assertions.assertEquals(
-        List.of(2, 3, 4), response.data.stream().map(record -> record.value).collect(toList()));
+        List.of(2, 3, 4, 5, 6, 7, 8, 9),
+        response.data.stream().map(record -> record.value).collect(toList()));
   }
 
   @ParameterizedTest