Repository: skiptests/astraea
Pull request #313 (2022-04-29 14:22:50)

Add support for showing data folder of replicas
目前 `PartitionInfo` 以 `NodeInfo` 來表達 replica 儲存在哪些節點。

但由於 Balancer 在做非常細緻的負載平衡時可能需要連 replica 放在哪個 data folder 都列入考量，爲此打開這個 PR 以修改 `PartitionInfo` 的 replica 系列函數簽名的回傳值，由原本的回傳 `NodeInfo` 變成回傳 `ReplicaInfo`。


Patch:
diff --git a/app/src/main/java/org/astraea/argument/CompressionField.java b/app/src/main/java/org/astraea/argument/CompressionField.java
index 4e937574..9f02ebbe 100644
--- a/app/src/main/java/org/astraea/argument/CompressionField.java
+++ b/app/src/main/java/org/astraea/argument/CompressionField.java
@@ -4,17 +4,18 @@ import com.beust.jcommander.ParameterException;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import org.apache.kafka.common.record.CompressionType;
+import org.astraea.topic.Compression;
 
-public class CompressionField extends Field<CompressionType> {
+public class CompressionField extends Field<Compression> {
   /**
    * @param value Name of compression type. Accept lower-case name only ("none", "gzip", "snappy",
    *     "lz4", "zstd").
    */
   @Override
-  public CompressionType convert(String value) {
+  public Compression convert(String value) {
     try {
       // `CompressionType#forName` accept lower-case name only.
-      return CompressionType.forName(value);
+      return Compression.of(value);
     } catch (IllegalArgumentException e) {
       throw new ParameterException(
           "the "
diff --git a/app/src/main/java/org/astraea/consumer/Builder.java b/app/src/main/java/org/astraea/consumer/Builder.java
index c8a579a8..d62b1ca8 100644
--- a/app/src/main/java/org/astraea/consumer/Builder.java
+++ b/app/src/main/java/org/astraea/consumer/Builder.java
@@ -13,25 +13,18 @@ import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 
 public class Builder<Key, Value> {
-
-  public enum OffsetPolicy {
-    EARLIEST,
-    LATEST
-  }
-
-  private final Map<String, Object> configs = new HashMap<>();
+  private final Map<String, Object> configs =
+      new HashMap<>(
+          Map.of(ConsumerConfig.GROUP_ID_CONFIG, "groupId-" + System.currentTimeMillis()));
   private Deserializer<?> keyDeserializer = Deserializer.BYTE_ARRAY;
   private Deserializer<?> valueDeserializer = Deserializer.BYTE_ARRAY;
-  private OffsetPolicy offsetPolicy = OffsetPolicy.LATEST;
-  private String groupId = "groupId-" + System.currentTimeMillis();
   private final Set<String> topics = new HashSet<>();
   private ConsumerRebalanceListener listener = ignore -> {};
 
   Builder() {}
 
   public Builder<Key, Value> groupId(String groupId) {
-    this.groupId = Objects.requireNonNull(groupId);
-    return this;
+    return config(ConsumerConfig.GROUP_ID_CONFIG, Objects.requireNonNull(groupId));
   }
 
   /**
@@ -40,8 +33,7 @@ public class Builder<Key, Value> {
    * @return this builder
    */
   public Builder<Key, Value> fromBeginning() {
-    this.offsetPolicy = OffsetPolicy.EARLIEST;
-    return this;
+    return config(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
   }
 
   public Builder<Key, Value> topics(Set<String> topics) {
@@ -62,14 +54,18 @@ public class Builder<Key, Value> {
     return (Builder<Key, NewValue>) this;
   }
 
-  public Builder<Key, Value> configs(Map<String, Object> configs) {
+  public Builder<Key, Value> config(String key, String value) {
+    this.configs.put(key, value);
+    return this;
+  }
+
+  public Builder<Key, Value> configs(Map<String, String> configs) {
     this.configs.putAll(configs);
     return this;
   }
 
   public Builder<Key, Value> brokers(String brokers) {
-    this.configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, Objects.requireNonNull(brokers));
-    return this;
+    return config(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, Objects.requireNonNull(brokers));
   }
 
   public Builder<Key, Value> consumerRebalanceListener(ConsumerRebalanceListener listener) {
@@ -77,17 +73,12 @@ public class Builder<Key, Value> {
     return this;
   }
 
+  public Builder<Key, Value> isolation(Isolation isolation) {
+    return config(ConsumerConfig.ISOLATION_LEVEL_CONFIG, isolation.nameOfKafka());
+  }
+
   @SuppressWarnings("unchecked")
   public Consumer<Key, Value> build() {
-    configs.put(ConsumerConfig.GROUP_ID_CONFIG, Objects.requireNonNull(groupId));
-    switch (offsetPolicy) {
-      case EARLIEST:
-        configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
-        break;
-      case LATEST:
-        configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
-        break;
-    }
     var kafkaConsumer =
         new KafkaConsumer<>(
             configs,
diff --git a/app/src/main/java/org/astraea/consumer/Consumer.java b/app/src/main/java/org/astraea/consumer/Consumer.java
index a4d2268e..3e936217 100644
--- a/app/src/main/java/org/astraea/consumer/Consumer.java
+++ b/app/src/main/java/org/astraea/consumer/Consumer.java
@@ -29,7 +29,7 @@ public interface Consumer<Key, Value> extends AutoCloseable {
     return builder().brokers(brokers).build();
   }
 
-  static Consumer<byte[], byte[]> of(Map<String, Object> configs) {
+  static Consumer<byte[], byte[]> of(Map<String, String> configs) {
     return builder().configs(configs).build();
   }
 }
diff --git a/app/src/main/java/org/astraea/consumer/Isolation.java b/app/src/main/java/org/astraea/consumer/Isolation.java
new file mode 100644
index 00000000..63b91501
--- /dev/null
+++ b/app/src/main/java/org/astraea/consumer/Isolation.java
@@ -0,0 +1,17 @@
+package org.astraea.consumer;
+
+import java.util.Locale;
+
+public enum Isolation {
+  READ_UNCOMMITTED,
+  READ_COMMITTED;
+
+  public static Isolation of(String name) {
+    return Isolation.valueOf(name.toUpperCase(Locale.ROOT));
+  }
+
+  /** @return the name parsed by kafka */
+  public String nameOfKafka() {
+    return name().toLowerCase(Locale.ROOT);
+  }
+}
diff --git a/app/src/main/java/org/astraea/cost/ClusterInfo.java b/app/src/main/java/org/astraea/cost/ClusterInfo.java
index 96fe70dd..b147ff26 100644
--- a/app/src/main/java/org/astraea/cost/ClusterInfo.java
+++ b/app/src/main/java/org/astraea/cost/ClusterInfo.java
@@ -25,16 +25,28 @@ public interface ClusterInfo {
       }
 
       @Override
-      public List<PartitionInfo> availablePartitions(String topic) {
+      public List<ReplicaInfo> availablePartitionLeaders(String topic) {
         return cluster.availablePartitionsForTopic(topic).stream()
-            .map(PartitionInfo::of)
+            .map(ReplicaInfo::of)
+            .map(
+                replicas ->
+                    replicas.stream().filter(ReplicaInfo::isLeader).findFirst().orElseThrow())
             .collect(Collectors.toUnmodifiableList());
       }
 
       @Override
-      public List<PartitionInfo> partitions(String topic) {
+      public List<ReplicaInfo> availablePartitions(String topic) {
+        return cluster.availablePartitionsForTopic(topic).stream()
+            .map(ReplicaInfo::of)
+            .flatMap(Collection::stream)
+            .collect(Collectors.toUnmodifiableList());
+      }
+
+      @Override
+      public List<ReplicaInfo> partitions(String topic) {
         return cluster.partitionsForTopic(topic).stream()
-            .map(PartitionInfo::of)
+            .map(ReplicaInfo::of)
+            .flatMap(Collection::stream)
             .collect(Collectors.toUnmodifiableList());
       }
 
@@ -71,7 +83,12 @@ public interface ClusterInfo {
       }
 
       @Override
-      public List<PartitionInfo> availablePartitions(String topic) {
+      public List<ReplicaInfo> availablePartitionLeaders(String topic) {
+        return cluster.availablePartitionLeaders(topic);
+      }
+
+      @Override
+      public List<ReplicaInfo> availablePartitions(String topic) {
         return cluster.availablePartitions(topic);
       }
 
@@ -81,7 +98,7 @@ public interface ClusterInfo {
       }
 
       @Override
-      public List<PartitionInfo> partitions(String topic) {
+      public List<ReplicaInfo> partitions(String topic) {
         return cluster.partitions(topic);
       }
 
@@ -131,12 +148,21 @@ public interface ClusterInfo {
   List<NodeInfo> nodes();
 
   /**
-   * Get the list of available partitions for this topic
+   * Get the list of replica leader information of each available partition for the given topic
+   *
+   * @param topic The Topic name
+   * @return A list of {@link ReplicaInfo}
+   */
+  List<ReplicaInfo> availablePartitionLeaders(String topic);
+
+  /**
+   * Get the list of replica information of each available partition/replica pair for the given
+   * topic
    *
    * @param topic The topic name
-   * @return A list of partitions
+   * @return A list of {@link ReplicaInfo}
    */
-  List<PartitionInfo> availablePartitions(String topic);
+  List<ReplicaInfo> availablePartitions(String topic);
 
   /**
    * All topic names
@@ -146,12 +172,12 @@ public interface ClusterInfo {
   Set<String> topics();
 
   /**
-   * Get the list of partitions for this topic
+   * Get the list of replica information of each partition/replica pair for the given topic
    *
    * @param topic The topic name
-   * @return A list of partitions
+   * @return A list of {@link ReplicaInfo}
    */
-  List<PartitionInfo> partitions(String topic);
+  List<ReplicaInfo> partitions(String topic);
 
   /**
    * @param brokerId broker id
diff --git a/app/src/main/java/org/astraea/cost/PartitionInfo.java b/app/src/main/java/org/astraea/cost/PartitionInfo.java
deleted file mode 100644
index 87c19003..00000000
--- a/app/src/main/java/org/astraea/cost/PartitionInfo.java
+++ /dev/null
@@ -1,80 +0,0 @@
-package org.astraea.cost;
-
-import java.util.Arrays;
-import java.util.List;
-import java.util.stream.Collectors;
-
-public interface PartitionInfo {
-
-  static PartitionInfo of(org.apache.kafka.common.PartitionInfo pf) {
-    return of(
-        pf.topic(),
-        pf.partition(),
-        NodeInfo.of(pf.leader()),
-        Arrays.stream(pf.replicas()).map(NodeInfo::of).collect(Collectors.toUnmodifiableList()),
-        Arrays.stream(pf.inSyncReplicas())
-            .map(NodeInfo::of)
-            .collect(Collectors.toUnmodifiableList()),
-        Arrays.stream(pf.offlineReplicas())
-            .map(NodeInfo::of)
-            .collect(Collectors.toUnmodifiableList()));
-  }
-
-  static PartitionInfo of(
-      String topic,
-      int partition,
-      NodeInfo leader,
-      List<NodeInfo> replicas,
-      List<NodeInfo> inSyncReplicas,
-      List<NodeInfo> offlineReplicas) {
-    return new PartitionInfo() {
-      @Override
-      public String topic() {
-        return topic;
-      }
-
-      @Override
-      public int partition() {
-        return partition;
-      }
-
-      @Override
-      public NodeInfo leader() {
-        return leader;
-      }
-
-      @Override
-      public List<NodeInfo> replicas() {
-        return List.copyOf(replicas);
-      }
-
-      @Override
-      public List<NodeInfo> inSyncReplica() {
-        return List.copyOf(inSyncReplicas);
-      }
-
-      @Override
-      public List<NodeInfo> offlineReplicas() {
-        return List.copyOf(offlineReplicas);
-      }
-    };
-  }
-
-  /** @return topic name */
-  String topic();
-
-  /** @return partition id */
-  int partition();
-
-  /** @return information of leader node */
-  NodeInfo leader();
-
-  /** @return a list of replicas */
-  List<NodeInfo> replicas();
-
-  /** @return a list of in-sync replicas */
-  List<NodeInfo> inSyncReplica();
-
-  /** @return a list of offline replicas */
-  List<NodeInfo> offlineReplicas();
-}
diff --git a/app/src/main/java/org/astraea/cost/ReplicaInfo.java b/app/src/main/java/org/astraea/cost/ReplicaInfo.java
new file mode 100644
index 00000000..13bb677b
--- /dev/null
+++ b/app/src/main/java/org/astraea/cost/ReplicaInfo.java
@@ -0,0 +1,153 @@
+package org.astraea.cost;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public interface ReplicaInfo {
+
+  static Set<ReplicaInfo> of(org.apache.kafka.common.PartitionInfo pf) {
+    final var replicas = List.of(pf.replicas());
+    final var leaderReplica = List.of(pf.leader());
+    final var inSyncReplicas = List.of(pf.inSyncReplicas());
+    final var offlineReplicas = List.of(pf.offlineReplicas());
+
+    return replicas.stream()
+        .map(
+            node ->
+                of(
+                    pf.topic(),
+                    pf.partition(),
+                    NodeInfo.of(node),
+                    leaderReplica.contains(node),
+                    inSyncReplicas.contains(node),
+                    offlineReplicas.contains(node)))
+        .collect(Collectors.toUnmodifiableSet());
+  }
+
+  static ReplicaInfo of(
+      String topic,
+      int partition,
+      NodeInfo nodeInfo,
+      boolean isLeader,
+      boolean isSynced,
+      boolean isOfflineReplica) {
+    return of(topic, partition, nodeInfo, isLeader, isSynced, isOfflineReplica, null);
+  }
+
+  static ReplicaInfo of(
+      String topic,
+      int partition,
+      NodeInfo nodeInfo,
+      boolean isLeader,
+      boolean isSynced,
+      boolean isOfflineReplica,
+      String dataFolder) {
+    return new ReplicaInfo() {
+      @Override
+      public String topic() {
+        return topic;
+      }
+
+      @Override
+      public int partition() {
+        return partition;
+      }
+
+      @Override
+      public NodeInfo nodeInfo() {
+        return nodeInfo;
+      }
+
+      @Override
+      public boolean isLeader() {
+        return isLeader;
+      }
+
+      @Override
+      public boolean inSync() {
+        return isSynced;
+      }
+
+      @Override
+      public boolean isOfflineReplica() {
+        return isOfflineReplica;
+      }
+
+      @Override
+      public Optional<String> dataFolder() {
+        return Optional.ofNullable(dataFolder);
+      }
+
+      @Override
+      public String toString() {
+        return "ReplicaInfo {"
+            + "topic=\""
+            + topic
+            + "\" partition="
+            + partition
+            + " replicaAtBroker="
+            + nodeInfo.id()
+            + " dataFolder="
+            + (dataFolder().map(x -> "\"" + x + "\"").orElse("unknown"))
+            + (isLeader() ? " leader" : "")
+            + (isFollower() ? " follower" : "")
+            + (inSync() ? ":synced" : "")
+            + (isOfflineReplica() ? ":offline" : "")
+            + "}";
+      }
+
+      @Override
+      public boolean equals(Object obj) {
+        if (obj instanceof ReplicaInfo) {
+          var that = ((ReplicaInfo) obj);
+
+          return this.topic().equals(that.topic())
+              && this.partition() == that.partition()
+              && this.nodeInfo().equals(that.nodeInfo())
+              && this.isLeader() == that.isLeader()
+              && this.isFollower() == that.isFollower()
+              && this.inSync() == that.inSync()
+              && this.isOfflineReplica() == that.isOfflineReplica()
+              && this.dataFolder().equals(that.dataFolder());
+        }
+        return false;
+      }
+    };
+  }
+
+  /** @return topic name */
+  String topic();
+
+  /** @return partition id */
+  int partition();
+
+  /** @return information of the node hosts this replica */
+  NodeInfo nodeInfo();
+
+  /** @return true if this replica is a leader replica */
+  boolean isLeader();
+
+  /** @return true if this replica is a follower replica */
+  default boolean isFollower() {
+    return !isLeader();
+  }
+
+  /** @return true if this replica is synced */
+  boolean inSync();
+
+  /** @return true if this replica is offline */
+  boolean isOfflineReplica();
+
+  /**
+   * The path to the data folder which hosts this replica. Since this information is not very openly
+   * available. An application might find it hard to retrieve this information(for example, the
+   * producer client might need to initialize an AdminClient to access this information). To provide
+   * this information or not is totally up to the caller.
+   *
+   * @return a {@link Optional<String>} that indicates the data folder path which stored this
+   *     replica on a specific Kafka node.
+   */
+  Optional<String> dataFolder();
+}
diff --git a/app/src/main/java/org/astraea/cost/brokersMetrics/NeutralIntegratedCost.java b/app/src/main/java/org/astraea/cost/brokersMetrics/NeutralIntegratedCost.java
new file mode 100644
index 00000000..623fb8be
--- /dev/null
+++ b/app/src/main/java/org/astraea/cost/brokersMetrics/NeutralIntegratedCost.java
@@ -0,0 +1,255 @@
+package org.astraea.cost.brokersMetrics;
+
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import org.astraea.cost.BrokerCost;
+import org.astraea.cost.ClusterInfo;
+import org.astraea.cost.CostFunction;
+import org.astraea.cost.HasBrokerCost;
+import org.astraea.cost.Periodic;
+import org.astraea.metrics.collector.Fetcher;
+
+/**
+ * The result is computed by four cost function.There are "BrokerInputCost", "BrokerOutputCost",
+ * "CpuCost" and "MemoryCost". We used a combination of AHP(Analytic Hierarchy Process) and entropy
+ * weighting to integrate these data into a score that is representative of Broker's load profile.
+ * Because AHP is a subjective evaluation method and entropy method is an objective evaluation
+ * method, so I named it NeutralIntegratedCost.
+ */
+public class NeutralIntegratedCost extends Periodic<Map<Integer, Double>> implements HasBrokerCost {
+  private final List<HasBrokerCost> metricsCost =
+      List.of(new BrokerInputCost(), new BrokerOutputCost(), new CpuCost(), new MemoryCost());
+  private final Map<Integer, BrokerMetrics> brokersMetric = new HashMap<>();
+  private final EntropyEmpowerment entropyEmpowerment = new EntropyEmpowerment();
+  private final AHPEmpowerment ahpEmpowerment = new AHPEmpowerment();
+
+  @Override
+  public BrokerCost brokerCost(ClusterInfo clusterInfo) {
+    var score =
+        tryUpdate(
+            () -> {
+              var costMetrics =
+                  clusterInfo.allBeans().entrySet().stream()
+                      .collect(Collectors.toMap(Map.Entry::getKey, entry -> 0.0));
+              costMetrics.forEach(
+                  (key, value) -> {
+                    if (!brokersMetric.containsKey(key)) {
+                      brokersMetric.put(key, new BrokerMetrics());
+                    }
+                  });
+
+              metricsCost.forEach(
+                  hasBrokerCost -> {
+                    if (hasBrokerCost instanceof BrokerInputCost) {
+                      hasBrokerCost
+                          .brokerCost(clusterInfo)
+                          .value()
+                          .forEach(
+                              (brokerID, value) -> brokersMetric.get(brokerID).inputScore = value);
+                    } else if (hasBrokerCost instanceof BrokerOutputCost) {
+                      hasBrokerCost
+                          .brokerCost(clusterInfo)
+                          .value()
+                          .forEach(
+                              (brokerID, value) -> brokersMetric.get(brokerID).outputScore = value);
+                    } else if (hasBrokerCost instanceof CpuCost) {
+                      hasBrokerCost
+                          .brokerCost(clusterInfo)
+                          .value()
+                          .forEach(
+                              (brokerID, value) -> brokersMetric.get(brokerID).cpuScore = value);
+                    } else if (hasBrokerCost instanceof MemoryCost) {
+                      hasBrokerCost
+                          .brokerCost(clusterInfo)
+                          .value()
+                          .forEach(
+                              (brokerID, value) -> brokersMetric.get(brokerID).memoryScore = value);
+                    }
+                  });
+
+              var integratedEmpowerment =
+                  entropyEmpowerment.empowerment(brokersMetric).entrySet().stream()
+                      .collect(
+                          Collectors.toMap(
+                              Map.Entry::getKey,
+                              entry ->
+                                  entry.getValue() * 0.5
+                                      + ahpEmpowerment.empowerment().get(entry.getKey()) * 0.5));
+
+              return brokersMetric.entrySet().stream()
+                  .collect(
+                      Collectors.toMap(
+                          Map.Entry::getKey,
+                          entry ->
+                              entry.getValue().inputScore
+                                      * integratedEmpowerment.get(
+                                          Metrics.inputThroughput.metricName)
+                                  + entry.getValue().outputScore
+                                      * integratedEmpowerment.get(
+                                          Metrics.outputThroughput.metricName)
+                                  + entry.getValue().cpuScore
+                                      * integratedEmpowerment.get(Metrics.cpu.metricName)
+                                  + entry.getValue().memoryScore
+                                      * integratedEmpowerment.get(Metrics.memory.metricName)));
+            });
+
+    return () -> score;
+  }
+
+  @Override
+  public Fetcher fetcher() {
+    return Fetcher.of(metricsCost.stream().map(CostFunction::fetcher).collect(Collectors.toList()));
+  }
+
+  static class BrokerMetrics {
+    double inputScore = 0.0;
+    double outputScore = 0.0;
+    double cpuScore = 0.0;
+    double memoryScore = 0.0;
+
+    BrokerMetrics() {}
+  }
+
+  /**
+   * Entropy weight method (EWM) is a commonly used weighting method that measures value dispersion
+   * in decision-making. The greater the degree of dispersion, the greater the degree of
+   * differentiation, and more information can be derived. Meanwhile, higher weight should be given
+   * to the index, and vice versa.
+   */
+  static class EntropyEmpowerment {
+    public Map<String, Double> empowerment(Map<Integer, BrokerMetrics> brokerMetrics) {
+      var metrics = new HashMap<String, Double>();
+      metrics.put(
+          Metrics.inputThroughput.metricName,
+          entropy(
+              brokerMetrics.entrySet().stream()
+                  .collect(
+                      Collectors.toMap(Map.Entry::getKey, entry -> entry.getValue().inputScore))));
+      metrics.put(
+          Metrics.outputThroughput.metricName,
+          entropy(
+              brokerMetrics.entrySet().stream()
+                  .collect(
+                      Collectors.toMap(Map.Entry::getKey, entry -> entry.getValue().outputScore))));
+      metrics.put(
+          Metrics.memory.metricName,
+          entropy(
+              brokerMetrics.entrySet().stream()
+                  .collect(
+                      Collectors.toMap(Map.Entry::getKey, entry -> entry.getValue().memoryScore))));
+      metrics.put(
+          Metrics.cpu.metricName,
+          entropy(
+              brokerMetrics.entrySet().stream()
+                  .collect(
+                      Collectors.toMap(Map.Entry::getKey, entry -> entry.getValue().cpuScore))));
+      var sum = metrics.values().stream().mapToDouble(i -> i).sum();
+
+      return metrics.entrySet().stream()
+          .collect(Collectors.toMap(Map.Entry::getKey, entry -> entry.getValue() / sum));
+    }
+
+    /**
+     * To the cost indexes
+     *
+     * @param metrics JMX metrics for each broker.
+     * @return Standardization of result
+     */
+    Map<Integer, Double> negativeIndicator(Map<Integer, Double> metrics) {
+      Comparator<Double> comparator = Comparator.comparing(Double::doubleValue);
+      var max = metrics.values().stream().max(comparator).orElse(0.0);
+      var min = metrics.values().stream().min(comparator).orElse(0.0);
+      return metrics.entrySet().stream()
+          .collect(
+              Collectors.toMap(Map.Entry::getKey, entry -> (max - entry.getValue()) / (max - min)));
+    }
+
+    /**
+     * To the benefit indexes.
+     *
+     * @param metrics JMX metrics for each broker.
+     * @return Standardization of result
+     */
+    Map<Integer, Double> positiveIndicator(Map<Integer, Double> metrics) {
+      Comparator<Double> comparator = Comparator.comparing(Double::doubleValue);
+      var max = metrics.values().stream().max(comparator).orElse(0.0) + 0.01;
+      double min = metrics.values().stream().min(comparator).orElse(0.0);
+      return metrics.entrySet().stream()
+          .collect(
+              Collectors.toMap(
+                  Map.Entry::getKey, entry -> (entry.getValue() + 0.01 - min) / (max - min)));
+    }
+
+    Map<Integer, Double> metricsWeight(Map<Integer, Double> indicator) {
+      var sum = indicator.values().stream().mapToDouble(i -> i).sum();
+      return indicator.entrySet().stream()
+          .collect(Collectors.toMap(Map.Entry::getKey, entry -> entry.getValue() / sum));
+    }
+
+    double entropy(Map<Integer, Double> metrics) {
+      return 1
+          - metricsWeight(positiveIndicator(metrics)).values().stream()
+                  .filter(weight -> !weight.equals(0.0))
+                  .mapToDouble(weight -> weight * Math.log(weight))
+                  .sum()
+              / (-Math.log(metrics.size()));
+    }
+  }
+
+  /**
+   * AHP is a structured technique for organizing and solving complex decision-making problems based
+   * on mathematics and psychology. AHP provides a comprehensive and logical framework to quantify
+   * each structural decision-making element within a hierarchical structure. The AHP begins with
+   * choosing the decision criteria.
+   *
+   * <p>This is the pairwise comparison matrix currently being used by Astraea Partitioner.
+   *
+   * <p>||==============||===============||================||==========||=============||========||
+   * || Metrics --------||InputThroughput||OutputThroughput|| CpuUsage || MemoryUsage || Weight ||
+   * || InputThroughput-||------ 1 ------||------ 4 -------||--- 6 ----||----- 6 -----|| 0.6003 ||
+   * || outputThroughput||----- 1/4 -----||------ 1 -------||--- 4 ----||----- 4 -----|| 0.2434 ||
+   * || CpuUsage--------||----- 1/6 -----||----- 1/4 ------||--- 1 ----||---- 1/2 ----|| 0.0914 ||
+   * || MemoryUsage-----||----- 1/6 -----||----- 1/4 ------||--- 2 ----||----- 1 -----|| 0.0649 ||
+   * ||=================||===============||================||==========||=============||========||
+   */
+  private static class AHPEmpowerment {
+    private final Map<String, Double> balanceThroughput =
+        Map.of(
+            Metrics.inputThroughput.metricName(),
+            0.6003,
+            Metrics.outputThroughput.metricName(),
+            0.2434,
+            Metrics.cpu.metricName(),
+            0.0649,
+            Metrics.memory.metricName(),
+            0.0914);
+
+    public Map<String, Double> empowerment() {
+      return balanceThroughput;
+    }
+  }
+
+  private enum Metrics {
+    inputThroughput("inputThroughput"),
+    outputThroughput("outputThroughput"),
+    memory("memory"),
+    cpu("cpu");
+
+    private final String metricName;
+
+    Metrics(String name) {
+      this.metricName = name;
+    }
+
+    public String metricName() {
+      return metricName;
+    }
+  }
+
+  EntropyEmpowerment entropyEmpowerment() {
+    return entropyEmpowerment;
+  }
+}
diff --git a/app/src/main/java/org/astraea/partitioner/StrictCostDispatcher.java b/app/src/main/java/org/astraea/partitioner/StrictCostDispatcher.java
index f25426b6..e93249dc 100644
--- a/app/src/main/java/org/astraea/partitioner/StrictCostDispatcher.java
+++ b/app/src/main/java/org/astraea/partitioner/StrictCostDispatcher.java
@@ -12,7 +12,7 @@ import org.astraea.Utils;
 import org.astraea.cost.ClusterInfo;
 import org.astraea.cost.CostFunction;
 import org.astraea.cost.HasBrokerCost;
-import org.astraea.cost.PartitionInfo;
+import org.astraea.cost.ReplicaInfo;
 import org.astraea.metrics.collector.BeanCollector;
 import org.astraea.metrics.collector.Fetcher;
 import org.astraea.metrics.collector.Receiver;
@@ -48,20 +48,20 @@ public class StrictCostDispatcher implements Dispatcher {
 
   @Override
   public int partition(String topic, byte[] key, byte[] value, ClusterInfo clusterInfo) {
-    var partitions = clusterInfo.availablePartitions(topic);
+    var partitionLeaders = clusterInfo.availablePartitionLeaders(topic);
     // just return first partition if there is no available partitions
-    if (partitions.isEmpty()) return 0;
+    if (partitionLeaders.isEmpty()) return 0;
 
     // just return the only one available partition
-    if (partitions.size() == 1) return partitions.iterator().next().partition();
+    if (partitionLeaders.size() == 1) return partitionLeaders.iterator().next().partition();
 
     // add new receivers for new brokers
-    partitions.stream()
-        .filter(p -> !receivers.containsKey(p.leader().id()))
+    partitionLeaders.stream()
+        .filter(p -> !receivers.containsKey(p.nodeInfo().id()))
         .forEach(
             p ->
                 receivers.put(
-                    p.leader().id(), receiver(p.leader().host(), jmxPort(p.leader().id()))));
+                    p.nodeInfo().id(), receiver(p.nodeInfo().host(), jmxPort(p.nodeInfo().id()))));
 
     // get latest beans for each node
     var beans =
@@ -76,18 +76,20 @@ public class StrictCostDispatcher implements Dispatcher {
             .map(f -> f.brokerCost(ClusterInfo.of(clusterInfo, beans)).value())
             .collect(Collectors.toUnmodifiableList());
 
-    return bestPartition(partitions, scores).map(e -> e.getKey().partition()).orElse(0);
+    return bestPartition(partitionLeaders, scores).map(e -> e.getKey().partition()).orElse(0);
   }
 
   // visible for testing
-  static Optional<Map.Entry<PartitionInfo, Double>> bestPartition(
-      List<PartitionInfo> partitions, List<Map<Integer, Double>> scores) {
+  static Optional<Map.Entry<ReplicaInfo, Double>> bestPartition(
+      List<ReplicaInfo> partitions, List<Map<Integer, Double>> scores) {
     return partitions.stream()
         .map(
             p ->
                 Map.entry(
                     p,
-                    scores.stream().mapToDouble(s -> s.getOrDefault(p.leader().id(), 0.0D)).sum()))
+                    scores.stream()
+                        .mapToDouble(s -> s.getOrDefault(p.nodeInfo().id(), 0.0D))
+                        .sum()))
         .min(Map.Entry.comparingByValue());
   }
 
diff --git a/app/src/main/java/org/astraea/partitioner/smoothPartitioner/SmoothWeightDispatcher.java b/app/src/main/java/org/astraea/partitioner/smoothPartitioner/SmoothWeightDispatcher.java
index 48c03a2f..53e35834 100644
--- a/app/src/main/java/org/astraea/partitioner/smoothPartitioner/SmoothWeightDispatcher.java
+++ b/app/src/main/java/org/astraea/partitioner/smoothPartitioner/SmoothWeightDispatcher.java
@@ -51,22 +51,23 @@ public class SmoothWeightDispatcher implements Dispatcher {
 
   @Override
   public int partition(String topic, byte[] key, byte[] value, ClusterInfo clusterInfo) {
-    var partitions = clusterInfo.availablePartitions(topic);
+    var partitionLeaders = clusterInfo.availablePartitionLeaders(topic);
     // just return first partition if there is no available partitions
-    if (partitions.isEmpty()) return 0;
+    if (partitionLeaders.isEmpty()) return 0;
 
     // just return the only one available partition
-    if (partitions.size() == 1) return partitions.iterator().next().partition();
+    if (partitionLeaders.size() == 1) return partitionLeaders.iterator().next().partition();
 
     if (Utils.overSecond(lastFetchTime, 1) && lock.tryLock()) {
       try {
         // add new receivers for new brokers
-        partitions.stream()
-            .filter(p -> !receivers.containsKey(p.leader().id()))
+        partitionLeaders.stream()
+            .filter(p -> !receivers.containsKey(p.nodeInfo().id()))
             .forEach(
                 p ->
                     receivers.put(
-                        p.leader().id(), receiver(p.leader().host(), jmxPort(p.leader().id()))));
+                        p.nodeInfo().id(),
+                        receiver(p.nodeInfo().host(), jmxPort(p.nodeInfo().id()))));
 
         // fetch the latest beans for each node
         var beans =
@@ -99,8 +100,8 @@ public class SmoothWeightDispatcher implements Dispatcher {
     // Make "smooth weight choosing" on the score.
     var targetBroker = smoothWeightCal.getAndChoose();
     var targetPartitions =
-        partitions.stream()
-            .filter(p -> p.leader().id() == targetBroker)
+        partitionLeaders.stream()
+            .filter(p -> p.nodeInfo().id() == targetBroker)
             .collect(Collectors.toList());
 
     return targetPartitions.get(random.nextInt(targetPartitions.size())).partition();
diff --git a/app/src/main/java/org/astraea/performance/Performance.java b/app/src/main/java/org/astraea/performance/Performance.java
index e99671f8..95df6ec3 100644
--- a/app/src/main/java/org/astraea/performance/Performance.java
+++ b/app/src/main/java/org/astraea/performance/Performance.java
@@ -15,12 +15,9 @@ import java.util.concurrent.ExecutionException;
 import java.util.function.BiConsumer;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
-import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.internals.DefaultPartitioner;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.WakeupException;
-import org.apache.kafka.common.record.CompressionType;
 import org.astraea.Utils;
 import org.astraea.argument.CompressionField;
 import org.astraea.argument.NonEmptyStringField;
@@ -33,8 +30,10 @@ import org.astraea.concurrent.Executor;
 import org.astraea.concurrent.State;
 import org.astraea.concurrent.ThreadPool;
 import org.astraea.consumer.Consumer;
+import org.astraea.consumer.Isolation;
 import org.astraea.producer.Producer;
 import org.astraea.producer.TransactionalProducer;
+import org.astraea.topic.Compression;
 import org.astraea.topic.TopicAdmin;
 import org.astraea.utils.DataSize;
 import org.astraea.utils.DataUnit;
@@ -115,7 +114,8 @@ public class Performance {
                                     .brokers(param.brokers)
                                     .topics(Set.of(param.topic))
                                     .groupId(groupId)
-                                    .configs(param.consumerProps())
+                                    .configs(param.configs)
+                                    .isolation(param.isolation())
                                     .consumerRebalanceListener(
                                         ignore -> manager.countDownGetAssignment())
                                     .build(),
@@ -128,11 +128,13 @@ public class Performance {
                         i ->
                             producerExecutor(
                                 Producer.builder()
-                                    .configs(param.producerProps())
+                                    .configs(param.configs)
+                                    .compression(param.compression)
                                     .partitionClassName(param.partitioner)
                                     .build(),
                                 Producer.builder()
-                                    .configs(param.producerProps())
+                                    .configs(param.configs)
+                                    .compression(param.compression)
                                     .partitionClassName(param.partitioner)
                                     .buildTransactional(),
                                 param,
@@ -192,57 +194,71 @@ public class Performance {
       List<Integer> partitions,
       Manager manager) {
     return new Executor() {
+
+      private State doTransaction() {
+        var rand = new Random();
+        var senders =
+            IntStream.range(0, param.transactionSize)
+                .mapToObj(i -> manager.payload())
+                .filter(Optional::isPresent)
+                .map(
+                    p ->
+                        producer
+                            .sender()
+                            .topic(param.topic)
+                            .partition(partitions.get(rand.nextInt(partitions.size())))
+                            .key(manager.getKey())
+                            .value(p.get())
+                            .timestamp(System.currentTimeMillis()))
+                .collect(Collectors.toList());
+
+        // No records to send
+        if (senders.isEmpty()) return State.DONE;
+        transactionalProducer
+            .transaction(senders)
+            .forEach(
+                future ->
+                    future.whenComplete(
+                        (m, e) ->
+                            observer.accept(
+                                System.currentTimeMillis() - m.timestamp(),
+                                m.serializedValueSize())));
+        return State.RUNNING;
+      }
+
+      private State doSend() {
+        var rand = new Random();
+        var payload = manager.payload();
+        if (payload.isEmpty()) return State.DONE;
+
+        long start = System.currentTimeMillis();
+        producer
+            .sender()
+            .topic(param.topic)
+            .partition(partitions.get(rand.nextInt(partitions.size())))
+            .key(manager.getKey())
+            .value(payload.get())
+            .timestamp(start)
+            .run()
+            .whenComplete(
+                (m, e) ->
+                    observer.accept(System.currentTimeMillis() - start, m.serializedValueSize()));
+        return State.RUNNING;
+      }
+
       @Override
       public State execute() throws InterruptedException {
         // Wait for all consumers get assignment.
         manager.awaitPartitionAssignment();
-        var rand = new Random();
         // Do transactional send.
-        if (param.transaction()) {
-          var senders =
-              IntStream.range(0, param.transactionSize)
-                  .mapToObj(i -> manager.payload())
-                  .filter(Optional::isPresent)
-                  .map(
-                      p ->
-                          producer
-                              .sender()
-                              .topic(param.topic)
-                              .partition(partitions.get(rand.nextInt(partitions.size())))
-                              .key(manager.getKey())
-                              .value(p.get())
-                              .timestamp(System.currentTimeMillis()))
-                  .collect(Collectors.toList());
-
-          // No records to send
-          if (senders.isEmpty()) return State.DONE;
-          transactionalProducer
-              .transaction(senders)
-              .forEach(
-                  future ->
-                      future.whenComplete(
-                          (m, e) ->
-                              observer.accept(
-                                  System.currentTimeMillis() - m.timestamp(),
-                                  m.serializedValueSize())));
-        } else {
-          var payload = manager.payload();
-          if (payload.isEmpty()) return State.DONE;
-
-          long start = System.currentTimeMillis();
-          producer
-              .sender()
-              .topic(param.topic)
-              .partition(partitions.get(rand.nextInt(partitions.size())))
-              .key(manager.getKey())
-              .value(payload.get())
-              .timestamp(start)
-              .run()
-              .whenComplete(
-                  (m, e) ->
-                      observer.accept(System.currentTimeMillis() - start, m.serializedValueSize()));
+        switch (param.isolation()) {
+          case READ_COMMITTED:
+            return doTransaction();
+          case READ_UNCOMMITTED:
+            return doSend();
+          default:
+            return State.RUNNING;
         }
-        return State.RUNNING;
       }
 
       @Override
@@ -336,25 +352,12 @@ public class Performance {
         validateWith = StringMapField.class)
     Map<String, String> configs = Map.of();
 
-    public Map<String, Object> producerProps() {
-      var props = props();
-      props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, compression.name);
-      props.putAll(configs);
-      return props;
-    }
-
-    public Map<String, Object> consumerProps() {
-      var props = props();
-      if (transaction()) props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
-      return props;
-    }
-
     @Parameter(
         names = {"--compression"},
         description =
             "String: the compression algorithm used by producer. Available algorithm are none, gzip, snappy, lz4, and zstd",
         converter = CompressionField.class)
-    CompressionType compression = CompressionType.NONE;
+    Compression compression = Compression.NONE;
 
     @Parameter(
         names = {"--transaction.size"},
@@ -362,8 +365,8 @@ public class Performance {
         validateWith = PositiveLongField.class)
     int transactionSize = 1;
 
-    public boolean transaction() {
-      return transactionSize > 1;
+    Isolation isolation() {
+      return transactionSize > 1 ? Isolation.READ_COMMITTED : Isolation.READ_UNCOMMITTED;
     }
 
     @Parameter(
diff --git a/app/src/main/java/org/astraea/producer/Builder.java b/app/src/main/java/org/astraea/producer/Builder.java
index 9f157719..4d82ff6c 100644
--- a/app/src/main/java/org/astraea/producer/Builder.java
+++ b/app/src/main/java/org/astraea/producer/Builder.java
@@ -17,6 +17,7 @@ import org.apache.kafka.common.errors.AuthorizationException;
 import org.apache.kafka.common.errors.OutOfOrderSequenceException;
 import org.apache.kafka.common.errors.ProducerFencedException;
 import org.astraea.consumer.Header;
+import org.astraea.topic.Compression;
 
 public class Builder<Key, Value> {
   private final Map<String, Object> configs = new HashMap<>();
@@ -37,20 +38,27 @@ public class Builder<Key, Value> {
     return (Builder<Key, NewValue>) this;
   }
 
-  public Builder<Key, Value> configs(Map<String, Object> configs) {
+  public Builder<Key, Value> config(String key, String value) {
+    this.configs.put(key, value);
+    return this;
+  }
+
+  public Builder<Key, Value> configs(Map<String, String> configs) {
     this.configs.putAll(configs);
     return this;
   }
 
   public Builder<Key, Value> brokers(String brokers) {
-    this.configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, Objects.requireNonNull(brokers));
-    return this;
+    return config(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, Objects.requireNonNull(brokers));
   }
 
   public Builder<Key, Value> partitionClassName(String partitionClassName) {
-    this.configs.put(
+    return config(
         ProducerConfig.PARTITIONER_CLASS_CONFIG, Objects.requireNonNull(partitionClassName));
-    return this;
+  }
+
+  public Builder<Key, Value> compression(Compression compression) {
+    return config(ProducerConfig.COMPRESSION_TYPE_CONFIG, compression.nameOfKafka());
   }
 
   private static <Key, Value> CompletionStage<Metadata> doSend(
diff --git a/app/src/main/java/org/astraea/producer/Producer.java b/app/src/main/java/org/astraea/producer/Producer.java
index 2d49f8c5..11ddbbd0 100644
--- a/app/src/main/java/org/astraea/producer/Producer.java
+++ b/app/src/main/java/org/astraea/producer/Producer.java
@@ -21,7 +21,7 @@ public interface Producer<Key, Value> extends AutoCloseable {
     return builder().brokers(brokers).build();
   }
 
-  static Producer<byte[], byte[]> of(Map<String, Object> configs) {
+  static Producer<byte[], byte[]> of(Map<String, String> configs) {
     return builder().configs(configs).build();
   }
 }
diff --git a/app/src/main/java/org/astraea/topic/Builder.java b/app/src/main/java/org/astraea/topic/Builder.java
index e89c7e29..25946585 100644
--- a/app/src/main/java/org/astraea/topic/Builder.java
+++ b/app/src/main/java/org/astraea/topic/Builder.java
@@ -398,6 +398,12 @@ public class Builder {
       return this;
     }
 
+    @Override
+    public Creator config(String key, String value) {
+      this.configs.put(key, value);
+      return this;
+    }
+
     @Override
     public Creator configs(Map<String, String> configs) {
       this.configs.putAll(configs);
diff --git a/app/src/main/java/org/astraea/topic/Compression.java b/app/src/main/java/org/astraea/topic/Compression.java
new file mode 100644
index 00000000..30e19b3a
--- /dev/null
+++ b/app/src/main/java/org/astraea/topic/Compression.java
@@ -0,0 +1,20 @@
+package org.astraea.topic;
+
+import java.util.Locale;
+
+public enum Compression {
+  NONE,
+  GZIP,
+  SNAPPY,
+  LZ4,
+  ZSTD;
+
+  public static Compression of(String name) {
+    return Compression.valueOf(name.toUpperCase(Locale.ROOT));
+  }
+
+  /** @return the name parsed by kafka */
+  public String nameOfKafka() {
+    return name().toLowerCase(Locale.ROOT);
+  }
+}
diff --git a/app/src/main/java/org/astraea/topic/Creator.java b/app/src/main/java/org/astraea/topic/Creator.java
index 42d8fafe..1940c28f 100644
--- a/app/src/main/java/org/astraea/topic/Creator.java
+++ b/app/src/main/java/org/astraea/topic/Creator.java
@@ -17,7 +17,7 @@ public interface Creator {
    * @return this creator
    */
   default Creator compacted() {
-    return configs(Map.of(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT));
+    return config(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT);
   }
 
   /**
@@ -28,10 +28,15 @@ public interface Creator {
    */
   default Creator compactionMaxLag(Duration value) {
     return compacted()
-        .configs(
-            Map.of(TopicConfig.MAX_COMPACTION_LAG_MS_CONFIG, String.valueOf(value.toMillis())));
+        .config(TopicConfig.MAX_COMPACTION_LAG_MS_CONFIG, String.valueOf(value.toMillis()));
   }
 
+  default Creator compression(Compression compression) {
+    return config(TopicConfig.COMPRESSION_TYPE_CONFIG, compression.nameOfKafka());
+  }
+
+  Creator config(String key, String value);
+
   /**
    * @param configs used to create new topic
    * @return this creator
diff --git a/app/src/main/java/org/astraea/topic/TopicExplorer.java b/app/src/main/java/org/astraea/topic/TopicExplorer.java
index c3b47c56..ea1d35f0 100644
--- a/app/src/main/java/org/astraea/topic/TopicExplorer.java
+++ b/app/src/main/java/org/astraea/topic/TopicExplorer.java
@@ -233,101 +233,97 @@ public class TopicExplorer {
       nextLevel(
           NEXT_LEVEL,
           () -> {
+            if (consumerGroups.isEmpty()) {
+              treePrintln("no consumer group.");
+              return;
+            }
+
             // find all the consumer groups that are related to this topic
             var consumerGroupList =
                 consumerGroups.values().stream()
-                    .filter(
-                        consumerGroup ->
-                            consumerGroup.consumeProgress().keySet().stream()
-                                .anyMatch(tp -> tp.topic().equals(topic)))
                     .map(ConsumerGroup::groupId)
                     .distinct()
                     .map(consumerGroups::get)
                     .collect(Collectors.toUnmodifiableList());
 
-            if (consumerGroupList.isEmpty()) {
-              treePrintln("no consumer group.");
-            } else {
-              consumerGroupList.forEach(
-                  consumerGroup -> {
-
-                    // print out consumer group
-                    treePrintln("Consumer Group \"%s\"", consumerGroup.groupId());
-                    nextLevel(
-                        "  ",
-                        () -> {
-                          IntStream.range(0, partitionInfos.size())
-                              .mapToObj(x -> new ConsumeProgress(topic, x, consumerGroup, info))
-                              .forEach(x -> treePrintln("%s", x));
-
-                          // print out the active member of this consumer group
-                          treePrintln("Members:");
-                          nextLevel(
-                              "  ",
-                              () -> {
-                                var memberAssignment =
-                                    consumerGroup.activeMembers().stream()
-                                        .collect(
-                                            Collectors.toUnmodifiableMap(
-                                                x -> x,
-                                                x ->
-                                                    consumerGroup
-                                                        .assignment()
-                                                        .getOrDefault(x, Set.of())
-                                                        .stream()
-                                                        .filter(tp -> tp.topic().equals(topic))
-                                                        .map(TopicPartition::partition)
-                                                        .collect(Collectors.toUnmodifiableSet())));
-
-                                int totalAssignedPartition =
-                                    memberAssignment.values().stream().mapToInt(Set::size).sum();
-
-                                // how to tell if this consumer group is working on the specific
-                                // 1. there must be some active member(of course).
-                                // 2. at least one partition is assigned to some member. It is
-                                // possible that a consumer group used to work on a specific topic
-                                // in the past. But no longer work at it at this moment. In such a
-                                // situation, Kafka API will still tell you the consumer group
-                                // belongs to that topic(of course there are offsets stored here so
-                                // Kafka cannot just delete it). But you will realize even though
-                                // there are some live members here but none of them are assigned to
-                                // the topic. Because they didn't declare to work on that topic(used
-                                // to, but not now).
-                                if (memberAssignment.isEmpty() || totalAssignedPartition == 0)
-                                  treePrintln("no active member.");
-                                else {
-                                  memberAssignment.keySet().stream()
-                                      .sorted(Comparator.comparing(Member::memberId))
-                                      .forEach(
-                                          member -> {
-                                            treePrintln("member \"%s\"", member.memberId());
-                                            nextLevel(
-                                                NEXT_LEVEL,
-                                                () -> {
-                                                  if (memberAssignment.get(member).size() == 0)
-                                                    treePrintln("no partition assigned.");
-                                                  else
-                                                    treePrintln(
-                                                        "working on partition %s.",
-                                                        memberAssignment.get(member).stream()
-                                                            .sorted()
-                                                            .map(Object::toString)
-                                                            .collect(Collectors.joining(", ")));
-                                                  treePrintln(
-                                                      "clientId: \"%s\"", member.clientId());
-                                                  treePrintln("host: \"%s\"", member.host());
-                                                  if (member.groupInstanceId().isPresent())
-                                                    treePrintln(
-                                                        "groupInstanceId: \"%s\"",
-                                                        member.groupInstanceId().get());
-                                                  else treePrintln("groupInstanceId: none");
-                                                });
-                                          });
-                                }
-                              });
-                        });
-                  });
-            }
+            consumerGroupList.forEach(
+                consumerGroup -> {
+
+                  // print out consumer group
+                  treePrintln("Consumer Group \"%s\"", consumerGroup.groupId());
+                  nextLevel(
+                      "  ",
+                      () -> {
+                        IntStream.range(0, partitionInfos.size())
+                            .mapToObj(x -> new ConsumeProgress(topic, x, consumerGroup, info))
+                            .forEach(x -> treePrintln("%s", x));
+
+                        // print out the active member of this consumer group
+                        treePrintln("Members:");
+                        nextLevel(
+                            "  ",
+                            () -> {
+                              var memberAssignment =
+                                  consumerGroup.activeMembers().stream()
+                                      .collect(
+                                          Collectors.toUnmodifiableMap(
+                                              x -> x,
+                                              x ->
+                                                  consumerGroup
+                                                      .assignment()
+                                                      .getOrDefault(x, Set.of())
+                                                      .stream()
+                                                      .filter(tp -> tp.topic().equals(topic))
+                                                      .map(TopicPartition::partition)
+                                                      .collect(Collectors.toUnmodifiableSet())));
+
+                              int totalAssignedPartition =
+                                  memberAssignment.values().stream().mapToInt(Set::size).sum();
+
+                              // how to tell if this consumer group is working on the specific
+                              // 1. there must be some active member(of course).
+                              // 2. at least one partition is assigned to some member. It is
+                              // possible that a consumer group used to work on a specific topic
+                              // in the past. But no longer work at it at this moment. In such a
+                              // situation, Kafka API will still tell you the consumer group
+                              // belongs to that topic(of course there are offsets stored here so
+                              // Kafka cannot just delete it). But you will realize even though
+                              // there are some live members here but none of them are assigned to
+                              // the topic. Because they didn't declare to work on that topic(used
+                              // to, but not now).
+                              if (memberAssignment.isEmpty() || totalAssignedPartition == 0) {
+                                treePrintln("no active member.");
+                                return;
+                              }
+                              memberAssignment.keySet().stream()
+                                  .sorted(Comparator.comparing(Member::memberId))
+                                  .forEach(
+                                      member -> {
+                                        treePrintln("member \"%s\"", member.memberId());
+                                        nextLevel(
+                                            NEXT_LEVEL,
+                                            () -> {
+                                              if (memberAssignment.get(member).size() == 0)
+                                                treePrintln("no partition assigned.");
+                                              else
+                                                treePrintln(
+                                                    "working on partition %s.",
+                                                    memberAssignment.get(member).stream()
+                                                        .sorted()
+                                                        .map(Object::toString)
+                                                        .collect(Collectors.joining(", ")));
+                                              treePrintln("clientId: \"%s\"", member.clientId());
+                                              treePrintln("host: \"%s\"", member.host());
+                                              if (member.groupInstanceId().isPresent())
+                                                treePrintln(
+                                                    "groupInstanceId: \"%s\"",
+                                                    member.groupInstanceId().get());
+                                              else treePrintln("groupInstanceId: none");
+                                            });
+                                      });
+                            });
+                      });
+                });
           });
     }
 
@@ -389,33 +385,44 @@ public class TopicExplorer {
       }
 
       private long current() {
-        return group.consumeProgress().get(new TopicPartition(topic, index));
+        return group.consumeProgress().getOrDefault(new TopicPartition(topic, index), -1L);
+      }
+
+      private boolean hasCommittedOffset() {
+        return current() >= 0;
       }
 
       private String progressBar() {
-        int totalBlocks = 20;
-        int filledBlocks =
-            Math.min(
-                (int) (20.0 * ((double) (current() - earliest()) / (latest() - earliest()))), 20);
-        int emptyBlocks = totalBlocks - filledBlocks;
-        return Stream.concat(
-                Collections.nCopies(filledBlocks, "#").stream(),
-                Collections.nCopies(emptyBlocks, " ").stream())
-            .collect(Collectors.joining("", "[", "]"));
+
+        if (hasCommittedOffset()) {
+          int totalBlocks = 20;
+          int filledBlocks =
+              Math.min(
+                  (int) (20.0 * ((double) (current() - earliest()) / (latest() - earliest()))), 20);
+          int emptyBlocks = totalBlocks - filledBlocks;
+          return Stream.concat(
+                  Collections.nCopies(filledBlocks, "#").stream(),
+                  Collections.nCopies(emptyBlocks, " ").stream())
+              .collect(Collectors.joining("", "[", "]"));
+        }
+        return "-1";
       }
 
       @Override
       public String toString() {
-        var partitionDigits = Math.max((int) (Math.log10(map.get(topic).size() - 1)) + 1, 1);
-        return String.format(
-            "consume progress of partition #%"
-                + partitionDigits
-                + "d %s (earliest/current/latest offset %d/%d/%d)",
-            index,
-            progressBar(),
-            earliest(),
-            current(),
-            latest());
+        if (hasCommittedOffset()) {
+          var partitionDigits = Math.max((int) (Math.log10(map.get(topic).size() - 1)) + 1, 1);
+          return String.format(
+              "consume progress of partition #%"
+                  + partitionDigits
+                  + "d %s (earliest/current/latest offset %d/%d/%d)",
+              index,
+              progressBar(),
+              earliest(),
+              current(),
+              latest());
+        }
+        return "no committed offset for partition " + index;
       }
     }
 


Test patch:
diff --git a/app/src/test/java/org/astraea/argument/CompressionFieldTest.java b/app/src/test/java/org/astraea/argument/CompressionFieldTest.java
index be6c69e4..dc28a507 100644
--- a/app/src/test/java/org/astraea/argument/CompressionFieldTest.java
+++ b/app/src/test/java/org/astraea/argument/CompressionFieldTest.java
@@ -3,7 +3,7 @@ package org.astraea.argument;
 import com.beust.jcommander.Parameter;
 import com.beust.jcommander.ParameterException;
 import java.util.stream.Stream;
-import org.apache.kafka.common.record.CompressionType;
+import org.astraea.topic.Compression;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
@@ -14,14 +14,14 @@ public class CompressionFieldTest {
         names = {"--field"},
         converter = CompressionField.class,
         validateWith = CompressionField.class)
-    public CompressionType value;
+    public Compression value;
   }
 
   @Test
   void testConversion() {
     var arg = new CompressionField();
-    Stream.of(CompressionType.values())
-        .forEach(type -> Assertions.assertEquals(type, arg.convert(type.name)));
+    Stream.of(Compression.values())
+        .forEach(type -> Assertions.assertEquals(type, arg.convert(type.name())));
     Assertions.assertThrows(ParameterException.class, () -> arg.convert("aaa"));
   }
 
@@ -29,6 +29,6 @@ public class CompressionFieldTest {
   void testParse() {
     var arg =
         org.astraea.argument.Argument.parse(new FakeParameter(), new String[] {"--field", "gzip"});
-    Assertions.assertEquals(CompressionType.GZIP, arg.value);
+    Assertions.assertEquals(Compression.GZIP, arg.value);
   }
 }
diff --git a/app/src/test/java/org/astraea/cost/ClusterInfoTest.java b/app/src/test/java/org/astraea/cost/ClusterInfoTest.java
index de532ed5..32d4d2f8 100644
--- a/app/src/test/java/org/astraea/cost/ClusterInfoTest.java
+++ b/app/src/test/java/org/astraea/cost/ClusterInfoTest.java
@@ -14,7 +14,7 @@ public class ClusterInfoTest {
   @Test
   void testNode() {
     var node = NodeInfoTest.node();
-    var partition = PartitionInfoTest.partitionInfo();
+    var partition = ReplicaInfoTest.partitionInfo();
     var kafkaCluster = Mockito.mock(Cluster.class);
     Mockito.when(kafkaCluster.availablePartitionsForTopic(partition.topic()))
         .thenReturn(List.of(partition));
@@ -29,9 +29,12 @@ public class ClusterInfoTest {
     Assertions.assertEquals(1, clusterInfo.availablePartitions(partition.topic()).size());
     Assertions.assertEquals(1, clusterInfo.partitions(partition.topic()).size());
     Assertions.assertEquals(
-        NodeInfo.of(node), clusterInfo.availablePartitions(partition.topic()).get(0).leader());
+        NodeInfo.of(node), clusterInfo.availablePartitions(partition.topic()).get(0).nodeInfo());
     Assertions.assertEquals(
-        NodeInfo.of(node), clusterInfo.partitions(partition.topic()).get(0).leader());
+        NodeInfo.of(node),
+        clusterInfo.availablePartitionLeaders(partition.topic()).get(0).nodeInfo());
+    Assertions.assertEquals(
+        NodeInfo.of(node), clusterInfo.partitions(partition.topic()).get(0).nodeInfo());
   }
 
   @Test
diff --git a/app/src/test/java/org/astraea/cost/FakeClusterInfo.java b/app/src/test/java/org/astraea/cost/FakeClusterInfo.java
index e2707feb..fd0f1d75 100644
--- a/app/src/test/java/org/astraea/cost/FakeClusterInfo.java
+++ b/app/src/test/java/org/astraea/cost/FakeClusterInfo.java
@@ -13,7 +13,12 @@ public class FakeClusterInfo implements ClusterInfo {
   }
 
   @Override
-  public List<PartitionInfo> availablePartitions(String topic) {
+  public List<ReplicaInfo> availablePartitionLeaders(String topic) {
+    return List.of();
+  }
+
+  @Override
+  public List<ReplicaInfo> availablePartitions(String topic) {
     return List.of();
   }
 
@@ -23,7 +28,7 @@ public class FakeClusterInfo implements ClusterInfo {
   }
 
   @Override
-  public List<PartitionInfo> partitions(String topic) {
+  public List<ReplicaInfo> partitions(String topic) {
     return List.of();
   }
 
diff --git a/app/src/test/java/org/astraea/cost/MemoryWarningCostTest.java b/app/src/test/java/org/astraea/cost/MemoryWarningCostTest.java
index c38dca8b..78cf5c16 100644
--- a/app/src/test/java/org/astraea/cost/MemoryWarningCostTest.java
+++ b/app/src/test/java/org/astraea/cost/MemoryWarningCostTest.java
@@ -34,14 +34,11 @@ public class MemoryWarningCostTest {
           }
 
           @Override
-          public List<PartitionInfo> availablePartitions(String topic) {
+          public List<ReplicaInfo> availablePartitions(String topic) {
             return List.of(
-                PartitionInfo.of(
-                    "t", 0, NodeInfo.of(1, "host1", 9092), List.of(), List.of(), List.of()),
-                PartitionInfo.of(
-                    "t", 0, NodeInfo.of(2, "host2", 9092), List.of(), List.of(), List.of()),
-                PartitionInfo.of(
-                    "t", 0, NodeInfo.of(3, "host3", 9092), List.of(), List.of(), List.of()));
+                ReplicaInfo.of("t", 0, NodeInfo.of(1, "host1", 9092), true, true, false),
+                ReplicaInfo.of("t", 0, NodeInfo.of(2, "host2", 9092), false, true, false),
+                ReplicaInfo.of("t", 0, NodeInfo.of(3, "host3", 9092), false, true, false));
           }
         };
 
diff --git a/app/src/test/java/org/astraea/cost/PartitionInfoTest.java b/app/src/test/java/org/astraea/cost/PartitionInfoTest.java
deleted file mode 100644
index 3a3aac74..00000000
--- a/app/src/test/java/org/astraea/cost/PartitionInfoTest.java
+++ /dev/null
@@ -1,24 +0,0 @@
-package org.astraea.cost;
-
-import org.apache.kafka.common.Node;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Test;
-
-public class PartitionInfoTest {
-
-  static org.apache.kafka.common.PartitionInfo partitionInfo() {
-    return new org.apache.kafka.common.PartitionInfo(
-        "ttt", 100, NodeInfoTest.node(), new Node[0], new Node[0]);
-  }
-
-  @Test
-  void testAllGetters() {
-    var kafkaPartition = partitionInfo();
-
-    var partitionInfo = PartitionInfo.of(kafkaPartition);
-
-    Assertions.assertEquals(kafkaPartition.topic(), partitionInfo.topic());
-    Assertions.assertEquals(kafkaPartition.partition(), partitionInfo.partition());
-    Assertions.assertEquals(NodeInfo.of(kafkaPartition.leader()), partitionInfo.leader());
-  }
-}
diff --git a/app/src/test/java/org/astraea/cost/ReplicaInfoTest.java b/app/src/test/java/org/astraea/cost/ReplicaInfoTest.java
new file mode 100644
index 00000000..eb92b69c
--- /dev/null
+++ b/app/src/test/java/org/astraea/cost/ReplicaInfoTest.java
@@ -0,0 +1,60 @@
+package org.astraea.cost;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+import org.apache.kafka.common.Node;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class ReplicaInfoTest {
+
+  static org.apache.kafka.common.PartitionInfo partitionInfo() {
+    return new org.apache.kafka.common.PartitionInfo(
+        "ttt",
+        100,
+        NodeInfoTest.node(),
+        new Node[] {NodeInfoTest.node()},
+        new Node[] {NodeInfoTest.node()});
+  }
+
+  @Test
+  void testAllGetters() {
+    var kafkaPartition = partitionInfo();
+    var replicaInfo = ReplicaInfo.of(kafkaPartition);
+    var followerReplicaNode =
+        (Supplier<Set<NodeInfo>>)
+            () ->
+                Arrays.stream(kafkaPartition.replicas())
+                    .map(NodeInfo::of)
+                    .filter(node -> !node.equals(NodeInfo.of(kafkaPartition.leader())))
+                    .collect(Collectors.toUnmodifiableSet());
+    final List<ReplicaInfo> leader =
+        replicaInfo.stream().filter(ReplicaInfo::isLeader).collect(Collectors.toList());
+    var followers =
+        replicaInfo.stream().filter(ReplicaInfo::isFollower).collect(Collectors.toList());
+
+    // test topic/partition
+    replicaInfo.forEach(
+        replica -> Assertions.assertEquals(kafkaPartition.topic(), replica.topic()));
+    replicaInfo.forEach(
+        replica -> Assertions.assertEquals(kafkaPartition.partition(), replica.partition()));
+
+    // test leader
+    Assertions.assertEquals(1, leader.size());
+    Assertions.assertEquals(
+        NodeInfo.of(kafkaPartition.leader()), leader.iterator().next().nodeInfo());
+
+    // test followers
+    Assertions.assertEquals(
+        kafkaPartition.replicas().length - 1,
+        followers.size(),
+        "The follower replica node count should match");
+    Assertions.assertEquals(
+        followerReplicaNode.get(),
+        followers.stream().map(ReplicaInfo::nodeInfo).collect(Collectors.toUnmodifiableSet()),
+        "The follower replica node set should match");
+  }
+}
diff --git a/app/src/test/java/org/astraea/cost/ThroughputCostTest.java b/app/src/test/java/org/astraea/cost/ThroughputCostTest.java
index e7982cf6..8b5ae285 100644
--- a/app/src/test/java/org/astraea/cost/ThroughputCostTest.java
+++ b/app/src/test/java/org/astraea/cost/ThroughputCostTest.java
@@ -36,7 +36,7 @@ public class ThroughputCostTest {
     Mockito.when(cluster.allBeans()).thenReturn(Map.of());
     Mockito.when(cluster.topics()).thenReturn(Set.of("t"));
     Mockito.when(cluster.availablePartitions("t"))
-        .thenReturn(List.of(PartitionInfo.of("t", 0, node, List.of(), List.of(), List.of())));
+        .thenReturn(List.of(ReplicaInfo.of("t", 0, node, true, true, false)));
 
     var cost =
         throughputCost.brokerCost(ClusterInfo.of(cluster, Map.of(10, List.of(bean)))).value();
diff --git a/app/src/test/java/org/astraea/cost/brokersMetrics/CpuCostTest.java b/app/src/test/java/org/astraea/cost/brokersMetrics/CpuCostTest.java
index 67884c5a..8c6dfe92 100644
--- a/app/src/test/java/org/astraea/cost/brokersMetrics/CpuCostTest.java
+++ b/app/src/test/java/org/astraea/cost/brokersMetrics/CpuCostTest.java
@@ -7,7 +7,7 @@ import java.util.Set;
 import org.astraea.cost.ClusterInfo;
 import org.astraea.cost.FakeClusterInfo;
 import org.astraea.cost.NodeInfo;
-import org.astraea.cost.PartitionInfo;
+import org.astraea.cost.ReplicaInfo;
 import org.astraea.metrics.HasBeanObject;
 import org.astraea.metrics.java.OperatingSystemInfo;
 import org.junit.jupiter.api.Assertions;
@@ -37,14 +37,11 @@ public class CpuCostTest {
           }
 
           @Override
-          public List<PartitionInfo> availablePartitions(String topic) {
+          public List<ReplicaInfo> availablePartitions(String topic) {
             return List.of(
-                PartitionInfo.of(
-                    "t", 0, NodeInfo.of(1, "host1", 9092), List.of(), List.of(), List.of()),
-                PartitionInfo.of(
-                    "t", 0, NodeInfo.of(2, "host2", 9092), List.of(), List.of(), List.of()),
-                PartitionInfo.of(
-                    "t", 0, NodeInfo.of(3, "host3", 9092), List.of(), List.of(), List.of()));
+                ReplicaInfo.of("t", 0, NodeInfo.of(1, "host1", 9092), true, true, false),
+                ReplicaInfo.of("t", 0, NodeInfo.of(2, "host2", 9092), false, true, false),
+                ReplicaInfo.of("t", 0, NodeInfo.of(3, "host3", 9092), false, true, false));
           }
         };
 
@@ -75,14 +72,11 @@ public class CpuCostTest {
           }
 
           @Override
-          public List<PartitionInfo> availablePartitions(String topic) {
+          public List<ReplicaInfo> availablePartitions(String topic) {
             return List.of(
-                PartitionInfo.of(
-                    "t", 0, NodeInfo.of(1, "host1", 9092), List.of(), List.of(), List.of()),
-                PartitionInfo.of(
-                    "t", 0, NodeInfo.of(2, "host2", 9092), List.of(), List.of(), List.of()),
-                PartitionInfo.of(
-                    "t", 0, NodeInfo.of(3, "host3", 9092), List.of(), List.of(), List.of()));
+                ReplicaInfo.of("t", 0, NodeInfo.of(1, "host1", 9092), true, true, false),
+                ReplicaInfo.of("t", 0, NodeInfo.of(2, "host2", 9092), false, true, false),
+                ReplicaInfo.of("t", 0, NodeInfo.of(3, "host3", 9092), false, true, false));
           }
         };
     scores = cpuCost.brokerCost(clusterInfo2).value();
diff --git a/app/src/test/java/org/astraea/cost/brokersMetrics/IntegratedScoreTest.java b/app/src/test/java/org/astraea/cost/brokersMetrics/IntegratedScoreTest.java
new file mode 100644
index 00000000..cdc82a34
--- /dev/null
+++ b/app/src/test/java/org/astraea/cost/brokersMetrics/IntegratedScoreTest.java
@@ -0,0 +1,41 @@
+package org.astraea.cost.brokersMetrics;
+
+import java.util.Map;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class IntegratedScoreTest {
+  @Test
+  void testEntropyEmpowerment() {
+    var neutralIntegratedCost = new NeutralIntegratedCost();
+    var brokers =
+        Map.of(
+            0,
+            new NeutralIntegratedCost.BrokerMetrics(),
+            1,
+            new NeutralIntegratedCost.BrokerMetrics(),
+            2,
+            new NeutralIntegratedCost.BrokerMetrics());
+    brokers.get(0).cpuScore = 0.1252;
+    brokers.get(0).inputScore = 550.0;
+    brokers.get(0).outputScore = 545.0;
+    brokers.get(0).memoryScore = 0.254;
+
+    brokers.get(1).inputScore = 750.0;
+    brokers.get(1).outputScore = 755.0;
+    brokers.get(1).memoryScore = 0.268;
+    brokers.get(1).cpuScore = 0.1342;
+
+    brokers.get(2).inputScore = 550.0;
+    brokers.get(2).outputScore = 550.0;
+    brokers.get(2).memoryScore = 0.235;
+    brokers.get(2).cpuScore = 0.1522;
+
+    var metrics = neutralIntegratedCost.entropyEmpowerment().empowerment(brokers);
+
+    Assertions.assertEquals(0.46689278143812796, metrics.get("inputThroughput"));
+    Assertions.assertEquals(0.4200836938888968, metrics.get("outputThroughput"));
+    Assertions.assertEquals(0.05781512136314425, metrics.get("memory"));
+    Assertions.assertEquals(0.055208403309831, metrics.get("cpu"));
+  }
+}
diff --git a/app/src/test/java/org/astraea/cost/brokersMetrics/MemoryCostTest.java b/app/src/test/java/org/astraea/cost/brokersMetrics/MemoryCostTest.java
index 393ff05e..538b0c80 100644
--- a/app/src/test/java/org/astraea/cost/brokersMetrics/MemoryCostTest.java
+++ b/app/src/test/java/org/astraea/cost/brokersMetrics/MemoryCostTest.java
@@ -8,7 +8,7 @@ import java.util.Set;
 import org.astraea.cost.ClusterInfo;
 import org.astraea.cost.FakeClusterInfo;
 import org.astraea.cost.NodeInfo;
-import org.astraea.cost.PartitionInfo;
+import org.astraea.cost.ReplicaInfo;
 import org.astraea.metrics.HasBeanObject;
 import org.astraea.metrics.java.HasJvmMemory;
 import org.junit.jupiter.api.Assertions;
@@ -38,14 +38,11 @@ public class MemoryCostTest {
           }
 
           @Override
-          public List<PartitionInfo> availablePartitions(String topic) {
+          public List<ReplicaInfo> availablePartitions(String topic) {
             return List.of(
-                PartitionInfo.of(
-                    "t", 0, NodeInfo.of(1, "host1", 9092), List.of(), List.of(), List.of()),
-                PartitionInfo.of(
-                    "t", 0, NodeInfo.of(2, "host2", 9092), List.of(), List.of(), List.of()),
-                PartitionInfo.of(
-                    "t", 0, NodeInfo.of(3, "host3", 9092), List.of(), List.of(), List.of()));
+                ReplicaInfo.of("t", 0, NodeInfo.of(1, "host1", 9092), true, false, false),
+                ReplicaInfo.of("t", 0, NodeInfo.of(2, "host2", 9092), false, true, false),
+                ReplicaInfo.of("t", 0, NodeInfo.of(3, "host3", 9092), false, true, false));
           }
         };
 
@@ -76,14 +73,11 @@ public class MemoryCostTest {
           }
 
           @Override
-          public List<PartitionInfo> availablePartitions(String topic) {
+          public List<ReplicaInfo> availablePartitions(String topic) {
             return List.of(
-                PartitionInfo.of(
-                    "t", 0, NodeInfo.of(1, "host1", 9092), List.of(), List.of(), List.of()),
-                PartitionInfo.of(
-                    "t", 0, NodeInfo.of(2, "host2", 9092), List.of(), List.of(), List.of()),
-                PartitionInfo.of(
-                    "t", 0, NodeInfo.of(3, "host3", 9092), List.of(), List.of(), List.of()));
+                ReplicaInfo.of("t", 0, NodeInfo.of(1, "host1", 9092), true, true, false),
+                ReplicaInfo.of("t", 0, NodeInfo.of(2, "host2", 9092), false, true, false),
+                ReplicaInfo.of("t", 0, NodeInfo.of(3, "host3", 9092), false, true, false));
           }
         };
     scores = memoryCost.brokerCost(clusterInfo2).value();
diff --git a/app/src/test/java/org/astraea/partitioner/StrictCostDispatcherTest.java b/app/src/test/java/org/astraea/partitioner/StrictCostDispatcherTest.java
index 405daeac..46091e45 100644
--- a/app/src/test/java/org/astraea/partitioner/StrictCostDispatcherTest.java
+++ b/app/src/test/java/org/astraea/partitioner/StrictCostDispatcherTest.java
@@ -10,7 +10,7 @@ import org.astraea.cost.BrokerCost;
 import org.astraea.cost.ClusterInfo;
 import org.astraea.cost.HasBrokerCost;
 import org.astraea.cost.NodeInfo;
-import org.astraea.cost.PartitionInfo;
+import org.astraea.cost.ReplicaInfo;
 import org.astraea.metrics.collector.Fetcher;
 import org.astraea.metrics.collector.Receiver;
 import org.junit.jupiter.api.Assertions;
@@ -19,39 +19,9 @@ import org.mockito.Mockito;
 
 public class StrictCostDispatcherTest {
 
-  private static PartitionInfo createFakePartitionInfo(
-      String topic, int partition, NodeInfo leaderNode) {
-    return new PartitionInfo() {
-      @Override
-      public String topic() {
-        return topic;
-      }
-
-      @Override
-      public int partition() {
-        return partition;
-      }
-
-      @Override
-      public NodeInfo leader() {
-        return leaderNode;
-      }
-
-      @Override
-      public List<NodeInfo> replicas() {
-        return null;
-      }
-
-      @Override
-      public List<NodeInfo> inSyncReplica() {
-        return null;
-      }
-
-      @Override
-      public List<NodeInfo> offlineReplicas() {
-        return null;
-      }
-    };
+  private static ReplicaInfo createFakeReplicaLeaderInfo(
+      String topic, int partition, NodeInfo replicaNode) {
+    return ReplicaInfo.of(topic, partition, replicaNode, true, true, false);
   }
 
   @Test
@@ -78,8 +48,8 @@ public class StrictCostDispatcherTest {
   void testBestPartition() {
     var partitions =
         List.of(
-            createFakePartitionInfo("t", 0, NodeInfo.of(10, "h0", 1000)),
-            createFakePartitionInfo("t", 1, NodeInfo.of(11, "h1", 1000)));
+            createFakeReplicaLeaderInfo("t", 0, NodeInfo.of(10, "h0", 1000)),
+            createFakeReplicaLeaderInfo("t", 1, NodeInfo.of(11, "h1", 1000)));
 
     var score = List.of(Map.of(10, 0.8D), Map.of(11, 0.7D));
 
@@ -93,10 +63,10 @@ public class StrictCostDispatcherTest {
   void testPartition() {
     // n0 is busy
     var n0 = NodeInfo.of(10, "host0", 12345);
-    var p0 = createFakePartitionInfo("aa", 1, n0);
+    var p0 = createFakeReplicaLeaderInfo("aa", 1, n0);
     // n1 is free
     var n1 = NodeInfo.of(11, "host1", 12345);
-    var p1 = createFakePartitionInfo("aa", 2, n1);
+    var p1 = createFakeReplicaLeaderInfo("aa", 2, n1);
 
     var receiver = Mockito.mock(Receiver.class);
     Mockito.when(receiver.current()).thenReturn(List.of());
@@ -135,16 +105,16 @@ public class StrictCostDispatcherTest {
     var clusterInfo = Mockito.mock(ClusterInfo.class);
 
     // there is no available partition
-    Mockito.when(clusterInfo.availablePartitions("aa")).thenReturn(List.of());
+    Mockito.when(clusterInfo.availablePartitionLeaders("aa")).thenReturn(List.of());
     Mockito.when(clusterInfo.topics()).thenReturn(Set.of("aa"));
     Assertions.assertEquals(0, dispatcher.partition("aa", new byte[0], new byte[0], clusterInfo));
 
     // there is only one available partition
-    Mockito.when(clusterInfo.availablePartitions("aa")).thenReturn(List.of(p0));
+    Mockito.when(clusterInfo.availablePartitionLeaders("aa")).thenReturn(List.of(p0));
     Assertions.assertEquals(1, dispatcher.partition("aa", new byte[0], new byte[0], clusterInfo));
 
     // there is no beans, so it just returns the partition.
-    Mockito.when(clusterInfo.availablePartitions("aa")).thenReturn(List.of(p0, p1));
+    Mockito.when(clusterInfo.availablePartitionLeaders("aa")).thenReturn(List.of(p0, p1));
     Assertions.assertEquals(2, dispatcher.partition("aa", new byte[0], new byte[0], clusterInfo));
   }
 }
diff --git a/app/src/test/java/org/astraea/partitioner/smoothPartitioner/PartitionerTest.java b/app/src/test/java/org/astraea/partitioner/smoothPartitioner/PartitionerTest.java
index c655bfa4..5c50861d 100644
--- a/app/src/test/java/org/astraea/partitioner/smoothPartitioner/PartitionerTest.java
+++ b/app/src/test/java/org/astraea/partitioner/smoothPartitioner/PartitionerTest.java
@@ -100,7 +100,8 @@ public class PartitionerTest extends RequireBrokerCluster {
             .keySerializer(Serializer.STRING)
             .configs(
                 initProConfig().entrySet().stream()
-                    .collect(Collectors.toMap(e -> e.getKey().toString(), Map.Entry::getValue)))
+                    .collect(
+                        Collectors.toMap(e -> e.getKey().toString(), e -> e.getValue().toString())))
             .build()) {
       var i = 0;
       while (i < 300) {
@@ -166,7 +167,7 @@ public class PartitionerTest extends RequireBrokerCluster {
                                             .collect(
                                                 Collectors.toMap(
                                                     e -> e.getKey().toString(),
-                                                    Map.Entry::getValue)))
+                                                    e -> e.getValue().toString())))
                                     .build(),
                                 topicName,
                                 key,
@@ -265,7 +266,8 @@ public class PartitionerTest extends RequireBrokerCluster {
             .keySerializer(Serializer.STRING)
             .configs(
                 props.entrySet().stream()
-                    .collect(Collectors.toMap(e -> e.getKey().toString(), Map.Entry::getValue)))
+                    .collect(
+                        Collectors.toMap(e -> e.getKey().toString(), e -> e.getValue().toString())))
             .build()) {
       var metadata =
           producer
diff --git a/app/src/test/java/org/astraea/performance/PerformanceTest.java b/app/src/test/java/org/astraea/performance/PerformanceTest.java
index e8e2c7a7..2aee96c2 100644
--- a/app/src/test/java/org/astraea/performance/PerformanceTest.java
+++ b/app/src/test/java/org/astraea/performance/PerformanceTest.java
@@ -12,6 +12,7 @@ import org.astraea.Utils;
 import org.astraea.concurrent.Executor;
 import org.astraea.concurrent.ThreadPool;
 import org.astraea.consumer.Consumer;
+import org.astraea.consumer.Isolation;
 import org.astraea.producer.Producer;
 import org.astraea.service.RequireBrokerCluster;
 import org.astraea.topic.TopicAdmin;
@@ -160,9 +161,9 @@ public class PerformanceTest extends RequireBrokerCluster {
   @Test
   void testTransactionSet() {
     var argument = new Performance.Argument();
-    Assertions.assertFalse(argument.transaction());
+    Assertions.assertEquals(Isolation.READ_UNCOMMITTED, argument.isolation());
     argument.transactionSize = 3;
-    Assertions.assertTrue(argument.transaction());
+    Assertions.assertEquals(Isolation.READ_COMMITTED, argument.isolation());
   }
 
   @Test
@@ -197,7 +198,7 @@ public class PerformanceTest extends RequireBrokerCluster {
     };
 
     var arg = org.astraea.argument.Argument.parse(new Performance.Argument(), arguments1);
-    Assertions.assertEquals("value", arg.producerProps().get("key").toString());
+    Assertions.assertEquals("value", arg.configs.get("key"));
 
     String[] arguments2 = {"--bootstrap.servers", "localhost:9092", "--topic", ""};
     Assertions.assertThrows(
diff --git a/app/src/test/java/org/astraea/producer/ProducerTest.java b/app/src/test/java/org/astraea/producer/ProducerTest.java
index f5cd9bf3..0a422a0e 100644
--- a/app/src/test/java/org/astraea/producer/ProducerTest.java
+++ b/app/src/test/java/org/astraea/producer/ProducerTest.java
@@ -3,13 +3,12 @@ package org.astraea.producer;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
-import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.astraea.consumer.Consumer;
 import org.astraea.consumer.Deserializer;
 import org.astraea.consumer.Header;
+import org.astraea.consumer.Isolation;
 import org.astraea.service.RequireBrokerCluster;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
@@ -81,14 +80,13 @@ public class ProducerTest extends RequireBrokerCluster {
       producer.transaction(senders);
     }
 
-    Map<String, Object> prop = Map.of(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
     try (var consumer =
         Consumer.builder()
             .brokers(bootstrapServers())
             .fromBeginning()
             .topics(Set.of(topicName))
             .keyDeserializer(Deserializer.STRING)
-            .configs(prop)
+            .isolation(Isolation.READ_COMMITTED)
             .build()) {
       var records = consumer.poll(Duration.ofSeconds(10));
       Assertions.assertEquals(3, records.size());
diff --git a/app/src/test/java/org/astraea/topic/CompressionTest.java b/app/src/test/java/org/astraea/topic/CompressionTest.java
new file mode 100644
index 00000000..5c9c954e
--- /dev/null
+++ b/app/src/test/java/org/astraea/topic/CompressionTest.java
@@ -0,0 +1,19 @@
+package org.astraea.topic;
+
+import java.util.Arrays;
+import org.apache.kafka.common.record.CompressionType;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class CompressionTest {
+
+  @Test
+  void testNameOfKafka() {
+    Assertions.assertEquals(Compression.values().length, CompressionType.values().length);
+    Arrays.stream(Compression.values())
+        .forEach(
+            c ->
+                Assertions.assertEquals(
+                    c.nameOfKafka(), CompressionType.forName(c.nameOfKafka()).name));
+  }
+}