Repository: skiptests/astraea
Pull request #578 (2022-08-11 18:13:47)

Simplify performance report
1. 移除了`current throughput` (原因請見https://github.com/skiptests/astraea/issues/577)
2. 刪除不必要的內部物件


Patch:
diff --git a/app/src/main/java/org/astraea/app/admin/Builder.java b/app/src/main/java/org/astraea/app/admin/Builder.java
index 4921acac..3ac68e1d 100644
--- a/app/src/main/java/org/astraea/app/admin/Builder.java
+++ b/app/src/main/java/org/astraea/app/admin/Builder.java
@@ -54,6 +54,7 @@ import org.apache.kafka.common.quota.ClientQuotaEntity;
 import org.apache.kafka.common.quota.ClientQuotaFilter;
 import org.apache.kafka.common.quota.ClientQuotaFilterComponent;
 import org.astraea.app.common.DataRate;
+import org.astraea.app.common.ExecutionRuntimeException;
 import org.astraea.app.common.Utils;
 
 public class Builder {
@@ -122,7 +123,11 @@ public class Builder {
                   .all()
                   .get();
             });
-      } catch (ElectionNotNeededException ignored) {
+      } catch (ExecutionRuntimeException executionRuntimeException) {
+        if (ElectionNotNeededException.class
+            != executionRuntimeException.getRootCause().getClass()) {
+          throw executionRuntimeException;
+        }
         // Swallow the ElectionNotNeededException.
         // This error occurred if the preferred leader of the given topic/partition is already the
         // leader. It is ok to swallow the exception since the preferred leader be the actual
@@ -565,11 +570,14 @@ public class Builder {
                   .all()
                   .get();
             });
-      } catch (IllegalArgumentException e) {
-        // Deleting all members can't work when there is no members already.
-        if (!ERROR_MSG_MEMBER_IS_EMPTY.equals(e.getMessage())) {
-          throw e;
+      } catch (ExecutionRuntimeException executionRuntimeException) {
+        var rootCause = executionRuntimeException.getRootCause();
+        if (IllegalArgumentException.class == rootCause.getClass()
+            && ERROR_MSG_MEMBER_IS_EMPTY.equals(rootCause.getMessage())) {
+          // Deleting all members can't work when there is no members already.
+          return;
         }
+        throw executionRuntimeException;
       }
     }
 
@@ -904,7 +912,11 @@ public class Builder {
                                 topicPartition.topic(), topicPartition.partition(), entry.getKey()),
                         Map.Entry::getValue));
         Utils.packException(() -> admin.alterReplicaLogDirs(payload).all().get());
