Repository: skiptests/astraea
Pull request #86 (2021-11-01 08:25:06)

Implementation of Metric Explorer
Context: #10

This PR implements a tool to fetch raw Mbean metrics from a JMX server.

## List Mbeans' metrics

`./gradlew run --args="metrics --jmx.server localhost:11111"`

https://gist.github.com/garyparrot/f13d858fe47b7da8cffa0b31d0ee90f6

## List Mbeans' object name

`./gradlew run --args="metrics --jmx.server localhost:11111 --view-object-name-list"`

https://gist.github.com/garyparrot/bfae020adf9d4f969ef4d4444078dd2c

## TODO

- [x] Update `README.md`
- [x] Add unit tests.
- [x] in `README.d`, `Metric Client` -> `Metric Explorer`
- [x] update demo


Patch:
diff --git a/README.md b/README.md
index f45af32a..9a459681 100644
--- a/README.md
+++ b/README.md
@@ -30,7 +30,7 @@ java -jar astraea-0.0.1-alpha.1-all.jar [tool] [args]
 
 The following scripts can build a kafka cluster by containers in one minute.
 
-### Set up zookeeper
+### Run Zookeeper
 
 ```shell
 ./docker/start_zookeeper.sh
@@ -46,7 +46,7 @@ run ./docker/start_broker.sh zookeeper.connect=192.168.50.178:17228 to join kafk
 
 You can define `ZOOKEEPER_VERSION` to change the binary version.
 
-### Set up (kafka) broker
+### Run Kafka Broker
 
 After the zk env is running, you can copy the command (see above example) from zk script output to set up kafka. For example:
 ```shell
@@ -57,35 +57,61 @@ The console will show the broker connection information and JMX address. For exa
 
 ```shell
 =================================================
-broker address: 192.168.50.224:11248
-jmx address: 192.168.50.224:15905
+broker address: 192.168.50.178:12747
+jmx address: 192.168.50.178:10216
+exporter address: 192.168.50.178:10558
+broker id: 677
 =================================================
 ```
 
+1. `broker address` is used by kafka client code. The alias is bootstrap server.
+2. `jmx address` exports the java metrics by JMX
+3. `exporter address` is the address of prometheus exporter.
+
 The command to set up broker can be executed multiple times to create a broker cluster. The env `KAFKA_VERSION` is used to
 define the release version of kafka. Or you can define `KAFKA_REVISION` to run kafka based on specific revision of source code.
 
+### Run Prometheus
+
+The exporters of brokers can offer metrics to Prometheus to monitor cluster status. For example, your kafka cluster has
+two brokers. The exporter address of first broker is `192.168.50.178:10558` and another is `192.168.50.178:10553`. You can
+use following command to run a prometheus service.
+
+```shell
+./docker/start_prometheus.sh 192.168.50.178:10558,192.168.50.178:10553
+```
+
+The console will show the http address of prometheus service.
+```shell
+=================================================
+config file: /tmp/prometheus-15483.yml
+prometheus address: http://192.168.50.178:15483
+=================================================
+```
+
 ---
 
 ## Performance Benchmark
 This tool is used test to following metrics.
 1. publish latency: the time of completing producer data request
 2. E2E latency: the time for a record to travel through Kafka
-3. input rate: sum of consumer inputs in MByte per second
-4. output rate: sum of producer outputs in MByte per second
+3. input rate: average of consumer inputs in MByte per second
+4. output rate: average of producer outputs in MByte per second
 
 Run the benchmark from source
 ```shell
-./gradlew run --args="Performance --bootstrap.servers localhost:9092 --topic topic --topicConfig partitions:10,replicationFactor:3 --producers 5 --consumers 1 --records 100000 --recordSize 10000"
+./gradlew run --args="performance --bootstrap.servers localhost:9092"
 ```
 ### Performance Benchmark Configurations
 1. --bootstrap.servers: the server to connect to
-2. --topic: the topic name
+2. --topic: the topic name. Default: testPerformance-{Time in millis}
 3. --partitions: topic config when creating new topic. Default: 1 
 4. --replicas: topic config when creating new topic. Default: 1
 5. --consumers: the number of consumers (threads). Default: 1
-6. --records: the total number of records sent by the producers. Default: 1000
-7. --record.size: the record size in byte. Default: 1024 byte
+6. --producers: the number of producers (threads). Default: 1
+7. --records: the total number of records sent by the producers. Default: 1000
+8. --record.size: the record size in byte. Default: 1024 byte
+9. --prop.file: the path to property file.
 
 ---
 
@@ -128,25 +154,50 @@ This project offers a way to run kafka official tool by container. For example:
 
 ---
 
-## Kafka Metric Client
+## Kafka Metric Explorer
 
 This tool can be used to access Kafka's MBean metrics via JMX.
 
 Run the tool from source code
 
 ```shell
+# fetch every Mbeans from specific JMX server.
 ./gradlew run --args="metrics --jmx.server 192.168.50.178:1099"
+
+# fetch any Mbean that its object name contains property "type=Memory".
+./gradlew run --args="metrics --jmx.server 192.168.50.178:1099 --property type=Memory"
+
+# fetch any Mbean that belongs to "kafka.network" domain name, 
+# and it's object name contains two properties "request=Metadata" and "name=LocalTimeMs".
+./gradlew run --args="metrics --jmx.server 192.168.50.178:1099 --domain kafka.network --property request=Metadata --property name=LocalTimeMs"
+
+# list all Mbeans' object name on specific JMX server.
+./gradlew run --args="metrics --jmx.server 192.168.50.178:1099 --view-object-name-list"
 ```
 
 Run the tool from release
 ```shell
+# fetch every Mbeans from specific JMX server.
 java -jar app-0.0.1-SNAPSHOT-all.jar metrics --jmx.server 192.168.50.178:1099
+
+# fetch any Mbean that its object name contains property "type=Memory".
+java -jar app-0.0.1-SNAPSHOT-all.jar metrics --jmx.server 192.168.50.178:1099 --property type=Memory
+
+# fetch any Mbean that belongs to "kafka.network" domain name,
+# and it's object name contains two properties "request=Metadata" and "name=LocalTimeMs".
+java -jar app-0.0.1-SNAPSHOT-all.jar metrics --jmx.server 192.168.50.178:1099 --domain kafka.network --property request=Metadata --property name=LocalTimeMs
+
+# list all Mbeans' object name on specific JMX server.
+java -jar app-0.0.1-SNAPSHOT-all.jar metrics --jmx.server 192.168.50.178:1099 --view-object-name-list
 ```
 
-### Metric Client Configurations
+### Metric Explorer Configurations
 
-1. --jmx.server: the address to connect to Kafka JMX remote server
-2. --metrics: the Mbean metric to fetch. Default: All metrics
+1. --jmx.server: the address to connect to Kafka JMX remote server.
+2. --domain: query Mbeans from the specific domain name (support wildcard "\*" and "?"). Default: "\*".
+3. --property: query mbeans with the specific property (support wildcard "\*" and "?"). You can specify this argument multiple times. Default: [].
+4. --strict-match: only Mbeans with its object name completely match the given criteria shows. Default: false.
+5. --view-object-name-list: show the list view of MBeans' domain name & properties. Default: false.
 
 ---
 
@@ -169,4 +220,4 @@ This tool offers an effective way to migrate all replicas from specific brokers
 ### Replica Collie Configurations
 1. --bootstrap.servers: the server to connect to
 2. --topics: the topics to be moved
-3. --admin.props.file: the file path containing the properties to be passed to kafka admin
\ No newline at end of file
+3. --admin.props.file: the file path containing the properties to be passed to kafka admin
diff --git a/app/src/main/java/org/astraea/App.java b/app/src/main/java/org/astraea/App.java
index 758a068a..a6ef0c43 100644
--- a/app/src/main/java/org/astraea/App.java
+++ b/app/src/main/java/org/astraea/App.java
@@ -5,7 +5,7 @@ import java.lang.reflect.InvocationTargetException;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
-import org.astraea.metrics.KafkaMetricClientApp;
+import org.astraea.metrics.MetricExplorer;
 import org.astraea.performance.Performance;
 import org.astraea.topic.ReplicaCollie;
 import org.astraea.topic.TopicExplorer;
