[BALANCER] Demonstration of ResourceBalancer & ResourceUsageHint

app/src/main/java/org/astraea/app/web/Scenario.java

@@ -16,13 +16,13 @@
  */
 package org.astraea.app.web;
 
-import java.util.Map;
 import java.util.concurrent.CompletionStage;
+import org.astraea.common.Configuration;
 import org.astraea.common.Utils;
 import org.astraea.common.admin.Admin;
 
 /** The subclass of this class should contain the logic to fulfill a scenario. */
-public interface Scenario {
+public interface Scenario<Result> {
 
   static Builder builder() {
     return new Builder();
@@ -56,54 +56,12 @@ public Builder binomialProbability(double binomialProbability) {
       return this;
     }
 
-    public Scenario build() {
+    public Scenario<SkewedPartitionScenario.Result> build() {
       return new SkewedPartitionScenario(
           topicName, numberOfPartitions, numberOfReplicas, binomialProbability);
     }
   }
 
   /** Apply this scenario to the Kafka cluster */
-  CompletionStage<Result> apply(Admin admin);
-
-  class Result {
-
-    private final String topicName;
-    private final int numberOfPartitions;
-    private final short numberOfReplicas;
-    private final Map<Integer, Long> leaderSum;
-    private final Map<Integer, Long> logSum;
-
-    public Result(
-        String topicName,
-        int numberOfPartitions,
-        short numberOfReplicas,
-        Map<Integer, Long> leaderSum,
-        Map<Integer, Long> logSum) {
-      this.topicName = topicName;
-      this.numberOfPartitions = numberOfPartitions;
-      this.numberOfReplicas = numberOfReplicas;
-      this.leaderSum = leaderSum;
-      this.logSum = logSum;
-    }
-
-    public String topicName() {
-      return topicName;
-    }
-
-    public int numberOfPartitions() {
-      return numberOfPartitions;
-    }
-
-    public short numberOfReplicas() {
-      return numberOfReplicas;
-    }
-
-    public Map<Integer, Long> leaderSum() {
-      return leaderSum;
-    }
-
-    public Map<Integer, Long> logSum() {
-      return logSum;
-    }
-  }
+  CompletionStage<Result> apply(Admin admin, Configuration scenarioConfig);
 }

app/src/main/java/org/astraea/app/web/SkewedPartitionScenario.java

@@ -29,12 +29,13 @@
 import org.apache.commons.math3.distribution.EnumeratedDistribution;
 import org.apache.commons.math3.distribution.IntegerDistribution;
 import org.apache.commons.math3.util.Pair;
+import org.astraea.common.Configuration;
 import org.astraea.common.admin.Admin;
 import org.astraea.common.admin.NodeInfo;
 import org.astraea.common.admin.TopicPartition;
 import org.astraea.common.admin.TopicPartitionReplica;
 
-public class SkewedPartitionScenario implements Scenario {
+public class SkewedPartitionScenario implements Scenario<SkewedPartitionScenario.Result> {
 
   final String topicName;
   final int partitions;
@@ -50,7 +51,7 @@ public SkewedPartitionScenario(
   }
 
   @Override
-  public CompletionStage<Result> apply(Admin admin) {
+  public CompletionStage<Result> apply(Admin admin, Configuration scenarioConfig) {
     return admin
         .creator()
         .topic(topicName)
@@ -138,4 +139,46 @@ public static List<Integer> sampledReplicaList(
     }
     return result;
   }
+
+  public static class Result {
+
+    private final String topicName;
+    private final int numberOfPartitions;
+    private final short numberOfReplicas;
+    private final Map<Integer, Long> leaderSum;
+    private final Map<Integer, Long> logSum;
+
+    public Result(
+        String topicName,
+        int numberOfPartitions,
+        short numberOfReplicas,
+        Map<Integer, Long> leaderSum,
+        Map<Integer, Long> logSum) {
+      this.topicName = topicName;
+      this.numberOfPartitions = numberOfPartitions;
+      this.numberOfReplicas = numberOfReplicas;
+      this.leaderSum = leaderSum;
+      this.logSum = logSum;
+    }
+
+    public String topicName() {
+      return topicName;
+    }
+
+    public int numberOfPartitions() {
+      return numberOfPartitions;
+    }
+
+    public short numberOfReplicas() {
+      return numberOfReplicas;
+    }
+
+    public Map<Integer, Long> leaderSum() {
+      return leaderSum;
+    }
+
+    public Map<Integer, Long> logSum() {
+      return logSum;
+    }
+  }
 }

app/src/main/java/org/astraea/app/web/TopicHandler.java

@@ -26,6 +26,7 @@
 import java.util.concurrent.CompletionStage;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
+import org.astraea.common.Configuration;
 import org.astraea.common.FutureUtils;
 import org.astraea.common.Utils;
 import org.astraea.common.admin.Admin;
@@ -175,7 +176,7 @@ public CompletionStage<Topics> post(Channel channel) {
                             .numberOfReplicas(numberOfReplicas)
                             .binomialProbability(topic.probability.get())
                             .build()
-                            .apply(admin)
+                            .apply(admin, Configuration.EMPTY)
                             .thenApply(ignored -> null)
                             .toCompletableFuture();
                       }

app/src/main/java/org/astraea/balancer/bench/CostProfilingImpl.java

@@ -17,20 +17,27 @@
 package org.astraea.balancer.bench;
 
 import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Collections;
+import java.util.LinkedList;
+import java.util.List;
 import java.util.LongSummaryStatistics;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import org.astraea.common.admin.ClusterBean;
 import org.astraea.common.admin.ClusterInfo;
 import org.astraea.common.balancer.AlgorithmConfig;
 import org.astraea.common.balancer.Balancer;
 import org.astraea.common.cost.ClusterCost;
+import org.astraea.common.cost.CompositeClusterCost;
 import org.astraea.common.cost.HasClusterCost;
 import org.astraea.common.cost.HasMoveCost;
 import org.astraea.common.cost.MoveCost;
+import org.astraea.common.cost.ResourceUsageHint;
 import org.astraea.common.metrics.collector.MetricSensor;
 
 class CostProfilingImpl implements BalancerBenchmark.CostProfilingBuilder {
@@ -85,7 +92,22 @@ public CompletableFuture<BalancerBenchmark.CostProfilingResult> start() {
             .clusterBean(clusterBean)
             .timeout(timeout)
             .clusterCost(
-                new HasClusterCost() {
+                new CompositeClusterCost() {
+                  @Override
+                  public Collection<? extends HasClusterCost> functions() {
+                    var queue = new LinkedList<HasClusterCost>(List.of(costFunction));
+                    var functions = new ArrayList<HasClusterCost>();
+
+                    while (!queue.isEmpty()) {
+                      var next = queue.pop();
+                      if (next instanceof CompositeClusterCost)
+                        functions.addAll(((CompositeClusterCost) next).functions());
+                      else functions.add(next);
+                    }
+
+                    return List.copyOf(functions);
+                  }
+
                   @Override
                   public ClusterCost clusterCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
                     final var start = System.nanoTime();
@@ -124,6 +146,11 @@ public Optional<MetricSensor> metricSensor() {
                     return moveCostFunction.metricSensor();
                   }
 
+                  @Override
+                  public Set<? extends ResourceUsageHint> resourceUsageHint() {
+                    return moveCostFunction.resourceUsageHint();
+                  }
+
                   @Override
                   public String toString() {
                     return moveCostFunction.toString();

app/src/test/java/org/astraea/app/BackboneImbalanceApplyTest.java

@@ -0,0 +1,106 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.astraea.app.web.BackboneImbalanceScenario;
+import org.astraea.common.Configuration;
+import org.astraea.common.admin.Admin;
+import org.astraea.common.json.JsonConverter;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+
+public class BackboneImbalanceApplyTest {
+
+  public static final String realCluster =
+      "192.168.103.177:25655,192.168.103.178:25655,192.168.103.179:25655,192.168.103.180:25655,192.168.103.181:25655,192.168.103.182:25655";
+  public static final List<String> clients =
+      List.of(
+          "192.168.103.184",
+          "192.168.103.142",
+          "192.168.103.183",
+          "192.168.103.141",
+          "192.168.103.143",
+          "192.168.103.144",
+          "192.168.103.145");
+
+  @Test
+  @Disabled
+  void testBackbone() {
+    try (Admin admin = Admin.of(realCluster)) {
+      var scenario = new BackboneImbalanceScenario();
+      var config =
+          Configuration.of(
+              Map.ofEntries(
+                  Map.entry(BackboneImbalanceScenario.CONFIG_PERF_ZIPFIAN_EXPONENT, "1.6"),
+                  Map.entry(BackboneImbalanceScenario.CONFIG_PERF_KEY_TABLE_SEED, "0"),
+                  Map.entry(
+                      BackboneImbalanceScenario.CONFIG_PERF_CLIENT_COUNT,
+                      Integer.toString(clients.size()))));
+      var result = scenario.apply(admin, config).toCompletableFuture().join();
+      // print summary
+      var converter = JsonConverter.defaultConverter();
+      System.out.println(converter.toJson(result));
+      // save result to json format
+      var ansibleInventory = converter.toJson(toAnsibleInventory(result));
+      var ansibleInventoryFile =
+          "/home/garyparrot/Programming/ansible/backbone-imbalance-scenario-inventory.json";
+      try (var stream = Files.newBufferedWriter(Path.of(ansibleInventoryFile))) {
+        stream.write(ansibleInventory);
+      } catch (IOException e) {
+        e.printStackTrace();
+      }
+    }
+  }
+
+  Map<?, ?> toAnsibleInventory(BackboneImbalanceScenario.Result result) {
+    var hosts =
+        IntStream.range(0, clients.size())
+            .boxed()
+            .collect(
+                Collectors.toUnmodifiableMap(
+                    clients::get,
+                    index -> {
+                      var clientHostname = clients.get(index);
+                      var clientPerf = result.perfCommands().get(index);
+                      return Map.ofEntries(
+                          Map.entry("ansible_host", clientHostname),
+                          Map.entry("ansible_user", "kafka"),
+                          Map.entry("expected_produce_rate", clientPerf.get("produce_rate")),
+                          Map.entry("expected_consume_rate", clientPerf.get("consume_rate")),
+                          Map.entry("key_table_seed", clientPerf.get("key_table_seed")),
+                          Map.entry("key_distribution", clientPerf.get("key_distribution")),
+                          Map.entry(
+                              "key_distribution_config", clientPerf.get("key_distribution_config")),
+                          Map.entry("throttle", clientPerf.get("throttle")),
+                          Map.entry("throttle_enable", !clientPerf.get("backbone").equals("true")),
+                          Map.entry("throughput", clientPerf.get("throughput")),
+                          Map.entry("throughput_enable", clientPerf.get("backbone").equals("true")),
+                          Map.entry("no_producer", clientPerf.get("no_producer")),
+                          Map.entry("no_consumer", clientPerf.get("no_consumer")),
+                          Map.entry("topics", clientPerf.get("topics")));
+                    }));
+
+    return Map.of("backbone_imbalance_hosts", Map.of("hosts", hosts));
+  }
+}