Repository: skiptests/astraea
Pull request #173 (2022-01-02 15:22:59)

Automation Performance
現在可以通過配置`config/automation.properties`文件，然後執行./gradlew run --args="automation"，達到可以去放心吃飯讓Performance自己跑的並行處理。

以下為執行效果：
![2022-01-02 23-08-18 的螢幕擷圖](https://user-images.githubusercontent.com/43372856/147880400-b6fca7e1-bf50-4c92-96c5-029e2239d172.png)




Patch:
diff --git a/.gitignore b/.gitignore
index 6bc550bc..3835bffb 100644
--- a/.gitignore
+++ b/.gitignore
@@ -15,7 +15,6 @@ zookeeper.dockerfile
 broker.dockerfile
 kafka_tool.dockerfile
 spark.dockerfile
-astraea.dockerfile
 
 # we don't put binary file to git repo
 gradle-wrapper.jar
\ No newline at end of file
diff --git a/README.md b/README.md
index 27dfe2f3..286e0cb0 100644
--- a/README.md
+++ b/README.md
@@ -221,12 +221,13 @@ Run the benchmark from source
 10. --prop.file: the path to property file.
 11. --partitioner: the partitioner to use in producers
 12. --jmx.servers: the jmx server addresses of the brokers 
+13. --key.distribution: Name of the distribution. Available distribution names: "uniform", "zipfian", "latest". Default: (No key)
 
 ---
 
 ## Topic Explorer
 
-This tool can expose both earliest offset and latest offset for all (public and private) topics.
+This tool can expose both earliest offset, consumer group offset and latest offset for all (public and private) topics.
 
 Run the tool from source code
 ```shell
diff --git a/app/src/main/java/org/astraea/App.java b/app/src/main/java/org/astraea/App.java
index a049a25a..23312892 100644
--- a/app/src/main/java/org/astraea/App.java
+++ b/app/src/main/java/org/astraea/App.java
@@ -5,6 +5,7 @@ import java.lang.reflect.InvocationTargetException;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
+import org.astraea.automation.Automation;
 import org.astraea.metrics.MetricExplorer;
 import org.astraea.performance.Performance;
 import org.astraea.topic.ReplicaCollie;
@@ -20,7 +21,8 @@ public class App {
           "replica", ReplicaCollie.class,
           "score", PartitionScore.class,
           "performance", Performance.class,
-          "monitor", ReplicaSyncingMonitor.class);
+          "monitor", ReplicaSyncingMonitor.class,
+          "automation", Automation.class);
 
   static void execute(Map<String, Class<?>> mains, List<String> args) throws Throwable {
 
diff --git a/app/src/main/java/org/astraea/automation/Automation.java b/app/src/main/java/org/astraea/automation/Automation.java
new file mode 100644
index 00000000..50fcd198
--- /dev/null
+++ b/app/src/main/java/org/astraea/automation/Automation.java
@@ -0,0 +1,97 @@
+package org.astraea.automation;
+
+import com.beust.jcommander.Parameter;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.ExecutionException;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.KafkaAdminClient;
+import org.astraea.argument.ArgumentUtil;
+import org.astraea.performance.Performance;
+
+/**
+ * By configuring the parameters in config/automation.properties, control the execution times of
+ * performance and its configuration parameters.
+ *
+ * <ol>
+ *   <li>--file: The address of the automation.properties in config folder.
+ * </ol>
+ */
+public class Automation {
+  private static final List<String> performanceProperties =
+      List.of(
+          "--bootstrap.servers",
+          "--record.size",
+          "--run.until",
+          "--compression",
+          "--consumers",
+          "--fixed.size",
+          "--jmx.servers",
+          "--partitioner",
+          "--partitions",
+          "--producers",
+          "--prop.file",
+          "--replicas",
+          "--topic");
+
+  public static void main(String[] args) {
+    try {
+      var properties = new Properties();
+      var arg = ArgumentUtil.parseArgument(new automationArgument(), args);
+      properties.load(new FileInputStream(arg.address));
+      var whetherDeleteTopic = properties.getProperty("--whetherDeleteTopic").equals("true");
+      var bootstrap = properties.getProperty("--bootstrap.servers");
+      var config = new Properties();
+      config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap);
+
+      var i = 0;
+      var times = 0;
+      if (properties.getProperty("--time").equals("Default")) times = 5;
+      else times = Integer.parseInt(properties.getProperty("--time"));
+
+      while (i < times) {
+        var str =
+            Performance.execute(
+                ArgumentUtil.parseArgument(
+                    new Performance.Argument(), performanceArgs(properties)));
+        i++;
+        if (whetherDeleteTopic) {
+          try (final AdminClient adminClient = KafkaAdminClient.create(config)) {
+            var topicName = str.get();
+            adminClient.deleteTopics(List.of(topicName));
+          }
+        }
+        System.out.println("=============== " + i + " time Performance Complete! ===============");
+      }
+    } catch (IOException | InterruptedException | ExecutionException e) {
+      e.printStackTrace();
+    }
+  }
+
+  @SuppressWarnings("unChecked")
+  private static String[] performanceArgs(Properties properties) {
+    var args = new ArrayList<String>();
+    performanceProperties.forEach(
+        str -> {
+          var property = properties.getProperty(str);
+          if (property != null && !property.equals("Default")) {
+            args.add(str);
+            args.add(property);
+          }
+        });
+    var strings = new String[args.size()];
+    return args.toArray(strings);
+  }
+
+  private static class automationArgument {
+    @Parameter(
+        names = {"--file"},
+        description = "String: automation.properties address",
+        validateWith = ArgumentUtil.NotEmptyString.class)
+    String address = "";
+  }
+}
diff --git a/app/src/main/java/org/astraea/performance/Distribution.java b/app/src/main/java/org/astraea/performance/Distribution.java
new file mode 100644
index 00000000..8148aad6
--- /dev/null
+++ b/app/src/main/java/org/astraea/performance/Distribution.java
@@ -0,0 +1,74 @@
+package org.astraea.performance;
+
+import com.beust.jcommander.IStringConverter;
+import com.beust.jcommander.ParameterException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import java.util.stream.IntStream;
+
+/** Randomly generate a long number with respect to some distribution */
+public interface Distribution {
+
+  long get();
+
+  class DistributionConverter implements IStringConverter<Distribution> {
+    @Override
+    public Distribution convert(String rawArgument) {
+      var args = rawArgument.split(":");
+      var name = rawArgument.split(":")[0];
+
+      switch (name) {
+        case "uniform":
+          if (args.length > 1) return uniform(Integer.parseInt(args[1]));
+          return uniform();
+        case "latest":
+          return latest();
+        case "zipfian":
+          if (args.length > 1) return zipfian(Integer.parseInt(args[1]));
+          return zipfian(30);
+        default:
+          throw new ParameterException(
+              "Unknown distribution \"" + name + "\". use \"uniform\", \"latest\", \"zipfian\".");
+      }
+    }
+  }
+
+  /** A distribution for providing a random long number from range [0, 2147483647) */
+  static Distribution uniform() {
+    return uniform(Integer.MAX_VALUE);
+  }
+
+  /** A distribution for providing a random long number from range [0, N) */
+  static Distribution uniform(int N) {
+    var rand = new Random();
+    return () -> (long) (rand.nextInt(N));
+  }
+
+  /** A distribution for providing different value every 2 seconds */
+  static Distribution latest() {
+    return () -> System.currentTimeMillis() / 2000L;
+  }
+
+  /**
+   * Building a zipfian distribution with PDF: 1/k/H_N, where H_N is the Nth harmonic number (= 1/1
+   * + 1/2 + ... + 1/N); k is the key id
+   */
+  static Distribution zipfian(int N) {
+    var rand = new Random();
+    final List<Double> cumulativeDensityTable = new ArrayList<>();
+    var H_N = IntStream.range(1, N + 1).mapToDouble(k -> 1D / k).sum();
+    cumulativeDensityTable.add(1D / H_N);
+    IntStream.range(1, N)
+        .forEach(
+            i ->
+                cumulativeDensityTable.add(cumulativeDensityTable.get(i - 1) + 1D / (i + 1) / H_N));
+    return () -> {
+      final double randNum = rand.nextDouble();
+      for (int i = 0; i < cumulativeDensityTable.size(); ++i) {
+        if (randNum < cumulativeDensityTable.get(i)) return (long) i;
+      }
+      return (long) cumulativeDensityTable.size() - 1L;
+    };
+  }
+}
diff --git a/app/src/main/java/org/astraea/performance/FileWriter.java b/app/src/main/java/org/astraea/performance/FileWriter.java
new file mode 100644
index 00000000..f59c00a8
--- /dev/null
+++ b/app/src/main/java/org/astraea/performance/FileWriter.java
@@ -0,0 +1,117 @@
+package org.astraea.performance;
+
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.stream.IntStream;
+import org.astraea.concurrent.ThreadPool;
+
+public class FileWriter implements ThreadPool.Executor {
+  private final Manager manager;
+  private final Tracker tracker;
+  private BufferedWriter writer;
+  private String CSVName;
+
+  public FileWriter(Manager manager, Tracker tracker) {
+    this.manager = manager;
+    this.tracker = tracker;
+    initFileWriter();
+  }
+
+  @Override
+  public ThreadPool.Executor.State execute() throws InterruptedException {
+    if (logToCSV()) return State.DONE;
+    Thread.sleep(1000);
+    return State.RUNNING;
+  }
+
+  public String CSVName() {
+    return CSVName;
+  }
+
+  private void initFileWriter() {
+    CSVName = "Performance" + new SimpleDateFormat("yyyyMMddHHmmss").format(new Date()) + ".csv";
+    try {
+      writer = new BufferedWriter(new java.io.FileWriter(CSVName));
+      writer.write(
+          "Time \\ Name, Consumed/Produced, Output throughput (MiB/sec), Input throughput (MiB/sec), "
+              + "Publish max latency (ms), Publish min latency (ms), "
+              + "End-to-end max latency (ms), End-to-end min latency (ms)");
+      IntStream.range(0, tracker.producerResult().bytes.size())
+          .forEach(
+              i -> {
+                try {
+                  writer.write(
+                      ",Producer["
+                          + i
+                          + "] average throughput (MB/sec), Producer["
+                          + i
+                          + "] average publish latency (ms)");
+                } catch (IOException ignore) {
+                }
+              });
+      IntStream.range(0, tracker.consumerResult().bytes.size())
+          .forEach(
+              i -> {
+                try {
+                  writer.write(
+                      ",Consumer["
+                          + i
+                          + "] average throughput (MB/sec), Consumer["
+                          + i
+                          + "] average ene-to-end latency (ms)");
+                } catch (IOException ignore) {
+                }
+              });
+      writer.newLine();
+    } catch (IOException ignore) {
+      writer = null;
+    }
+  }
+
+  private boolean logToCSV() {
+    var producerResult = tracker.producerResult();
+    var consumerResult = tracker.consumerResult();
+    if (writer == null || producerResult.completedRecords == 0L) return false;
+    var duration = tracker.duration();
+    var producerPercentage =
+        Math.min(
+            100D,
+            manager.exeTime().percentage(producerResult.completedRecords, duration.toMillis()));
+    var consumerPercentage = consumerResult.completedRecords * 100D / manager.producedRecords();
+    try {
+      writer.write(
+          duration.toHoursPart()
+              + "h"
+              + duration.toMinutesPart()
+              + "m"
+              + duration.toSecondsPart()
+              + "s");
+      writer.write(String.format(",%.2f%% / %.2f%%", consumerPercentage, producerPercentage));
+      writer.write("," + producerResult.averageBytes(duration));
+      writer.write("," + consumerResult.averageBytes(duration));
+      writer.write("," + producerResult.maxLatency + "," + producerResult.minLatency);
+      writer.write("," + consumerResult.maxLatency + "," + consumerResult.minLatency);
+      for (int i = 0; i < producerResult.bytes.size(); ++i) {
+        writer.write("," + Tracker.avg(duration, producerResult.bytes.get(i)));
+        writer.write("," + producerResult.averageLatencies.get(i));
+      }
+      for (int i = 0; i < consumerResult.bytes.size(); ++i) {
+        writer.write("," + Tracker.avg(duration, consumerResult.bytes.get(i)));
+        writer.write("," + consumerResult.averageLatencies.get(i));
+      }
+      writer.newLine();
+    } catch (IOException ignore) {
+    }
+    return manager.producedDone() && manager.consumedDone();
+  }
+
+  @Override
+  public void close() {
+    try {
+      writer.close();
+    } catch (IOException ignore) {
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/performance/Manager.java b/app/src/main/java/org/astraea/performance/Manager.java
index e84b809f..e8b786bf 100644
--- a/app/src/main/java/org/astraea/performance/Manager.java
+++ b/app/src/main/java/org/astraea/performance/Manager.java
@@ -1,8 +1,6 @@
 package org.astraea.performance;
 
-import java.util.List;
-import java.util.Optional;
-import java.util.Random;
+import java.util.*;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
@@ -25,6 +23,7 @@ public class Manager {
   private final List<Metrics> producerMetrics, consumerMetrics;
   private final long start = System.currentTimeMillis();
   private final AtomicLong payloadNum = new AtomicLong(0);
+  private final Distribution distribution;
 
   /**
    * Used to manage producing/consuming.
@@ -53,6 +52,7 @@ public class Manager {
     this.producerMetrics = producerMetrics;
     this.consumerMetrics = consumerMetrics;
     this.exeTime = argument.exeTime;
+    this.distribution = argument.distribution;
   }
 
   /**
@@ -109,4 +109,10 @@ public class Manager {
   public boolean consumedDone() {
     return producedDone() && consumedRecords() >= producedRecords();
   }
+
+  /** Randomly choose a key according to the distribution. */
+  public Optional<byte[]> getKey() {
+    if (distribution == null) return Optional.empty();
+    return Optional.of((String.valueOf(distribution.get())).getBytes());
+  }
 }
diff --git a/app/src/main/java/org/astraea/performance/Performance.java b/app/src/main/java/org/astraea/performance/Performance.java
index 17e7cbe3..9e63f9ae 100644
--- a/app/src/main/java/org/astraea/performance/Performance.java
+++ b/app/src/main/java/org/astraea/performance/Performance.java
@@ -5,9 +5,13 @@ import com.beust.jcommander.Parameter;
 import com.beust.jcommander.ParameterException;
 import java.io.IOException;
 import java.time.Duration;
+import java.util.Collection;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
 import java.util.function.BiConsumer;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
@@ -52,14 +56,15 @@ import org.astraea.utils.DataUnit;
  * To avoid records being produced too fast, producer wait for one millisecond after each send.
  */
 public class Performance {
-
+  /** Used in Automation, to achieve the end of one Performance and then start another. */
   public static void main(String[] args)
       throws InterruptedException, IOException, ExecutionException {
     execute(ArgumentUtil.parseArgument(new Argument(), args));
   }
 
-  public static void execute(final Argument param)
+  public static Future<String> execute(final Argument param)
       throws InterruptedException, IOException, ExecutionException {
+    var future = new CompletableFuture<String>();
     try (var topicAdmin = TopicAdmin.of(param.props())) {
       topicAdmin
           .creator()
@@ -80,6 +85,8 @@ public class Performance {
 
     var manager = new Manager(param, producerMetrics, consumerMetrics);
     var tracker = new Tracker(producerMetrics, consumerMetrics, manager);
+    Collection<ThreadPool.Executor> fileWriter =
+        (param.createCSV) ? List.of(new FileWriter(manager, tracker)) : List.of();
     var groupId = "groupId-" + System.currentTimeMillis();
     try (ThreadPool threadPool =
         ThreadPool.builder()
@@ -113,8 +120,11 @@ public class Performance {
                                 manager))
                     .collect(Collectors.toUnmodifiableList()))
             .executor(tracker)
+            .executors(fileWriter)
             .build()) {
       threadPool.waitAll();
+      future.complete(param.topic);
+      return future;
     }
   }
 
@@ -172,6 +182,7 @@ public class Performance {
         producer
             .sender()
             .topic(param.topic)
+            .key(manager.getKey().orElse(null))
             .value(payload.get())
             .timestamp(start)
             .run()
@@ -192,7 +203,7 @@ public class Performance {
     };
   }
 
-  static class Argument extends BasicArgumentWithPropFile {
+  public static class Argument extends BasicArgumentWithPropFile {
 
     @Parameter(
         names = {"--topic"},
@@ -266,12 +277,24 @@ public class Performance {
       return props;
     }
 
+    @Parameter(
+        names = {"--createCSV"},
+        description = "create the metrics into a csv file if this flag is set")
+    boolean createCSV = false;
+
     @Parameter(
         names = {"--compression"},
         description =
             "String: the compression algorithm used by producer. Available algorithm are none, gzip, snappy, lz4, and zstd",
         converter = CompressionArgument.class)
     CompressionType compression = CompressionType.NONE;
+
+    @Parameter(
+        names = {"--key.distribution"},
+        description =
+            "String: Distribution name. Available distribution names: \"uniform\", \"zipfian\", \"latest\". Default: (No key)",
+        converter = Distribution.DistributionConverter.class)
+    Distribution distribution = null;
   }
 
   static class CompressionArgument implements IStringConverter<CompressionType> {
diff --git a/app/src/main/java/org/astraea/performance/Tracker.java b/app/src/main/java/org/astraea/performance/Tracker.java
index 33380550..fbd25def 100644
--- a/app/src/main/java/org/astraea/performance/Tracker.java
+++ b/app/src/main/java/org/astraea/performance/Tracker.java
@@ -1,9 +1,7 @@
 package org.astraea.performance;
 
 import java.time.Duration;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
+import java.util.*;
 import org.astraea.concurrent.ThreadPool;
 
 /** Print out the given metrics. */
@@ -11,36 +9,41 @@ public class Tracker implements ThreadPool.Executor {
   private final List<Metrics> producerData;
   private final List<Metrics> consumerData;
   private final Manager manager;
+  // Snapshot of metrics
+  private Result producerResult;
+  private Result consumerResult;
   long start = 0L;
 
   public Tracker(List<Metrics> producerData, List<Metrics> consumerData, Manager manager) {
     this.producerData = producerData;
     this.consumerData = consumerData;
+    this.producerResult = result(producerData);
+    this.consumerResult = result(consumerData);
     this.manager = manager;
   }
 
   @Override
   public State execute() throws InterruptedException {
-    if (logProducers() & logConsumers()) return State.DONE;
+    producerResult = result(producerData);
+    consumerResult = result(consumerData);
+    if (logProducers(producerResult) & logConsumers(consumerResult)) return State.DONE;
     // Log after waiting for one second
     Thread.sleep(1000);
     return State.RUNNING;
   }
 
-  private Duration duration() {
+  Duration duration() {
     if (start == 0L) start = System.currentTimeMillis() - Duration.ofSeconds(1).toMillis();
     return Duration.ofMillis(System.currentTimeMillis() - start);
   }
 
-  private static double avg(Duration duration, long value) {
+  static double avg(Duration duration, long value) {
     return duration.toSeconds() <= 0
         ? 0
         : ((double) (value / duration.toSeconds())) / 1024D / 1024D;
   }
 
-  private boolean logProducers() {
-    var result = result(producerData);
-
+  private boolean logProducers(Result result) {
     if (result.completedRecords == 0) return false;
 
     var duration = duration();
@@ -70,10 +73,9 @@ public class Tracker implements ThreadPool.Executor {
     return percentage >= 100D;
   }
 
-  private boolean logConsumers() {
+  private boolean logConsumers(Result result) {
     // there is no consumer, so we just complete this log.
     if (consumerData.isEmpty()) return true;
-    var result = result(consumerData);
     if (result.completedRecords == 0) return false;
     var duration = duration();
 
@@ -94,6 +96,14 @@ public class Tracker implements ThreadPool.Executor {
     return manager.producedDone() && percentage >= 100D;
   }
 
+  public Result producerResult() {
+    return producerResult;
+  }
+
+  public Result consumerResult() {
+    return consumerResult;
+  }
+
   private static Result result(List<Metrics> metrics) {
     var completed = 0;
     var bytes = new ArrayList<Long>();
@@ -115,7 +125,7 @@ public class Tracker implements ThreadPool.Executor {
         max);
   }
 
-  private static class Result {
+  static class Result {
     public final long completedRecords;
     public final List<Long> bytes;
     public final List<Double> averageLatencies;
diff --git a/app/src/main/java/org/astraea/topic/Builder.java b/app/src/main/java/org/astraea/topic/Builder.java
index dc835dce..24b754f7 100644
--- a/app/src/main/java/org/astraea/topic/Builder.java
+++ b/app/src/main/java/org/astraea/topic/Builder.java
@@ -66,70 +66,59 @@ public class Builder {
     }
 
     @Override
-    public Map<TopicPartition, List<Group>> groups(Set<String> topics) {
-      var groups =
-          Utils.handleException(() -> admin.listConsumerGroups().valid().get()).stream()
-              .map(ConsumerGroupListing::groupId)
-              .collect(Collectors.toList());
-
-      var allPartitions = partitions(topics);
-
-      var result = new HashMap<TopicPartition, List<Group>>();
-      Utils.handleException(() -> admin.describeConsumerGroups(groups).all().get())
-          .forEach(
-              (groupId, groupDescription) -> {
-                var partitionOffsets =
-                    Utils.handleException(
-                        () ->
-                            admin
-                                .listConsumerGroupOffsets(groupId)
-                                .partitionsToOffsetAndMetadata()
-                                .get());
-
-                var partitionMembers =
-                    groupDescription.members().stream()
-                        .flatMap(
-                            m ->
-                                m.assignment().topicPartitions().stream()
-                                    .map(tp -> Map.entry(tp, m)))
-                        .collect(Collectors.groupingBy(Map.Entry::getKey))
-                        .entrySet()
-                        .stream()
-                        .collect(
-                            Collectors.toMap(
-                                Map.Entry::getKey,
-                                e ->
-                                    e.getValue().stream()
-                                        .map(Map.Entry::getValue)
-                                        .collect(Collectors.toList())));
-
-                allPartitions.forEach(
-                    tp -> {
-                      var offset =
-                          partitionOffsets.containsKey(tp)
-                              ? OptionalLong.of(partitionOffsets.get(tp).offset())
-                              : OptionalLong.empty();
+    public Map<String, ConsumerGroup> consumerGroup(Set<String> consumerGroupNames) {
+
+      final Set<String> groupIds =
+          !consumerGroupNames.isEmpty()
+              ? consumerGroupNames
+              : Utils.handleException(() -> admin.listConsumerGroups().all().get()).stream()
+                  .map(ConsumerGroupListing::groupId)
+                  .collect(Collectors.toUnmodifiableSet());
+
+      return Utils.handleException(
+          () -> {
+            var consumerGroupDescriptions = admin.describeConsumerGroups(groupIds).all().get();
+
+            var consumerGroupMetadata =
+                groupIds.stream()
+                    .map(x -> Map.entry(x, admin.listConsumerGroupOffsets(x)))
+                    .map(
+                        x ->
+                            Map.entry(
+                                x.getKey(),
+                                Utils.handleException(
+                                    () -> x.getValue().partitionsToOffsetAndMetadata().get())))
+                    .collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, Map.Entry::getValue));
+
+            var createMember =
+                (BiFunction<String, MemberDescription, Member>)
+                    (s, x) ->
+                        new Member(s, x.consumerId(), x.groupInstanceId(), x.clientId(), x.host());
+
+            return groupIds.stream()
+                .map(
+                    groupId -> {
                       var members =
-                          partitionMembers.getOrDefault(tp, List.of()).stream()
-                              .map(
-                                  m ->
-                                      new Member(
-                                          m.consumerId(),
-                                          m.groupInstanceId(),
-                                          m.clientId(),
-                                          m.host()))
-                              .collect(Collectors.toList());
-                      // This group is related to the partition only if it has either member or
-                      // offset.
-                      if (offset.isPresent() || !members.isEmpty()) {
-                        result
-                            .computeIfAbsent(tp, ignore -> new ArrayList<>())
-                            .add(new Group(groupId, offset, members));
-                      }
-                    });
-              });
-
-      return result;
+                          consumerGroupDescriptions.get(groupId).members().stream()
+                              .map(x -> createMember.apply(groupId, x))
+                              .collect(Collectors.toUnmodifiableList());
+                      var consumeOffset =
+                          consumerGroupMetadata.get(groupId).entrySet().stream()
+                              .collect(
+                                  Collectors.toUnmodifiableMap(
+                                      Map.Entry::getKey, x -> x.getValue().offset()));
+                      var assignment =
+                          consumerGroupDescriptions.get(groupId).members().stream()
+                              .collect(
+                                  Collectors.toUnmodifiableMap(
+                                      x -> createMember.apply(groupId, x),
+                                      x -> x.assignment().topicPartitions()));
+
+                      return Map.entry(
+                          groupId, new ConsumerGroup(groupId, members, consumeOffset, assignment));
+                    })
+                .collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, Map.Entry::getValue));
+          });
     }
 
     private Map<TopicPartition, Long> earliestOffset(Set<TopicPartition> partitions) {
diff --git a/app/src/main/java/org/astraea/topic/ConsumerGroup.java b/app/src/main/java/org/astraea/topic/ConsumerGroup.java
new file mode 100644
index 00000000..1618ca29
--- /dev/null
+++ b/app/src/main/java/org/astraea/topic/ConsumerGroup.java
@@ -0,0 +1,38 @@
+package org.astraea.topic;
+
+import java.util.*;
+import org.apache.kafka.common.TopicPartition;
+
+public class ConsumerGroup {
+  private final String groupId;
+  private final List<Member> activeMembers;
+  private final Map<TopicPartition, Long> consumeProgress;
+  private final Map<Member, Set<TopicPartition>> assignment;
+
+  public ConsumerGroup(
+      String groupId,
+      List<Member> activeMembers,
+      Map<TopicPartition, Long> consumeProgress,
+      Map<Member, Set<TopicPartition>> assignment) {
+    this.groupId = Objects.requireNonNull(groupId);
+    this.activeMembers = List.copyOf(activeMembers);
+    this.consumeProgress = Map.copyOf(consumeProgress);
+    this.assignment = Map.copyOf(assignment);
+  }
+
+  public String groupId() {
+    return groupId;
+  }
+
+  public Map<Member, Set<TopicPartition>> assignment() {
+    return assignment;
+  }
+
+  public Map<TopicPartition, Long> consumeProgress() {
+    return consumeProgress;
+  }
+
+  public List<Member> activeMembers() {
+    return activeMembers;
+  }
+}
diff --git a/app/src/main/java/org/astraea/topic/Group.java b/app/src/main/java/org/astraea/topic/Group.java
deleted file mode 100644
index 834ff6fa..00000000
--- a/app/src/main/java/org/astraea/topic/Group.java
+++ /dev/null
@@ -1,42 +0,0 @@
-package org.astraea.topic;
-
-import java.util.Collections;
-import java.util.List;
-import java.util.OptionalLong;
-
-public final class Group {
-  private final String groupId;
-  private final OptionalLong offset;
-  private final List<Member> members;
-
-  Group(String groupId, OptionalLong offset, List<Member> members) {
-    this.groupId = groupId;
-    this.offset = offset;
-    this.members = Collections.unmodifiableList(members);
-  }
-
-  @Override
-  public String toString() {
-    return "Group{"
-        + "groupId='"
-        + groupId
-        + '\''
-        + ", offset="
-        + (offset.isEmpty() ? "none" : offset.getAsLong())
-        + ", members="
-        + members
-        + '}';
-  }
-
-  public String groupId() {
-    return groupId;
-  }
-
-  public OptionalLong offset() {
-    return offset;
-  }
-
-  public List<Member> members() {
-    return members;
-  }
-}
diff --git a/app/src/main/java/org/astraea/topic/Member.java b/app/src/main/java/org/astraea/topic/Member.java
index 94027516..1dd5a939 100644
--- a/app/src/main/java/org/astraea/topic/Member.java
+++ b/app/src/main/java/org/astraea/topic/Member.java
@@ -1,14 +1,22 @@
 package org.astraea.topic;
 
+import java.util.Objects;
 import java.util.Optional;
 
 public final class Member {
+  private final String groupId;
   private final String memberId;
   private final Optional<String> groupInstanceId;
   private final String clientId;
   private final String host;
 
-  Member(String memberId, Optional<String> groupInstanceId, String clientId, String host) {
+  Member(
+      String groupId,
+      String memberId,
+      Optional<String> groupInstanceId,
+      String clientId,
+      String host) {
+    this.groupId = groupId;
     this.memberId = memberId;
     this.groupInstanceId = groupInstanceId;
     this.clientId = clientId;
@@ -32,6 +40,10 @@ public final class Member {
         + '}';
   }
 
+  public String groupId() {
+    return groupId;
+  }
+
   public String memberId() {
     return memberId;
   }
@@ -47,4 +59,21 @@ public final class Member {
   public String host() {
     return host;
   }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) return true;
+    if (o == null || getClass() != o.getClass()) return false;
+    Member member = (Member) o;
+    return Objects.equals(groupId, member.groupId)
+        && Objects.equals(memberId, member.memberId)
+        && Objects.equals(groupInstanceId, member.groupInstanceId)
+        && Objects.equals(clientId, member.clientId)
+        && Objects.equals(host, member.host);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(groupId, memberId, groupInstanceId, clientId, host);
+  }
 }
diff --git a/app/src/main/java/org/astraea/topic/TopicAdmin.java b/app/src/main/java/org/astraea/topic/TopicAdmin.java
index be815c56..d2cc1ffc 100644
--- a/app/src/main/java/org/astraea/topic/TopicAdmin.java
+++ b/app/src/main/java/org/astraea/topic/TopicAdmin.java
@@ -2,7 +2,6 @@ package org.astraea.topic;
 
 import java.io.Closeable;
 import java.util.*;
-import org.apache.kafka.clients.admin.*;
 import org.apache.kafka.common.TopicPartition;
 
 public interface TopicAdmin extends Closeable {
@@ -43,10 +42,11 @@ public interface TopicAdmin extends Closeable {
   Map<TopicPartition, Offset> offsets(Set<String> topics);
 
   /**
-   * @param topics topic names
-   * @return the partition having consumer group id and consumer group offset
+   * @param consumerGroupNames consumer group names. if empty set given, every consume group will
+   *     return.
+   * @return the member info of each consumer group
    */
-  Map<TopicPartition, List<Group>> groups(Set<String> topics);
+  Map<String, ConsumerGroup> consumerGroup(Set<String> consumerGroupNames);
 
   /**
    * @param topics topic names
diff --git a/app/src/main/java/org/astraea/topic/TopicExplorer.java b/app/src/main/java/org/astraea/topic/TopicExplorer.java
index db909c0b..5ad05426 100644
--- a/app/src/main/java/org/astraea/topic/TopicExplorer.java
+++ b/app/src/main/java/org/astraea/topic/TopicExplorer.java
@@ -2,84 +2,421 @@ package org.astraea.topic;
 
 import com.beust.jcommander.Parameter;
 import java.io.IOException;
+import java.io.PrintStream;
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
 import java.util.*;
+import java.util.function.*;
 import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+import org.apache.kafka.common.TopicPartition;
 import org.astraea.argument.ArgumentUtil;
 import org.astraea.argument.BasicArgumentWithPropFile;
+import org.astraea.utils.DataSize;
+import org.astraea.utils.DataUnit;
 
 public class TopicExplorer {
 
-  static class Result {
-    final String topic;
-    final int partition;
+  static class PartitionInfo {
+    final TopicPartition topicPartition;
     final long earliestOffset;
     final long latestOffset;
-    final List<Group> groups;
     final List<Replica> replicas;
 
-    Result(
-        String topic,
-        int partition,
-        long startOffset,
-        long endOffset,
-        List<Group> groups,
+    public PartitionInfo(
+        TopicPartition topicPartition,
+        long earliestOffset,
+        long latestOffset,
         List<Replica> replicas) {
-      this.topic = topic;
-      this.partition = partition;
-      this.earliestOffset = startOffset;
-      this.latestOffset = endOffset;
-      this.groups =
-          groups.stream().sorted(Comparator.comparing(Group::groupId)).collect(Collectors.toList());
+      this.topicPartition = topicPartition;
+      this.earliestOffset = earliestOffset;
+      this.latestOffset = latestOffset;
       this.replicas =
           replicas.stream()
               .sorted(Comparator.comparing(Replica::broker))
               .collect(Collectors.toList());
     }
+  }
 
-    @Override
-    public String toString() {
-      return "topic='"
-          + topic
-          + '\''
-          + ", partition="
-          + partition
-          + ", earliestOffset="
-          + earliestOffset
-          + ", latestOffset="
-          + latestOffset
-          + ", groups="
-          + groups
-          + ", replicas="
-          + replicas;
+  static class Result {
+    public final LocalDateTime time;
+    public final Map<String, List<PartitionInfo>> partitionInfo;
+    public final Map<String, ConsumerGroup> consumerGroups;
+
+    Result(
+        LocalDateTime time,
+        Map<String, List<PartitionInfo>> partitionInfo,
+        Map<String, ConsumerGroup> consumerGroups) {
+      this.time = time;
+      this.partitionInfo = partitionInfo;
+      this.consumerGroups = consumerGroups;
     }
   }
 
-  static List<Result> execute(TopicAdmin admin, Set<String> topics) {
+  static Result execute(TopicAdmin admin, Set<String> topics) {
     var replicas = admin.replicas(topics);
     var offsets = admin.offsets(topics);
-    var groups = admin.groups(topics);
-
-    return replicas.entrySet().stream()
-        .filter(e -> offsets.containsKey(e.getKey()))
-        .map(
-            e ->
-                new Result(
-                    e.getKey().topic(),
-                    e.getKey().partition(),
-                    offsets.get(e.getKey()).earliest(),
-                    offsets.get(e.getKey()).latest(),
-                    groups.getOrDefault(e.getKey(), List.of()),
-                    e.getValue()))
-        .sorted(
-            Comparator.comparing((Result r) -> r.topic).thenComparing((Result r) -> r.partition))
-        .collect(Collectors.toList());
+    var consumerGroups = admin.consumerGroup(Set.of());
+    var time = LocalDateTime.now();
+
+    // Given topic name, return the partition count
+    var topicPartitionCount =
+        replicas.keySet().stream()
+            .collect(Collectors.groupingBy(TopicPartition::topic))
+            .entrySet()
+            .stream()
+            .map(entry -> Map.entry(entry.getKey(), entry.getValue().size()))
+            .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+
+    var topicPartitionInfos =
+        topics.stream()
+            .collect(
+                Collectors.toMap(
+                    Function.identity(),
+                    (topic) ->
+                        IntStream.range(0, topicPartitionCount.get(topic))
+                            .mapToObj(partition -> new TopicPartition(topic, partition))
+                            .map(
+                                topicPartition ->
+                                    new PartitionInfo(
+                                        topicPartition,
+                                        offsets.get(topicPartition).earliest(),
+                                        offsets.get(topicPartition).latest(),
+                                        replicas.getOrDefault(topicPartition, List.of())))
+                            .collect(Collectors.toUnmodifiableList())));
+
+    return new Result(time, topicPartitionInfos, consumerGroups);
   }
 
   public static void main(String[] args) throws IOException {
     var argument = ArgumentUtil.parseArgument(new Argument(), args);
     try (var admin = TopicAdmin.of(argument.props())) {
-      execute(admin, argument.topics.isEmpty() ? admin.topicNames() : argument.topics)
-          .forEach(System.out::println);
+      var result = execute(admin, argument.topics.isEmpty() ? admin.topicNames() : argument.topics);
+      TreeOutput.print(result, System.out);
+    }
+  }
+
+  static class TreeOutput {
+
+    private final LocalDateTime time;
+    private final Map<String, List<PartitionInfo>> info;
+    private final Map<String, ConsumerGroup> consumerGroups;
+    private final PrintStream printStream;
+
+    private static final String NEXT_LEVEL = "| ";
+    private static final String TERMINATOR =
+        "|_____________________________________________________________________________________";
+
+    private TreeOutput(Result result, PrintStream printStream) {
+      this.time = Objects.requireNonNull(result.time);
+      this.info = Map.copyOf(result.partitionInfo);
+      this.consumerGroups = Map.copyOf(result.consumerGroups);
+      this.printStream = Objects.requireNonNull(printStream);
+    }
+
+    public static void print(Result result, PrintStream printStream) {
+      new TreeOutput(result, printStream).print();
+    }
+
+    private final Stack<String> treePrefix = new Stack<>();
+    private String cachedPrefix = "";
+
+    private void nextLevel(String prefix, Runnable runnable) {
+      treePrefix.push(prefix);
+      cachedPrefix = String.join("", treePrefix);
+      runnable.run();
+      treePrefix.pop();
+      cachedPrefix = String.join("", treePrefix);
+    }
+
+    private void nextLevel(String prefix, Runnable runnable, String terminator) {
+      treePrefix.push(prefix);
+      cachedPrefix = String.join("", treePrefix);
+      runnable.run();
+      treePrintln(terminator);
+      treePrefix.pop();
+      cachedPrefix = String.join("", treePrefix);
+    }
+
+    private void treePrintln(String format, Object... args) {
+      printStream.print(cachedPrefix);
+      printStream.printf(format, args);
+      printStream.println();
+    }
+
+    public void print() {
+      treePrefix.clear();
+      cachedPrefix = "";
+
+      treePrintln("[%s]", time.format(DateTimeFormatter.ISO_LOCAL_DATE_TIME));
+      nextLevel(
+          " ",
+          () ->
+              info.entrySet().stream()
+                  .sorted(Map.Entry.comparingByKey())
+                  .forEach(
+                      (entry) -> {
+                        treePrintln("Topic \"%s\"", entry.getKey());
+                        nextLevel(
+                            "",
+                            () ->
+                                nextLevel(
+                                    NEXT_LEVEL,
+                                    () -> {
+                                      printStatistics(entry.getKey(), entry.getValue());
+                                      printConsumerGroup(entry.getKey(), entry.getValue());
+                                      printPartitionReplica(entry.getValue());
+                                    }),
+                            TERMINATOR);
+                      }));
+    }
+
+    private void printStatistics(String topic, List<PartitionInfo> partitionInfos) {
+      treePrintln("Statistics of Topic \"%s\"", topic);
+      nextLevel(
+          NEXT_LEVEL,
+          () -> {
+            treePrintln(
+                "Topic Size: %s",
+                DataUnit.Byte.of(
+                    partitionInfos.stream()
+                        .flatMapToLong(x -> x.replicas.stream().mapToLong(Replica::size))
+                        .sum()));
+            treePrintln("Partition Count: %d", partitionInfos.size());
+            treePrintln(
+                "Partition Size Average: %s",
+                DataUnit.Byte.of(
+                    (long)
+                        partitionInfos.stream()
+                            .mapToLong(x -> x.replicas.stream().mapToLong(Replica::size).sum())
+                            .summaryStatistics()
+                            .getAverage()));
+            treePrintln(
+                "Replica Count: %d",
+                partitionInfos.stream().mapToInt(x -> x.replicas.size()).sum());
+            treePrintln(
+                "Replica Size Average: %s",
+                DataUnit.Byte.of(
+                    (long)
+                        partitionInfos.stream()
+                            .flatMapToLong(x -> x.replicas.stream().mapToLong(Replica::size))
+                            .summaryStatistics()
+                            .getAverage()));
+          });
+    }
+
+    private void printConsumerGroup(String topic, List<PartitionInfo> partitionInfos) {
+      treePrintln("Consumer Groups:");
+      nextLevel(
+          NEXT_LEVEL,
+          () -> {
+            // 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");
+                                                });
+                                          });
+                                }
+                              });
+                        });
+                  });
+            }
+          });
+    }
+
+    private void printPartitionReplica(List<PartitionInfo> partitionInfos) {
+      treePrintln("Partitions/Replicas:");
+      nextLevel(
+          NEXT_LEVEL,
+          () -> {
+            // print partition & replica info
+            partitionInfos.forEach(
+                partitionInfo -> {
+                  treePrintln(
+                      "Partition \"%d\" (size: %s) (offset range: [%d, %d])",
+                      partitionInfo.topicPartition.partition(),
+                      DataUnit.Byte.of(
+                          partitionInfo.replicas.stream().mapToLong(Replica::size).sum()),
+                      partitionInfo.earliestOffset,
+                      partitionInfo.latestOffset);
+                  nextLevel(
+                      "  ",
+                      () ->
+                          nextLevel(
+                              NEXT_LEVEL,
+                              () ->
+                                  partitionInfo.replicas.stream()
+                                      .sorted(Comparator.comparing(Replica::broker))
+                                      .forEach(
+                                          replica ->
+                                              treePrintln(
+                                                  "replica on broker %-4s %17s %s at \"%s\"",
+                                                  "#" + replica.broker(),
+                                                  ReplicaHelper.size(replica.size()),
+                                                  ReplicaHelper.descriptor(replica),
+                                                  replica.path()))));
+                });
+          });
+    }
+
+    private static class ConsumeProgress {
+      private final String topic;
+      private final int index;
+      private final ConsumerGroup group;
+      private final Map<String, List<PartitionInfo>> map;
+
+      private ConsumeProgress(
+          String topic, int partition, ConsumerGroup group, Map<String, List<PartitionInfo>> map) {
+        this.topic = topic;
+        this.index = partition;
+        this.group = group;
+        this.map = map;
+      }
+
+      private long earliest() {
+        return map.get(topic).get(index).earliestOffset;
+      }
+
+      private long latest() {
+        return map.get(topic).get(index).latestOffset;
+      }
+
+      private long current() {
+        return group.consumeProgress().get(new TopicPartition(topic, index));
+      }
+
+      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("", "[", "]"));
+      }
+
+      @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());
+      }
+    }
+
+    private static class ReplicaHelper {
+      static String dataSizeString(DataSize dataSize) {
+        return String.format(
+            "%.2f %s", dataSize.idealMeasurement().doubleValue(), dataSize.idealDataUnit());
+      }
+
+      static String size(long bytes) {
+        var dataSize = DataUnit.Byte.of(bytes);
+        return "(size=" + dataSizeString(dataSize) + ")";
+      }
+
+      static String descriptor(Replica replica) {
+        return Stream.of(
+                Optional.of(replica.leader() ? "leader" : "follower"),
+                Optional.ofNullable(replica.lag() > 0 ? "lagged" + size(replica.lag()) : null),
+                Optional.ofNullable(replica.inSync() ? null : "non-synced"),
+                Optional.ofNullable(replica.isFuture() ? "future" : null))
+            .flatMap(Optional::stream)
+            .collect(Collectors.joining(", ", "[", "]"));
+      }
     }
   }
 
diff --git a/app/src/main/java/org/astraea/utils/DataSize.java b/app/src/main/java/org/astraea/utils/DataSize.java
index 0fdfb90c..f37f1053 100644
--- a/app/src/main/java/org/astraea/utils/DataSize.java
+++ b/app/src/main/java/org/astraea/utils/DataSize.java
@@ -4,6 +4,7 @@ import com.beust.jcommander.IStringConverter;
 import java.math.BigDecimal;
 import java.math.BigInteger;
 import java.math.MathContext;
+import java.math.RoundingMode;
 import java.time.Duration;
 import java.time.temporal.ChronoUnit;
 import java.util.Comparator;
@@ -183,7 +184,9 @@ public class DataSize implements Comparable<DataSize> {
 
   /** Return a string represent current size in given data unit. */
   public String toString(DataUnit unit) {
-    return String.format("%s %s", this.bits.divide(unit.bits), unit.name());
+    var divide =
+        new BigDecimal(this.bits).divide(new BigDecimal(unit.bits), 3, RoundingMode.HALF_EVEN);
+    return String.format("%s %s", divide, unit.name());
   }
 
   /**
diff --git a/config/automation.properties b/config/automation.properties
new file mode 100644
index 00000000..6ecd7247
--- /dev/null
+++ b/config/automation.properties
@@ -0,0 +1,24 @@
+#Defaults is to use the default parameters
+
+#You can set as many times as you want it to run.
+--time = Default
+
+#This parameter must be configured so that the performance can be run.
+--bootstrap.servers = Default
+
+#Set it to true to delete the generated topic every time you perform a performance.
+--whetherDeleteTopic = false
+
+#Optional
+--record.size = Default
+--run.until = Default
+--compression = Default
+--consumers = Default
+--fixed.size = Default
+--jmx.servers = Default
+--partitioner = Default
+--partitions = Default
+--producers = Default
+--prop.file =Default
+--replicas = Default
+--topic = Default
diff --git a/docker/start_broker.sh b/docker/start_broker.sh
index a826cdfa..05aa1e59 100755
--- a/docker/start_broker.sh
+++ b/docker/start_broker.sh
@@ -4,15 +4,16 @@
 
 declare -r USER=astraea
 declare -r VERSION=${REVISION:-${VERSION:-2.8.1}}
-declare -r REPO=${REPO:-astraea/broker}
+declare -r REPO=${REPO:-ghcr.io/skiptests/astraea/broker}
 declare -r IMAGE_NAME="$REPO:$VERSION"
+declare -r BUILD=${BUILD:-false}
 declare -r RUN=${RUN:-true}
 declare -r DOCKER_FOLDER=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" &>/dev/null && pwd)
 declare -r DOCKERFILE=$DOCKER_FOLDER/broker.dockerfile
 declare -r DATA_FOLDER_IN_CONTAINER_PREFIX="/tmp/log-folder"
 declare -r EXPORTER_VERSION="0.16.1"
-declare -r EXPORTER_PORT="$(($(($RANDOM % 10000)) + 10000))"
-declare -r BROKER_PORT="$(($(($RANDOM % 10000)) + 10000))"
+declare -r EXPORTER_PORT=${EXPORTER_PORT:-$(($(($RANDOM % 10000)) + 10000))}
+declare -r BROKER_PORT=${BROKER_PORT:-$(($(($RANDOM % 10000)) + 10000))}
 declare -r CONTAINER_NAME="broker-$BROKER_PORT"
 declare -r BROKER_JMX_PORT="$(($(($RANDOM % 10000)) + 10000))"
 declare -r ADMIN_NAME="admin"
@@ -45,7 +46,8 @@ function showHelp() {
   echo "    HEAP_OPTS=\"-Xmx2G -Xms2G\"                set broker JVM memory"
   echo "    REVISION=trunk                           set revision of kafka source code to build container"
   echo "    VERSION=2.8.1                            set version of kafka distribution"
-  echo "    RUN=false                                set false if you want to build image only"
+  echo "    BUILD=false                              set true if you want to build image locally"
+  echo "    RUN=false                                set false if you want to build/pull image only"
   echo "    DATA_FOLDERS=/tmp/folder1,/tmp/folder2   set host folders used by broker"
 }
 
@@ -147,7 +149,22 @@ function generateDockerfile() {
 
 function buildImageIfNeed() {
   if [[ "$(docker images -q $IMAGE_NAME 2>/dev/null)" == "" ]]; then
-    docker build --no-cache -t "$IMAGE_NAME" -f "$DOCKERFILE" "$DOCKER_FOLDER"
+    local needToBuild="true"
+    if [[ "$BUILD" == "false" ]]; then
+      docker pull $IMAGE_NAME 2>/dev/null
+      if [[ "$?" == "0" ]]; then
+        needToBuild="false"
+      else
+        echo "Can't find $IMAGE_NAME from repo. Will build $IMAGE_NAME on the local"
+      fi
+    fi
+    if [[ "$needToBuild" == "true" ]]; then
+      generateDockerfile
+      docker build --no-cache -t "$IMAGE_NAME" -f "$DOCKERFILE" "$DOCKER_FOLDER"
+      if [[ "$?" != "0" ]]; then
+        exit 2
+      fi
+    fi
   fi
 }
 
@@ -201,7 +218,7 @@ function generateMountCommand() {
     declare -i count=0
 
     for folder in "${folders[@]}"; do
-      mount="-v $folder:$DATA_FOLDER_IN_CONTAINER_PREFIX-$count"
+      mount="$mount -v $folder:$DATA_FOLDER_IN_CONTAINER_PREFIX-$count"
       count=$((count + 1))
     done
   fi
diff --git a/docker/start_astraea.sh b/docker/start_kafka_tool.sh
similarity index 50%
rename from docker/start_astraea.sh
rename to docker/start_kafka_tool.sh
index 8d387d2c..d2afe920 100755
--- a/docker/start_astraea.sh
+++ b/docker/start_kafka_tool.sh
@@ -2,12 +2,24 @@
 
 # ===============================[global variables]===============================
 
-IMAGE_NAME="astraea/astraea:latest"
-DOCKER_FOLDER=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" &>/dev/null && pwd)
-DOCKER_FILE=$DOCKER_FOLDER/astraea.dockerfile
+declare -r VERSION=${REVISION:-${VERSION:-main}}
+declare -r REPO=${REPO:-ghcr.io/skiptests/astraea/kafka-tool}
+declare -r IMAGE_NAME="$REPO:$VERSION"
+declare -r DOCKER_FOLDER=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" &>/dev/null && pwd)
+declare -r DOCKERFILE=$DOCKER_FOLDER/kafka_tool.dockerfile
+declare -r BUILD=${BUILD:-false}
+declare -r RUN=${RUN:-true}
 
 # ===================================[functions]===================================
 
+function showHelp() {
+  echo "Usage: [ENV] start_kafka_tool.sh"
+  echo "ENV: "
+  echo "    REPO=astraea/kafka-tool    set the docker repo"
+  echo "    BUILD=false                set true if you want to build image locally"
+  echo "    RUN=false                  set false if you want to build/pull image only"
+}
+
 function checkDocker() {
   if [[ "$(which docker)" == "" ]]; then
     echo "you have to install docker"
@@ -31,14 +43,30 @@ RUN git clone https://github.com/skiptests/astraea
 
 # pre-build project to collect all dependencies
 WORKDIR /tmp/astraea
+RUN git checkout $VERSION
 RUN ./gradlew clean build -x test --no-daemon
 RUN cp \$(find ./app/build/libs/ -maxdepth 1 -type f -name app-*-all.jar) /tmp/app.jar
-" >"$DOCKER_FILE"
+" >"$DOCKERFILE"
 }
 
 function buildImageIfNeed() {
   if [[ "$(docker images -q $IMAGE_NAME 2>/dev/null)" == "" ]]; then
-    docker build --no-cache -t "$IMAGE_NAME" -f "$DOCKER_FILE" "$DOCKER_FOLDER"
+    local needToBuild="true"
+    if [[ "$BUILD" == "false" ]]; then
+      docker pull $IMAGE_NAME 2>/dev/null
+      if [[ "$?" == "0" ]]; then
+        needToBuild="false"
+      else
+        echo "Can't find $IMAGE_NAME from repo. Will build $IMAGE_NAME on the local"
+      fi
+    fi
+    if [[ "$needToBuild" == "true" ]]; then
+      generateDockerfile
+      docker build --no-cache -t "$IMAGE_NAME" -f "$DOCKERFILE" "$DOCKER_FOLDER"
+      if [[ "$?" != "0" ]]; then
+        exit 2
+      fi
+    fi
   fi
 }
 
@@ -54,8 +82,15 @@ function runContainer() {
 checkDocker
 generateDockerfile
 buildImageIfNeed
+
+if [[ "$RUN" != "true" ]]; then
+  echo "docker image: $IMAGE_NAME is created"
+  exit 0
+fi
+
 if [[ -n "$1" ]]; then
   runContainer "$*"
 else
+  showHelp
   runContainer "help"
 fi
diff --git a/docker/start_spark.sh b/docker/start_spark.sh
index 402ee4bc..7ae25739 100755
--- a/docker/start_spark.sh
+++ b/docker/start_spark.sh
@@ -4,11 +4,12 @@
 
 declare -r USER=astraea
 declare -r VERSION=${REVISION:-${VERSION:-3.1.2}}
-declare -r REPO=${REPO:-astraea/spark}
+declare -r REPO=${REPO:-ghcr.io/skiptests/astraea/spark}
 declare -r IMAGE_NAME="$REPO:$VERSION"
+declare -r BUILD=${BUILD:-false}
 declare -r RUN=${RUN:-true}
-declare -r SPARK_PORT="$(($(($RANDOM % 10000)) + 10000))"
-declare -r SPARK_UI_PORT="$(($(($RANDOM % 10000)) + 10000))"
+declare -r SPARK_PORT=${SPARK_PORT:-$(($(($RANDOM % 10000)) + 10000))}
+declare -r SPARK_UI_PORT=${SPARK_UI_PORT:-$(($(($RANDOM % 10000)) + 10000))}
 declare -r IVY_VERSION=2.5.0
 declare -r DELTA_VERSION=${DELTA_VERSION:-1.0.0}
 declare -r PYTHON_KAFKA_VERSION=${PYTHON_KAFKA_VERSION:-1.7.0}
@@ -27,12 +28,12 @@ function showHelp() {
   echo "Optional Arguments: "
   echo "    master-url=spar://node00:1111    start a spark worker. Or start a spark master if master-url is not defined"
   echo "ENV: "
-  echo "    REPO=astraea/spark               set the docker repo"
   echo "    VERSION=3.1.2                    set version of spark distribution"
   echo "    DELTA_VERSION=1.0.0              set version of delta distribution"
-  echo "    PYTHON_KAFKA_VERSION=1.7.0           set version of confluent kafka distribution"
-  echo "    RUN=false                        set false if you want to build image only"
-  echo "    PYTHON_DEPS=delta-spark=1.0.0    those dependencies will be pre-installed in the docker image"
+  echo "    PYTHON_KAFKA_VERSION=1.7.0       set version of confluent kafka distribution"
+  echo "    BUILD=false                      set true if you want to build image locally"
+  echo "    RUN=false                        set false if you want to build/pull image only"
+  echo "    PYTHON_DEPS=delta-spark=1.0.0    set the python dependencies which are pre-installed in the docker image"
 }
 
 function checkDocker() {
@@ -146,7 +147,22 @@ function generateDockerfile() {
 
 function buildImageIfNeed() {
   if [[ "$(docker images -q $IMAGE_NAME 2>/dev/null)" == "" ]]; then
-    docker build --no-cache -t "$IMAGE_NAME" -f "$DOCKERFILE" "$DOCKER_FOLDER"
+    local needToBuild="true"
+    if [[ "$BUILD" == "false" ]]; then
+      docker pull $IMAGE_NAME 2>/dev/null
+      if [[ "$?" == "0" ]]; then
+        needToBuild="false"
+      else
+        echo "Can't find $IMAGE_NAME from repo. Will build $IMAGE_NAME on the local"
+      fi
+    fi
+    if [[ "$needToBuild" == "true" ]]; then
+      generateDockerfile
+      docker build --no-cache -t "$IMAGE_NAME" -f "$DOCKERFILE" "$DOCKER_FOLDER"
+      if [[ "$?" != "0" ]]; then
+        exit 2
+      fi
+    fi
   fi
 }
 
diff --git a/docker/start_zookeeper.sh b/docker/start_zookeeper.sh
index 84f487f1..37271c57 100755
--- a/docker/start_zookeeper.sh
+++ b/docker/start_zookeeper.sh
@@ -4,14 +4,15 @@
 
 declare -r USER=astraea
 declare -r VERSION=${VERSION:-3.7.0}
-declare -r REPO=${REPO:-astraea/zookeeper}
+declare -r REPO=${REPO:-ghcr.io/skiptests/astraea/zookeeper}
 declare -r IMAGE_NAME="$REPO:$VERSION"
-declare -r PORT="$(($(($RANDOM % 10000)) + 10000))"
+declare -r ZOOKEEPER_PORT=${ZOOKEEPER_PORT:-$(($(($RANDOM % 10000)) + 10000))}
+declare -r BUILD=${BUILD:-false}
 declare -r RUN=${RUN:-true}
 declare -r DOCKER_FOLDER=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" &>/dev/null && pwd)
 declare -r DOCKERFILE=$DOCKER_FOLDER/zookeeper.dockerfile
 declare -r DATA_FOLDER_IN_CONTAINER="/tmp/zookeeper-dir"
-declare -r CONTAINER_NAME="zookeeper-$PORT"
+declare -r CONTAINER_NAME="zookeeper-$ZOOKEEPER_PORT"
 declare -r ADDRESS=$([[ "$(which ipconfig)" != "" ]] && ipconfig getifaddr en0 || hostname -i)
 
 # ===================================[functions]===================================
@@ -21,7 +22,8 @@ function showHelp() {
   echo "ENV: "
   echo "    REPO=astraea/zk            set the docker repo"
   echo "    VERSION=3.7.0              set version of zookeeper distribution"
-  echo "    RUN=false                  set false if you want to build image only"
+  echo "    BUILD=false                set true if you want to build image locally"
+  echo "    RUN=false                  set false if you want to build/pull image only"
   echo "    DATA_FOLDER=/tmp/folder1   set host folders used by zookeeper"
 }
 
@@ -68,7 +70,22 @@ RUN echo "clientPort=2181" >> ./conf/zoo.cfg
 
 function buildImageIfNeed() {
   if [[ "$(docker images -q $IMAGE_NAME 2>/dev/null)" == "" ]]; then
-    docker build --no-cache -t "$IMAGE_NAME" -f "$DOCKERFILE" "$DOCKER_FOLDER"
+    local needToBuild="true"
+    if [[ "$BUILD" == "false" ]]; then
+      docker pull $IMAGE_NAME 2>/dev/null
+      if [[ "$?" == "0" ]]; then
+        needToBuild="false"
+      else
+        echo "Can't find $IMAGE_NAME from repo. Will build $IMAGE_NAME on the local"
+      fi
+    fi
+    if [[ "$needToBuild" == "true" ]]; then
+      generateDockerfile
+      docker build --no-cache -t "$IMAGE_NAME" -f "$DOCKERFILE" "$DOCKER_FOLDER"
+      if [[ "$?" != "0" ]]; then
+        exit 2
+      fi
+    fi
   fi
 }
 
@@ -80,7 +97,6 @@ if [[ "$1" == "help" ]]; then
 fi
 
 checkDocker
-generateDockerfile
 buildImageIfNeed
 
 if [[ "$RUN" != "true" ]]; then
@@ -95,15 +111,15 @@ if [[ -n "$DATA_FOLDER" ]]; then
   echo "mount $DATA_FOLDER to container"
   docker run -d \
     --name $CONTAINER_NAME \
-    -p $PORT:2181 \
+    -p $ZOOKEEPER_PORT:2181 \
     -v "$DATA_FOLDER":$DATA_FOLDER_IN_CONTAINER \
     $IMAGE_NAME ./bin/zkServer.sh start-foreground
 else
   docker run -d \
-    -p $PORT:2181 \
+    -p $ZOOKEEPER_PORT:2181 \
     $IMAGE_NAME ./bin/zkServer.sh start-foreground
 fi
 
 echo "================================================="
-echo "run $DOCKER_FOLDER/start_broker.sh zookeeper.connect=$ADDRESS:$PORT to join kafka broker"
+echo "run $DOCKER_FOLDER/start_broker.sh zookeeper.connect=$ADDRESS:$ZOOKEEPER_PORT to join kafka broker"
 echo "================================================="


Test patch:
diff --git a/app/src/test/java/org/astraea/performance/DistributionTest.java b/app/src/test/java/org/astraea/performance/DistributionTest.java
new file mode 100644
index 00000000..2cb38401
--- /dev/null
+++ b/app/src/test/java/org/astraea/performance/DistributionTest.java
@@ -0,0 +1,23 @@
+package org.astraea.performance;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class DistributionTest {
+  @Test
+  void testLatest() throws InterruptedException {
+    var distribution = Distribution.latest();
+    Assertions.assertEquals(distribution.get(), distribution.get());
+
+    long first = distribution.get();
+    Thread.sleep(2000);
+
+    Assertions.assertNotEquals(first, distribution.get());
+  }
+
+  @Test
+  void testZipfian() {
+    var distribution = Distribution.zipfian(5);
+    Assertions.assertTrue(distribution.get() < 5);
+  }
+}
diff --git a/app/src/test/java/org/astraea/performance/ManagerTest.java b/app/src/test/java/org/astraea/performance/ManagerTest.java
index 37b42817..57f72668 100644
--- a/app/src/test/java/org/astraea/performance/ManagerTest.java
+++ b/app/src/test/java/org/astraea/performance/ManagerTest.java
@@ -53,4 +53,25 @@ public class ManagerTest {
     // Assertion failed with probability < 1/102400
     Assertions.assertFalse(same);
   }
+
+  @Test
+  void testGetKey() {
+    var argument = new Performance.Argument();
+
+    argument.distribution = Distribution.uniform();
+    var manager = new Manager(argument, List.of(), List.of());
+    Assertions.assertTrue(manager.getKey().isPresent());
+
+    argument.distribution = Distribution.zipfian(10);
+    manager = new Manager(argument, List.of(), List.of());
+    Assertions.assertTrue(manager.getKey().isPresent());
+
+    argument.distribution = Distribution.latest();
+    manager = new Manager(argument, List.of(), List.of());
+    Assertions.assertTrue(manager.getKey().isPresent());
+
+    argument.distribution = null;
+    manager = new Manager(argument, List.of(), List.of());
+    Assertions.assertTrue(manager.getKey().isEmpty());
+  }
 }
diff --git a/app/src/test/java/org/astraea/topic/TopicAdminTest.java b/app/src/test/java/org/astraea/topic/TopicAdminTest.java
index ce80d02d..4d60cf14 100644
--- a/app/src/test/java/org/astraea/topic/TopicAdminTest.java
+++ b/app/src/test/java/org/astraea/topic/TopicAdminTest.java
@@ -112,19 +112,6 @@ public class TopicAdminTest extends RequireBrokerCluster {
     }
   }
 
-  @Test
-  void testGroups() throws InterruptedException {
-    var topicName = "testGroups";
-    try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
-      topicAdmin.creator().topic(topicName).numberOfPartitions(3).create();
-      // wait for syncing topic creation
-      TimeUnit.SECONDS.sleep(5);
-      Assertions.assertTrue(topicAdmin.topicNames().contains(topicName));
-      var groups = topicAdmin.groups(Set.of(topicName));
-      Assertions.assertEquals(0, groups.size());
-    }
-  }
-
   @Test
   void testOffsets() throws InterruptedException {
     var topicName = "testOffsets";
@@ -144,6 +131,26 @@ public class TopicAdminTest extends RequireBrokerCluster {
     }
   }
 
+  @Test
+  void testConsumerGroups() throws InterruptedException {
+    var topicName = "testConsumerGroups-Topic";
+    var consumerGroup = "testConsumerGroups-Group";
+    try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
+      topicAdmin.creator().topic(topicName).numberOfPartitions(3).create();
+      Consumer.builder()
+          .brokers(bootstrapServers())
+          .topics(Set.of(topicName))
+          .groupId(consumerGroup)
+          .build();
+      // wait for syncing topic creation
+      TimeUnit.SECONDS.sleep(5);
+      var consumerGroupMap = topicAdmin.consumerGroup(Set.of(consumerGroup));
+      Assertions.assertEquals(1, consumerGroupMap.size());
+      Assertions.assertTrue(consumerGroupMap.containsKey(consumerGroup));
+      Assertions.assertEquals(consumerGroup, consumerGroupMap.get(consumerGroup).groupId());
+    }
+  }
+
   @Test
   @DisabledOnOs(WINDOWS)
   void testMigrateSinglePartition() throws InterruptedException {
diff --git a/app/src/test/java/org/astraea/topic/TopicExplorerTest.java b/app/src/test/java/org/astraea/topic/TopicExplorerTest.java
index 45191831..c7c20d3a 100644
--- a/app/src/test/java/org/astraea/topic/TopicExplorerTest.java
+++ b/app/src/test/java/org/astraea/topic/TopicExplorerTest.java
@@ -1,29 +1,196 @@
 package org.astraea.topic;
 
-import java.io.IOException;
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.io.*;
+import java.time.Duration;
+import java.time.LocalDateTime;
+import java.time.temporal.ChronoUnit;
 import java.util.*;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.TopicPartition;
+import org.astraea.consumer.Consumer;
 import org.astraea.service.RequireBrokerCluster;
-import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
 
 public class TopicExplorerTest extends RequireBrokerCluster {
 
+  // List of created consumer work.
+  // Each consumer has to keep fetching from the broker to prevent been killed.
+  // The `TopicExploreTest#consumer` created a thread that keep doing the fetching job.
+  // This list keeps the thread works, shutdown these thread when necessary.
+  private final List<Closeable> createdConsumerWorks = new ArrayList<>();
+
+  @AfterEach
+  private void closeConsumers() {
+    createdConsumerWorks.forEach(
+        x -> {
+          try {
+            x.close();
+          } catch (IOException e) {
+            e.printStackTrace();
+          }
+        });
+    createdConsumerWorks.clear();
+  }
+
+  private void consumer(Set<String> topics, String groupName, String instanceName) {
+    // start the consuming work and insert the new consumer work into the list
+    createdConsumerWorks.add(
+        new Closeable() {
+          private final AtomicBoolean shutdown = new AtomicBoolean(false);
+
+          public Closeable start() {
+            new Thread(
+                    () -> {
+                      var consumer =
+                          Consumer.builder()
+                              .brokers(bootstrapServers())
+                              .topics(topics)
+                              .groupId(groupName)
+                              .configs(
+                                  Map.of(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, instanceName))
+                              .fromBeginning()
+                              .build();
+
+                      while (!shutdown.get()) consumer.poll(Duration.ofSeconds(1));
+
+                      consumer.close();
+                    })
+                .start();
+            return this;
+          }
+
+          @Override
+          public void close() {
+            shutdown.set(true);
+          }
+        }.start());
+  }
+
   @Test
-  void testExecute() throws IOException, InterruptedException {
-    var topicName = "testExecute";
+  void testExecute() throws InterruptedException {
+    // arrange
+    var topicName = "TopicExplorerTest-testExecute-Topic";
+    var groupName = "TopicExplorerTest-testExecute-ConsumerGroup";
     try (var admin = TopicAdmin.of(bootstrapServers())) {
-      admin.creator().topic(topicName).numberOfPartitions(1).numberOfReplicas((short) 1).create();
-      // wait for topic
-      TimeUnit.SECONDS.sleep(5);
+      admin.creator().topic(topicName).numberOfPartitions(3).numberOfReplicas((short) 1).create();
+      consumer(Set.of(topicName), groupName, "alpha");
+      consumer(Set.of(topicName), groupName, "beta");
+      consumer(Set.of(topicName), groupName, "gamma");
+
+      TimeUnit.SECONDS.sleep(10); // wait for broker process resource creation
+
+      // act
       var result = TopicExplorer.execute(admin, Set.of(topicName));
-      Assertions.assertEquals(1, result.size());
-      Assertions.assertEquals(topicName, result.get(0).topic);
-      Assertions.assertEquals(0, result.get(0).partition);
-      Assertions.assertEquals(0, result.get(0).earliestOffset);
-      Assertions.assertEquals(0, result.get(0).latestOffset);
-      Assertions.assertEquals(0, result.get(0).groups.size());
-      Assertions.assertEquals(1, result.get(0).replicas.size());
+
+      // assert
+      assertTrue(
+          ChronoUnit.SECONDS.between(result.time, LocalDateTime.now()) < 10,
+          "assert the execution timestamp come from a reasonable passed time (10sec before)");
+      assertTrue(result.consumerGroups.containsKey(groupName));
+      assertEquals(3, result.consumerGroups.get(groupName).activeMembers().size());
+      assertTrue(
+          result.consumerGroups.get(groupName).activeMembers().stream()
+              .anyMatch(x -> x.groupInstanceId().orElseThrow().equals("alpha")));
+      assertTrue(
+          result.consumerGroups.get(groupName).activeMembers().stream()
+              .anyMatch(x -> x.groupInstanceId().orElseThrow().equals("beta")));
+      assertTrue(
+          result.consumerGroups.get(groupName).activeMembers().stream()
+              .anyMatch(x -> x.groupInstanceId().orElseThrow().equals("gamma")));
+      assertTrue(result.partitionInfo.containsKey(topicName));
+      assertEquals(3, result.partitionInfo.get(topicName).size());
     }
   }
+
+  @Test
+  void testOutput() {
+    // arrange
+    var now = LocalDateTime.now();
+    var mockOutput = new ByteArrayOutputStream();
+    var printStream = new PrintStream(mockOutput);
+    var groupMembers =
+        List.of(
+            new Member(
+                "my-consumer-group-1",
+                "memberId-1",
+                Optional.of("instance-1"),
+                "clientId-1",
+                "host1"),
+            new Member(
+                "my-consumer-group-1",
+                "memberId-2",
+                Optional.of("instance-2"),
+                "clientId-2",
+                "host2"),
+            new Member(
+                "my-consumer-group-1",
+                "memberId-3",
+                Optional.of("instance-3"),
+                "clientId-3",
+                "host3"));
+    var partitionInfo =
+        List.of(
+            new TopicExplorer.PartitionInfo(
+                new TopicPartition("my-topic", 0),
+                0,
+                100,
+                List.of(new Replica(55, 15, 100, true, false, true, "/tmp/path0"))));
+    var result =
+        new TopicExplorer.Result(
+            now,
+            Map.of("my-topic", partitionInfo),
+            Map.of(
+                "my-consumer-group-1",
+                new ConsumerGroup(
+                    "my-consumer-group-1",
+                    groupMembers,
+                    Map.of(new TopicPartition("my-topic", 0), 50L),
+                    Map.of(groupMembers.get(0), Set.of(new TopicPartition("my-topic", 0))))));
+
+    // act
+    TopicExplorer.TreeOutput.print(result, printStream);
+
+    // assert
+    final String output = mockOutput.toString();
+    System.out.println(output);
+    assertTrue(output.matches("(?ms).+Topic.+my-topic.+"), "topic name printed");
+    assertTrue(
+        output.matches("(?ms).+Consumer Group.+my-consumer-group.+"),
+        "consumer group name printed");
+    assertTrue(output.matches("(?ms).+offset.+0/50/100.+"), "offset printed");
+    assertTrue(output.matches("(?ms).+member.+memberId-1.+"), "member info printed");
+    assertTrue(output.matches("(?ms).+member.+memberId-2.+"), "member info printed");
+    assertTrue(output.matches("(?ms).+member.+memberId-3.+"), "member info printed");
+    assertTrue(output.matches("(?ms).+working on partition 0.+"), "assignment printed");
+    assertTrue(output.matches("(?ms).+no partition assigned.+"), "assignment printed");
+    assertTrue(output.matches("(?ms).+clientId.+clientId-1.+"), "host printed");
+    assertTrue(output.matches("(?ms).+clientId.+clientId-2.+"), "host printed");
+    assertTrue(output.matches("(?ms).+clientId.+clientId-3.+"), "host printed");
+    assertTrue(output.matches("(?ms).+host.+host1.+"), "host printed");
+    assertTrue(output.matches("(?ms).+host.+host2.+"), "host printed");
+    assertTrue(output.matches("(?ms).+host.+host3.+"), "host printed");
+    assertTrue(output.matches("(?ms).+groupInstanceId.+instance-1.+"), "group instance id printed");
+    assertTrue(output.matches("(?ms).+groupInstanceId.+instance-2.+"), "group instance id printed");
+    assertTrue(output.matches("(?ms).+groupInstanceId.+instance-3.+"), "group instance id printed");
+    assertTrue(output.matches("(?ms).+Topic Size.+100.+Byte.+"), "statistics");
+    assertTrue(output.matches("(?ms).+Partition Count.+1.+"), "statistics");
+    assertTrue(output.matches("(?ms).+Partition Size Average.+100.+Byte.+"), "statistics");
+    assertTrue(output.matches("(?ms).+Replica Count.+1.+"), "statistics");
+    assertTrue(output.matches("(?ms).+Replica Size Average.+100.+Byte.+"), "statistics");
+    assertTrue(output.matches("(?ms).+Partition.+0.+"), "partition info printed");
+    assertTrue(output.matches("(?ms).+size.+100.+Byte.+"), "statistics");
+    assertTrue(output.matches("(?ms).+offset range.+0.+100.+"), "partition offset printed");
+    assertTrue(output.matches("(?ms).+replica on broker.+55.+"), "broker id printed");
+    assertTrue(output.matches("(?ms).+size.+100.+Byte.+"), "replica size printed");
+    assertTrue(output.matches("(?ms).+\\[.*leader.*].+"), "descriptor printed");
+    assertTrue(output.matches("(?ms).+\\[.*lagged.+size.+15.+Byte.*].+"), "descriptor printed");
+    assertTrue(output.matches("(?ms).+\\[.*non-synced.*].+"), "descriptor printed");
+    assertTrue(output.matches("(?ms).+\\[.*future.*].+"), "descriptor printed");
+    assertTrue(output.matches("(?ms).+/tmp/path0.+"), "path printed");
+  }
 }