-      } catch (ReplicaNotAvailableException ignore) {
+      } catch (ExecutionRuntimeException executionRuntimeException) {
+        if (ReplicaNotAvailableException.class
+            != executionRuntimeException.getRootCause().getClass()) {
+          throw executionRuntimeException;
+        }
         // The call is probably trying to declare the preferred data directory. Swallow the
         // exception since this is a supported operation. See the Javadoc of
         // AdminClient#alterReplicaLogDirs for details.
diff --git a/app/src/main/java/org/astraea/app/common/ExecutionRuntimeException.java b/app/src/main/java/org/astraea/app/common/ExecutionRuntimeException.java
new file mode 100644
index 00000000..2a535014
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/common/ExecutionRuntimeException.java
@@ -0,0 +1,43 @@
+/*
+ * 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.common;
+
+import java.util.concurrent.ExecutionException;
+
+/** ExecutionRuntimeException convert ExecutionException to a RuntimeException. */
+public class ExecutionRuntimeException extends RuntimeException {
+
+  private final ExecutionException executionException;
+
+  public ExecutionRuntimeException(ExecutionException executionException) {
+    super(executionException.getMessage(), executionException);
+    this.executionException = executionException;
+  }
+
+  /** @return the root cause of ExecutionException */
+  public Throwable getRootCause() {
+    return unpack(executionException);
+  }
+
+  private static Throwable unpack(Throwable exception) {
+    Throwable current = exception;
+    while (current instanceof ExecutionException) {
+      current = current.getCause();
+    }
+    return current;
+  }
+}
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 d2306f9c..2a2243ba 100644
--- a/app/src/main/java/org/astraea/app/common/Utils.java
+++ b/app/src/main/java/org/astraea/app/common/Utils.java
@@ -39,17 +39,10 @@ import org.astraea.app.partitioner.Configuration;
 
 public final class Utils {
 
-  private static Throwable unpack(Throwable exception) {
-    Throwable current = exception;
-    while (current instanceof ExecutionException) {
-      current = current.getCause();
-    }
-    return current;
-  }
-
   /**
-   * Convert the exception thrown by getter to RuntimeException. This method can eliminate the
-   * exception from Java signature.
+   * Convert the exception thrown by getter to RuntimeException, except ExecutionException.
+   * ExecutionException will be converted to ExecutionRuntimeException , in order to preserve the
+   * stacktrace of ExecutionException. This method can eliminate the exception from Java signature.
    *
    * @param getter to execute
    * @param <R> type of returned object
@@ -58,11 +51,12 @@ public final class Utils {
   public static <R> R packException(Getter<R> getter) {
     try {
       return getter.get();
+    } catch (RuntimeException exception) {
+      throw exception;
+    } catch (ExecutionException exception) {
+      throw new ExecutionRuntimeException(exception);
     } catch (Throwable exception) {
-      var current = unpack(exception);
-      if (current instanceof RuntimeException) throw (RuntimeException) current;
-      if (current == null) throw new RuntimeException("unknown error");
-      throw new RuntimeException(current);
+      throw new RuntimeException(exception);
     }
   }
 
@@ -267,5 +261,15 @@ public final class Utils {
     }
   }
 
+  public static double averageMB(Duration duration, long value) {
+    return average(duration, value) / 1024D / 1024D;
+  }
+
+  public static double average(Duration duration, long value) {
+    if (duration.toSeconds() > 0) return ((double) (value / duration.toSeconds()));
+    if (duration.toMillis() > 0) return (double) (value / duration.toMillis()) * 1000;
+    return (double) (value / duration.toNanos()) * 1000000000L;
+  }
+
   private Utils() {}
 }
diff --git a/app/src/main/java/org/astraea/app/performance/ConsumerThread.java b/app/src/main/java/org/astraea/app/performance/ConsumerThread.java
index fffa0469..c0d9ed6a 100644
--- a/app/src/main/java/org/astraea/app/performance/ConsumerThread.java
+++ b/app/src/main/java/org/astraea/app/performance/ConsumerThread.java
@@ -28,25 +28,17 @@ import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import org.apache.kafka.common.errors.WakeupException;
 import org.astraea.app.common.Utils;
-import org.astraea.app.consumer.Consumer;
+import org.astraea.app.consumer.SubscribedConsumer;
 
 public interface ConsumerThread extends AbstractThread {
 
   static List<ConsumerThread> create(
-      int consumers, Supplier<Consumer<byte[], byte[]>> consumerSupplier) {
+      int consumers, Supplier<SubscribedConsumer<byte[], byte[]>> consumerSupplier) {
     if (consumers == 0) return List.of();
-    var reports =
-        IntStream.range(0, consumers)
-            .mapToObj(ignored -> new Report())
-            .collect(Collectors.toUnmodifiableList());
     var closeLatches =
         IntStream.range(0, consumers)
             .mapToObj(ignored -> new CountDownLatch(1))
             .collect(Collectors.toUnmodifiableList());
-    var closeFlags =
-        IntStream.range(0, consumers)
-            .mapToObj(ignored -> new AtomicBoolean(false))
-            .collect(Collectors.toUnmodifiableList());
     var executors = Executors.newFixedThreadPool(consumers);
     // monitor
     CompletableFuture.runAsync(
@@ -62,9 +54,9 @@ public interface ConsumerThread extends AbstractThread {
         .mapToObj(
             index -> {
               var consumer = consumerSupplier.get();
-              var report = reports.get(index);
+              var report = new Report();
               var closeLatch = closeLatches.get(index);
-              var closed = closeFlags.get(index);
+              var closed = new AtomicBoolean(false);
               var subscribed = new AtomicBoolean(true);
               executors.execute(
                   () -> {
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 b205055a..011dbc6b 100644
--- a/app/src/main/java/org/astraea/app/performance/Performance.java
+++ b/app/src/main/java/org/astraea/app/performance/Performance.java
@@ -162,7 +162,8 @@ public class Performance {
                             .map(ProducerThread::report)
                             .mapToLong(Report::records)
                             .sum(),
-                    tracker));
+                    producerReports,
+                    consumerReports));
 
     var fileWriterFuture =
         fileWriter.map(CompletableFuture::runAsync).orElse(CompletableFuture.completedFuture(null));
diff --git a/app/src/main/java/org/astraea/app/performance/ProducerThread.java b/app/src/main/java/org/astraea/app/performance/ProducerThread.java
index 0a0fcaf5..dfbafc02 100644
--- a/app/src/main/java/org/astraea/app/performance/ProducerThread.java
+++ b/app/src/main/java/org/astraea/app/performance/ProducerThread.java
@@ -39,18 +39,10 @@ public interface ProducerThread extends AbstractThread {
       int producers,
       Supplier<Producer<byte[], byte[]>> producerSupplier) {
     if (producers <= 0) return List.of();
-    var reports =
-        IntStream.range(0, producers)
-            .mapToObj(ignored -> new Report())
-            .collect(Collectors.toUnmodifiableList());
     var closeLatches =
         IntStream.range(0, producers)
             .mapToObj(ignored -> new CountDownLatch(1))
             .collect(Collectors.toUnmodifiableList());
-    var closedFlags =
-        IntStream.range(0, producers)
-            .mapToObj(ignored -> new AtomicBoolean(false))
-            .collect(Collectors.toUnmodifiableList());
     var executors = Executors.newFixedThreadPool(producers);
     // monitor
     CompletableFuture.runAsync(
@@ -66,9 +58,9 @@ public interface ProducerThread extends AbstractThread {
         .mapToObj(
             index -> {
               var producer = producerSupplier.get();
-              var report = reports.get(index);
+              var report = new Report();
               var closeLatch = closeLatches.get(index);
-              var closed = closedFlags.get(index);
+              var closed = new AtomicBoolean(false);
               executors.execute(
                   () -> {
                     try {
diff --git a/app/src/main/java/org/astraea/app/performance/Report.java b/app/src/main/java/org/astraea/app/performance/Report.java
index 45bb46a8..1c92ba25 100644
--- a/app/src/main/java/org/astraea/app/performance/Report.java
+++ b/app/src/main/java/org/astraea/app/performance/Report.java
@@ -41,7 +41,6 @@ public class Report {
   private long max = 0;
   private long min = Long.MAX_VALUE;
   private long totalBytes = 0;
-  private long currentBytes = 0;
   private final Map<TopicPartition, Long> currentOffsets = new HashMap<>();
 
   /** Simultaneously add latency and bytes. */
@@ -51,7 +50,6 @@ public class Report {
     min = Math.min(min, latency);
     max = Math.max(max, latency);
     avgLatency += (((double) latency) - avgLatency) / (double) records;
-    currentBytes += bytes;
     totalBytes += bytes;
     var tp = TopicPartition.of(topic, partition);
     currentOffsets.put(tp, Math.max(offset, offset(tp)));
@@ -79,12 +77,6 @@ public class Report {
     return totalBytes;
   }
 
-  public synchronized long clearAndGetCurrentBytes() {
-    var ans = currentBytes;
-    currentBytes = 0;
-    return ans;
-  }
-
   public synchronized long offset(TopicPartition tp) {
     return currentOffsets.getOrDefault(tp, 0L);
   }
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 129d4ac2..bbe880be 100644
--- a/app/src/main/java/org/astraea/app/performance/ReportFormat.java
+++ b/app/src/main/java/org/astraea/app/performance/ReportFormat.java
@@ -26,6 +26,7 @@ import java.nio.file.Path;
 import java.text.SimpleDateFormat;
 import java.time.Duration;
 import java.util.Date;
+import java.util.List;
 import java.util.function.Supplier;
 import java.util.stream.IntStream;
 import org.astraea.app.common.DataSize;
@@ -63,7 +64,8 @@ public enum ReportFormat {
       Supplier<Boolean> consumerDone,
       Supplier<Boolean> producerDone,
       Supplier<Long> producedRecords,
-      TrackerThread tracker)
+      List<Report> producerReports,
+      List<Report> consumerReports)
       throws IOException {
     var filePath =
         FileSystems.getDefault()
@@ -76,13 +78,18 @@ public enum ReportFormat {
     var writer = new BufferedWriter(new FileWriter(filePath.toFile()));
     switch (reportFormat) {
       case CSV:
-        initCSVFormat(
-            writer, tracker.producerResult().bytes.size(), tracker.consumerResult().bytes.size());
+        initCSVFormat(writer, producerReports.size(), consumerReports.size());
         return () -> {
           try {
             while (true) {
-              if (logToCSV(writer, exeTime, consumerDone, producerDone, producedRecords, tracker))
-                return;
+              if (logToCSV(
+                  writer,
+                  exeTime,
+                  consumerDone,
+                  producerDone,
+                  producedRecords,
+                  producerReports,
+                  consumerReports)) return;
               Utils.sleep(Duration.ofSeconds(1));
             }
           } finally {
@@ -94,8 +101,14 @@ public enum ReportFormat {
         return () -> {
           try {
             while (true) {
-              if (logToJSON(writer, exeTime, consumerDone, producerDone, producedRecords, tracker))
-                return;
+              if (logToJSON(
+                  writer,
+                  exeTime,
+                  consumerDone,
+                  producerDone,
+                  producedRecords,
+                  producerReports,
+                  consumerReports)) return;
               Utils.sleep(Duration.ofSeconds(1));
             }
           } finally {
@@ -120,7 +133,7 @@ public enum ReportFormat {
                 writer.write(
                     ",Producer["
                         + i
-                        + "] current throughput (MiB/sec), Producer["
+                        + "] bytes produced, Producer["
                         + i
                         + "] average publish latency (ms)");
               } catch (IOException ignore) {
@@ -133,7 +146,7 @@ public enum ReportFormat {
                 writer.write(
                     ",Consumer["
                         + i
-                        + "] current throughput (MiB/sec), Consumer["
+                        + "] bytes consumed, Consumer["
                         + i
                         + "] average ene-to-end latency (ms)");
               } catch (IOException ignore) {
@@ -148,9 +161,10 @@ public enum ReportFormat {
       Supplier<Boolean> consumerDone,
       Supplier<Boolean> producerDone,
       Supplier<Long> producedRecords,
-      TrackerThread tracker) {
-    var result = processResult(exeTime, tracker, producedRecords);
-    if (result.producerResult.completedRecords == 0) return false;
+      List<Report> producerReports,
+      List<Report> consumerReports) {
+    var result = processResult(exeTime, producerReports, consumerReports, producedRecords);
+    if (producerReports.stream().mapToLong(Report::records).sum() == 0) return false;
     try {
       writer.write(
           result.duration.toHoursPart()
@@ -162,31 +176,23 @@ public enum ReportFormat {
           String.format(",%.2f%% / %.2f%%", result.consumerPercentage, result.producerPercentage));
       writer.write(
           ","
-              + DataSize.Byte.of(result.producerResult.totalCurrentBytes())
-                  .measurement(DataUnit.MiB)
-                  .doubleValue());
+              + producerReports.stream().mapToLong(Report::max).max().orElse(0)
+              + ","
+              + producerReports.stream().mapToLong(Report::min).min().orElse(0));
       writer.write(
           ","
-              + DataSize.Byte.of(result.consumerResult.totalCurrentBytes())
-                  .measurement(DataUnit.MiB)
-                  .doubleValue());
-      writer.write("," + result.producerResult.maxLatency + "," + result.producerResult.minLatency);
-      writer.write("," + result.consumerResult.maxLatency + "," + result.consumerResult.minLatency);
-      for (int i = 0; i < result.producerResult.bytes.size(); ++i) {
+              + 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(result.producerResult.currentBytes.get(i))
-                    .measurement(DataUnit.MiB)
-                    .doubleValue());
-        writer.write("," + result.producerResult.averageLatencies.get(i));
+            "," + DataSize.Byte.of(r.totalBytes()).measurement(DataUnit.MiB).doubleValue());
+        writer.write("," + r.avgLatency());
       }
-      for (int i = 0; i < result.consumerResult.bytes.size(); ++i) {
+      for (var r : consumerReports) {
         writer.write(
-            ","
-                + DataSize.Byte.of(result.consumerResult.currentBytes.get(i))
-                    .measurement(DataUnit.MiB)
-                    .doubleValue());
-        writer.write("," + result.consumerResult.averageLatencies.get(i));
+            "," + DataSize.Byte.of(r.totalBytes()).measurement(DataUnit.MiB).doubleValue());
+        writer.write("," + r.avgLatency());
       }
       writer.newLine();
     } catch (IOException ignore) {
@@ -201,9 +207,10 @@ public enum ReportFormat {
       Supplier<Boolean> consumerDone,
       Supplier<Boolean> producerDone,
       Supplier<Long> producedRecords,
-      TrackerThread tracker) {
-    var result = processResult(exeTime, tracker, producedRecords);
-    if (result.producerResult.completedRecords == 0) return false;
+      List<Report> producerReports,
+      List<Report> consumerReports) {
+    var result = processResult(exeTime, producerReports, consumerReports, producedRecords);
+    if (producerReports.stream().mapToLong(Report::records).sum() == 0) return false;
     try {
       writer.write(
           String.format(
@@ -216,29 +223,38 @@ public enum ReportFormat {
               "\"consumerPercentage\": %.2f%%, \"Producer Percentage\": %.2f%%",
               result.consumerPercentage, result.producerPercentage));
       writer.write(
-          ", \"outputThroughput\": " + result.producerResult.averageBytes(result.duration));
-      writer.write(", \"inputThroughput\": " + result.consumerResult.averageBytes(result.duration));
-      writer.write(", \"publishMaxLatency\": " + result.producerResult.maxLatency);
-      writer.write(", \"publishMinLatency\": " + result.producerResult.minLatency);
-      writer.write(", \"E2EMaxLatency\": " + result.consumerResult.maxLatency);
-      writer.write(", \"E2EMinLatency\": " + result.consumerResult.minLatency);
+          ", \"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 (int i = 0; i < result.producerResult.bytes.size(); ++i) {
-        writer.write(TrackerThread.avg(result.duration, result.producerResult.bytes.get(i)) + ", ");
-      }
+      for (var r : producerReports)
+        writer.write(Utils.averageMB(result.duration, r.totalBytes()) + ", ");
       writer.write("], \"producerLatency\": [");
-      for (int i = 0; i < result.producerResult.bytes.size(); ++i) {
-        writer.write(result.producerResult.averageLatencies.get(i) + ", ");
-      }
+      for (var r : producerReports)
+        writer.write(Utils.averageMB(result.duration, (long) r.avgLatency()) + ", ");
       writer.write("], \"consumerThroughput\": [");
-      for (int i = 0; i < result.consumerResult.bytes.size(); ++i) {
-        writer.write(TrackerThread.avg(result.duration, result.consumerResult.bytes.get(i)) + ", ");
-      }
+      for (var r : consumerReports)
+        writer.write(Utils.averageMB(result.duration, r.totalBytes()) + ", ");
       writer.write("], \"consumerLatency\": [");
-      for (int i = 0; i < result.consumerResult.bytes.size(); ++i) {
-        writer.write(result.consumerResult.averageLatencies.get(i) + ", ");
-      }
+      for (var r : consumerReports)
+        writer.write(Utils.averageMB(result.duration, (long) r.avgLatency()) + ", ");
       writer.write("]}");
       writer.newLine();
     } catch (IOException ignore) {
@@ -247,33 +263,26 @@ public enum ReportFormat {
   }
 
   private static ProcessedResult processResult(
-      ExeTime exeTime, TrackerThread tracker, Supplier<Long> producedRecords) {
-    var producerResult = tracker.producerResult();
-    var consumerResult = tracker.consumerResult();
-    var duration = Duration.ofMillis(System.currentTimeMillis() - tracker.startTime());
+      ExeTime exeTime,
+      List<Report> producerReports,
+      List<Report> consumerReports,
+      Supplier<Long> producedRecords) {
+    var duration = Duration.ofMillis(System.currentTimeMillis() - System.currentTimeMillis());
     return new ProcessedResult(
-        consumerResult,
-        producerResult,
         duration,
-        Math.min(100D, exeTime.percentage(producerResult.completedRecords, duration.toMillis())),
-        consumerResult.completedRecords * 100D / producedRecords.get());
+        Math.min(
+            100D,
+            exeTime.percentage(
+                producerReports.stream().mapToLong(Report::records).sum(), duration.toMillis())),
+        consumerReports.stream().mapToLong(Report::records).sum() * 100D / producedRecords.get());
   }
 
   static class ProcessedResult {
-    public final TrackerThread.Result consumerResult;
-    public final TrackerThread.Result producerResult;
     public final Duration duration;
     public final double consumerPercentage;
     public final double producerPercentage;
 
-    ProcessedResult(
-        TrackerThread.Result consumerResult,
-        TrackerThread.Result producerResult,
-        Duration duration,
-        double consumerPercentage,
-        double producerPercentage) {
-      this.consumerResult = consumerResult;
-      this.producerResult = producerResult;
+    ProcessedResult(Duration duration, double consumerPercentage, double producerPercentage) {
       this.duration = duration;
       this.consumerPercentage = consumerPercentage;
       this.producerPercentage = producerPercentage;
diff --git a/app/src/main/java/org/astraea/app/performance/TrackerThread.java b/app/src/main/java/org/astraea/app/performance/TrackerThread.java
index eb04df31..6ba5e110 100644
--- a/app/src/main/java/org/astraea/app/performance/TrackerThread.java
+++ b/app/src/main/java/org/astraea/app/performance/TrackerThread.java
@@ -17,14 +17,11 @@
 package org.astraea.app.performance;
 
 import java.time.Duration;
-import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.function.Function;
-import org.astraea.app.common.DataSize;
+import java.util.function.Supplier;
 import org.astraea.app.common.Utils;
 
 /** Print out the given metrics. */
@@ -34,14 +31,14 @@ public interface TrackerThread extends AbstractThread {
       List<Report> producerReports, List<Report> consumerReports, ExeTime exeTime) {
     var start = System.currentTimeMillis() - Duration.ofSeconds(1).toMillis();
 
-    Function<Result, Boolean> logProducers =
-        result -> {
-          if (result.completedRecords == 0) return false;
+    Supplier<Boolean> logProducers =
+        () -> {
+          var records = producerReports.stream().mapToLong(Report::records).sum();
+          if (records == 0) return false;
 
           var duration = Duration.ofMillis(System.currentTimeMillis() - start);
           // Print completion rate (by number of records) or (by time)
-          var percentage =
-              Math.min(100D, exeTime.percentage(result.completedRecords, duration.toMillis()));
+          var percentage = Math.min(100D, exeTime.percentage(records, duration.toMillis()));
 
           System.out.println(
               "Time: "
@@ -53,29 +50,35 @@ public interface TrackerThread extends AbstractThread {
                   + "sec");
           System.out.printf("producers completion rate: %.2f%%%n", percentage);
           System.out.printf(
-              "  average throughput: %.3f MB/second%n", result.averageBytes(duration));
-          System.out.printf(
-              "  current throughput: %s/second%n", DataSize.Byte.of(result.totalCurrentBytes()));
-          System.out.println("  publish max latency: " + result.maxLatency + " ms");
-          System.out.println("  publish mim latency: " + result.minLatency + " ms");
-          for (int i = 0; i < result.bytes.size(); ++i) {
+              "  average throughput: %.3f MB/second%n",
+              Utils.averageMB(
+                  duration, producerReports.stream().mapToLong(Report::totalBytes).sum()));
+          producerReports.stream()
+              .mapToLong(Report::max)
+              .max()
+              .ifPresent(i -> System.out.println("  publish max latency: " + i + " ms"));
+          producerReports.stream()
+              .mapToLong(Report::min)
+              .min()
+              .ifPresent(i -> System.out.println("  publish mim latency: " + i + " ms"));
+          for (int i = 0; i < producerReports.size(); ++i) {
             System.out.printf(
                 "  producer[%d] average throughput: %.3f MB%n",
-                i, avg(duration, result.bytes.get(i)));
+                i, Utils.averageMB(duration, producerReports.get(i).totalBytes()));
             System.out.printf(
                 "  producer[%d] average publish latency: %.3f ms%n",
-                i, result.averageLatencies.get(i));
+                i, producerReports.get(i).avgLatency());
           }
           System.out.println("\n");
           return percentage >= 100;
         };
 
-    Function<Result, Boolean> logConsumers =
-        result -> {
+    Supplier<Boolean> logConsumers =
+        () -> {
           // there is no consumer, so we just complete this log.
           if (consumerReports.isEmpty()) return true;
 
-          if (result.completedRecords == 0) return false;
+          if (consumerReports.stream().mapToLong(Report::records).sum() == 0) return false;
 
           var duration = Duration.ofMillis(System.currentTimeMillis() - start);
 
@@ -87,18 +90,24 @@ public interface TrackerThread extends AbstractThread {
           var percentage = 100 * (double) consumerOffset / producerOffset;
           System.out.printf("consumer completion rate: %.2f%%%n", percentage);
           System.out.printf(
-              "  average throughput: %.3f MB/second%n", result.averageBytes(duration));
-          System.out.printf(
-              "  current throughput: %s/second%n", DataSize.Byte.of(result.totalCurrentBytes()));
-          System.out.println("  end-to-end max latency: " + result.maxLatency + " ms");
-          System.out.println("  end-to-end mim latency: " + result.minLatency + " ms");
-          for (int i = 0; i < result.bytes.size(); ++i) {
+              "  average throughput: %.3f MB/second%n",
+              Utils.averageMB(
+                  duration, consumerReports.stream().mapToLong(Report::totalBytes).sum()));
+          consumerReports.stream()
+              .mapToLong(Report::max)
+              .max()
+              .ifPresent(i -> System.out.println("  end-to-end max latency: " + i + " ms"));
+          consumerReports.stream()
+              .mapToLong(Report::min)
+              .min()
+              .ifPresent(i -> System.out.println("  end-to-end mim latency: " + i + " ms"));
+          for (int i = 0; i < consumerReports.size(); ++i) {
             System.out.printf(
                 "  consumer[%d] average throughput: %.3f MB%n",
-                i, avg(duration, result.bytes.get(i)));
+                i, Utils.averageMB(duration, consumerReports.get(i).totalBytes()));
             System.out.printf(
                 "  consumer[%d] average ene-to-end latency: %.3f ms%n",
-                i, result.averageLatencies.get(i));
+                i, consumerReports.get(i).avgLatency());
           }
           System.out.println("\n");
           // Target number of records consumed OR consumed all that produced
@@ -112,9 +121,7 @@ public interface TrackerThread extends AbstractThread {
         () -> {
           try {
             while (!closed.get()) {
-              var producerResult = result(producerReports);
-              var consumerResult = result(consumerReports);
-              if (logProducers.apply(producerResult) & logConsumers.apply(consumerResult)) return;
+              if (logProducers.get() & logConsumers.get()) return;
 
               // Log after waiting for one second
               Utils.sleep(Duration.ofSeconds(1));
@@ -126,16 +133,6 @@ public interface TrackerThread extends AbstractThread {
 
     return new TrackerThread() {
 
-      @Override
-      public Result producerResult() {
-        return result(producerReports);
-      }
-
-      @Override
-      public Result consumerResult() {
-        return result(consumerReports);
-      }
-
       @Override
       public long startTime() {
         return start;
@@ -159,75 +156,5 @@ public interface TrackerThread extends AbstractThread {
     };
   }
 
-  static double avg(Duration duration, long value) {
-    return duration.toSeconds() <= 0
-        ? 0
-        : ((double) (value / duration.toSeconds())) / 1024D / 1024D;
-  }
-
-  private static Result result(List<Report> metrics) {
-    var completed = 0;
-    var bytes = new ArrayList<Long>();
-    var currentBytes = new ArrayList<Long>();
-    var averageLatencies = new ArrayList<Double>();
-    var max = 0L;
-    var min = Long.MAX_VALUE;
-    for (var data : metrics) {
-      completed += data.records();
-      bytes.add(data.totalBytes());
-      currentBytes.add(data.clearAndGetCurrentBytes());
-      averageLatencies.add(data.avgLatency());
-      max = Math.max(max, data.max());
-      min = Math.min(min, data.min());
-    }
-    return new Result(
-        completed,
-        Collections.unmodifiableList(bytes),
-        Collections.unmodifiableList(currentBytes),
-        Collections.unmodifiableList(averageLatencies),
-        min,
-        max);
-  }
-
-  class Result {
-    public final long completedRecords;
-    public final List<Long> bytes;
-    public final List<Long> currentBytes;
-    public final List<Double> averageLatencies;
-    public final long minLatency;
-    public final long maxLatency;
-
-    Result(
-        long completedRecords,
-        List<Long> bytes,
-        List<Long> currentBytes,
-        List<Double> averageLatencies,
-        long minLatency,
-        long maxLatency) {
-      this.completedRecords = completedRecords;
-      this.bytes = bytes;
-      this.currentBytes = currentBytes;
-      this.averageLatencies = averageLatencies;
-      this.minLatency = minLatency;
-      this.maxLatency = maxLatency;
-    }
-
-    double averageBytes(Duration duration) {
-      return avg(duration, totalBytes());
-    }
-
-    long totalBytes() {
-      return bytes.stream().mapToLong(i -> i).sum();
-    }
-
-    long totalCurrentBytes() {
-      return currentBytes.stream().mapToLong(i -> i).sum();
-    }
-  }
-
-  Result producerResult();
-
-  Result consumerResult();
-
   long startTime();
 }
diff --git a/app/src/main/java/org/astraea/app/web/TopicHandler.java b/app/src/main/java/org/astraea/app/web/TopicHandler.java
index ddd87c09..e83dd89b 100644
--- a/app/src/main/java/org/astraea/app/web/TopicHandler.java
+++ b/app/src/main/java/org/astraea/app/web/TopicHandler.java
@@ -28,6 +28,7 @@ import java.util.stream.StreamSupport;
 import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
 import org.astraea.app.admin.Admin;
 import org.astraea.app.admin.Config;
+import org.astraea.app.common.ExecutionRuntimeException;
 
 class TopicHandler implements Handler {
 
@@ -105,8 +106,11 @@ class TopicHandler implements Handler {
       try {
         // if the topic creation is synced, we return the details.
         return get(Set.of(request.value(TOPIC_NAME_KEY)), ignored -> true);
-      } catch (UnknownTopicOrPartitionException e) {
-        // swallow
+      } catch (ExecutionRuntimeException executionRuntimeException) {
+        if (UnknownTopicOrPartitionException.class
+            != executionRuntimeException.getRootCause().getClass()) {
+          throw executionRuntimeException;
+        }
       }
     }
     // Otherwise, return only name


Test patch:
diff --git a/app/src/test/java/org/astraea/app/admin/AdminTest.java b/app/src/test/java/org/astraea/app/admin/AdminTest.java
index 9aeb6c0c..3fb5faed 100644
--- a/app/src/test/java/org/astraea/app/admin/AdminTest.java
+++ b/app/src/test/java/org/astraea/app/admin/AdminTest.java
@@ -37,6 +37,7 @@ import java.util.stream.StreamSupport;
 import org.apache.kafka.common.config.TopicConfig;
 import org.apache.kafka.common.errors.GroupNotEmptyException;
 import org.astraea.app.common.DataRate;
+import org.astraea.app.common.ExecutionRuntimeException;
 import org.astraea.app.common.Utils;
 import org.astraea.app.consumer.Consumer;
 import org.astraea.app.consumer.Deserializer;
@@ -975,7 +976,11 @@ public class AdminTest extends RequireBrokerCluster {
     try (var admin = Admin.of(bootstrapServers())) {
       Assertions.assertTrue(admin.consumerGroupIds().contains(groupId));
       // the static member is existent
-      Assertions.assertThrows(GroupNotEmptyException.class, () -> admin.removeGroup(groupId));
+      var astraeaExecutionRuntimeException =
+          Assertions.assertThrows(
+              ExecutionRuntimeException.class, () -> admin.removeGroup(groupId));
+      Assertions.assertEquals(
+          GroupNotEmptyException.class, astraeaExecutionRuntimeException.getRootCause().getClass());
       // cleanup members
       admin.removeAllMembers(groupId);
       admin.removeGroup(groupId);
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 888b855e..0eb0f484 100644
--- a/app/src/test/java/org/astraea/app/common/UtilsTest.java
+++ b/app/src/test/java/org/astraea/app/common/UtilsTest.java
@@ -35,13 +35,17 @@ public class UtilsTest {
 
   @Test
   void testHandleException() {
-    Assertions.assertThrows(
-        IllegalArgumentException.class,
-        () ->
-            Utils.packException(
-                () -> {
-                  throw new ExecutionException(new IllegalArgumentException());
-                }));
+    var executionRuntimeException =
+        Assertions.assertThrows(
+            ExecutionRuntimeException.class,
+            () ->
+                Utils.packException(
+                    () -> {
+                      throw new ExecutionException(new IllegalArgumentException());
+                    }));
+
+    Assertions.assertEquals(
+        IllegalArgumentException.class, executionRuntimeException.getRootCause().getClass());
 
     Assertions.assertThrows(
         IllegalArgumentException.class,
diff --git a/app/src/test/java/org/astraea/app/performance/MetricsTest.java b/app/src/test/java/org/astraea/app/performance/MetricsTest.java
index 3705aab7..c7d2f9c3 100644
--- a/app/src/test/java/org/astraea/app/performance/MetricsTest.java
+++ b/app/src/test/java/org/astraea/app/performance/MetricsTest.java
@@ -48,15 +48,4 @@ public class MetricsTest {
     metrics.record("topic", 0, 100, 0L, 1000);
     Assertions.assertEquals(1000, metrics.totalBytes());
   }
-
-  @Test
-  void testCurrentBytes() {
-    var metrics = new Report();
-
-    Assertions.assertEquals(0, metrics.clearAndGetCurrentBytes());
-    metrics.record("topic", 0, 100, 0L, 100);
-    metrics.record("topic", 0, 100, 0L, 101);
-    Assertions.assertEquals(201, metrics.clearAndGetCurrentBytes());
-    Assertions.assertEquals(0, metrics.clearAndGetCurrentBytes());
-  }
 }
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 cefc8ba1..5ac404aa 100644
--- a/app/src/test/java/org/astraea/app/web/RecordHandlerTest.java
+++ b/app/src/test/java/org/astraea/app/web/RecordHandlerTest.java
@@ -47,6 +47,7 @@ import java.util.stream.IntStream;
 import java.util.stream.Stream;
 import org.astraea.app.admin.Admin;
 import org.astraea.app.admin.TopicPartition;
+import org.astraea.app.common.ExecutionRuntimeException;
 import org.astraea.app.common.Utils;
 import org.astraea.app.consumer.Consumer;
 import org.astraea.app.consumer.Deserializer;
@@ -71,10 +72,13 @@ public class RecordHandlerTest extends RequireBrokerCluster {
         IllegalArgumentException.class,
         () -> handler.post(PostRequest.of(Map.of(RECORDS, "[]"))),
         "records should contain at least one record");
-    Assertions.assertThrows(
-        IllegalArgumentException.class,
-        () -> handler.post(PostRequest.of(Map.of(RECORDS, "[{}]"))),
-        "topic must be set");
+    var executionRuntimeException =
+        Assertions.assertThrows(
+            ExecutionRuntimeException.class,
+            () -> handler.post(PostRequest.of(Map.of(RECORDS, "[{}]"))),
+            "topic must be set");
+    Assertions.assertEquals(
+        IllegalArgumentException.class, executionRuntimeException.getRootCause().getClass());
   }
 
   @Test