Repository: skiptests/astraea
Pull request #593 (2022-08-17 09:12:01)

Reconstruct `ReportFormat`
#579 
1. 將完成度移除
2. 改成使用當地時間紀錄
3. 修改 json 輸出格式

把資料名稱和資料統一放在一起了，方便新增、刪除要紀錄的數據。


Patch:
diff --git a/app/src/main/java/org/astraea/app/balancer/BalanceProcessDemo.java b/app/src/main/java/org/astraea/app/balancer/BalanceProcessDemo.java
index d83c962a..469efa2a 100644
--- a/app/src/main/java/org/astraea/app/balancer/BalanceProcessDemo.java
+++ b/app/src/main/java/org/astraea/app/balancer/BalanceProcessDemo.java
@@ -50,10 +50,7 @@ public class BalanceProcessDemo {
       // 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"));
+      var targetAllocation = rebalancePlan.rebalancePlan();
 
       System.out.println("[Target Allocation]");
       System.out.println(ClusterLogAllocation.toString(targetAllocation));
diff --git a/app/src/main/java/org/astraea/app/balancer/BalancerUtils.java b/app/src/main/java/org/astraea/app/balancer/BalancerUtils.java
new file mode 100644
index 00000000..919fa684
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/balancer/BalancerUtils.java
@@ -0,0 +1,185 @@
+/*
+ * 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.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Function;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.astraea.app.admin.ClusterBean;
+import org.astraea.app.admin.ClusterInfo;
+import org.astraea.app.admin.NodeInfo;
+import org.astraea.app.admin.ReplicaInfo;
+import org.astraea.app.admin.TopicPartition;
+import org.astraea.app.balancer.log.ClusterLogAllocation;
+import org.astraea.app.cost.HasClusterCost;
+import org.astraea.app.metrics.HasBeanObject;
+
+class BalancerUtils {
+
+  /**
+   * Create a {@link ClusterInfo} with its log placement replaced by {@link ClusterLogAllocation}.
+   * Every log will be marked as online & synced. Based on the given content in {@link
+   * ClusterLogAllocation}, some logs might not have its data directory specified. Noted that this
+   * method doesn't check if the given logs is suitable & exists in the cluster info base. the beans
+   * alongside the based cluster info might be out-of-date or even completely meaningless.
+   *
+   * @param clusterInfo the based cluster info
+   * @param allocation the log allocation to replace {@link ClusterInfo}'s log placement. If the
+   *     allocation implementation is {@link ClusterLogAllocation} then the given instance will be
+   *     locked.
+   * @return a {@link ClusterInfo} with its log placement replaced.
+   */
+  public static ClusterInfo merge(ClusterInfo clusterInfo, ClusterLogAllocation allocation) {
+    return new ClusterInfo() {
+      // TODO: maybe add a field to tell if this cluster info is mocked.
+
+      @Override
+      public List<NodeInfo> nodes() {
+        return clusterInfo.nodes();
+      }
+
+      @Override
+      public Set<String> dataDirectories(int brokerId) {
+        return clusterInfo.dataDirectories(brokerId);
+      }
+
+      @Override
+      public Set<String> topics() {
+        return allocation.topicPartitions().stream()
+            .map(TopicPartition::topic)
+            .collect(Collectors.toUnmodifiableSet());
+      }
+
+      @Override
+      public List<ReplicaInfo> availableReplicaLeaders(String topic) {
+        return replicas(topic).stream()
+            .filter(ReplicaInfo::isLeader)
+            .collect(Collectors.toUnmodifiableList());
+      }
+
+      @Override
+      public List<ReplicaInfo> availableReplicas(String topic) {
+        // there is no offline sense for a fake cluster info, so everything is online.
+        return replicas(topic);
+      }
+
+      @Override
+      public List<ReplicaInfo> replicas(String topic) {
+        Map<Integer, NodeInfo> nodeIdMap =
+            nodes().stream()
+                .collect(Collectors.toUnmodifiableMap(NodeInfo::id, Function.identity()));
+        var result =
+            allocation.topicPartitions().stream()
+                .filter(tp -> tp.topic().equals(topic))
+                .map(tp -> Map.entry(tp, allocation.logPlacements(tp)))
+                .flatMap(
+                    entry -> {
+                      var tp = entry.getKey();
+                      var logs = entry.getValue();
+
+                      return IntStream.range(0, logs.size())
+                          .mapToObj(
+                              i ->
+                                  ReplicaInfo.of(
+                                      tp.topic(),
+                                      tp.partition(),
+                                      nodeIdMap.get(logs.get(i).broker()),
+                                      i == 0,
+                                      true,
+                                      false,
+                                      logs.get(i).logDirectory().orElse(null)));
+                    })
+                .collect(Collectors.toUnmodifiableList());
+
+        if (result.isEmpty()) throw new NoSuchElementException();
+
+        return result;
+      }
+    };
+  }
+
+  public static Thread progressWatch(String title, double totalTasks, Supplier<Double> accTasks) {
+    AtomicInteger counter = new AtomicInteger();
+
+    Supplier<String> nextProgressBar =
+        () -> {
+          int blockCount = 20;
+          double percentagePerBlock = 1.0 / blockCount;
+          double now = accTasks.get();
+          double currentProgress = now / totalTasks;
+          int fulfilled = Math.min((int) (currentProgress / percentagePerBlock), blockCount);
+          int rollingBlock = blockCount - fulfilled >= 1 ? 1 : 0;
+          int emptyBlocks = blockCount - rollingBlock - fulfilled;
+
+          String rollingText = "-\\|/";
+          String filled = String.join("", Collections.nCopies(fulfilled, "-"));
+          String rolling =
+              String.join(
+                  "",
+                  Collections.nCopies(
+                      rollingBlock, "" + rollingText.charAt(counter.getAndIncrement() % 4)));
+          String empty = String.join("", Collections.nCopies(emptyBlocks, " "));
+          return String.format("[%s%s%s] (%.2f/%.2f)", filled, rolling, empty, now, totalTasks);
+        };
+
+    Runnable progressWatch =
+        () -> {
+          while (!Thread.currentThread().isInterrupted()) {
+            System.out.print("[" + title + "] " + nextProgressBar.get() + '\r');
+            try {
+              TimeUnit.MILLISECONDS.sleep(500);
+            } catch (InterruptedException e) {
+              break;
+            }
+          }
+          System.out.println("[" + title + "] " + nextProgressBar.get() + '\r');
+          System.out.println();
+        };
+
+    return new Thread(progressWatch);
+  }
+
+  static double evaluateCost(
+      ClusterInfo clusterInfo,
+      Map<HasClusterCost, Map<Integer, Collection<HasBeanObject>>> metrics) {
+    var scores =
+        metrics.keySet().stream()
+            .map(
+                cf -> {
+                  var theMetrics = metrics.get(cf);
+                  var clusterBean = ClusterBean.of(theMetrics);
+                  return Map.entry(cf, cf.clusterCost(clusterInfo, clusterBean).value());
+                })
+            .collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, Map.Entry::getValue));
+    return aggregateFunction(scores);
+  }
+
+  /** the lower, the better. */
+  static double aggregateFunction(Map<HasClusterCost, Double> scores) {
+    // use the simple summation result, treat every cost equally.
+    return scores.values().stream().mapToDouble(x -> x).sum();
+  }
+}
diff --git a/app/src/main/java/org/astraea/app/balancer/RebalancePlanProposal.java b/app/src/main/java/org/astraea/app/balancer/RebalancePlanProposal.java
index 12d5256a..f8c38e44 100644
--- a/app/src/main/java/org/astraea/app/balancer/RebalancePlanProposal.java
+++ b/app/src/main/java/org/astraea/app/balancer/RebalancePlanProposal.java
@@ -20,12 +20,11 @@ import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
-import java.util.Optional;
 import org.astraea.app.balancer.log.ClusterLogAllocation;
 
 public interface RebalancePlanProposal {
 
-  Optional<ClusterLogAllocation> rebalancePlan();
+  ClusterLogAllocation rebalancePlan();
 
   List<String> info();
 
@@ -40,45 +39,27 @@ public interface RebalancePlanProposal {
     List<String> info = Collections.synchronizedList(new ArrayList<>());
     List<String> warnings = Collections.synchronizedList(new ArrayList<>());
 
-    // guard by this
-    private boolean built = false;
-
-    public synchronized Build noRebalancePlan() {
-      ensureNotBuiltYet();
-      this.allocation = null;
-      return this;
-    }
-
     public synchronized Build withRebalancePlan(ClusterLogAllocation clusterLogAllocation) {
-      ensureNotBuiltYet();
       this.allocation = Objects.requireNonNull(clusterLogAllocation);
       return this;
     }
 
     public synchronized Build addWarning(String warning) {
-      ensureNotBuiltYet();
       this.warnings.add(warning);
       return this;
     }
 
     public synchronized Build addInfo(String info) {
-      ensureNotBuiltYet();
       this.info.add(info);
       return this;
     }
 
-    private synchronized void ensureNotBuiltYet() {
-      if (built) throw new IllegalStateException("This builder already built.");
-    }
-
     public synchronized RebalancePlanProposal build() {
-      final var allocationRef = allocation;
+      final var allocationRef =
+          Objects.requireNonNull(allocation, () -> "No log allocation specified for this proposal");
       final var infoRef = info;
       final var warningRef = warnings;
 
-      ensureNotBuiltYet();
-
-      built = true;
       allocation = null;
       info = null;
       warnings = null;
@@ -86,8 +67,8 @@ public interface RebalancePlanProposal {
       return new RebalancePlanProposal() {
 
         @Override
-        public Optional<ClusterLogAllocation> rebalancePlan() {
-          return Optional.ofNullable(allocationRef);
+        public ClusterLogAllocation rebalancePlan() {
+          return allocationRef;
         }
 
         @Override
diff --git a/app/src/main/java/org/astraea/app/balancer/generator/RebalancePlanGenerator.java b/app/src/main/java/org/astraea/app/balancer/generator/RebalancePlanGenerator.java
index 26c46ca2..0f30dd46 100644
--- a/app/src/main/java/org/astraea/app/balancer/generator/RebalancePlanGenerator.java
+++ b/app/src/main/java/org/astraea/app/balancer/generator/RebalancePlanGenerator.java
@@ -29,6 +29,13 @@ public interface RebalancePlanGenerator {
    * same plan for the same input argument. There can be some randomization that takes part in this
    * process.
    *
+   * <p>If the generator implementation thinks it can't find any rebalance proposal(which the plan
+   * might improve the cluster). Then the implementation should return a Stream with exactly one
+   * rebalance plan proposal in it, where the proposed allocation will be exactly the same as the
+   * {@code baseAllocation} parameter. This means there is no movement or alteration will occur. And
+   * the implementation should place some detailed information in the info/warning/error string, to
+   * indicate the reason for no meaningful plan.
+   *
    * @param clusterInfo the cluster state, implementation can take advantage of the data inside to
    *     proposal the plan it feels confident to improve the cluster.
    * @return a {@link Stream} generating rebalance plan regarding the given {@link ClusterInfo}
@@ -42,6 +49,13 @@ public interface RebalancePlanGenerator {
    * same plan for the same input argument. There can be some randomization that takes part in this
    * process.
    *
+   * <p>If the generator implementation thinks it can't find any rebalance proposal(which the plan
+   * might improve the cluster). Then the implementation should return a Stream with exactly one
+   * rebalance plan proposal in it, where the proposed allocation will be exactly the same as the
+   * {@code baseAllocation} parameter. This means there is no movement or alteration that will
+   * occur. And The implementation should place some detailed information in the info/warning/error
+   * string, to indicate the reason for no meaningful plan.
+   *
    * @param clusterInfo the cluster state, implementation can take advantage of the data inside to
    *     proposal the plan it feels confident to improve the cluster.
    * @param baseAllocation the cluster log allocation as the based of proposal generation.
diff --git a/app/src/main/java/org/astraea/app/balancer/generator/ShufflePlanGenerator.java b/app/src/main/java/org/astraea/app/balancer/generator/ShufflePlanGenerator.java
index 0de7f2d7..a098c53c 100644
--- a/app/src/main/java/org/astraea/app/balancer/generator/ShufflePlanGenerator.java
+++ b/app/src/main/java/org/astraea/app/balancer/generator/ShufflePlanGenerator.java
@@ -63,32 +63,36 @@ public class ShufflePlanGenerator implements RebalancePlanGenerator {
   @Override
   public Stream<RebalancePlanProposal> generate(
       ClusterInfo clusterInfo, ClusterLogAllocation baseAllocation) {
+    final var brokerIds =
+        clusterInfo.nodes().stream().map(NodeInfo::id).collect(Collectors.toUnmodifiableSet());
+
+    if (brokerIds.size() == 0) {
+      return Stream.of(
+          RebalancePlanProposal.builder()
+              .withRebalancePlan(baseAllocation)
+              .addWarning("There is no broker")
+              .build());
+    }
+
+    if (brokerIds.size() == 1) {
+      return Stream.of(
+          RebalancePlanProposal.builder()
+              .withRebalancePlan(baseAllocation)
+              .addWarning("Only one broker exists, unable to do some migration")
+              .build());
+    }
+
+    if (clusterInfo.topics().size() == 0) {
+      return Stream.of(
+          RebalancePlanProposal.builder()
+              .withRebalancePlan(baseAllocation)
+              .addWarning("No non-ignored topic to working on.")
+              .build());
+    }
+
     return Stream.generate(
         () -> {
           final var rebalancePlanBuilder = RebalancePlanProposal.builder();
-          final var brokerIds =
-              clusterInfo.nodes().stream()
-                  .map(NodeInfo::id)
-                  .collect(Collectors.toUnmodifiableSet());
-
-          if (brokerIds.size() == 0)
-            return rebalancePlanBuilder
-                .addWarning("Why there is no broker?")
-                .noRebalancePlan()
-                .build();
-
-          if (brokerIds.size() == 1)
-            return rebalancePlanBuilder
-                .addWarning("Only one broker exists. There is no reason to rebalance.")
-                .noRebalancePlan()
-                .build();
-
-          if (clusterInfo.topics().size() == 0)
-            return rebalancePlanBuilder
-                .addWarning("No non-ignored topic to working on.")
-                .noRebalancePlan()
-                .build();
-
           final var shuffleCount = numberOfShuffle.get();
 
           rebalancePlanBuilder.addInfo(
diff --git a/app/src/main/java/org/astraea/app/performance/Performance.java b/app/src/main/java/org/astraea/app/performance/Performance.java
index 346dfd41..1050ae01 100644
--- a/app/src/main/java/org/astraea/app/performance/Performance.java
+++ b/app/src/main/java/org/astraea/app/performance/Performance.java
@@ -155,14 +155,8 @@ public class Performance {
                 ReportFormat.createFileWriter(
                     param.reportFormat,
                     param.CSVPath,
-                    param.exeTime,
                     () -> consumerThreads.stream().allMatch(AbstractThread::closed),
                     () -> producerThreads.stream().allMatch(AbstractThread::closed),
-                    () ->
-                        producerThreads.stream()
-                            .map(ProducerThread::report)
-                            .mapToLong(Report::records)
-                            .sum(),
                     producerReports,
                     consumerReports));
 
diff --git a/app/src/main/java/org/astraea/app/performance/ReportFormat.java b/app/src/main/java/org/astraea/app/performance/ReportFormat.java
index 6520c00f..5e3b80d4 100644
--- a/app/src/main/java/org/astraea/app/performance/ReportFormat.java
+++ b/app/src/main/java/org/astraea/app/performance/ReportFormat.java
@@ -25,12 +25,13 @@ import java.nio.file.FileSystems;
 import java.nio.file.Path;
 import java.text.SimpleDateFormat;
 import java.time.Duration;
+import java.time.LocalTime;
+import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
 import java.util.Date;
 import java.util.List;
 import java.util.function.Supplier;
 import java.util.stream.IntStream;
-import org.astraea.app.common.DataSize;
-import org.astraea.app.common.DataUnit;
 import org.astraea.app.common.Utils;
 
 public enum ReportFormat {
@@ -60,10 +61,8 @@ public enum ReportFormat {
   public static Runnable createFileWriter(
       ReportFormat reportFormat,
       Path path,
-      ExeTime exeTime,
       Supplier<Boolean> consumerDone,
       Supplier<Boolean> producerDone,
-      Supplier<Long> producedRecords,
       List<ProducerThread.Report> producerReports,
       List<ConsumerThread.Report> consumerReports)
       throws IOException {
@@ -76,20 +75,14 @@ public enum ReportFormat {
                     + "."
                     + reportFormat);
     var writer = new BufferedWriter(new FileWriter(filePath.toFile()));
+    var elements = latencyAndIO(producerReports, consumerReports);
     switch (reportFormat) {
       case CSV:
-        initCSVFormat(writer, producerReports.size(), consumerReports.size());
+        initCSVFormat(writer, elements);
         return () -> {
           try {
-            while (true) {
-              if (logToCSV(
-                  writer,
-                  exeTime,
-                  consumerDone,
-                  producerDone,
-                  producedRecords,
-                  producerReports,
-                  consumerReports)) return;
+            while (!(producerDone.get() && consumerDone.get())) {
+              logToCSV(writer, elements);
               Utils.sleep(Duration.ofSeconds(1));
             }
           } finally {
@@ -100,15 +93,8 @@ public enum ReportFormat {
         writer.write("{");
         return () -> {
           try {
-            while (true) {
-              if (logToJSON(
-                  writer,
-                  exeTime,
-                  consumerDone,
-                  producerDone,
-                  producedRecords,
-                  producerReports,
-                  consumerReports)) return;
+            while (!(producerDone.get() && consumerDone.get())) {
+              logToJSON(writer, elements);
               Utils.sleep(Duration.ofSeconds(1));
             }
           } finally {
@@ -120,175 +106,110 @@ public enum ReportFormat {
     }
   }
 
-  private static void initCSVFormat(BufferedWriter writer, int producerCounts, int consumerCounts)
+  static void initCSVFormat(BufferedWriter writer, List<CSVContentElement> elements)
       throws IOException {
-    writer.write(
-        "Time \\ Name, Consumed/Produced, Output throughput (MiB/sec), Input throughput (MiB/sec), "
-            + "Publish max latency (ms), Publish min latency (ms), "
-            + "End-to-end max latency (ms), End-to-end min latency (ms)");
-    IntStream.range(0, producerCounts)
-        .forEach(
-            i -> {
-              try {
-                writer.write(
-                    ",Producer["
-                        + i
-                        + "] bytes produced, Producer["
-                        + i
-                        + "] average publish latency (ms)");
-              } catch (IOException ignore) {
-              }
-            });
-    IntStream.range(0, consumerCounts)
-        .forEach(
-            i -> {
-              try {
-                writer.write(
-                    ",Consumer["
-                        + i
-                        + "] bytes consumed, Consumer["
-                        + i
-                        + "] average ene-to-end latency (ms)");
-              } catch (IOException ignore) {
-              }
-            });
+    elements.forEach(element -> Utils.packException(() -> writer.write(element.title() + ", ")));
     writer.newLine();
   }
 
-  private static boolean logToCSV(
-      BufferedWriter writer,
-      ExeTime exeTime,
-      Supplier<Boolean> consumerDone,
-      Supplier<Boolean> producerDone,
-      Supplier<Long> producedRecords,
-      List<ProducerThread.Report> producerReports,
-      List<ConsumerThread.Report> consumerReports) {
-    var result = processResult(exeTime, producerReports, consumerReports, producedRecords);
-    if (producerReports.stream().mapToLong(Report::records).sum() == 0) return false;
+  static void logToCSV(BufferedWriter writer, List<CSVContentElement> elements) {
     try {
-      writer.write(
-          result.duration.toHoursPart()
-              + ":"
-              + result.duration.toMinutesPart()
-              + ":"
-              + result.duration.toSecondsPart());
-      writer.write(
-          String.format(",%.2f%% / %.2f%%", result.consumerPercentage, result.producerPercentage));
-      writer.write(
-          ","
-              + producerReports.stream().mapToLong(Report::max).max().orElse(0)
-              + ","
-              + producerReports.stream().mapToLong(Report::min).min().orElse(0));
-      writer.write(
-          ","
-              + consumerReports.stream().mapToLong(Report::max).max().orElse(0)
-              + ","
-              + consumerReports.stream().mapToLong(Report::min).min().orElse(0));
-      for (var r : producerReports) {
-        writer.write(
-            "," + DataSize.Byte.of(r.totalBytes()).measurement(DataUnit.MiB).doubleValue());
-        writer.write("," + r.avgLatency());
-      }
-      for (var r : consumerReports) {
-        writer.write(
-            "," + DataSize.Byte.of(r.totalBytes()).measurement(DataUnit.MiB).doubleValue());
-        writer.write("," + r.avgLatency());
-      }
+      elements.forEach(element -> Utils.packException(() -> writer.write(element.value() + ", ")));
       writer.newLine();
     } catch (IOException ignore) {
     }
-    return producerDone.get() && consumerDone.get();
   }
 
   /** Write to writer. Output: "(timestamp)": { (many metrics ...) } */
-  private static boolean logToJSON(
-      BufferedWriter writer,
-      ExeTime exeTime,
-      Supplier<Boolean> consumerDone,
-      Supplier<Boolean> producerDone,
-      Supplier<Long> producedRecords,
-      List<ProducerThread.Report> producerReports,
-      List<ConsumerThread.Report> consumerReports) {
-    var result = processResult(exeTime, producerReports, consumerReports, producedRecords);
-    if (producerReports.stream().mapToLong(Report::records).sum() == 0) return false;
+  static void logToJSON(BufferedWriter writer, List<CSVContentElement> elements) {
     try {
-      writer.write(
-          String.format(
-              "\"%dh%dm%ds\": {",
-              result.duration.toHoursPart(),
-              result.duration.toMinutesPart(),
-              result.duration.toSecondsPart()));
-      writer.write(
-          String.format(
-              "\"consumerPercentage\": %.2f%%, \"Producer Percentage\": %.2f%%",
-              result.consumerPercentage, result.producerPercentage));
-      writer.write(
-          ", \"outputThroughput\": "
-              + Utils.averageMB(
-                  result.duration, producerReports.stream().mapToLong(Report::totalBytes).sum()));
-      writer.write(
-          ", \"inputThroughput\": "
-              + Utils.averageMB(
-                  result.duration, consumerReports.stream().mapToLong(Report::totalBytes).sum()));
-      writer.write(
-          ", \"publishMaxLatency\": "
-              + producerReports.stream().mapToLong(Report::max).max().orElse(0));
-      writer.write(
-          ", \"publishMinLatency\": "
-              + producerReports.stream().mapToLong(Report::min).min().orElse(0));
-      writer.write(
-          ", \"E2EMaxLatency\": "
-              + consumerReports.stream().mapToLong(Report::max).max().orElse(0));
-      writer.write(
-          ", \"E2EMinLatency\": "
-              + consumerReports.stream().mapToLong(Report::min).min().orElse(0));
-
-      writer.write(", \"producerThroughput\": [");
-      for (var r : producerReports)
-        writer.write(Utils.averageMB(result.duration, r.totalBytes()) + ", ");
-      writer.write("], \"producerLatency\": [");
-      for (var r : producerReports)
-        writer.write(Utils.averageMB(result.duration, (long) r.avgLatency()) + ", ");
-      writer.write("], \"consumerThroughput\": [");
-      for (var r : consumerReports)
-        writer.write(Utils.averageMB(result.duration, r.totalBytes()) + ", ");
-      writer.write("], \"consumerLatency\": [");
-      for (var r : consumerReports)
-        writer.write(Utils.averageMB(result.duration, (long) r.avgLatency()) + ", ");
-      writer.write("]}");
+      writer.write(LocalTime.now().format(DateTimeFormatter.ISO_LOCAL_TIME) + "\":{");
+      elements.forEach(
+          element ->
+              Utils.packException(
+                  () -> writer.write("\"" + element.title() + "\":" + element.value() + ",")));
+      writer.write("}");
       writer.newLine();
     } catch (IOException ignore) {
     }
-    return producerDone.get() && consumerDone.get();
   }
 
-  private static ProcessedResult processResult(
-      ExeTime exeTime,
-      List<ProducerThread.Report> producerReports,
-      List<ConsumerThread.Report> consumerReports,
-      Supplier<Long> producedRecords) {
-    var duration = Duration.ofMillis(System.currentTimeMillis() - System.currentTimeMillis());
-    return new ProcessedResult(
-        duration,
-        Math.min(
-            100D,
-            exeTime.percentage(
-                producerReports.stream().mapToLong(Report::records).sum(), duration.toMillis())),
-        consumerReports.stream().mapToLong(Report::records).sum() * 100D / producedRecords.get());
-  }
+  // Visible for test
+  interface CSVContentElement {
+    String title();
 
-  static class ProcessedResult {
-    public final Duration duration;
-    public final double consumerPercentage;
-    public final double producerPercentage;
+    String value();
 
-    ProcessedResult(Duration duration, double consumerPercentage, double producerPercentage) {
-      this.duration = duration;
-      this.consumerPercentage = consumerPercentage;
-      this.producerPercentage = producerPercentage;
+    static CSVContentElement create(String title, Supplier<String> value) {
+      return new CSVContentElement() {
+        @Override
+        public String title() {
+          return title;
+        }
+
+        @Override
+        public String value() {
+          return value.get();
+        }
+      };
     }
   }
 
+  private static List<CSVContentElement> latencyAndIO(
+      List<ProducerThread.Report> producerReports, List<ConsumerThread.Report> consumerReports) {
+    var elements = new ArrayList<CSVContentElement>();
+    elements.add(
+        CSVContentElement.create(
+            "Time",
+            () ->
+                LocalTime.now().getHour()
+                    + ":"
+                    + LocalTime.now().getMinute()
+                    + ":"
+                    + LocalTime.now().getSecond()));
+    elements.add(
+        CSVContentElement.create(
+            "Publish Max latency (ms)",
+            () -> Long.toString(producerReports.stream().mapToLong(Report::max).max().orElse(0))));
+    elements.add(
+        CSVContentElement.create(
+            "Publish min latency (ms)",
+            () -> Long.toString(producerReports.stream().mapToLong(Report::min).min().orElse(0))));
+    elements.add(
+        CSVContentElement.create(
+            "End-to-end max latency (ms)",
+            () -> Long.toString(consumerReports.stream().mapToLong(Report::max).max().orElse(0))));
+    elements.add(
+        CSVContentElement.create(
+            "End-to-end min latency (ms)",
+            () -> Long.toString(consumerReports.stream().mapToLong(Report::min).min().orElse(0))));
+    IntStream.range(0, producerReports.size())
+        .forEach(
+            i -> {
+              elements.add(
+                  CSVContentElement.create(
+                      "Producer[" + i + "] bytes produced",
+                      () -> Long.toString(producerReports.get(i).totalBytes())));
+              elements.add(
+                  CSVContentElement.create(
+                      "Producer[" + i + "] average publish latency (ms)",
+                      () -> Double.toString(producerReports.get(i).avgLatency())));
+            });
+    IntStream.range(0, consumerReports.size())
+        .forEach(
+            i -> {
+              elements.add(
+                  CSVContentElement.create(
+                      "Consumer[" + i + "] bytes produced",
+                      () -> Long.toString(consumerReports.get(i).totalBytes())));
+              elements.add(
+                  CSVContentElement.create(
+                      "Consumer[" + i + "] average publish latency (ms)",
+                      () -> Double.toString(consumerReports.get(i).avgLatency())));
+            });
+    return elements;
+  }
+
   @Override
   public String toString() {
     return name;


Test patch:
diff --git a/app/src/test/java/org/astraea/app/admin/ReplicationThrottlerTest.java b/app/src/test/java/org/astraea/app/admin/ReplicationThrottlerTest.java
index dd46b878..db899e92 100644
--- a/app/src/test/java/org/astraea/app/admin/ReplicationThrottlerTest.java
+++ b/app/src/test/java/org/astraea/app/admin/ReplicationThrottlerTest.java
@@ -31,6 +31,12 @@ import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable;
 
 public class ReplicationThrottlerTest extends RequireBrokerCluster {
 
+  /**
+   * There is a bug in the Kafka broker implementation. This will cause the replication throttle to
+   * act unstable at the beginning of replication. For more details, see <a
+   * href="https://github.com/apache/kafka/pull/12528">this</a>. This test attempt to ensure the
+   * throttle works smoothly. It is disabled since it won't pass until the bug is resolved.
+   */
   @EnabledIfEnvironmentVariable(named = "RunReplicationThrottler", matches = "^yes$")
   @Test
   void runReplicationThrottler() {
diff --git a/app/src/test/java/org/astraea/app/balancer/BalancerUtilsTest.java b/app/src/test/java/org/astraea/app/balancer/BalancerUtilsTest.java
new file mode 100644
index 00000000..7edbb13d
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/balancer/BalancerUtilsTest.java
@@ -0,0 +1,135 @@
+/*
+ * 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.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.PartitionInfo;
+import org.astraea.app.admin.ClusterInfo;
+import org.astraea.app.admin.TopicPartition;
+import org.astraea.app.balancer.log.ClusterLogAllocation;
+import org.astraea.app.balancer.log.LogPlacement;
+import org.astraea.app.cost.ReplicaDiskInCost;
+import org.astraea.app.cost.ReplicaLeaderCost;
+import org.astraea.app.metrics.BeanObject;
+import org.astraea.app.metrics.HasBeanObject;
+import org.astraea.app.metrics.broker.HasValue;
+import org.astraea.app.metrics.broker.LogMetrics;
+import org.astraea.app.metrics.broker.ServerMetrics;
+import org.astraea.app.partitioner.Configuration;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class BalancerUtilsTest {
+  private static final HasValue OLD_TP1_0 =
+      fakePartitionBeanObject("Log", LogMetrics.Log.SIZE.metricName(), "test-1", "0", 1000, 1000L);
+  private static final HasValue NEW_TP1_0 =
+      fakePartitionBeanObject(
+          "Log", LogMetrics.Log.SIZE.metricName(), "test-1", "0", 500000, 10000L);
+  private static final HasValue OLD_TP1_1 =
+      fakePartitionBeanObject("Log", LogMetrics.Log.SIZE.metricName(), "test-1", "1", 500, 1000L);
+  private static final HasValue NEW_TP1_1 =
+      fakePartitionBeanObject(
+          "Log", LogMetrics.Log.SIZE.metricName(), "test-1", "1", 100000000, 10000L);
+  private static final HasValue LEADER_BROKER1 =
+      fakeBrokerBeanObject(
+          "ReplicaManager", ServerMetrics.ReplicaManager.LEADER_COUNT.metricName(), 2, 10000L);
+  private static final HasValue LEADER_BROKER2 =
+      fakeBrokerBeanObject(
+          "ReplicaManager", ServerMetrics.ReplicaManager.LEADER_COUNT.metricName(), 4, 10000L);
+  private static final Collection<HasBeanObject> broker1 =
+      List.of(OLD_TP1_0, NEW_TP1_0, LEADER_BROKER1);
+  private static final Collection<HasBeanObject> broker2 =
+      List.of(OLD_TP1_1, NEW_TP1_1, LEADER_BROKER2);
+  private static final Map<Integer, Collection<HasBeanObject>> beanObjects =
+      Map.of(0, broker1, 1, broker2);
+
+  @Test
+  void testMockClusterInfoAllocation() {
+    var tp1 = TopicPartition.of("testMockCluster", 1);
+    var tp2 = TopicPartition.of("testMockCluster", 0);
+    var logPlacement1 = List.of(LogPlacement.of(0), LogPlacement.of(1));
+    var logPlacement2 = List.of(LogPlacement.of(1), LogPlacement.of(2));
+    var nodes =
+        new Node[] {
+          new Node(0, "localhost", 9092),
+          new Node(1, "localhost", 9092),
+          new Node(2, "localhost", 9092)
+        };
+    var partitionInfo = new PartitionInfo("test-1", 1, nodes[0], nodes, nodes);
+    var clusterInfo =
+        ClusterInfo.of(
+            new Cluster(
+                "",
+                List.of(nodes[0], nodes[1], nodes[2]),
+                List.of(partitionInfo),
+                Set.of(),
+                Set.of(),
+                Node.noNode()));
+    var cla = ClusterLogAllocation.of(Map.of(tp1, logPlacement1, tp2, logPlacement2));
+    var mockClusterInfo = BalancerUtils.merge(clusterInfo, cla);
+    Assertions.assertEquals(mockClusterInfo.replicas("testMockCluster").size(), 4);
+    Assertions.assertEquals(mockClusterInfo.nodes().size(), 3);
+    Assertions.assertEquals(mockClusterInfo.topics().size(), 1);
+    Assertions.assertTrue(mockClusterInfo.topics().contains("testMockCluster"));
+  }
+
+  @Test
+  void testEvaluateCost() {
+    var node1 = new Node(0, "localhost", 9092);
+    var node2 = new Node(1, "localhost", 9092);
+    var partitionInfo1 =
+        new PartitionInfo("test-1", 0, node1, new Node[] {node1}, new Node[] {node1});
+    var partitionInfo2 =
+        new PartitionInfo("test-1", 1, node2, new Node[] {node2}, new Node[] {node2});
+    var clusterInfo =
+        ClusterInfo.of(
+            new Cluster(
+                "",
+                List.of(node1, node2),
+                List.of(partitionInfo1, partitionInfo2),
+                Set.of(),
+                Set.of(),
+                Node.noNode()));
+
+    var cf1 = new ReplicaLeaderCost();
+    var cf2 = new ReplicaDiskInCost(Configuration.of(Map.of("metrics.duration", "5")));
+    var cost = BalancerUtils.evaluateCost(clusterInfo, Map.of(cf1, beanObjects, cf2, beanObjects));
+    Assertions.assertEquals(1.3234028368582615, cost);
+  }
+
+  private static LogMetrics.Log.Meter fakePartitionBeanObject(
+      String type, String name, String topic, String partition, long size, long time) {
+    return new LogMetrics.Log.Meter(
+        new BeanObject(
+            "kafka.log",
+            Map.of("name", name, "type", type, "topic", topic, "partition", partition),
+            Map.of("Value", size),
+            time));
+  }
+
+  private static ServerMetrics.ReplicaManager.Meter fakeBrokerBeanObject(
+      String type, String name, long value, long time) {
+    return new ServerMetrics.ReplicaManager.Meter(
+        new BeanObject(
+            "kafka.server", Map.of("type", type, "name", name), Map.of("Value", value), time));
+  }
+}
diff --git a/app/src/test/java/org/astraea/app/balancer/RebalancePlanProposalTest.java b/app/src/test/java/org/astraea/app/balancer/RebalancePlanProposalTest.java
index e8e82352..bd48c226 100644
--- a/app/src/test/java/org/astraea/app/balancer/RebalancePlanProposalTest.java
+++ b/app/src/test/java/org/astraea/app/balancer/RebalancePlanProposalTest.java
@@ -38,8 +38,7 @@ class RebalancePlanProposalTest {
             .addWarning("Warning2")
             .build();
 
-    Assertions.assertTrue(build.rebalancePlan().isPresent());
-    final var thatAllocation = build.rebalancePlan().orElseThrow();
+    final var thatAllocation = build.rebalancePlan();
     final var thisTps = thisAllocation.topicPartitions();
     final var thatTps = thatAllocation.topicPartitions();
     Assertions.assertEquals(thisTps, thatTps);
@@ -54,28 +53,4 @@ class RebalancePlanProposalTest {
     Assertions.assertEquals("Warning1", build.warnings().get(1));
     Assertions.assertEquals("Warning2", build.warnings().get(2));
   }
-
-  @Test
-  void testNoBuildTwice() {
-    // A builder should only build once. If a builder can build multiple times then it will have to
-    // do much copy work once a new build is requested. This will harm performance.
-    final var fakeClusterInfo = ClusterInfoProvider.fakeClusterInfo(10, 10, 10, 10);
-    final var logAllocation = ClusterLogAllocation.of(fakeClusterInfo);
-    final var build = RebalancePlanProposal.builder().withRebalancePlan(logAllocation);
-
-    Assertions.assertDoesNotThrow(build::build);
-    Assertions.assertThrows(IllegalStateException.class, build::build);
-  }
-
-  @Test
-  void testNoModifyAfterBuild() {
-    final var fakeClusterInfo = ClusterInfoProvider.fakeClusterInfo(10, 10, 10, 10);
-    final var logAllocation = ClusterLogAllocation.of(fakeClusterInfo);
-    final var build = RebalancePlanProposal.builder().withRebalancePlan(logAllocation);
-
-    RebalancePlanProposal proposal = build.build();
-    Assertions.assertThrows(
-        IllegalStateException.class, () -> build.addInfo("modify after built."));
-    Assertions.assertFalse(proposal.info().contains("modify after built."));
-  }
 }
diff --git a/app/src/test/java/org/astraea/app/balancer/generator/ShufflePlanGeneratorTest.java b/app/src/test/java/org/astraea/app/balancer/generator/ShufflePlanGeneratorTest.java
index a2b1d583..ee52531d 100644
--- a/app/src/test/java/org/astraea/app/balancer/generator/ShufflePlanGeneratorTest.java
+++ b/app/src/test/java/org/astraea/app/balancer/generator/ShufflePlanGeneratorTest.java
@@ -56,7 +56,7 @@ class ShufflePlanGeneratorTest {
         .limit(100)
         .forEach(
             proposal -> {
-              final var that = proposal.rebalancePlan().orElseThrow();
+              final var that = proposal.rebalancePlan();
               final var thisTps = allocation.topicPartitions();
               final var thatTps = that.topicPartitions();
               final var thisMap =
@@ -78,8 +78,12 @@ class ShufflePlanGeneratorTest {
     final var proposal = shufflePlanGenerator.generate(fakeClusterInfo).iterator().next();
 
     System.out.println(proposal);
-    Assertions.assertFalse(proposal.rebalancePlan().isPresent());
+    Assertions.assertTrue(
+        ClusterLogAllocation.findNonFulfilledAllocation(
+                ClusterLogAllocation.of(fakeClusterInfo), proposal.rebalancePlan())
+            .isEmpty());
     Assertions.assertTrue(proposal.warnings().size() >= 1);
+    Assertions.assertEquals(1, shufflePlanGenerator.generate(fakeClusterInfo).limit(10).count());
   }
 
   @Test
@@ -90,8 +94,8 @@ class ShufflePlanGeneratorTest {
     final var proposal = shufflePlanGenerator.generate(fakeClusterInfo).iterator().next();
 
     System.out.println(proposal);
-    Assertions.assertFalse(proposal.rebalancePlan().isPresent());
     Assertions.assertTrue(proposal.warnings().size() >= 1);
+    Assertions.assertEquals(1, shufflePlanGenerator.generate(fakeClusterInfo).limit(10).count());
   }
 
   @Test
@@ -102,8 +106,12 @@ class ShufflePlanGeneratorTest {
     final var proposal = shufflePlanGenerator.generate(fakeClusterInfo).iterator().next();
 
     System.out.println(proposal);
-    Assertions.assertFalse(proposal.rebalancePlan().isPresent());
+    Assertions.assertTrue(
+        ClusterLogAllocation.findNonFulfilledAllocation(
+                ClusterLogAllocation.of(fakeClusterInfo), proposal.rebalancePlan())
+            .isEmpty());
     Assertions.assertTrue(proposal.warnings().size() >= 1);
+    Assertions.assertEquals(1, shufflePlanGenerator.generate(fakeClusterInfo).limit(10).count());
   }
 
   @ParameterizedTest(name = "[{0}] {1} nodes, {2} topics, {3} partitions, {4} replicas")
diff --git a/app/src/test/java/org/astraea/app/performance/ReportFormatTest.java b/app/src/test/java/org/astraea/app/performance/ReportFormatTest.java
new file mode 100644
index 00000000..b61687c0
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/performance/ReportFormatTest.java
@@ -0,0 +1,54 @@
+/*
+ * 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.performance;
+
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.StringWriter;
+import java.util.List;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class ReportFormatTest {
+  @Test
+  void testInitCSV() throws IOException {
+    var stringWriter = new StringWriter();
+    var writer = new BufferedWriter(stringWriter);
+    var elements = List.of(titleValue("t1", "1"), titleValue("t2", "2"));
+
+    ReportFormat.initCSVFormat(writer, elements);
+    writer.flush();
+
+    Assertions.assertEquals("t1, t2, \n", stringWriter.toString());
+  }
+
+  @Test
+  void testLogToCSV() throws IOException {
+    StringWriter stringWriter = new StringWriter();
+    var writer = new BufferedWriter(stringWriter);
+    var elements = List.of(titleValue("t1", "1"), titleValue("t2", "2"));
+
+    ReportFormat.logToCSV(writer, elements);
+    writer.flush();
+
+    Assertions.assertEquals("1, 2, \n", stringWriter.toString());
+  }
+
+  private static ReportFormat.CSVContentElement titleValue(String title, String value) {
+    return ReportFormat.CSVContentElement.create(title, () -> value);
+  }
+}