Repository: skiptests/astraea
Pull request #569 (2022-08-10 08:05:23)

Log size partition cost
#553 從其中切割出，爲NodeTopicSize加入partition cost。


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 7a140361..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);
     }
   }
 
@@ -86,11 +80,11 @@ public final class Utils {
    * @param runner to execute
    */
   public static void swallowException(Runner runner) {
-    Utils.packException(
-        () -> {
-          runner.run();
-          return null;
-        });
+    try {
+      runner.run();
+    } catch (Exception ex) {
+      ex.printStackTrace();
+    }
   }
 
   public interface Getter<R> {
@@ -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/consumer/Builder.java b/app/src/main/java/org/astraea/app/consumer/Builder.java
index 2e517a0c..4ad6794f 100644
--- a/app/src/main/java/org/astraea/app/consumer/Builder.java
+++ b/app/src/main/java/org/astraea/app/consumer/Builder.java
@@ -164,23 +164,16 @@ public abstract class Builder<Key, Value> {
     DISTANCE_FROM_LATEST(
         (kafkaConsumer, distanceFromLatest) -> {
           // this mode is not supported by kafka, so we have to calculate the offset first
-          // 1) poll data until the assignment is completed
-          while (kafkaConsumer.assignment().isEmpty()) {
-            kafkaConsumer.poll(Duration.ofMillis(500));
-          }
           var partitions = kafkaConsumer.assignment();
-          // 2) get the end offsets from all subscribed partitions
+          // 1) get the end offsets from all subscribed partitions
           var endOffsets = kafkaConsumer.endOffsets(partitions);
-          // 3) calculate and then seek to the correct offset (end offset - recent offset)
+          // 2) calculate and then seek to the correct offset (end offset - recent offset)
           endOffsets.forEach(
               (tp, latest) ->
                   kafkaConsumer.seek(tp, Math.max(0, latest - (long) distanceFromLatest)));
         }),
     DISTANCE_FROM_BEGINNING(
         (kafkaConsumer, distanceFromBeginning) -> {
-          while (kafkaConsumer.assignment().isEmpty()) {
-            kafkaConsumer.poll(Duration.ofMillis(500));
-          }
           var partitions = kafkaConsumer.assignment();
           var beginningOffsets = kafkaConsumer.beginningOffsets(partitions);
           beginningOffsets.forEach(
@@ -189,17 +182,18 @@ public abstract class Builder<Key, Value> {
     @SuppressWarnings("unchecked")
     SEEK_TO(
         (kafkaConsumer, seekTo) -> {
-          while (kafkaConsumer.assignment().isEmpty()) {
-            kafkaConsumer.poll(Duration.ofMillis(500));
-          }
           if (seekTo instanceof Long) {
             var partitions = kafkaConsumer.assignment();
             partitions.forEach(tp -> kafkaConsumer.seek(tp, (long) seekTo));
             return;
           }
           if (seekTo instanceof Map) {
+            var partitions = kafkaConsumer.assignment();
             ((Map<TopicPartition, Long>) seekTo)
-                .forEach((tp, offset) -> kafkaConsumer.seek(TopicPartition.to(tp), offset));
+                .entrySet().stream()
+                    // don't seek the partition which is not belonged to this consumer
+                    .filter(e -> partitions.contains(TopicPartition.to(e.getKey())))
+                    .forEach(e -> kafkaConsumer.seek(TopicPartition.to(e.getKey()), e.getValue()));
             return;
           }
           throw new IllegalArgumentException(
diff --git a/app/src/main/java/org/astraea/app/consumer/ConsumerRebalanceListener.java b/app/src/main/java/org/astraea/app/consumer/ConsumerRebalanceListener.java
index 4415a0aa..161265a8 100644
--- a/app/src/main/java/org/astraea/app/consumer/ConsumerRebalanceListener.java
+++ b/app/src/main/java/org/astraea/app/consumer/ConsumerRebalanceListener.java
@@ -17,6 +17,7 @@
 package org.astraea.app.consumer;
 
 import java.util.Collection;
+import java.util.List;
 import java.util.Set;
 import java.util.stream.Collectors;
 import org.astraea.app.admin.TopicPartition;
@@ -33,7 +34,7 @@ public interface ConsumerRebalanceListener {
   void onPartitionAssigned(Set<TopicPartition> partitions);
 
   static org.apache.kafka.clients.consumer.ConsumerRebalanceListener of(
-      ConsumerRebalanceListener listener) {
+      List<ConsumerRebalanceListener> listeners) {
     return new org.apache.kafka.clients.consumer.ConsumerRebalanceListener() {
       @Override
       public void onPartitionsRevoked(Collection<org.apache.kafka.common.TopicPartition> ignore) {}
@@ -41,8 +42,10 @@ public interface ConsumerRebalanceListener {
       @Override
       public void onPartitionsAssigned(
           Collection<org.apache.kafka.common.TopicPartition> partitions) {
-        listener.onPartitionAssigned(
-            partitions.stream().map(TopicPartition::from).collect(Collectors.toSet()));
+        listeners.forEach(
+            l ->
+                l.onPartitionAssigned(
+                    partitions.stream().map(TopicPartition::from).collect(Collectors.toSet())));
       }
     };
   }
diff --git a/app/src/main/java/org/astraea/app/consumer/PartitionsBuilder.java b/app/src/main/java/org/astraea/app/consumer/PartitionsBuilder.java
index 9c5a90ca..916f2df6 100644
--- a/app/src/main/java/org/astraea/app/consumer/PartitionsBuilder.java
+++ b/app/src/main/java/org/astraea/app/consumer/PartitionsBuilder.java
@@ -88,6 +88,18 @@ public class PartitionsBuilder<Key, Value> extends Builder<Key, Value> {
     return this;
   }
 
+  @Override
+  public PartitionsBuilder<Key, Value> seek(SeekStrategy seekStrategy, long value) {
+    super.seek(seekStrategy, value);
+    return this;
+  }
+
+  @Override
+  public PartitionsBuilder<Key, Value> seek(Map<TopicPartition, Long> offsets) {
+    super.seek(offsets);
+    return this;
+  }
+
   @SuppressWarnings("unchecked")
   @Override
   public AssignedConsumer<Key, Value> build() {
diff --git a/app/src/main/java/org/astraea/app/consumer/SubscribedConsumer.java b/app/src/main/java/org/astraea/app/consumer/SubscribedConsumer.java
index 3690f074..369d190d 100644
--- a/app/src/main/java/org/astraea/app/consumer/SubscribedConsumer.java
+++ b/app/src/main/java/org/astraea/app/consumer/SubscribedConsumer.java
@@ -17,7 +17,10 @@
 package org.astraea.app.consumer;
 
 import java.time.Duration;
+import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
+import org.astraea.app.admin.TopicPartition;
 
 /**
  * This inherited consumer offers function related to consumer group.
@@ -42,4 +45,10 @@ public interface SubscribedConsumer<Key, Value> extends Consumer<Key, Value> {
 
   /** @return group instance id (static member) */
   Optional<String> groupInstanceId();
+
+  /**
+   * @return the historical subscription. key is the time of getting assignments. value is the
+   *     assignments.
+   */
+  Map<Long, Set<TopicPartition>> historicalSubscription();
 }
diff --git a/app/src/main/java/org/astraea/app/consumer/TopicsBuilder.java b/app/src/main/java/org/astraea/app/consumer/TopicsBuilder.java
index 523b3aef..16f19949 100644
--- a/app/src/main/java/org/astraea/app/consumer/TopicsBuilder.java
+++ b/app/src/main/java/org/astraea/app/consumer/TopicsBuilder.java
@@ -19,17 +19,25 @@ package org.astraea.app.consumer;
 import static java.util.Objects.requireNonNull;
 
 import java.time.Duration;
+import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.astraea.app.admin.TopicPartition;
+import org.astraea.app.common.Utils;
 
 public class TopicsBuilder<Key, Value> extends Builder<Key, Value> {
   private final Set<String> topics;
   private ConsumerRebalanceListener listener = ignore -> {};
 
+  private boolean enableTrace = false;
+
   TopicsBuilder(Set<String> topics) {
     this.topics = requireNonNull(topics);
   }
@@ -47,6 +55,18 @@ public class TopicsBuilder<Key, Value> extends Builder<Key, Value> {
     return this;
   }
 
+  @Override
+  public TopicsBuilder<Key, Value> seek(SeekStrategy seekStrategy, long value) {
+    super.seek(seekStrategy, value);
+    return this;
+  }
+
+  @Override
+  public TopicsBuilder<Key, Value> seek(Map<TopicPartition, Long> offsets) {
+    super.seek(offsets);
+    return this;
+  }
+
   /**
    * make the consumer read data from beginning. By default, it reads the latest data.
    *
@@ -103,6 +123,17 @@ public class TopicsBuilder<Key, Value> extends Builder<Key, Value> {
     return this;
   }
 
+  /**
+   * enable to trace the historical subscription. see {@link
+   * SubscribedConsumer#historicalSubscription()}
+   *
+   * @return this builder
+   */
+  public TopicsBuilder<Key, Value> enableTrace() {
+    this.enableTrace = true;
+    return this;
+  }
+
   public TopicsBuilder<Key, Value> disableAutoCommitOffsets() {
     return config(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
   }
@@ -118,11 +149,38 @@ public class TopicsBuilder<Key, Value> extends Builder<Key, Value> {
             configs,
             Deserializer.of((Deserializer<Key>) keyDeserializer),
             Deserializer.of((Deserializer<Value>) valueDeserializer));
-    kafkaConsumer.subscribe(topics, ConsumerRebalanceListener.of(listener));
+
+    var tracker =
+        new ConsumerRebalanceListener() {
+          private final Map<Long, Set<TopicPartition>> history = new ConcurrentHashMap<>();
+
+          @Override
+          public void onPartitionAssigned(Set<TopicPartition> partitions) {
+            if (enableTrace) history.put(System.currentTimeMillis(), Set.copyOf(partitions));
+          }
+        };
+
+    if (seekStrategy != SeekStrategy.NONE) {
+      // make sure this consumer is assigned before seeking
+      var latch = new CountDownLatch(1);
+      kafkaConsumer.subscribe(
+          topics,
+          ConsumerRebalanceListener.of(List.of(listener, ignored -> latch.countDown(), tracker)));
+      while (latch.getCount() != 0) {
+        // the offset will be reset, so it is fine to poll data
+        // TODO: should we disable auto-commit here?
+        kafkaConsumer.poll(Duration.ofMillis(500));
+        Utils.sleep(Duration.ofSeconds(1));
+      }
+    } else {
+      // nothing to seek so we just subscribe topics
+      kafkaConsumer.subscribe(topics, ConsumerRebalanceListener.of(List.of(listener, tracker)));
+    }
 
     seekStrategy.apply(kafkaConsumer, seekValue);
 
-    return new SubscribedConsumerImpl<>(kafkaConsumer, topics, listener);
+    return new SubscribedConsumerImpl<>(
+        kafkaConsumer, topics, listener, Collections.unmodifiableMap(tracker.history));
   }
 
   private static class SubscribedConsumerImpl<Key, Value> extends Builder.BaseConsumer<Key, Value>
@@ -130,13 +188,17 @@ public class TopicsBuilder<Key, Value> extends Builder<Key, Value> {
     private final Set<String> topics;
     private final ConsumerRebalanceListener listener;
 
+    private final Map<Long, Set<TopicPartition>> history;
+
     public SubscribedConsumerImpl(
         org.apache.kafka.clients.consumer.Consumer<Key, Value> kafkaConsumer,
         Set<String> topics,
-        ConsumerRebalanceListener listener) {
+        ConsumerRebalanceListener listener,
+        Map<Long, Set<TopicPartition>> history) {
       super(kafkaConsumer);
       this.topics = topics;
       this.listener = listener;
+      this.history = history;
     }
 
     @Override
@@ -158,9 +220,14 @@ public class TopicsBuilder<Key, Value> extends Builder<Key, Value> {
       return kafkaConsumer.groupMetadata().groupInstanceId();
     }
 
+    @Override
+    public Map<Long, Set<TopicPartition>> historicalSubscription() {
+      return history;
+    }
+
     @Override
     protected void doResubscribe() {
-      kafkaConsumer.subscribe(topics, ConsumerRebalanceListener.of(listener));
+      kafkaConsumer.subscribe(topics, ConsumerRebalanceListener.of(List.of(listener)));
     }
   }
 }
diff --git a/app/src/main/java/org/astraea/app/cost/Dispersion.java b/app/src/main/java/org/astraea/app/cost/Dispersion.java
new file mode 100644
index 00000000..effd87a1
--- /dev/null
+++ b/app/src/main/java/org/astraea/app/cost/Dispersion.java
@@ -0,0 +1,45 @@
+/*
+ * 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.cost;
+
+import java.util.Collection;
+
+/** used to aggregate a sequence into a number */
+public interface Dispersion {
+  /**
+   * use correlation coefficient to a aggregate a sequence * @return correlation coefficient
+   * Dispersion
+   */
+  static Dispersion correlationCoefficient() {
+    return brokerCost -> {
+      var dataRateMean = brokerCost.stream().mapToDouble(x -> x).sum() / brokerCost.size();
+      var dataRateSD =
+          Math.sqrt(
+              brokerCost.stream().mapToDouble(score -> Math.pow((score - dataRateMean), 2)).sum()
+                  / brokerCost.size());
+      return dataRateSD / dataRateMean;
+    };
+  }
+
+  /**
+   * aggregated the values into a number
+   *
+   * @param scores origin data
+   * @return aggregated data
+   */
+  double calculate(Collection<Double> scores);
+}
diff --git a/app/src/main/java/org/astraea/app/cost/NodeTopicSizeCost.java b/app/src/main/java/org/astraea/app/cost/NodeTopicSizeCost.java
index dab447f5..b616b0fe 100644
--- a/app/src/main/java/org/astraea/app/cost/NodeTopicSizeCost.java
+++ b/app/src/main/java/org/astraea/app/cost/NodeTopicSizeCost.java
@@ -16,16 +16,21 @@
  */
 package org.astraea.app.cost;
 
+import java.util.Collection;
 import java.util.Map;
 import java.util.Optional;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
+import org.astraea.app.admin.TopicPartition;
+import org.astraea.app.admin.TopicPartitionReplica;
+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.collector.Fetcher;
 
-public class NodeTopicSizeCost implements HasBrokerCost {
-
+public class NodeTopicSizeCost implements HasBrokerCost, HasPartitionCost {
   @Override
   public Optional<Fetcher> fetcher() {
     return Optional.of(LogMetrics.Log.SIZE::fetch);
@@ -48,4 +53,51 @@ public class NodeTopicSizeCost implements HasBrokerCost {
                             .sum()));
     return () -> result;
   }
+
+  @Override
+  public PartitionCost partitionCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
+    return new PartitionCost() {
+      @Override
+      public Map<TopicPartition, Double> value(String topic) {
+        var replicas =
+            clusterInfo.availableReplicaLeaders(topic).stream()
+                .map(r -> TopicPartitionReplica.of(r.topic(), r.partition(), r.nodeInfo().id()))
+                .collect(Collectors.toUnmodifiableSet());
+        return clusterBean.mapByReplica().entrySet().stream()
+            .filter(
+                topicPartitionCollectionEntry ->
+                    replicas.contains(topicPartitionCollectionEntry.getKey()))
+            .collect(
+                Collectors.toMap(
+                    topicPartitionCollectionEntry ->
+                        TopicPartition.of(
+                            topicPartitionCollectionEntry.getKey().topic(),
+                            topicPartitionCollectionEntry.getKey().partition()),
+                    toDouble));
+      }
+
+      @Override
+      public Map<TopicPartition, Double> value(int brokerId) {
+        return clusterBean.mapByReplica().entrySet().stream()
+            .filter(
+                topicPartitionCollectionEntry ->
+                    topicPartitionCollectionEntry.getKey().brokerId() == brokerId)
+            .collect(
+                Collectors.toMap(
+                    topicPartitionCollectionEntry ->
+                        TopicPartition.of(
+                            topicPartitionCollectionEntry.getKey().topic(),
+                            topicPartitionCollectionEntry.getKey().partition()),
+                    toDouble));
+      }
+    };
+  }
+
+  private final Function<Map.Entry<TopicPartitionReplica, Collection<HasBeanObject>>, Double>
+      toDouble =
+          e ->
+              LogMetrics.Log.meters(e.getValue(), LogMetrics.Log.SIZE).stream()
+                  .mapToDouble(HasValue::value)
+                  .findAny()
+                  .orElse(0.0D);
 }
diff --git a/app/src/main/java/org/astraea/app/cost/ReplicaDiskInCost.java b/app/src/main/java/org/astraea/app/cost/ReplicaDiskInCost.java
index 49f8a3f1..a75cf5eb 100644
--- a/app/src/main/java/org/astraea/app/cost/ReplicaDiskInCost.java
+++ b/app/src/main/java/org/astraea/app/cost/ReplicaDiskInCost.java
@@ -18,13 +18,11 @@ package org.astraea.app.cost;
 
 import java.time.Duration;
 import java.util.Comparator;
-import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.stream.Collectors;
 import org.astraea.app.admin.ClusterBean;
 import org.astraea.app.admin.ClusterInfo;
-import org.astraea.app.admin.NodeInfo;
 import org.astraea.app.admin.TopicPartition;
 import org.astraea.app.admin.TopicPartitionReplica;
 import org.astraea.app.metrics.HasBeanObject;
@@ -37,48 +35,38 @@ import org.astraea.app.partitioner.Configuration;
  * responds to the replica log size of brokers. The calculation method of the score is the rate of
  * increase of log size per unit time divided by the upper limit of broker bandwidth.
  */
-public class ReplicaDiskInCost implements HasBrokerCost, HasPartitionCost {
+public class ReplicaDiskInCost implements HasClusterCost, HasBrokerCost, HasPartitionCost {
   private final Duration duration;
+  private final Dispersion dispersion = Dispersion.correlationCoefficient();
+  static final double OVERFLOW_SCORE = 9999.0;
 
   public ReplicaDiskInCost(Configuration configuration) {
     duration =
         Duration.ofSeconds(Integer.parseInt(configuration.string("metrics.duration").orElse("30")));
   }
 
+  @Override
+  public ClusterCost clusterCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
+    var brokerCost = brokerCost(clusterInfo, clusterBean).value();
+    // when retention occur, brokerCost will be set to -1 , and return a big score to reject this
+    // plan.
+    if (brokerCost.containsValue(-1.0)) return () -> OVERFLOW_SCORE;
+    return () -> dispersion.calculate(brokerCost.values());
+  }
+
   @Override
   public BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
-    final Map<Integer, List<TopicPartitionReplica>> topicPartitionOfEachBroker =
-        clusterInfo.topics().stream()
-            .flatMap(topic -> clusterInfo.replicas(topic).stream())
-            .map(
-                replica ->
-                    TopicPartitionReplica.of(
-                        replica.topic(), replica.partition(), replica.nodeInfo().id()))
-            .collect(Collectors.groupingBy(TopicPartitionReplica::brokerId));
-    final var actual =
+    var partitionCost = partitionCost(clusterInfo, clusterBean);
+    var brokerLoad =
         clusterInfo.nodes().stream()
-            .collect(
-                Collectors.toUnmodifiableMap(
-                    NodeInfo::id,
-                    node -> topicPartitionOfEachBroker.getOrDefault(node.id(), List.of())));
-
-    final var topicPartitionDataRate = topicPartitionDataRate(clusterBean, duration);
-
-    final var brokerLoad =
-        actual.entrySet().stream()
             .map(
-                entry ->
+                node ->
                     Map.entry(
-                        entry.getKey(),
-                        entry.getValue().stream()
-                            .mapToDouble(
-                                x ->
-                                    topicPartitionDataRate.get(
-                                        TopicPartition.of(x.topic(), x.partition())))
+                        node.id(),
+                        partitionCost.value(node.id()).values().stream()
+                            .mapToDouble(rate -> rate)
                             .sum()))
-            .map(entry -> Map.entry(entry.getKey(), entry.getValue()))
             .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
-
     return () -> brokerLoad;
   }
 
@@ -140,7 +128,24 @@ public class ReplicaDiskInCost implements HasBrokerCost, HasPartitionCost {
                                 Collectors.toUnmodifiableMap(
                                     Map.Entry::getKey, Map.Entry::getValue))))
             .collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, Map.Entry::getValue));
+    // when retention occur, set all partitionScore to -1.
+    if (replicaIn.containsValue(-1.0)) {
+      return new PartitionCost() {
+        @Override
+        public Map<TopicPartition, Double> value(String topic) {
+          return scoreForTopic.get(topic).keySet().stream()
+              .map(x -> Map.entry(x, -1.0))
+              .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+        }
 
+        @Override
+        public Map<TopicPartition, Double> value(int brokerId) {
+          return scoreForBroker.get(brokerId).keySet().stream()
+              .map(x -> Map.entry(x, -1.0))
+              .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+        }
+      };
+    }
     return new PartitionCost() {
       @Override
       public Map<TopicPartition, Double> value(String topic) {
@@ -168,17 +173,6 @@ public class ReplicaDiskInCost implements HasBrokerCost, HasPartitionCost {
    *     doesn't have the sufficient old metric then an exception will likely be thrown.
    * @return a map contain the maximum increase rate of each topic/partition log
    */
-  public static Map<TopicPartition, Double> topicPartitionDataRate(
-      ClusterBean clusterBean, Duration sampleWindow) {
-    return replicaDataRate(clusterBean, sampleWindow).entrySet().stream()
-        .map(
-            x -> {
-              var tpr = x.getKey();
-              return Map.entry(TopicPartition.of(tpr.topic(), tpr.partition()), x.getValue());
-            })
-        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue, (x1, x2) -> x1));
-  }
-
   public static Map<TopicPartitionReplica, Double> replicaDataRate(
       ClusterBean clusterBean, Duration sampleWindow) {
     return clusterBean.mapByReplica().entrySet().parallelStream()
@@ -207,6 +201,8 @@ public class ReplicaDiskInCost implements HasBrokerCost, HasPartitionCost {
                       / 1024.0
                       / ((double) (latestSize.createdTimestamp() - windowSize.createdTimestamp())
                           / 1000);
+              // when retention occur, set all data rate to -1.
+              if (dataRate < 0) dataRate = -1.0;
               return Map.entry(metrics.getKey(), dataRate);
             })
         .collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, Map.Entry::getValue));
diff --git a/app/src/main/java/org/astraea/app/cost/ReplicaLeaderCost.java b/app/src/main/java/org/astraea/app/cost/ReplicaLeaderCost.java
index 93afe4cd..ccf62c7f 100644
--- a/app/src/main/java/org/astraea/app/cost/ReplicaLeaderCost.java
+++ b/app/src/main/java/org/astraea/app/cost/ReplicaLeaderCost.java
@@ -29,19 +29,7 @@ import org.astraea.app.metrics.collector.Fetcher;
 
 /** more replica leaders -> higher cost */
 public class ReplicaLeaderCost implements HasBrokerCost, HasClusterCost {
-  static double coefficientVariation(Map<Integer, Double> brokerScore) {
-    var dataRateMean =
-        brokerScore.values().stream().mapToDouble(x -> (double) x).sum() / brokerScore.size();
-    var dataRateSD =
-        Math.sqrt(
-            brokerScore.values().stream()
-                    .mapToDouble(score -> Math.pow((score - dataRateMean), 2))
-                    .sum()
-                / brokerScore.size());
-    var cv = dataRateSD / dataRateMean;
-    if (cv > 1) return 1.0;
-    return cv;
-  }
+  private final Dispersion dispersion = Dispersion.correlationCoefficient();
 
   @Override
   public BrokerCost brokerCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
@@ -54,7 +42,7 @@ public class ReplicaLeaderCost implements HasBrokerCost, HasClusterCost {
   @Override
   public ClusterCost clusterCost(ClusterInfo clusterInfo, ClusterBean clusterBean) {
     var brokerScore = brokerCost(clusterInfo, clusterBean).value();
-    return () -> coefficientVariation(brokerScore);
+    return () -> dispersion.calculate(brokerScore.values());
   }
 
   Map<Integer, Integer> leaderCount(ClusterInfo ignored, ClusterBean clusterBean) {
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 930bac37..c0d9ed6a 100644
--- a/app/src/main/java/org/astraea/app/performance/ConsumerThread.java
+++ b/app/src/main/java/org/astraea/app/performance/ConsumerThread.java
@@ -23,29 +23,23 @@ import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
 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(List<Consumer<byte[], byte[]>> consumers) {
-    if (consumers.isEmpty()) return List.of();
-    var reports =
-        IntStream.range(0, consumers.size())
-            .mapToObj(ignored -> new Report())
-            .collect(Collectors.toUnmodifiableList());
+  static List<ConsumerThread> create(
+      int consumers, Supplier<SubscribedConsumer<byte[], byte[]>> consumerSupplier) {
+    if (consumers == 0) return List.of();
     var closeLatches =
-        IntStream.range(0, consumers.size())
+        IntStream.range(0, consumers)
             .mapToObj(ignored -> new CountDownLatch(1))
             .collect(Collectors.toUnmodifiableList());
-    var closeFlags =
-        IntStream.range(0, consumers.size())
-            .mapToObj(ignored -> new AtomicBoolean(false))
-            .collect(Collectors.toUnmodifiableList());
-    var executors = Executors.newFixedThreadPool(consumers.size());
+    var executors = Executors.newFixedThreadPool(consumers);
     // monitor
     CompletableFuture.runAsync(
         () -> {
@@ -56,13 +50,13 @@ public interface ConsumerThread extends AbstractThread {
             Utils.swallowException(() -> executors.awaitTermination(30, TimeUnit.SECONDS));
           }
         });
-    return IntStream.range(0, consumers.size())
+    return IntStream.range(0, consumers)
         .mapToObj(
             index -> {
-              var consumer = consumers.get(index);
-              var report = reports.get(index);
+              var consumer = consumerSupplier.get();
+              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 bd5a2338..011dbc6b 100644
--- a/app/src/main/java/org/astraea/app/performance/Performance.java
+++ b/app/src/main/java/org/astraea/app/performance/Performance.java
@@ -31,11 +31,11 @@ import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
-import java.util.stream.IntStream;
 import org.astraea.app.admin.Admin;
 import org.astraea.app.admin.Compression;
 import org.astraea.app.admin.TopicPartition;
 import org.astraea.app.argument.CompressionField;
+import org.astraea.app.argument.DurationField;
 import org.astraea.app.argument.NonEmptyStringField;
 import org.astraea.app.argument.NonNegativeShortField;
 import org.astraea.app.argument.PathField;
@@ -122,22 +122,19 @@ public class Performance {
             param.transactionSize,
             dataSupplier(param),
             partitionSupplier,
-            IntStream.range(0, param.producers)
-                .mapToObj(ignored -> param.createProducer())
-                .collect(Collectors.toUnmodifiableList()));
+            param.producers,
+            param::createProducer);
     var consumerThreads =
         ConsumerThread.create(
-            IntStream.range(0, param.consumers)
-                .mapToObj(
-                    ignored ->
-                        Consumer.forTopics(Set.of(param.topic))
-                            .bootstrapServers(param.bootstrapServers())
-                            .groupId(groupId)
-                            .configs(param.configs())
-                            .isolation(param.isolation())
-                            .seek(latestOffsets)
-                            .build())
-                .collect(Collectors.toUnmodifiableList()));
+            param.consumers,
+            () ->
+                Consumer.forTopics(Set.of(param.topic))
+                    .bootstrapServers(param.bootstrapServers())
+                    .groupId(groupId)
+                    .configs(param.configs())
+                    .isolation(param.isolation())
+                    .seek(latestOffsets)
+                    .build());
 
     var producerReports =
         producerThreads.stream()
@@ -165,9 +162,25 @@ public class Performance {
                             .map(ProducerThread::report)
                             .mapToLong(Report::records)
                             .sum(),
-                    tracker));
-
-    fileWriter.ifPresent(CompletableFuture::runAsync);
+                    producerReports,
+                    consumerReports));
+
+    var fileWriterFuture =
+        fileWriter.map(CompletableFuture::runAsync).orElse(CompletableFuture.completedFuture(null));
+
+    var chaos =
+        param.chaosDuration == null
+            ? CompletableFuture.completedFuture(null)
+            : CompletableFuture.runAsync(
+                () -> {
+                  while (!consumerThreads.stream().allMatch(AbstractThread::closed)) {
+                    var thread =
+                        consumerThreads.get((int) (Math.random() * consumerThreads.size()));
+                    thread.unsubscribe();
+                    Utils.sleep(param.chaosDuration);
+                    thread.resubscribe();
+                  }
+                });
 
     CompletableFuture.runAsync(
         () -> {
@@ -191,6 +204,8 @@ public class Performance {
 
     consumerThreads.forEach(AbstractThread::waitForDone);
     tracker.waitForDone();
+    fileWriterFuture.join();
+    chaos.join();
     return param.topic;
   }
 
@@ -348,5 +363,13 @@ public class Performance {
         description = "Output format for the report",
         converter = ReportFormat.ReportFormatConverter.class)
     ReportFormat reportFormat = ReportFormat.CSV;
+
+    @Parameter(
+        names = {"--chaos.frequency"},
+        description =
+            "time to run the chaos monkey. It will kill consumer arbitrarily. There is no monkey by default",
+        validateWith = DurationField.class,
+        converter = DurationField.class)
+    Duration chaosDuration = 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 4f4c3b5a..dfbafc02 100644
--- a/app/src/main/java/org/astraea/app/performance/ProducerThread.java
+++ b/app/src/main/java/org/astraea/app/performance/ProducerThread.java
@@ -36,21 +36,14 @@ public interface ProducerThread extends AbstractThread {
       int batchSize,
       DataSupplier dataSupplier,
       Supplier<Integer> partitionSupplier,
-      List<Producer<byte[], byte[]>> producers) {
-    if (producers.isEmpty()) return List.of();
-    var reports =
-        IntStream.range(0, producers.size())
-            .mapToObj(ignored -> new Report())
-            .collect(Collectors.toUnmodifiableList());
+      int producers,
+      Supplier<Producer<byte[], byte[]>> producerSupplier) {
+    if (producers <= 0) return List.of();
     var closeLatches =
-        IntStream.range(0, producers.size())
+        IntStream.range(0, producers)
             .mapToObj(ignored -> new CountDownLatch(1))
             .collect(Collectors.toUnmodifiableList());
-    var closedFlags =
-        IntStream.range(0, producers.size())
-            .mapToObj(ignored -> new AtomicBoolean(false))
-            .collect(Collectors.toUnmodifiableList());
-    var executors = Executors.newFixedThreadPool(producers.size());
+    var executors = Executors.newFixedThreadPool(producers);
     // monitor
     CompletableFuture.runAsync(
         () -> {
@@ -61,13 +54,13 @@ public interface ProducerThread extends AbstractThread {
             Utils.swallowException(() -> executors.awaitTermination(30, TimeUnit.SECONDS));
           }
         });
-    return IntStream.range(0, producers.size())
+    return IntStream.range(0, producers)
         .mapToObj(
             index -> {
-              var producer = producers.get(index);
-              var report = reports.get(index);
+              var producer = producerSupplier.get();
+              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 0a000fa1..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,
@@ -123,4 +127,14 @@ public class UtilsTest {
     Assertions.assertThrows(
         RuntimeException.class, () -> Utils.constructCostFunction(aClass, config));
   }
+
+  @Test
+  void testSwallowException() {
+    Assertions.assertDoesNotThrow(
+        () ->
+            Utils.swallowException(
+                () -> {
+                  throw new IllegalArgumentException();
+                }));
+  }
 }
diff --git a/app/src/test/java/org/astraea/app/consumer/ConsumerTest.java b/app/src/test/java/org/astraea/app/consumer/ConsumerTest.java
index ca056022..ee198e96 100644
--- a/app/src/test/java/org/astraea/app/consumer/ConsumerTest.java
+++ b/app/src/test/java/org/astraea/app/consumer/ConsumerTest.java
@@ -26,9 +26,14 @@ import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.BiConsumer;
+import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
 import org.apache.kafka.common.errors.WakeupException;
@@ -403,4 +408,111 @@ public class ConsumerTest extends RequireBrokerCluster {
       consumer.resubscribe();
     }
   }
+
+  @Test
+  void testCreateConsumersConcurrent() throws ExecutionException, InterruptedException {
+    var partitions = 3;
+    var topic = Utils.randomString(10);
+    try (var admin = Admin.of(bootstrapServers())) {
+      admin.creator().topic(topic).numberOfPartitions(partitions).create();
+      Utils.sleep(Duration.ofSeconds(3));
+    }
+
+    // one consume is idle
+    var groupId = Utils.randomString(10);
+    var consumers = partitions + 1;
+    var log = new ConcurrentHashMap<Integer, Integer>();
+    var closed = new AtomicBoolean(false);
+    var fs =
+        Utils.sequence(
+            IntStream.range(0, consumers)
+                .mapToObj(
+                    index ->
+                        CompletableFuture.runAsync(
+                            () -> {
+                              try (var consumer =
+                                  Consumer.forTopics(Set.of(topic))
+                                      .groupId(groupId)
+                                      .bootstrapServers(bootstrapServers())
+                                      .seek(SEEK_TO, 0)
+                                      .consumerRebalanceListener(ps -> log.put(index, ps.size()))
+                                      .build()) {
+                                while (!closed.get()) consumer.poll(Duration.ofSeconds(2));
+                              }
+                            }))
+                .collect(Collectors.toUnmodifiableList()));
+    Utils.waitFor(() -> log.size() == consumers, Duration.ofSeconds(15));
+    Utils.waitFor(
+        () -> log.values().stream().filter(ps -> ps == 0).count() == 1, Duration.ofSeconds(15));
+    closed.set(true);
+    fs.get();
+  }
+
+  @Test
+  void testHistoricalSubscription() {
+    var partitions = 3;
+    var topic = Utils.randomString(10);
+    try (var admin = Admin.of(bootstrapServers())) {
+      admin.creator().topic(topic).numberOfPartitions(partitions).create();
+      Utils.sleep(Duration.ofSeconds(3));
+    }
+
+    var groupId = Utils.randomString(10);
+    var closed = new AtomicBoolean(false);
+
+    var consumers =
+        IntStream.range(0, 2)
+            .mapToObj(
+                ignored ->
+                    Consumer.forTopics(Set.of(topic))
+                        .bootstrapServers(bootstrapServers())
+                        .groupId(groupId)
+                        .enableTrace()
+                        .build())
+            .collect(Collectors.toUnmodifiableList());
+
+    var fs =
+        Utils.sequence(
+            consumers.stream()
+                .map(
+                    c ->
+                        CompletableFuture.runAsync(
+                            () -> {
+                              try (c) {
+                                while (!closed.get()) c.poll(Duration.ofSeconds(1));
+                              }
+                            }))
+                .collect(Collectors.toUnmodifiableList()));
+
+    Utils.waitFor(() -> consumers.stream().allMatch(c -> c.historicalSubscription().size() >= 1));
+
+    // create another consumer to trigger balance
+    try (var consumer =
+        Consumer.forTopics(Set.of(topic))
+            .bootstrapServers(bootstrapServers())
+            .groupId(groupId)
+            .enableTrace()
+            .build()) {
+
+      Utils.waitFor(
+          () -> {
+            consumer.poll(Duration.ofSeconds(1));
+            return consumers.stream().allMatch(c -> c.historicalSubscription().size() > 1);
+          });
+    }
+
+    // produce data to make sure the balance get done.
+    produceData(topic, 100);
+
+    try (var consumer =
+        Consumer.forTopics(Set.of(topic))
+            .bootstrapServers(bootstrapServers())
+            .groupId(groupId)
+            .fromBeginning()
+            .build()) {
+      Utils.waitFor(() -> !consumer.poll(Duration.ofSeconds(1)).isEmpty());
+      Assertions.assertTrue(
+          consumers.stream().anyMatch(c -> c.historicalSubscription().size() > 1));
+    }
+  }
 }
diff --git a/app/src/test/java/org/astraea/app/cost/DispersionTest.java b/app/src/test/java/org/astraea/app/cost/DispersionTest.java
new file mode 100644
index 00000000..703baad7
--- /dev/null
+++ b/app/src/test/java/org/astraea/app/cost/DispersionTest.java
@@ -0,0 +1,31 @@
+/*
+ * 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.cost;
+
+import java.util.List;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class DispersionTest {
+
+  @Test
+  void testCorrelationCoefficient() {
+    var dispersion = Dispersion.correlationCoefficient();
+    var scores = List.of(0.2, 0.4, 0.7);
+    Assertions.assertEquals(0.47418569253607507, dispersion.calculate(scores));
+  }
+}
diff --git a/app/src/test/java/org/astraea/app/cost/NodeTopicSizeCostTest.java b/app/src/test/java/org/astraea/app/cost/NodeTopicSizeCostTest.java
index 2fe69603..b5e0a56a 100644
--- a/app/src/test/java/org/astraea/app/cost/NodeTopicSizeCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/NodeTopicSizeCostTest.java
@@ -16,10 +16,16 @@
  */
 package org.astraea.app.cost;
 
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 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.metrics.BeanObject;
 import org.astraea.app.metrics.broker.LogMetrics;
 import org.junit.jupiter.api.Assertions;
@@ -27,19 +33,30 @@ import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
 
 class NodeTopicSizeCostTest {
+  private final BeanObject bean =
+      new BeanObject(
+          "domain", Map.of("topic", "t", "partition", "10", "name", "SIZE"), Map.of("Value", 777));
 
   @Test
-  void testCost() {
-    var bean =
-        new BeanObject(
-            "domain",
-            Map.of("topic", "t", "partition", "10", "name", "SIZE"),
-            Map.of("Value", 777));
+  void testBrokerCost() {
     var meter = new LogMetrics.Log.Meter(bean);
     var cost = new NodeTopicSizeCost();
     var result =
-        cost.brokerCost(Mockito.mock(ClusterInfo.class), ClusterBean.of(Map.of(1, List.of(meter))));
+        cost.brokerCost(mock(ClusterInfo.class), ClusterBean.of(Map.of(1, List.of(meter))));
     Assertions.assertEquals(1, result.value().size());
     Assertions.assertEquals(777, result.value().entrySet().iterator().next().getValue());
   }
+
+  @Test
+  void testPartitionCost() {
+    var meter = new LogMetrics.Log.Meter(bean);
+    var cost = new NodeTopicSizeCost();
+    var clusterInfo = Mockito.mock(ClusterInfo.class);
+    when(clusterInfo.topics()).thenReturn(Set.of("t"));
+    when(clusterInfo.availableReplicas("t"))
+        .thenReturn(List.of(ReplicaInfo.of("t", 10, NodeInfo.of(0, "0", 0), true, false, false)));
+    var result = cost.partitionCost(clusterInfo, ClusterBean.of(Map.of(0, List.of(meter))));
+    Assertions.assertEquals(1, result.value(0).size());
+    Assertions.assertEquals(777, result.value(0).entrySet().iterator().next().getValue());
+  }
 }
diff --git a/app/src/test/java/org/astraea/app/cost/ReplicaDiskInCostTest.java b/app/src/test/java/org/astraea/app/cost/ReplicaDiskInCostTest.java
index bb8ed0f8..158be525 100644
--- a/app/src/test/java/org/astraea/app/cost/ReplicaDiskInCostTest.java
+++ b/app/src/test/java/org/astraea/app/cost/ReplicaDiskInCostTest.java
@@ -96,6 +96,14 @@ class ReplicaDiskInCostTest extends RequireBrokerCluster {
     Assertions.assertEquals(33.37843418121338, brokerLoad.get(3));
   }
 
+  @Test
+  void testClusterCost() {
+    var configuration = Configuration.of(Map.of("metrics.duration", "3"));
+    var loadCostFunction = new ReplicaDiskInCost(configuration);
+    var brokerLoad = loadCostFunction.clusterCost(clusterInfo(), clusterBean()).value();
+    Assertions.assertEquals(0.20721255412897746, brokerLoad);
+  }
+
   private ClusterInfo clusterInfo() {
     ClusterInfo clusterInfo = Mockito.mock(ClusterInfo.class);
     Mockito.when(clusterInfo.nodes())
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/performance/PerformanceTest.java b/app/src/test/java/org/astraea/app/performance/PerformanceTest.java
index f61cfdcd..62a70cd9 100644
--- a/app/src/test/java/org/astraea/app/performance/PerformanceTest.java
+++ b/app/src/test/java/org/astraea/app/performance/PerformanceTest.java
@@ -17,6 +17,7 @@
 package org.astraea.app.performance;
 
 import com.beust.jcommander.ParameterException;
+import java.time.Duration;
 import java.util.concurrent.CountDownLatch;
 import java.util.function.BiConsumer;
 import org.astraea.app.argument.Argument;
@@ -156,4 +157,13 @@ public class PerformanceTest extends RequireBrokerCluster {
     Assertions.assertThrows(
         ParameterException.class, () -> Argument.parse(new Performance.Argument(), arguments14));
   }
+
+  @Test
+  void testChaosFrequency() {
+    var args =
+        Argument.parse(
+            new Performance.Argument(),
+            new String[] {"--bootstrap.servers", "localhost:9092", "--chaos.frequency", "10s"});
+    Assertions.assertEquals(Duration.ofSeconds(10), args.chaosDuration);
+  }
 }
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