@@ -14,7 +14,7 @@ public class App {
   private static final Map<String, Class<?>> MAIN_CLASSES =
       Map.of(
           "offset", TopicExplorer.class,
-          "metrics", KafkaMetricClientApp.class,
+          "metrics", MetricExplorer.class,
           "replica", ReplicaCollie.class,
           "performance", Performance.class);
 
diff --git a/app/src/main/java/org/astraea/argument/BasicAdminArgument.java b/app/src/main/java/org/astraea/argument/BasicArgumentWithPropFile.java
similarity index 64%
rename from app/src/main/java/org/astraea/argument/BasicAdminArgument.java
rename to app/src/main/java/org/astraea/argument/BasicArgumentWithPropFile.java
index a64d81fd..d7f7ae9d 100644
--- a/app/src/main/java/org/astraea/argument/BasicAdminArgument.java
+++ b/app/src/main/java/org/astraea/argument/BasicArgumentWithPropFile.java
@@ -3,18 +3,18 @@ package org.astraea.argument;
 import com.beust.jcommander.Parameter;
 import java.util.Map;
 
-public abstract class BasicAdminArgument extends BasicArgument {
+public abstract class BasicArgumentWithPropFile extends BasicArgument {
   @Parameter(
-      names = {"--admin.props.file"},
+      names = {"--prop.file"},
       description = "the file path containing the properties to be passed to kafka admin",
       validateWith = ArgumentUtil.NotEmptyString.class)
-  public String propsFile;
+  public String propFile;
 
   /**
    * @return the kafka props consists of bootstrap servers and all props from file (if it is
    *     existent)
    */
-  public Map<String, Object> adminProps() {
-    return properties(propsFile);
+  public Map<String, Object> props() {
+    return properties(propFile);
   }
 }
diff --git a/app/src/main/java/org/astraea/consumer/Builder.java b/app/src/main/java/org/astraea/consumer/Builder.java
index 43dd1d76..07b47bbb 100644
--- a/app/src/main/java/org/astraea/consumer/Builder.java
+++ b/app/src/main/java/org/astraea/consumer/Builder.java
@@ -20,6 +20,7 @@ public class Builder<Key, Value> {
   private OffsetPolicy offsetPolicy = OffsetPolicy.LATEST;
   private String groupId = "groupId-" + System.currentTimeMillis();
   private final Set<String> topics = new HashSet<>();
+  private ConsumerRebalanceListener listener = ignore -> {};
 
   Builder() {}
 
@@ -61,6 +62,11 @@ public class Builder<Key, Value> {
     return this;
   }
 
+  public Builder<Key, Value> consumerRebalanceListener(ConsumerRebalanceListener listener) {
+    this.listener = Objects.requireNonNull(listener);
+    return this;
+  }
+
   @SuppressWarnings("unchecked")
   public Consumer<Key, Value> build() {
     configs.put(ConsumerConfig.GROUP_ID_CONFIG, Objects.requireNonNull(groupId));
@@ -77,7 +83,7 @@ public class Builder<Key, Value> {
             configs,
             Deserializer.of((Deserializer<Key>) keyDeserializer),
             Deserializer.of((Deserializer<Value>) valueDeserializer));
-    kafkaConsumer.subscribe(topics);
+    kafkaConsumer.subscribe(topics, ConsumerRebalanceListener.of(listener));
     return new Consumer<>() {
       @Override
       public Collection<Record<Key, Value>> poll(Duration timeout) {
diff --git a/app/src/main/java/org/astraea/consumer/ConsumerRebalanceListener.java b/app/src/main/java/org/astraea/consumer/ConsumerRebalanceListener.java
new file mode 100644
index 00000000..12d39f89
--- /dev/null
+++ b/app/src/main/java/org/astraea/consumer/ConsumerRebalanceListener.java
@@ -0,0 +1,31 @@
+package org.astraea.consumer;
+
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Set;
+import org.apache.kafka.common.TopicPartition;
+
+@FunctionalInterface
+public interface ConsumerRebalanceListener {
+
+  /**
+   * A callback method the user can implement to provide handling of customized offsets on
+   * completion of a successful partition re-assignment. See {@link
+   * org.apache.kafka.clients.consumer.ConsumerRebalanceListener#onPartitionsAssigned(Collection)}
+   * for more detail.
+   */
+  void onPartitionAssigned(Set<TopicPartition> partitions);
+
+  static org.apache.kafka.clients.consumer.ConsumerRebalanceListener of(
+      ConsumerRebalanceListener listener) {
+    return new org.apache.kafka.clients.consumer.ConsumerRebalanceListener() {
+      @Override
+      public void onPartitionsRevoked(Collection<TopicPartition> ignore) {}
+
+      @Override
+      public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
+        listener.onPartitionAssigned(new HashSet<>(partitions));
+      }
+    };
+  }
+}
diff --git a/app/src/main/java/org/astraea/metrics/KafkaMetricClientApp.java b/app/src/main/java/org/astraea/metrics/KafkaMetricClientApp.java
deleted file mode 100644
index 64ed43df..00000000
--- a/app/src/main/java/org/astraea/metrics/KafkaMetricClientApp.java
+++ /dev/null
@@ -1,88 +0,0 @@
-package org.astraea.metrics;
-
-import com.beust.jcommander.Parameter;
-import java.net.MalformedURLException;
-import java.time.LocalTime;
-import java.time.format.DateTimeFormatter;
-import java.time.format.FormatStyle;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Set;
-import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
-import javax.management.remote.JMXServiceURL;
-import org.astraea.argument.ArgumentUtil;
-import org.astraea.metrics.jmx.MBeanClient;
-import org.astraea.metrics.kafka.BrokerTopicMetricsResult;
-import org.astraea.metrics.kafka.KafkaMetrics;
-
-public final class KafkaMetricClientApp {
-
-  public static String JMX_URI_FORMAT = "service:jmx:rmi:///jndi/rmi://" + "%s" + "/jmxrmi";
-
-  public static String createJmxUrl(String address) {
-    return String.format(JMX_URI_FORMAT, address);
-  }
-
-  public static void main(String[] args) throws MalformedURLException {
-    var parameters = ArgumentUtil.parseArgument(new Argument(), args);
-
-    String argumentJmxServerNetworkAddress = parameters.address;
-    Set<String> argumentTargetMetrics = parameters.metrics;
-
-    JMXServiceURL serviceURL = new JMXServiceURL(createJmxUrl(argumentJmxServerNetworkAddress));
-    try (MBeanClient mBeanClient = new MBeanClient(serviceURL)) {
-
-      // find the actual metrics to fetch.
-      List<KafkaMetrics.BrokerTopic> metrics =
-          argumentTargetMetrics.stream()
-              .map(KafkaMetrics.BrokerTopic::of)
-              .collect(Collectors.toUnmodifiableList());
-
-      while (!Thread.interrupted()) {
-        // fetch
-        List<BrokerTopicMetricsResult> collect =
-            metrics.stream()
-                .map(x -> x.fetch(mBeanClient))
-                .collect(Collectors.toUnmodifiableList());
-
-        // output
-        System.out.println(
-            "["
-                + LocalTime.now().format(DateTimeFormatter.ofLocalizedTime(FormatStyle.MEDIUM))
-                + "]");
-        for (BrokerTopicMetricsResult result : collect) System.out.println(result);
-        System.out.println();
-
-        try {
-          TimeUnit.SECONDS.sleep(1);
-        } catch (InterruptedException e) {
-          break;
-        }
-      }
-
-    } catch (Exception e) {
-      e.printStackTrace();
-    }
-  }
-
-  static class Argument {
-    @Parameter(
-        names = {"--jmx.server"},
-        description = "The address to connect to JMX remote server",
-        validateWith = ArgumentUtil.NotEmptyString.class,
-        required = true)
-    String address;
-
-    @Parameter(
-        names = {"--metrics"},
-        variableArity = true,
-        converter = ArgumentUtil.StringSetConverter.class,
-        description =
-            "The metric names you want. If no metric name specified in argument, all metrics will be selected.")
-    Set<String> metrics =
-        Arrays.stream(KafkaMetrics.BrokerTopic.values())
-            .map(KafkaMetrics.BrokerTopic::metricName)
-            .collect(Collectors.toUnmodifiableSet());
-  }
-}
diff --git a/app/src/main/java/org/astraea/metrics/MetricExplorer.java b/app/src/main/java/org/astraea/metrics/MetricExplorer.java
new file mode 100644
index 00000000..26ac4cd1
--- /dev/null
+++ b/app/src/main/java/org/astraea/metrics/MetricExplorer.java
@@ -0,0 +1,398 @@
+package org.astraea.metrics;
+
+import com.beust.jcommander.IParameterValidator;
+import com.beust.jcommander.IStringConverter;
+import com.beust.jcommander.Parameter;
+import com.beust.jcommander.ParameterException;
+import java.net.MalformedURLException;
+import java.util.*;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+import javax.management.openmbean.CompositeDataSupport;
+import javax.management.remote.JMXServiceURL;
+import org.astraea.argument.ArgumentUtil;
+import org.astraea.metrics.jmx.BeanObject;
+import org.astraea.metrics.jmx.BeanQuery;
+import org.astraea.metrics.jmx.MBeanClient;
+
+public class MetricExplorer {
+
+  static void execute(MBeanClient mBeanClient, Argument args) {
+    // compile query
+    BeanQuery.BeanQueryBuilder builder =
+        BeanQuery.builder(
+            args.fromDomainName,
+            args.properties.stream()
+                .map(Argument.CorrectPropertyFormat::toEntry)
+                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)));
+    if (!args.strictMatch) builder.usePropertyListPattern();
+
+    // execute query
+    Collection<BeanObject> beanObjects = mBeanClient.queryBeans(builder.build());
+
+    // display result
+    if (args.viewObjectNameList) displayObjectNameList(beanObjects);
+    else displayBeanObjects(beanObjects);
+  }
+
+  private static void displayBeanObjects(Collection<BeanObject> beanObjects) {
+    var propertyOrderMap = propertyOrder(beanObjects);
+    var collect = sortBeanObjects(beanObjects, propertyOrderMap);
+
+    for (BeanObject beanObject : collect) {
+      String properties =
+          beanObject.getProperties().entrySet().stream()
+              .sorted(Comparator.comparing(x -> propertyOrderMap.get(x.getKey())))
+              .map(entry -> entry.getKey() + "=" + entry.getValue())
+              .collect(Collectors.joining(","));
+
+      System.out.printf("[%s:%s]\n", beanObject.domainName(), properties);
+
+      Stream<String> strings = DataUtils.elaborateData(beanObject.getAttributes());
+      strings = DataUtils.streamAppendWith(" ", 4, strings);
+      strings.forEach(System.out::println);
+    }
+  }
+
+  private static void displayObjectNameList(Collection<BeanObject> beanObjects) {
+    var propertyOrderMap = propertyOrder(beanObjects);
+    var collect = sortBeanObjects(beanObjects, propertyOrderMap);
+
+    collect.forEach(
+        x ->
+            System.out.printf(
+                "%s:%s\n",
+                x.domainName(),
+                x.getProperties().entrySet().stream()
+                    .sorted(Comparator.comparingLong(z -> propertyOrderMap.get(z.getKey())))
+                    .map(z -> z.getKey() + "=" + z.getValue())
+                    .collect(Collectors.joining(","))));
+  }
+
+  /**
+   * generate the order of property entry, the lower number the higher priority
+   *
+   * @param beanObjects Mbeans objects, the order is decided by some property or statistics based on
+   *     the given Mbeans.
+   * @return a {@link Map} describe the priority of each property key
+   */
+  static Map<String, Long> propertyOrder(Collection<BeanObject> beanObjects) {
+
+    // count the frequency of each property key name
+    // property who has higher frequency will close to the tree root when printing result
+    Map<String, Long> propertyFrequencyMap =
+        beanObjects.stream()
+            .flatMap(x -> x.getProperties().entrySet().stream())
+            .map(Map.Entry::getKey)
+            .collect(Collectors.groupingBy(x -> x, Collectors.counting()));
+
+    // the order for sorting properties
+    Map<String, Long> propertyOrderMap =
+        propertyFrequencyMap.entrySet().stream()
+            .map(entry -> Map.entry(entry.getKey(), beanObjects.size() + 1 - entry.getValue()))
+            .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+
+    // the property key "name" will always has the lowest priority, just like how JMC does it.
+    propertyOrderMap.put("name", beanObjects.size() + 1L);
+
+    return propertyOrderMap;
+  }
+
+  /**
+   * sort the given bean objects based on the given property order. The order is first decide by the
+   * alphabet order of domain name, then decided by the given property key order.
+   *
+   * @param beanObjects the Mbeans to be sorted.
+   * @param propertyOrderMap the order of property key, the lower number the higher priority.
+   * @return a new {@link List} consist of all the given Mbeans, all Mbeans are sort by given
+   *     property order.
+   */
+  static List<BeanObject> sortBeanObjects(
+      Collection<BeanObject> beanObjects, Map<String, Long> propertyOrderMap) {
+    return beanObjects.stream()
+        .sorted(
+            Comparator.comparing(BeanObject::domainName)
+                .thenComparing(
+                    (o1, o2) -> {
+                      List<Map.Entry<String, String>> order1 =
+                          o1.getProperties().entrySet().stream()
+                              .sorted(Comparator.comparing(x -> propertyOrderMap.get(x.getKey())))
+                              .collect(Collectors.toList());
+                      List<Map.Entry<String, String>> order2 =
+                          o2.getProperties().entrySet().stream()
+                              .sorted(Comparator.comparing(x -> propertyOrderMap.get(x.getKey())))
+                              .collect(Collectors.toList());
+
+                      // how the comparison works:
+                      // 1. sort all properties, now property key/value with higher key frequency
+                      // will show first.
+                      // 2. sort by property key/value amount
+                      // 3. compare the 1'st property key
+                      // 4. compare the 2'nd property key
+                      // 5. etc...
+                      // 6. compare the 1'st property value
+                      // 7. compare the 2'nd property value
+                      // 8. etc...
+                      // 9. object name equal
+
+                      if (order1.size() != order2.size()) {
+                        return -Integer.compare(order1.size(), order2.size());
+                      } else {
+                        var key =
+                            Comparator.comparing(
+                                (Map.Entry<String, String> x) -> propertyOrderMap.get(x.getKey()));
+                        var value = Map.Entry.<String, String>comparingByValue();
+
+                        for (int i = 0; i < order1.size(); i++) {
+                          int result = key.compare(order1.get(i), order2.get(i));
+                          if (result != 0) return result;
+                        }
+
+                        for (int i = 0; i < order1.size(); i++) {
+                          int result = value.compare(order1.get(i), order2.get(i));
+                          if (result != 0) return result;
+                        }
+
+                        return 0;
+                      }
+                    }))
+        .collect(Collectors.toList());
+  }
+
+  public static void main(String[] args) throws Exception {
+    var arguments = ArgumentUtil.parseArgument(new Argument(), args);
+
+    try (MBeanClient mBeanClient = new MBeanClient(arguments.jmxServer)) {
+      execute(mBeanClient, arguments);
+    }
+  }
+
+  static class Argument {
+    @Parameter(
+        names = {"--jmx.server"},
+        description =
+            "The JMX server address to connect to, support [hostname:port] style or JMX URI format",
+        validateWith = ArgumentUtil.NotEmptyString.class,
+        converter = Argument.JmxServerUrlConverter.class,
+        required = true)
+    JMXServiceURL jmxServer;
+
+    @Parameter(
+        names = {"--domain"},
+        description =
+            "Show Mbeans from specific domain pattern, support wildcard(*) and any char(?)")
+    String fromDomainName = "*";
+
+    @Parameter(
+        names = {"--property"},
+        description = "Fetch mbeans with specific property, support wildcard(*) and any char(?)",
+        validateWith = CorrectPropertyFormat.class)
+    List<String> properties = List.of();
+
+    @Parameter(
+        names = {"--strict-match"},
+        description =
+            "Only MBeans metrics with properties completely match the given criteria shows")
+    boolean strictMatch = false;
+
+    @Parameter(
+        names = {"--view-object-name-list"},
+        description = "Show the list view of MBeans' domain name & properties")
+    boolean viewObjectNameList = false;
+
+    public static class JmxServerUrlConverter implements IStringConverter<JMXServiceURL> {
+
+      /** This regex used to test if a string look like a JMX URL */
+      static Pattern patternOfJmxUrlStart = Pattern.compile("^service:jmx:rmi://");
+
+      /**
+       * Convert the given string to a {@link JMXServiceURL}.
+       *
+       * <p>Following string formats are supported
+       *
+       * <ol>
+       *   <li>From hostname and port {@code "kafka1.example.com:9875"}
+       *   <li>From JMX Service URL string {@code "service:jmx:rmi:///kafka1.example.come ..."}
+       * </ol>
+       */
+      @Override
+      public JMXServiceURL convert(String value) {
+        Matcher matchJmxUrlStart = patternOfJmxUrlStart.matcher(value);
+
+        if (matchJmxUrlStart.find()) {
+          return convertFromJmxUrl(value);
+        } else {
+          return convertFromAddress(value);
+        }
+      }
+
+      private JMXServiceURL convertFromJmxUrl(String jmxUrlString) {
+        try {
+          return new JMXServiceURL(jmxUrlString);
+        } catch (MalformedURLException e) {
+          throw new IllegalArgumentException("Cannot parse given URL: " + jmxUrlString, e);
+        }
+      }
+
+      private JMXServiceURL convertFromAddress(String addressOfJmxServer) {
+        String url = String.format("service:jmx:rmi:///jndi/rmi://%s/jmxrmi", addressOfJmxServer);
+        try {
+          return new JMXServiceURL(url);
+        } catch (MalformedURLException e) {
+          throw new IllegalArgumentException("Illegal JMX Url: " + url, e);
+        }
+      }
+    }
+
+    public static class CorrectPropertyFormat implements IParameterValidator {
+
+      static Pattern propertyPattern = Pattern.compile("^(?<key>[^=]+)=(?<value>.+)$");
+
+      @Override
+      public void validate(String name, String value) throws ParameterException {
+        if (!propertyPattern.matcher(value).find())
+          throw new ParameterException(
+              name
+                  + "with value ["
+                  + value
+                  + "] should match format: "
+                  + propertyPattern.pattern());
+      }
+
+      static Map.Entry<String, String> toEntry(String property) {
+        Matcher matcher = propertyPattern.matcher(property);
+        if (!matcher.find()) throw new IllegalArgumentException("Not property format: " + property);
+        return Map.entry(matcher.group("key"), matcher.group("value"));
+      }
+    }
+  }
+
+  static class DataUtils {
+
+    static int stringLengthLimit = 500;
+
+    static Stream<String> elaborateData(Object target) {
+      try {
+        if (target == null) {
+          return Stream.of("null");
+        } else if (target instanceof List<?>) {
+          return elaborateList((List<?>) target);
+        } else if (target instanceof Map<?, ?>) {
+          return elaborateMap((Map<?, ?>) target);
+        } else if (target instanceof CompositeDataSupport) {
+          return elaborateCompositeDataSupport(((CompositeDataSupport) target));
+        } else if (target.getClass().isArray()) {
+          if (target.getClass() == long[].class) return elaborateArray((long[]) target);
+          else return elaborateArray((Object[]) target);
+        }
+
+        // I have no idea how to display this object, just toString it.
+        return Stream.of(stringify(target));
+      } catch (Exception e) {
+        return Stream.of(e.toString());
+      }
+    }
+
+    static String stringify(Object object) {
+      String string = object.toString();
+      if (string.length() > stringLengthLimit) {
+        return string.substring(0, stringLengthLimit)
+            + String.format("%n... (%d characters truncated)", string.length() - stringLengthLimit);
+      } else {
+        return string;
+      }
+    }
+
+    static Stream<String> elaborateMap(Map<?, ?> data) {
+      try {
+        return data.entrySet().stream()
+            .sorted(Comparator.comparing(Object::toString))
+            .flatMap(DataUtils::elaborateMapEntry);
+      } catch (Exception e) {
+        return Stream.of(e.toString());
+      }
+    }
+
+    static Stream<String> elaborateMapEntry(Map.Entry<?, ?> entry) {
+      try {
+        List<String> key = elaborateData(entry.getKey()).collect(Collectors.toList());
+        List<String> value = elaborateData(entry.getValue()).collect(Collectors.toList());
+
+        if (key.size() == 1 && value.size() > 1)
+          return Stream.concat(
+              Stream.of(String.format("\"%s\":", key.get(0))),
+              streamAppendWith(" ", 4, value.stream()));
+
+        return Stream.of(String.format("\"%s\": %s", key.get(0), value.get(0)));
+      } catch (Exception e) {
+        return Stream.of(e.toString());
+      }
+    }
+
+    static Stream<String> elaborateArray(Object[] array) {
+      try {
+        return IntStream.range(0, array.length)
+            .boxed()
+            .flatMap(
+                i -> {
+                  List<String> collect = elaborateData(array[i]).collect(Collectors.toList());
+                  if (collect.size() == 1)
+                    return Stream.of(String.format("%d: %s", i, collect.get(0)));
+                  else
+                    return Stream.concat(
+                        Stream.of(i + ":"), streamAppendWith(" ", 4, collect.stream()));
+                });
+      } catch (Exception e) {
+        return Stream.of(e.toString());
+      }
+    }
+
+    static Stream<String> elaborateArray(long[] array) {
+      try {
+        return IntStream.range(0, array.length).mapToObj(x -> x + ": " + array[x]);
+      } catch (Exception e) {
+        return Stream.of(e.toString());
+      }
+    }
+
+    static Stream<String> elaborateList(List<?> list) {
+      try {
+        return IntStream.range(0, list.size())
+            .boxed()
+            .flatMap(
+                i -> {
+                  List<String> collect = elaborateData(list.get(i)).collect(Collectors.toList());
+                  if (collect.size() == 1)
+                    return Stream.of(String.format("%d: %s", i, collect.get(0)));
+                  else
+                    return Stream.concat(
+                        Stream.of(i + ":"), streamAppendWith(" ", 4, collect.stream()));
+                });
+      } catch (Exception e) {
+        return Stream.of(e.toString());
+      }
+    }
+
+    static Stream<String> elaborateCompositeDataSupport(CompositeDataSupport data) {
+      try {
+        return data.getCompositeType().keySet().stream()
+            .sorted()
+            .flatMap(key -> DataUtils.elaborateMapEntry(Map.entry(key, data.get(key))));
+      } catch (Exception e) {
+        return Stream.of(e.toString());
+      }
+    }
+
+    static Stream<String> streamAppendWith(String s, int size, Stream<String> source) {
+      try {
+        String append = String.join("", Collections.nCopies(size, s));
+        return source.map(x -> append + x);
+      } catch (Exception e) {
+        return Stream.of(e.toString());
+      }
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/partitioner/partitionerFactory/LinkPartitioner.java b/app/src/main/java/org/astraea/partitioner/partitionerFactory/LinkPartitioner.java
index 0404404b..d3b40786 100644
--- a/app/src/main/java/org/astraea/partitioner/partitionerFactory/LinkPartitioner.java
+++ b/app/src/main/java/org/astraea/partitioner/partitionerFactory/LinkPartitioner.java
@@ -78,6 +78,8 @@ public class LinkPartitioner implements Partitioner {
         if (maxWeightServer == null || item.getValue()[1] > maxWeightServer.getValue()[1]) {
           maxWeightServer = item;
         }
+        currentBrokerHashMap.put(
+            item.getKey(), new int[] {item.getValue()[0], item.getValue()[1] + item.getValue()[0]});
       }
       assert maxWeightServer != null;
       currentBrokerHashMap.put(
diff --git a/app/src/main/java/org/astraea/performance/Metrics.java b/app/src/main/java/org/astraea/performance/Metrics.java
index 1d091d1e..20690fd4 100644
--- a/app/src/main/java/org/astraea/performance/Metrics.java
+++ b/app/src/main/java/org/astraea/performance/Metrics.java
@@ -1,7 +1,11 @@
 package org.astraea.performance;
 
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+
 /** Used to record statistics. This is thread safe. */
 public class Metrics {
+  private final long startTime = System.currentTimeMillis();
   private double avgLatency;
   private long num;
   private long max;
@@ -18,50 +22,49 @@ public class Metrics {
 
   /** Simultaneously add latency and bytes. */
   public synchronized void put(long latency, long bytes) {
+    ++num;
     putLatency(latency);
     addBytes(bytes);
   }
   /** Add a new value to latency metric. */
-  public synchronized void putLatency(long latency) {
-    if (min > latency) min = latency;
-    if (max < latency) max = latency;
-    ++num;
+  private synchronized void putLatency(long latency) {
+    min = Math.min(min, latency);
+    max = Math.max(max, latency);
     avgLatency += (((double) latency) - avgLatency) / (double) num;
   }
   /** Add a new value to bytes. */
-  public synchronized void addBytes(long bytes) {
+  private synchronized void addBytes(long bytes) {
     this.bytes += bytes;
   }
 
-  /** Get the number of latency put. */
+  /** @return Get the number of latency put. */
   public synchronized long num() {
     return num;
   }
-  /** Get the maximum of latency put. */
+  /** @return Get the maximum of latency put. */
   public synchronized long max() {
     return max;
   }
-  /** Get the minimum of latency put. */
+  /** @return Get the minimum of latency put. */
   public synchronized long min() {
     return min;
   }
-  /** Get the average latency. */
+  /** @return Get the average latency. */
   public synchronized double avgLatency() {
     return avgLatency;
   }
-  /** Reset to 0 and returns the old value of bytes */
-  public synchronized long bytesThenReset() {
-    long tmp = this.bytes;
-    this.bytes = 0;
-    return tmp;
+
+  /** @return the average bytes (in second) */
+  public synchronized double avgBytes() {
+    var value = new BigDecimal(bytes);
+    var scale = new BigDecimal(1024 * 1024);
+    var time = BigDecimal.valueOf((double) (System.currentTimeMillis() - startTime) / 1000);
+    if (time.doubleValue() == 0.0) return 0.0;
+    return value.divide(scale, 3, RoundingMode.UP).divide(time, 3, RoundingMode.UP).doubleValue();
   }
-  /** Set all attributes to default value */
-  public synchronized void reset() {
-    avgLatency = 0;
-    num = 0;
-    max = 0;
-    // 初始為最大的integer值
-    min = Long.MAX_VALUE;
-    bytes = 0;
+
+  /** @return total send/received bytes */
+  public synchronized long bytes() {
+    return bytes;
   }
 }
diff --git a/app/src/main/java/org/astraea/performance/Performance.java b/app/src/main/java/org/astraea/performance/Performance.java
index 340c4bb3..8142173d 100644
--- a/app/src/main/java/org/astraea/performance/Performance.java
+++ b/app/src/main/java/org/astraea/performance/Performance.java
@@ -5,15 +5,16 @@ import java.io.IOException;
 import java.time.Duration;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
+import org.apache.kafka.clients.producer.internals.DefaultPartitioner;
 import org.apache.kafka.common.errors.WakeupException;
 import org.astraea.argument.ArgumentUtil;
-import org.astraea.argument.BasicArgument;
+import org.astraea.argument.BasicArgumentWithPropFile;
 import org.astraea.concurrent.ThreadPool;
-import org.astraea.consumer.Builder;
 import org.astraea.consumer.Consumer;
 import org.astraea.producer.Producer;
 import org.astraea.topic.TopicAdmin;
@@ -53,7 +54,7 @@ public class Performance {
 
   public static void execute(final Argument param)
       throws InterruptedException, IOException, ExecutionException {
-    try (var topicAdmin = TopicAdmin.of(param.perfProps())) {
+    try (var topicAdmin = TopicAdmin.of(param.props())) {
       topicAdmin
           .creator()
           .numberOfReplicas(param.replicas)
@@ -75,6 +76,7 @@ public class Performance {
     var producerRecords = new AtomicLong(param.records);
     var tracker = new Tracker(producerMetrics, consumerMetrics, param.records);
     var groupId = "groupId-" + System.currentTimeMillis();
+    var getAssignment = new CountDownLatch(param.consumers);
     try (ThreadPool threadPool =
         ThreadPool.builder()
             .executors(
@@ -85,9 +87,9 @@ public class Performance {
                                 Consumer.builder()
                                     .brokers(param.brokers)
                                     .topics(Set.of(param.topic))
-                                    .offsetPolicy(Builder.OffsetPolicy.EARLIEST)
                                     .groupId(groupId)
-                                    .configs(param.perfProps())
+                                    .configs(param.props())
+                                    .consumerRebalanceListener(ignore -> getAssignment.countDown())
                                     .build(),
                                 consumerMetrics.get(i),
                                 consumerRecords))
@@ -97,10 +99,14 @@ public class Performance {
                     .mapToObj(
                         i ->
                             producerExecutor(
-                                Producer.of(param.perfProps()),
+                                Producer.builder()
+                                    .configs(param.props())
+                                    .partitionClassName(param.partitioner)
+                                    .build(),
                                 param,
                                 producerMetrics.get(i),
-                                producerRecords))
+                                producerRecords,
+                                getAssignment))
                     .collect(Collectors.toUnmodifiableList()))
             .executor(tracker)
             .build()) {
@@ -144,11 +150,17 @@ public class Performance {
   }
 
   static ThreadPool.Executor producerExecutor(
-      Producer<byte[], byte[]> producer, Argument param, Metrics metrics, AtomicLong records) {
+      Producer<byte[], byte[]> producer,
+      Argument param,
+      Metrics metrics,
+      AtomicLong records,
+      CountDownLatch getAssignment) {
     byte[] payload = new byte[param.recordSize];
     return new ThreadPool.Executor() {
       @Override
-      public State execute() {
+      public State execute() throws InterruptedException {
+        // Wait for all consumers get assignment.
+        getAssignment.await();
         var currentRecords = records.getAndDecrement();
         if (currentRecords <= 0) return State.DONE;
         long start = System.currentTimeMillis();
@@ -156,6 +168,7 @@ public class Performance {
             .sender()
             .topic(param.topic)
             .value(payload)
+            .timestamp(start)
             .run()
             .whenComplete(
                 (m, e) -> metrics.put(System.currentTimeMillis() - start, payload.length));
@@ -169,7 +182,7 @@ public class Performance {
     };
   }
 
-  static class Argument extends BasicArgument {
+  static class Argument extends BasicArgumentWithPropFile {
 
     @Parameter(
         names = {"--topic"},
@@ -215,13 +228,23 @@ public class Performance {
     int recordSize = 1024;
 
     @Parameter(
-        names = {"--prop.file"},
-        description = "String: path to the properties file",
+        names = {"--jmx.servers"},
+        description =
+            "String: server to get jmx metrics <jmx_server>@<broker_id>[,<jmx_server>@<broker_id>]*",
+        validateWith = ArgumentUtil.NotEmptyString.class)
+    String jmxServers = "";
+
+    @Parameter(
+        names = {"--partitioner"},
+        description = "String: the full class name of the desired partitioner",
         validateWith = ArgumentUtil.NotEmptyString.class)
-    String propFile;
+    String partitioner = DefaultPartitioner.class.getName();
 
-    public Map<String, Object> perfProps() {
-      return properties(propFile);
+    @Override
+    public Map<String, Object> props() {
+      Map<String, Object> prop = properties(propFile);
+      if (!this.jmxServers.isEmpty()) prop.put("jmx_servers", this.jmxServers);
+      return prop;
     }
   }
 }
diff --git a/app/src/main/java/org/astraea/performance/Tracker.java b/app/src/main/java/org/astraea/performance/Tracker.java
index 963451c5..3653faa5 100644
--- a/app/src/main/java/org/astraea/performance/Tracker.java
+++ b/app/src/main/java/org/astraea/performance/Tracker.java
@@ -17,55 +17,68 @@ public class Tracker implements ThreadPool.Executor {
 
   @Override
   public State execute() throws InterruptedException {
-    long completed = 0;
-    long bytes = 0L;
-    long max = 0;
-    long min = Long.MAX_VALUE;
+    if (logProducers() & logConsumers()) return State.DONE;
+    // 等待1秒，再抓資料輸出
+    Thread.sleep(1000);
+    return State.RUNNING;
+  }
+
+  private boolean logProducers() {
+    var completed = 0L;
+    var bytes = 0D;
+    var max = 0L;
+    var min = Long.MAX_VALUE;
 
-    /* producer */
     for (Metrics data : producerData) {
       completed += data.num();
-      bytes += data.bytesThenReset();
-      if (max < data.max()) max = data.max();
-      if (min > data.min()) min = data.min();
-    }
-    if (completed == 0) {
-      Thread.sleep(1000);
-      return State.RUNNING;
+      bytes += data.avgBytes();
+      max = Math.max(max, data.max());
+      min = Math.min(min, data.min());
     }
+
+    if (completed == 0) return false;
+
     System.out.printf("producers完成度: %.2f%%%n", ((double) completed * 100.0 / (double) records));
-    System.out.printf("  輸出%.3fMB/second%n", ((double) bytes / (1 << 20)));
+    System.out.printf("  輸出%.3fMB/second%n", bytes);
     System.out.println("  發送max latency: " + max + "ms");
     System.out.println("  發送mim latency: " + min + "ms");
     for (int i = 0; i < producerData.size(); ++i) {
+      System.out.printf(
+          "  producer[%d]的發送average bytes: %.3fMB%n", i, producerData.get(i).avgBytes());
       System.out.printf(
           "  producer[%d]的發送average latency: %.3fms%n", i, producerData.get(i).avgLatency());
     }
-    /* consumer */
-    completed = 0;
-    bytes = 0L;
-    max = 0;
-    min = Long.MAX_VALUE;
+    System.out.println("\n");
+    return completed >= records;
+  }
+
+  private boolean logConsumers() {
+    // there is no consumer, so we just complete this log.
+    if (consumerData.isEmpty()) return true;
+    var completed = 0;
+    var bytes = 0D;
+    var max = 0L;
+    var min = Long.MAX_VALUE;
     for (Metrics data : consumerData) {
       completed += data.num();
-      bytes += data.bytesThenReset();
-      if (max < data.max()) max = data.max();
-      if (min > data.min()) min = data.min();
+      bytes += data.avgBytes();
+      max = Math.max(max, data.max());
+      min = Math.min(min, data.min());
     }
+
+    if (completed == 0) return false;
+
     System.out.printf("consumer完成度: %.2f%%%n", ((double) completed * 100.0 / (double) records));
-    System.out.printf("  輸入%.3fMB/second%n", ((double) bytes / (1 << 20)));
+    System.out.printf("  輸入%.3fMB/second%n", bytes);
     System.out.println("  端到端max latency: " + max + "ms");
     System.out.println("  端到端mim latency: " + min + "ms");
     for (int i = 0; i < consumerData.size(); ++i) {
+      System.out.printf(
+          "  consumer[%d]的端到端average bytes: %.3fMB%n", i, consumerData.get(i).avgBytes());
       System.out.printf(
           "  consumer[%d]的端到端average latency: %.3fms%n", i, consumerData.get(i).avgLatency());
     }
-
     System.out.println("\n");
-    // The consumer has consumed the target records
-    if (completed >= records) return State.DONE;
-    // 等待1秒，再抓資料輸出
-    Thread.sleep(1000);
-    return State.RUNNING;
+    return completed >= records;
   }
 }
diff --git a/app/src/main/java/org/astraea/topic/Builder.java b/app/src/main/java/org/astraea/topic/Builder.java
index 1e62ff97..ef3e7a20 100644
--- a/app/src/main/java/org/astraea/topic/Builder.java
+++ b/app/src/main/java/org/astraea/topic/Builder.java
@@ -1,6 +1,7 @@
 package org.astraea.topic;
 
 import java.util.*;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 import org.apache.kafka.clients.admin.*;
 import org.apache.kafka.common.Node;
@@ -288,19 +289,27 @@ public class Builder {
 
     @Override
     public Creator creator() {
-      return new CreatorImpl(admin);
+      return new CreatorImpl(
+          admin, topic -> this.replicas(Set.of(topic)), topic -> topics().get(topic));
     }
   }
 
   private static class CreatorImpl implements Creator {
     private final Admin admin;
+    private final Function<String, Map<TopicPartition, List<Replica>>> replicasGetter;
+    private final Function<String, Map<String, String>> configsGetter;
     private String topic;
     private int numberOfPartitions = 1;
     private short numberOfReplicas = 1;
     private final Map<String, String> configs = new HashMap<>();
 
-    CreatorImpl(Admin admin) {
+    CreatorImpl(
+        Admin admin,
+        Function<String, Map<TopicPartition, List<Replica>>> replicasGetter,
+        Function<String, Map<String, String>> configsGetter) {
       this.admin = admin;
+      this.replicasGetter = replicasGetter;
+      this.configsGetter = configsGetter;
     }
 
     @Override
@@ -329,6 +338,48 @@ public class Builder {
 
     @Override
     public void create() {
+      if (Utils.handleException(() -> admin.listTopics().names().get()).contains(topic)) {
+        var partitionReplicas = replicasGetter.apply(topic);
+        partitionReplicas.forEach(
+            (tp, replicas) -> {
+              if (replicas.size() != numberOfReplicas)
+                throw new IllegalArgumentException(
+                    topic
+                        + " is existent but its replicas: "
+                        + replicas.size()
+                        + " is not equal to expected: "
+                        + numberOfReplicas);
+            });
+        var result =
+            Utils.handleException(() -> admin.describeTopics(Set.of(topic)).all().get().get(topic));
+        if (result.partitions().size() != numberOfPartitions)
+          throw new IllegalArgumentException(
+              topic
+                  + " is existent but its partitions: "
+                  + result.partitions().size()
+                  + " is not equal to expected: "
+                  + numberOfReplicas);
+
+        var actualConfigs = configsGetter.apply(topic);
+        this.configs.forEach(
+            (key, value) -> {
+              if (!actualConfigs.get(key).equals(value))
+                throw new IllegalArgumentException(
+                    topic
+                        + " is existent but its config: <"
+                        + key
+                        + ", "
+                        + actualConfigs.get(key)
+                        + "> is not equal to expected: "
+                        + key
+                        + ", "
+                        + value);
+            });
+
+        // ok, the existent topic is totally equal to what we want to create.
+        return;
+      }
+
       Utils.handleException(
           () ->
               admin
diff --git a/app/src/main/java/org/astraea/topic/ReplicaCollie.java b/app/src/main/java/org/astraea/topic/ReplicaCollie.java
index bf1cdd43..3ec87d6d 100644
--- a/app/src/main/java/org/astraea/topic/ReplicaCollie.java
+++ b/app/src/main/java/org/astraea/topic/ReplicaCollie.java
@@ -6,7 +6,7 @@ import java.util.*;
 import java.util.stream.Collectors;
 import org.apache.kafka.common.TopicPartition;
 import org.astraea.argument.ArgumentUtil;
-import org.astraea.argument.BasicAdminArgument;
+import org.astraea.argument.BasicArgumentWithPropFile;
 
 public class ReplicaCollie {
 
@@ -66,7 +66,7 @@ public class ReplicaCollie {
 
   public static void main(String[] args) throws IOException {
     var argument = ArgumentUtil.parseArgument(new Argument(), args);
-    try (var admin = TopicAdmin.of(argument.adminProps())) {
+    try (var admin = TopicAdmin.of(argument.props())) {
       execute(admin, argument)
           .forEach(
               (tp, assignments) ->
@@ -82,7 +82,7 @@ public class ReplicaCollie {
     }
   }
 
-  static class Argument extends BasicAdminArgument {
+  static class Argument extends BasicArgumentWithPropFile {
     @Parameter(
         names = {"--topics"},
         description = "Those topics' partitions will get reassigned. Empty menas all topics",
diff --git a/app/src/main/java/org/astraea/topic/TopicExplorer.java b/app/src/main/java/org/astraea/topic/TopicExplorer.java
index 3aaee868..db909c0b 100644
--- a/app/src/main/java/org/astraea/topic/TopicExplorer.java
+++ b/app/src/main/java/org/astraea/topic/TopicExplorer.java
@@ -5,7 +5,7 @@ import java.io.IOException;
 import java.util.*;
 import java.util.stream.Collectors;
 import org.astraea.argument.ArgumentUtil;
-import org.astraea.argument.BasicAdminArgument;
+import org.astraea.argument.BasicArgumentWithPropFile;
 
 public class TopicExplorer {
 
@@ -77,13 +77,13 @@ public class TopicExplorer {
 
   public static void main(String[] args) throws IOException {
     var argument = ArgumentUtil.parseArgument(new Argument(), args);
-    try (var admin = TopicAdmin.of(argument.adminProps())) {
+    try (var admin = TopicAdmin.of(argument.props())) {
       execute(admin, argument.topics.isEmpty() ? admin.topicNames() : argument.topics)
           .forEach(System.out::println);
     }
   }
 
-  static class Argument extends BasicAdminArgument {
+  static class Argument extends BasicArgumentWithPropFile {
     @Parameter(
         names = {"--topics"},
         description = "the topics to show all offset-related information. Empty means all topics",
diff --git a/docker/start_broker.sh b/docker/start_broker.sh
index f33c3cf8..8477b31e 100755
--- a/docker/start_broker.sh
+++ b/docker/start_broker.sh
@@ -23,7 +23,6 @@ function showHelp() {
   echo "    num.network.threads=10        set JVM memory"
   echo "    memory=\"-Xmx2G -Xms2G\"        set JVM memory"
 }
-
 # =====================================================================
 
 if [[ "$(which docker)" == "" ]]; then
@@ -35,7 +34,9 @@ if [[ -z "$KAFKA_VERSION" ]]; then
   KAFKA_VERSION=2.8.1
 fi
 
-USER=astraea
+kafka_user=astraea
+exporter_version="0.16.1"
+exporter_port="$(($(($RANDOM % 10000)) + 10000))"
 image_name=astraea/broker:$KAFKA_VERSION
 if [[ -n "$KAFKA_REVISION" ]]; then
   image_name=astraea/broker:$KAFKA_REVISION
@@ -44,6 +45,10 @@ broker_id="$(($RANDOM % 1000))"
 address=$(getAddress)
 broker_port="$(($(($RANDOM % 10000)) + 10000))"
 broker_jmx_port="$(($(($RANDOM % 10000)) + 10000))"
+admin_name="admin"
+admin_password="admin-secret"
+user_name="user"
+user_password="user-secret"
 jmx_opts="-Dcom.sun.management.jmxremote \
   -Dcom.sun.management.jmxremote.authenticate=false \
   -Dcom.sun.management.jmxremote.ssl=false \
@@ -60,7 +65,7 @@ while [[ $# -gt 0 ]]; do
     showHelp
     exit 2
   fi
-  echo "$1" >>"$config_file"
+  echo "$1" >> "$config_file"
   shift
 done
 
@@ -71,15 +76,31 @@ if [[ "$(cat $config_file | grep zookeeper.connect)" == "" ]]; then
 fi
 
 # set JVM heap
-KAFKA_HEAP="${KAFKA_HEAP:-"-Xmx2G -Xms2G"}"
+heap="${heap:-"-Xmx2G -Xms2G"}"
 
 # listeners will be generated automatically
 if [[ "$(cat $config_file | grep listeners)" != "" ]]; then
   echo "you should not define listeners"
   exit 2
 else
-  echo "listeners=PLAINTEXT://:9092" >>"$config_file"
-  echo "advertised.listeners=PLAINTEXT://${address}:$broker_port" >>"$config_file"
+  if [[ "$SASL" == "true" ]]; then
+    echo "listeners=SASL_PLAINTEXT://:9092" >> "$config_file"
+    echo "advertised.listeners=SASL_PLAINTEXT://${address}:$broker_port" >> "$config_file"
+    echo "security.inter.broker.protocol=SASL_PLAINTEXT" >> "$config_file"
+    echo "sasl.mechanism.inter.broker.protocol=PLAIN" >> "$config_file"
+    echo "sasl.enabled.mechanisms=PLAIN" >> "$config_file"
+    echo "listener.name.sasl_plaintext.plain.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
+             username="$admin_name" \
+             password="$admin_password" \
+             user_${admin_name}="$admin_password" \
+             user_${user_name}="${user_password}";" >> "$config_file"
+    echo "authorizer.class.name=kafka.security.authorizer.AclAuthorizer" >> "$config_file"
+    # allow brokers to communicate each other
+    echo "super.users=User:admin" >> "$config_file"
+  else
+    echo "listeners=PLAINTEXT://:9092" >>"$config_file"
+    echo "advertised.listeners=PLAINTEXT://${address}:$broker_port" >>"$config_file"
+  fi
 fi
 
 # log.dirs is not exposed so it should be generated automatically
@@ -87,7 +108,7 @@ if [[ "$(cat $config_file | grep log.dirs)" != "" ]]; then
   echo "you should not define log.dirs"
   exit 2
 else
-  echo "log.dirs=/tmp/kafka-logs" >>"$config_file"
+  echo "log.dirs=/tmp/kafka-logs" >> "$config_file"
 fi
 
 # auto-generate broker id if it does not exist
@@ -95,32 +116,32 @@ if [[ "$(cat $config_file | grep broker.id)" != "" ]]; then
   echo "you should not define broker.id"
   exit 2
 else
-  echo "broker.id=${broker_id}" >>"$config_file"
+  echo "broker.id=${broker_id}" >> "$config_file"
 fi
 
 # =============================[performance configs]=============================
 if [[ "$(cat $config_file | grep num.io.threads)" == "" ]]; then
-  echo "num.io.threads=8" >>"$config_file"
+  echo "num.io.threads=8" >> "$config_file"
 fi
 
 if [[ "$(cat $config_file | grep num.network.threads)" == "" ]]; then
-  echo "num.network.threads=8" >>"$config_file"
+  echo "num.network.threads=8" >> "$config_file"
 fi
 
 if [[ "$(cat $config_file | grep num.partitions)" == "" ]]; then
-  echo "num.partitions=8" >>"$config_file"
+  echo "num.partitions=8" >> "$config_file"
 fi
 
 if [[ "$(cat $config_file | grep transaction.state.log.replication.factor)" == "" ]]; then
-  echo "transaction.state.log.replication.factor=1" >>"$config_file"
+  echo "transaction.state.log.replication.factor=1" >> "$config_file"
 fi
 
 if [[ "$(cat $config_file | grep offsets.topic.replication.factor)" == "" ]]; then
-  echo "offsets.topic.replication.factor=1" >>"$config_file"
+  echo "offsets.topic.replication.factor=1" >> "$config_file"
 fi
 
 if [[ "$(cat $config_file | grep transaction.state.log.min.isr)" == "" ]]; then
-  echo "transaction.state.log.min.isr=1" >>"$config_file"
+  echo "transaction.state.log.min.isr=1" >> "$config_file"
 fi
 # ==============================================================================
 
@@ -133,19 +154,25 @@ FROM ubuntu:20.04
 RUN apt-get update && apt-get upgrade -y && DEBIAN_FRONTEND=noninteractive apt-get install -y openjdk-11-jdk wget git curl
 
 # add user
-RUN groupadd $USER && useradd -ms /bin/bash -g $USER $USER
+RUN groupadd $kafka_user && useradd -ms /bin/bash -g $kafka_user $kafka_user
 
 # change user
-USER $USER
+USER $kafka_user
+
+# download jmx exporter
+RUN mkdir /tmp/jmx_exporter
+WORKDIR /tmp/jmx_exporter
+RUN wget https://raw.githubusercontent.com/prometheus/jmx_exporter/master/example_configs/kafka-2_0_0.yml
+RUN wget https://repo1.maven.org/maven2/io/prometheus/jmx/jmx_prometheus_javaagent/${exporter_version}/jmx_prometheus_javaagent-${exporter_version}.jar
 
 # build kafka from source code
 RUN git clone https://github.com/apache/kafka /tmp/kafka
 WORKDIR /tmp/kafka
 RUN git checkout $KAFKA_REVISION
 RUN ./gradlew clean releaseTarGz
-RUN mkdir /home/$USER/kafka
-RUN tar -zxvf \$(find ./core/build/distributions/ -maxdepth 1 -type f -name kafka_*SNAPSHOT.tgz) -C /home/$USER/kafka --strip-components=1
-WORKDIR "/home/$USER/kafka"
+RUN mkdir /home/$kafka_user/kafka
+RUN tar -zxvf \$(find ./core/build/distributions/ -maxdepth 1 -type f -name kafka_*SNAPSHOT.tgz) -C /home/$kafka_user/kafka --strip-components=1
+WORKDIR "/home/$kafka_user/kafka"
 
 Dockerfile
 
@@ -157,31 +184,56 @@ FROM ubuntu:20.04
 RUN apt-get update && apt-get upgrade -y && apt-get install -y openjdk-11-jdk wget
 
 # add user
-RUN groupadd $USER && useradd -ms /bin/bash -g $USER $USER
+RUN groupadd $kafka_user && useradd -ms /bin/bash -g $kafka_user $kafka_user
 
 # change user
-USER $USER
+USER $kafka_user
+
+# download jmx exporter
+RUN mkdir /tmp/jmx_exporter
+WORKDIR /tmp/jmx_exporter
+RUN wget https://raw.githubusercontent.com/prometheus/jmx_exporter/master/example_configs/kafka-2_0_0.yml
+RUN wget https://repo1.maven.org/maven2/io/prometheus/jmx/jmx_prometheus_javaagent/${exporter_version}/jmx_prometheus_javaagent-${exporter_version}.jar
 
 # download kafka
 WORKDIR /tmp
 RUN wget https://archive.apache.org/dist/kafka/${KAFKA_VERSION}/kafka_2.13-${KAFKA_VERSION}.tgz
-RUN mkdir /home/$USER/kafka
-RUN tar -zxvf kafka_2.13-${KAFKA_VERSION}.tgz -C /home/$USER/kafka --strip-components=1
-WORKDIR "/home/$USER/kafka"
+RUN mkdir /home/$kafka_user/kafka
+RUN tar -zxvf kafka_2.13-${KAFKA_VERSION}.tgz -C /home/$kafka_user/kafka --strip-components=1
+WORKDIR "/home/$kafka_user/kafka"
 
 Dockerfile
 fi
 
 docker run -d \
-  -e KAFKA_HEAP_OPTS="$KAFKA_HEAP" \
+  -e KAFKA_HEAP_OPTS="$heap" \
   -e KAFKA_JMX_OPTS="$jmx_opts" \
+  -e KAFKA_OPTS="-javaagent:/tmp/jmx_exporter/jmx_prometheus_javaagent-${exporter_version}.jar=$exporter_port:/tmp/jmx_exporter/kafka-2_0_0.yml" \
   -v $config_file:/tmp/broker.properties:ro \
   -p $broker_port:9092 \
   -p $broker_jmx_port:$broker_jmx_port \
+  -p $exporter_port:$exporter_port \
   $image_name ./bin/kafka-server-start.sh /tmp/broker.properties
 
 echo "================================================="
 echo "broker address: ${address}:$broker_port"
 echo "jmx address: ${address}:$broker_jmx_port"
+echo "exporter address: ${address}:$exporter_port"
 echo "broker id: $broker_id"
+if [[ "$SASL" == "true" ]]; then
+  user_jaas_file=/tmp/user-jaas-${broker_port}.conf
+  echo "
+  sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username=${user_name} password=${user_password};
+  security.protocol=SASL_PLAINTEXT
+  sasl.mechanism=PLAIN
+  " > $user_jaas_file
+
+  admin_jaas_file=/tmp/admin-jaas-${broker_port}.conf
+  echo "
+  sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username=${admin_name} password=${admin_password};
+  security.protocol=SASL_PLAINTEXT
+  sasl.mechanism=PLAIN
+  " > $admin_jaas_file
+  echo "SASL_PLAINTEXT is enabled. user config: $user_jaas_file admin config: $admin_jaas_file"
+fi
 echo "================================================="
diff --git a/docker/start_kafka_tool.sh b/docker/start_kafka_tool.sh
index 0baf57ad..7b68c854 100755
--- a/docker/start_kafka_tool.sh
+++ b/docker/start_kafka_tool.sh
@@ -1,8 +1,10 @@
 #!/bin/bash
 
+# =============================[functions]=============================
 function showHelp() {
   echo "Usage: [ KAFKA SCRIPT ] [ OPTIONS ]"
 }
+# =====================================================================
 
 if [[ "$(which docker)" == "" ]]; then
   echo "you have to install docker"
@@ -26,7 +28,7 @@ image_name=astraea/kafka-tool:$KAFKA_VERSION
 # set JVM heap
 KAFKA_HEAP="${KAFKA_HEAP:-"-Xmx2G -Xms2G"}"
 
-USER=astraea
+kafka_user=astraea
 
 docker build -t $image_name - <<Dockerfile
 FROM ubuntu:20.04
@@ -35,17 +37,17 @@ FROM ubuntu:20.04
 RUN apt-get update && apt-get upgrade -y && apt-get install -y openjdk-11-jdk wget
 
 # add user
-RUN groupadd $USER && useradd -ms /bin/bash -g $USER $USER
+RUN groupadd $kafka_user && useradd -ms /bin/bash -g $kafka_user $kafka_user
 
 # change user
-USER $USER
+USER $kafka_user
 
 # download kafka
 WORKDIR /tmp
 RUN wget https://archive.apache.org/dist/kafka/${KAFKA_VERSION}/kafka_2.13-${KAFKA_VERSION}.tgz
-RUN mkdir /home/$USER/kafka
-RUN tar -zxvf kafka_2.13-${KAFKA_VERSION}.tgz -C /home/$USER/kafka --strip-components=1
-WORKDIR "/home/$USER/kafka"
+RUN mkdir /home/$kafka_user/kafka
+RUN tar -zxvf kafka_2.13-${KAFKA_VERSION}.tgz -C /home/$kafka_user/kafka --strip-components=1
+WORKDIR "/home/$kafka_user/kafka"
 
 Dockerfile
 
diff --git a/docker/start_prometheus.sh b/docker/start_prometheus.sh
new file mode 100755
index 00000000..5ef1a9be
--- /dev/null
+++ b/docker/start_prometheus.sh
@@ -0,0 +1,70 @@
+#!/bin/bash
+
+# =============================[functions]=============================
+function getAddress() {
+  if [[ "$(which ipconfig)" != "" ]]; then
+    address=$(ipconfig getifaddr en0)
+  else
+    address=$(hostname -i)
+  fi
+  if [[ "$address" == "127.0.0.1" ]]; then
+    echo "the address: 127.0.0.1 can't be used in this script. Please check /etc/hosts"
+    exit 2
+  fi
+  echo "$address"
+}
+
+function showHelp() {
+  echo "Usage: [targets]"
+}
+# =====================================================================
+
+if [[ "$(which docker)" == "" ]]; then
+  echo "you have to install docker"
+  exit 2
+fi
+
+if [[ "$1" == "" ]]; then
+  showHelp
+  exit 2
+fi
+
+targets=""
+IFS=',' read -ra ADDR <<< "$1"
+for i in "${ADDR[@]}"; do
+  if [[ "$targets" == "" ]]; then
+    targets="'$i'"
+  else
+    targets="$targets,'$i'"
+  fi
+done
+echo "$targets"
+
+image_name=prom/prometheus
+prometheus_port="$(($(($RANDOM % 10000)) + 10000))"
+file=/tmp/prometheus-${prometheus_port}.yml
+address=$(getAddress)
+
+cat <<EOT > "$file"
+global:
+  scrape_interval: 15s
+  external_labels:
+    monitor: 'kafka-monitor'
+
+scrape_configs:
+  - job_name: 'brokers'
+    scrape_interval: 5s
+    static_configs:
+      - targets: [$targets]
+
+EOT
+
+docker run -d \
+    -p $prometheus_port:9090 \
+    -v $file:/etc/prometheus/prometheus.yml \
+    prom/prometheus
+
+echo "================================================="
+echo "config file: $file"
+echo "prometheus address: http://${address}:$prometheus_port"
+echo "================================================="
\ No newline at end of file
diff --git a/docker/start_zookeeper.sh b/docker/start_zookeeper.sh
index 7c3ed789..191ff457 100755
--- a/docker/start_zookeeper.sh
+++ b/docker/start_zookeeper.sh
@@ -24,7 +24,7 @@ if [[ -z "$ZOOKEEPER_VERSION" ]]; then
   ZOOKEEPER_VERSION=3.6.3
 fi
 
-USER=astraea
+zookeeper_user=astraea
 image_name=astraea/zookeeper:$ZOOKEEPER_VERSION
 zk_port="$(($(($RANDOM % 10000)) + 10000))"
 address=$(getAddress)
@@ -36,17 +36,17 @@ FROM ubuntu:20.04
 RUN apt-get update && apt-get upgrade -y && apt-get install -y openjdk-11-jdk wget
 
 # add user
-RUN groupadd $USER && useradd -ms /bin/bash -g $USER $USER
+RUN groupadd $zookeeper_user && useradd -ms /bin/bash -g $zookeeper_user $zookeeper_user
 
 # change user
-USER $USER
+USER $zookeeper_user
 
 # download zookeeper
 WORKDIR /tmp
 RUN wget https://archive.apache.org/dist/zookeeper/zookeeper-${ZOOKEEPER_VERSION}/apache-zookeeper-${ZOOKEEPER_VERSION}-bin.tar.gz
-RUN mkdir /home/$USER/zookeeper
-RUN tar -zxvf apache-zookeeper-${ZOOKEEPER_VERSION}-bin.tar.gz -C /home/$USER/zookeeper --strip-components=1
-WORKDIR /home/$USER/zookeeper
+RUN mkdir /home/$zookeeper_user/zookeeper
+RUN tar -zxvf apache-zookeeper-${ZOOKEEPER_VERSION}-bin.tar.gz -C /home/$zookeeper_user/zookeeper --strip-components=1
+WORKDIR /home/$zookeeper_user/zookeeper
 
 # create config file
 RUN echo "tickTime=2000" >> ./conf/zoo.cfg


Test patch:
diff --git a/app/src/test/java/org/astraea/consumer/RebalanceListenerTest.java b/app/src/test/java/org/astraea/consumer/RebalanceListenerTest.java
new file mode 100644
index 00000000..ce50a8d6
--- /dev/null
+++ b/app/src/test/java/org/astraea/consumer/RebalanceListenerTest.java
@@ -0,0 +1,34 @@
+package org.astraea.consumer;
+
+import java.time.Duration;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.astraea.Utils;
+import org.astraea.concurrent.ThreadPool;
+import org.astraea.service.RequireBrokerCluster;
+import org.junit.jupiter.api.Test;
+
+public class RebalanceListenerTest extends RequireBrokerCluster {
+  @Test
+  void testConsumerRebalanceListener() {
+    var getAssignment = new AtomicInteger(0);
+    var topicName = "testRebalanceListener-" + System.currentTimeMillis();
+    try (var consumer =
+        Consumer.builder()
+            .brokers(bootstrapServers())
+            .topics(Set.of(topicName))
+            .consumerRebalanceListener(ignore -> getAssignment.incrementAndGet())
+            .build()) {
+      try (ThreadPool threadPool =
+          ThreadPool.builder()
+              .executor(
+                  () -> {
+                    consumer.poll(Duration.ofSeconds(10));
+                    return ThreadPool.Executor.State.DONE;
+                  })
+              .build()) {
+        Utils.waitFor(() -> getAssignment.get() == 1, Duration.ofSeconds(10));
+      }
+    }
+  }
+}
diff --git a/app/src/test/java/org/astraea/metrics/MetricExplorerTest.java b/app/src/test/java/org/astraea/metrics/MetricExplorerTest.java
new file mode 100644
index 00000000..7aa324c6
--- /dev/null
+++ b/app/src/test/java/org/astraea/metrics/MetricExplorerTest.java
@@ -0,0 +1,123 @@
+package org.astraea.metrics;
+
+import static org.astraea.metrics.MetricExplorer.*;
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import java.io.ByteArrayOutputStream;
+import java.io.PrintStream;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import org.astraea.argument.ArgumentUtil;
+import org.astraea.metrics.jmx.BeanObject;
+import org.astraea.metrics.jmx.MBeanClient;
+import org.junit.jupiter.api.function.Executable;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+class MetricExplorerTest {
+
+  @ParameterizedTest
+  @CsvSource(
+      delimiterString = "->",
+      value = {
+        "service:jmx:rmi://localhost:1234/jndi/rmi://localhost:2344/jmxrmi     -> match",
+        "service:jmx:rmi:///jndi/rmi://:9999/jmxrmi                            -> match",
+        "service:jmx:rmi://myhost/jndi/rmi://myhost:1099/myhost/myjmxconnector -> match",
+        "service:jmx:rmi:///jndi/rmi://localhost:9987/jmxrmi                   -> match",
+        "aaaaaaaaaaaaaaaaaaa                                                   -> no-match",
+        "bbbbbbbbbbbbbbbbbbb                                                   -> no-match",
+      })
+  void testPatternJmxUrlStart(String url, String expected) {
+    // arrange
+    Pattern pattern = Argument.JmxServerUrlConverter.patternOfJmxUrlStart;
+
+    // act
+    Matcher matcher = pattern.matcher(url);
+
+    // assert
+    assertEquals(expected, matcher.find() ? "match" : "no-match");
+  }
+
+  @ParameterizedTest
+  @CsvSource(
+      delimiterString = "->",
+      value = {
+        "type=Memory                      -> type            -> Memory",
+        "type=OperatingSystem             -> type            -> OperatingSystem",
+        "key=value                        -> key             -> value",
+        "logDirectory=\"/tmp/kafka-logs\" -> logDirectory    -> \"/tmp/kafka-logs\"",
+        "face=\"= =\"                     -> face            -> \"= =\"",
+      })
+  void testPatternProperty(String property, String key, String value) {
+    // arrange
+    Pattern pattern = Argument.CorrectPropertyFormat.propertyPattern;
+
+    // act
+    Matcher matcher = pattern.matcher(property);
+
+    // assert
+    assertTrue(matcher.find());
+    assertEquals(key, matcher.group("key"));
+    assertEquals(value, matcher.group("value"));
+  }
+
+  @ParameterizedTest
+  @ValueSource(
+      strings = {
+        "--jmx.server localhost:5566 --domain example.com",
+        "--jmx.server localhost:5566 --domain example.com --view-object-name-list",
+      })
+  void executeDoesPrintSomething(String args) {
+    // arrange
+    var argument = ArgumentUtil.parseArgument(new Argument(), args.split(" "));
+    var mockMBeanClient = mock(MBeanClient.class);
+    when(mockMBeanClient.queryBeans(any()))
+        .thenReturn(List.of(new BeanObject("example.com", Map.of("key", "value"), Map.of())));
+    var stdout = System.out;
+    var mockOutput = new ByteArrayOutputStream();
+    System.setOut(new PrintStream(mockOutput));
+
+    // act
+    Executable executable = () -> MetricExplorer.execute(mockMBeanClient, argument);
+
+    // assert
+    assertDoesNotThrow(executable);
+    assertTrue(mockOutput.toString().contains("example.com"));
+    assertTrue(mockOutput.toString().contains("key"));
+    assertTrue(mockOutput.toString().contains("value"));
+
+    // restore
+    System.setOut(stdout);
+  }
+
+  @ParameterizedTest
+  @CsvSource(
+      delimiterString = "(is",
+      value = {
+        "--jmx.server localhost:5566                                       (is ok",
+        "--jmx.server localhost:5566 --domain kafka.log                    (is ok",
+        "--jmx.server localhost:5566 --property type=Memory                (is ok",
+        "--jmx.server localhost:5566 --strict-match --property type=Memory (is ok",
+        "--jmx.server localhost:5566 --view-object-name-list               (is ok",
+        "wuewuewuewue                                                      (is not ok",
+        "--view-object-name-list                                           (is not ok",
+      })
+  void ensureArgumentWorking(String argumentString, String outcome) {
+    // arrange
+    String[] arguments = argumentString.split(" ");
+
+    // act
+    Executable doParsing = () -> ArgumentUtil.parseArgument(new Argument(), arguments);
+
+    // assert
+    if (outcome.equals("ok")) {
+      assertDoesNotThrow(doParsing);
+    } else if (outcome.equals("not ok")) {
+      assertThrows(Exception.class, doParsing);
+    }
+  }
+}
diff --git a/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricClientAppTest.java b/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricClientAppTest.java
deleted file mode 100644
index a8d7ac5c..00000000
--- a/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricClientAppTest.java
+++ /dev/null
@@ -1,33 +0,0 @@
-package org.astraea.metrics.kafka;
-
-import static org.junit.jupiter.api.Assertions.*;
-
-import com.beust.jcommander.ParameterException;
-import java.io.OutputStream;
-import java.io.PrintStream;
-import org.astraea.metrics.KafkaMetricClientApp;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-
-class KafkaMetricClientAppTest {
-
-  private final PrintStream stderr = System.err;
-
-  @BeforeEach
-  void beforeEach() {
-    // swallow stderr output to null stream, avoid help() info from KafkaMetricClientApp
-    System.setErr(new PrintStream(OutputStream.nullOutputStream()));
-  }
-
-  @AfterEach
-  void afterEach() {
-    // restore stderr stream
-    System.setErr(stderr);
-  }
-
-  @Test
-  void testBadArguments() {
-    assertThrows(ParameterException.class, () -> KafkaMetricClientApp.main(new String[0]));
-  }
-}
diff --git a/app/src/test/java/org/astraea/performance/MetricsTest.java b/app/src/test/java/org/astraea/performance/MetricsTest.java
index 940b4164..103f1769 100644
--- a/app/src/test/java/org/astraea/performance/MetricsTest.java
+++ b/app/src/test/java/org/astraea/performance/MetricsTest.java
@@ -1,13 +1,11 @@
 package org.astraea.performance;
 
 import java.util.Random;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.LongAdder;
-import org.astraea.concurrent.ThreadPool;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 public class MetricsTest {
+
   @Test
   void testAverage() {
     Random rand = new Random();
@@ -20,55 +18,18 @@ public class MetricsTest {
     for (int i = 0; i < num; ++i) {
       long next = rand.nextInt();
       avg += ((double) next - avg) / (i + 1);
-      metrics.putLatency(next);
+      metrics.put(next, 0);
     }
 
     Assertions.assertEquals(avg, metrics.avgLatency());
   }
 
-  // Simultaneously add and get
   @Test
   void testBytes() {
-    final Metrics metrics = new Metrics();
-    final LongAdder longAdder = new LongAdder();
-    final AtomicInteger adderCount = new AtomicInteger(0);
-    final AtomicInteger getterCount = new AtomicInteger(0);
-    final long input = 100;
-    final int loopCount = 10000;
-
-    try (ThreadPool threadPool =
-        ThreadPool.builder()
-            .executor(
-                () -> {
-                  metrics.addBytes(input);
-                  if (adderCount.incrementAndGet() < loopCount)
-                    return ThreadPool.Executor.State.RUNNING;
-                  else return ThreadPool.Executor.State.DONE;
-                })
-            .executor(
-                () -> {
-                  longAdder.add(metrics.bytesThenReset());
-                  if (getterCount.incrementAndGet() < loopCount)
-                    return ThreadPool.Executor.State.RUNNING;
-                  else return ThreadPool.Executor.State.DONE;
-                })
-            .build()) {
-      threadPool.waitAll();
-    }
-    longAdder.add(metrics.bytesThenReset());
-
-    Assertions.assertEquals(loopCount * input, longAdder.sum());
-  }
+    var metrics = new Metrics();
 
-  @Test
-  public void testReset() {
-    final Metrics metrics = new Metrics();
-    metrics.addBytes(10);
-    metrics.putLatency(11);
-
-    metrics.reset();
-
-    Assertions.assertEquals(0, metrics.bytesThenReset());
-    Assertions.assertEquals(0, metrics.avgLatency());
+    Assertions.assertEquals(0, metrics.bytes());
+    metrics.put(0, 1000);
+    Assertions.assertEquals(1000, metrics.bytes());
   }
 }
diff --git a/app/src/test/java/org/astraea/performance/PerformanceTest.java b/app/src/test/java/org/astraea/performance/PerformanceTest.java
index 8514dc9f..33b9b39c 100644
--- a/app/src/test/java/org/astraea/performance/PerformanceTest.java
+++ b/app/src/test/java/org/astraea/performance/PerformanceTest.java
@@ -1,6 +1,7 @@
 package org.astraea.performance;
 
 import java.util.Set;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicLong;
 import org.astraea.Utils;
@@ -23,11 +24,12 @@ public class PerformanceTest extends RequireBrokerCluster {
             Producer.builder().brokers(bootstrapServers()).build(),
             param,
             metrics,
-            new AtomicLong(10))) {
+            new AtomicLong(10),
+            new CountDownLatch(0))) {
       executor.execute();
 
       Utils.waitFor(() -> metrics.num() == 1);
-      Assertions.assertEquals(1024, metrics.bytesThenReset());
+      Assertions.assertEquals(1024, metrics.bytes());
     }
   }
 
@@ -43,7 +45,7 @@ public class PerformanceTest extends RequireBrokerCluster {
       executor.execute();
 
       Assertions.assertEquals(0, metrics.num());
-      Assertions.assertEquals(0, metrics.bytesThenReset());
+      Assertions.assertEquals(0, metrics.bytes());
 
       try (var producer = Producer.builder().brokers(bootstrapServers()).build()) {
         producer.sender().topic(topicName).value(new byte[1024]).run().toCompletableFuture().get();
@@ -51,7 +53,7 @@ public class PerformanceTest extends RequireBrokerCluster {
       executor.execute();
 
       Assertions.assertEquals(1, metrics.num());
-      Assertions.assertNotEquals(1024, metrics.bytesThenReset());
+      Assertions.assertNotEquals(1024, metrics.bytes());
     }
   }
 }
diff --git a/app/src/test/java/org/astraea/performance/TrackerTest.java b/app/src/test/java/org/astraea/performance/TrackerTest.java
new file mode 100644
index 00000000..fcc3aa14
--- /dev/null
+++ b/app/src/test/java/org/astraea/performance/TrackerTest.java
@@ -0,0 +1,31 @@
+package org.astraea.performance;
+
+import java.util.List;
+import org.astraea.concurrent.ThreadPool;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TrackerTest {
+  @Test
+  public void testTerminate() throws InterruptedException {
+    var producerData = List.of(new Metrics());
+    var consumerData = List.of(new Metrics());
+    List<Metrics> empty = List.of();
+    int records = 1;
+
+    try (Tracker tracker = new Tracker(producerData, consumerData, records)) {
+      Assertions.assertEquals(ThreadPool.Executor.State.RUNNING, tracker.execute());
+      producerData.get(0).put(1, 1);
+      consumerData.get(0).put(1, 1);
+      Assertions.assertEquals(ThreadPool.Executor.State.DONE, tracker.execute());
+    }
+
+    // Zero consumer
+    producerData = List.of(new Metrics());
+    try (Tracker tracker = new Tracker(producerData, empty, records)) {
+      Assertions.assertEquals(ThreadPool.Executor.State.RUNNING, tracker.execute());
+      producerData.get(0).put(1, 1);
+      Assertions.assertEquals(ThreadPool.Executor.State.DONE, tracker.execute());
+    }
+  }
+}
diff --git a/app/src/test/java/org/astraea/topic/TopicAdminTest.java b/app/src/test/java/org/astraea/topic/TopicAdminTest.java
index 166ccf22..f94f4b50 100644
--- a/app/src/test/java/org/astraea/topic/TopicAdminTest.java
+++ b/app/src/test/java/org/astraea/topic/TopicAdminTest.java
@@ -7,6 +7,7 @@ import java.util.Set;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 import org.apache.kafka.common.config.TopicConfig;
 import org.astraea.Utils;
 import org.astraea.producer.Producer;
@@ -25,8 +26,53 @@ public class TopicAdminTest extends RequireBrokerCluster {
           .topic(topicName)
           .configs(Map.of(TopicConfig.COMPRESSION_TYPE_CONFIG, "lz4"))
           .create();
-      Assertions.assertEquals(
-          "lz4", topicAdmin.topics().get(topicName).get(TopicConfig.COMPRESSION_TYPE_CONFIG));
+      Utils.waitFor(
+          () ->
+              topicAdmin
+                  .topics()
+                  .getOrDefault(topicName, Map.of())
+                  .get(TopicConfig.COMPRESSION_TYPE_CONFIG)
+                  .equals("lz4"));
+    }
+  }
+
+  @Test
+  void testCreateTopicRepeatedly() throws IOException {
+    var topicName = "testCreateTopicRepeatedly";
+    try (var topicAdmin = TopicAdmin.of(bootstrapServers())) {
+      Runnable createTopic =
+          () ->
+              topicAdmin
+                  .creator()
+                  .configs(Map.of(TopicConfig.COMPRESSION_TYPE_CONFIG, "lz4"))
+                  .numberOfReplicas((short) 1)
+                  .numberOfPartitions(3)
+                  .topic(topicName)
+                  .create();
+
+      createTopic.run();
+      Utils.waitFor(() -> topicAdmin.topics().containsKey(topicName));
+      IntStream.range(0, 10).forEach(i -> createTopic.run());
+
+      // changing number of partitions can producer error
+      Assertions.assertThrows(
+          IllegalArgumentException.class,
+          () -> topicAdmin.creator().numberOfPartitions(1).topic(topicName).create());
+
+      // changing number of replicas can producer error
+      Assertions.assertThrows(
+          IllegalArgumentException.class,
+          () -> topicAdmin.creator().numberOfReplicas((short) 2).topic(topicName).create());
+
+      // changing config can producer error
+      Assertions.assertThrows(
+          IllegalArgumentException.class,
+          () ->
+              topicAdmin
+                  .creator()
+                  .configs(Map.of(TopicConfig.COMPRESSION_TYPE_CONFIG, "gzip"))
+                  .topic(topicName)
+                  .create());
     }
   }