Repository: skiptests/astraea
Pull request #15 (2021-09-11 05:56:41)

Implemetation of MBean client
Context: This PR is related to task 1 in #10 

This PR implements two things.

1. ``BeanObject``, represent an MBean at the remote MBean server.
2. ``MBeanClient``, the client we used to fetch & query MBean from an MBean server.

### Code Example

```java
package org.astraea;

import java.util.Set;
import javax.management.remote.JMXServiceURL;
import org.astraea.metrics.jmx.BeanObject;
import org.astraea.metrics.jmx.BeanQuery;
import org.astraea.metrics.jmx.MBeanClient;

public class Ignore {
  public static String JMX_URI_FORMAT =
      "service:jmx:rmi:///jndi/rmi://" + "%s" + ":" + "%d" + "/jmxrmi";

  public static String createJmxUrl(String hostname, String port) {
    return String.format(JMX_URI_FORMAT, hostname, Integer.parseInt(port));
  }

  public static void main(String[] args) throws Exception {
    JMXServiceURL jmxServiceURL = new JMXServiceURL(createJmxUrl("localhost", "19835"));
    MBeanClient client = new MBeanClient(jmxServiceURL);
    client.connect();

    BeanObject fetchResult =
        client.fetchObjectAttribute(
            BeanObject.fromDomainName("java.lang")
                .selectProperty("type", "Memory")
                .fetchAttribute("HeapMemoryUsage")
                .fetchAttribute("NonHeapMemoryUsage"));

    fetchResult
        .getAttributeView()
        .forEach(
            (key, value) -> {
              System.out.printf("%s: %s%n", key, value);
            });

    Set<BeanObject> beanObjects =
        client.queryObject(
            BeanQuery.forDomainName("kafka.network")
                .whereProperty("type", "RequestMetrics")
                .whereProperty("request", "Delete*")
                .whereProperty("name", "LocalTimeMs"));

    System.out.printf("%nQuery Result:%n");
    beanObjects.forEach(
        bean -> {
          System.out.printf("%s%n", bean.objectName());
        });

    client.close();
  }
}
```

```
HeapMemoryUsage: javax.management.openmbean.CompositeDataSupport(compositeType=javax.management.openmbean.CompositeType(name=java.lang.management.MemoryUsage,items=((itemName=committed,itemType=javax.management.openmbean.SimpleType(name=java.lang.Long)),(itemName=init,itemType=javax.management.openmbean.SimpleType(name=java.lang.Long)),(itemName=max,itemType=javax.management.openmbean.SimpleType(name=java.lang.Long)),(itemName=used,itemType=javax.management.openmbean.SimpleType(name=java.lang.Long)))),contents={committed=2147483648, init=2147483648, max=2147483648, used=242296752})
NonHeapMemoryUsage: javax.management.openmbean.CompositeDataSupport(compositeType=javax.management.openmbean.CompositeType(name=java.lang.management.MemoryUsage,items=((itemName=committed,itemType=javax.management.openmbean.SimpleType(name=java.lang.Long)),(itemName=init,itemType=javax.management.openmbean.SimpleType(name=java.lang.Long)),(itemName=max,itemType=javax.management.openmbean.SimpleType(name=java.lang.Long)),(itemName=used,itemType=javax.management.openmbean.SimpleType(name=java.lang.Long)))),contents={committed=100085760, init=7667712, max=-1, used=92189200})

Query Result:
kafka.network:name=LocalTimeMs,request=DeleteTopics,type=RequestMetrics
kafka.network:name=LocalTimeMs,request=DeleteAcls,type=RequestMetrics
kafka.network:name=LocalTimeMs,request=DeleteRecords,type=RequestMetrics
kafka.network:name=LocalTimeMs,request=DeleteGroups,type=RequestMetrics
```

chia7712: The example LGTM. Could you add it to `App`? It would be nice that we have a way to show (raw) metrics of brokers.
garyparrot: I have updated the code example at issue https://github.com/skiptests/astraea/pull/15#issue-731879543.

The new code example demonstrates the new way to operate with the immutable design ``BeanObject``. Also, the ability to query MBean has been added to the PR. The code example shows the usage.

There are still two tasks to do 
- [ ] build a JMX server locally and test all getter by local MBean server.
- [ ] Add a small utility to App? A way to show (raw) metrics of brokers.

I might be off for the issue for a while, I have other battles to fight.
chia7712: > build a JMX server locally and test all getter by local MBean server.

等你有空把測試補完這隻PR才能審視or合併

> Add a small utility to App? A way to show (raw) metrics of brokers.

這可以作為另一隻PR

可以先花點時間看書看點資料，然後再回頭完善這隻PR (尤其是測試)

Patch:
diff --git a/README.md b/README.md
index f775c5c4..254aa989 100644
--- a/README.md
+++ b/README.md
@@ -6,15 +6,16 @@ a collection of tools used to balance Kafka data
 - Yi-Chen   Wang <warren215215@gmail.com>
 - Ching-Hong Fang <fjh7777@gmail.com>
 - Zheng-Xian Li <garyparrottt@gmail.com>
+- Xiang-Jun Sun <sean0651101@gmail.com>
 
-# Quickstart
+# Quickstart a Kafka Env
 
 There are two scripts which can setup env quickly by container
 
-## Set up zookeeper env
+## Set up zookeeper with default version
 
 ```shell
-./docker/start_zk.sh
+./docker/start_zookeeper.sh
 ```
 
 The above script creates a zookeeper instance by container. Also, it will show the command used to add broker instance. For example:
@@ -25,9 +26,15 @@ run ./docker/start_broker.sh zookeeper.connect=192.168.50.178:17228 to join kafk
 =================================================
 ```
 
-## Set up (kafka) broker env
+## Set up zookeeper with specific version
 
-After the zk env is running, you can copy the command (see above example) from zk script output to setup kafka. For example:
+```shell
+ZOOKEEPER_VERSION=3.6.3 ./docker/start_zookeeper.sh
+```
+
+## Set up (kafka) broker with default version
+
+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
 ./docker/start_broker.sh zookeeper.connect=192.168.50.178:17228
 ```
@@ -41,4 +48,38 @@ jmx address: 192.168.50.224:15905
 =================================================
 ```
 
-Noted that the command to set up broker can be executed multiple times to create a broker cluster.
\ No newline at end of file
+Noted that the command to set up broker can be executed multiple times to create a broker cluster.
+
+## Set up (kafka) broker with specific version
+
+```shell
+KAFKA_VERSION=2.8.0 ./docker/start_broker.sh zookeeper.connect=192.168.50.178:17228
+```
+
+# Kafka Tools
+
+This project offers many kafka tools to simplify the life for kafka users.
+
+## Latency Benchmark
+
+This tool is used to test following latencies.
+1. producer latency: the time of completing producer data request
+2. E2E latency: the time for a record to travel through Kafka
+
+Run the benchmark from source code
+```shell
+./gradlew run --args="Latency --bootstrap.servers 192.168.50.224:18878"
+```
+
+Run the benchmark from release
+```shell
+./bin/App Latency --bootstrap.servers 192.168.50.224:18878
+```
+
+### Latency Benchmark Configurations
+1. --bootstrap.servers: the brokers addresses
+2. --consumers: the number of consumers (threads). Default: 1
+3. --producers: the number of producers (threads). Default: 1
+4. --valueSize: the size of record value. Default: 100 bytes
+5. --duration: the duration to run this benchmark. Default: 5 seconds
+6. --flushDuration: the duration to flush producer records. Default: 2 seconds
diff --git a/app/build.gradle b/app/build.gradle
index 3e14b024..2a9bd8e8 100644
--- a/app/build.gradle
+++ b/app/build.gradle
@@ -16,17 +16,21 @@ repositories {
     mavenCentral()
 }
 
+def versions = [
+    kafka: project.properties['kafka.version'] ?: "2.8.0",
+    junit: project.properties['junit.version'] ?: "5.7.2",
+]
+
 dependencies {
-    // Use JUnit Jupiter for testing.
-    testImplementation 'org.junit.jupiter:junit-jupiter:5.7.2'
+    testImplementation "org.junit.jupiter:junit-jupiter:${versions["junit"]}"
 
-    // This dependency is used by the application.
-    implementation 'org.apache.kafka:kafka-clients:2.8.0'
+    implementation "org.apache.kafka:kafka-clients:${versions["kafka"]}"
 }
 
 application {
     // Define the main class for the application.
     mainClass = 'org.astraea.App'
+    applicationDefaultJvmArgs = ["-server", "-XX:+UseG1GC", "-Djava.awt.headless=true", "-Xms4G", "-Xmx4G"]
 }
 
 java {
diff --git a/app/src/main/java/org/astraea/App.java b/app/src/main/java/org/astraea/App.java
index eb8dd5d6..cf2d1fae 100644
--- a/app/src/main/java/org/astraea/App.java
+++ b/app/src/main/java/org/astraea/App.java
@@ -4,12 +4,10 @@ import java.util.Arrays;
 import java.util.List;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
+import org.astraea.performance.latency.End2EndLatency;
 
 public class App {
-  private static final List<Class<?>> MAIN_CLASSES =
-      Arrays.asList(
-          // add the classes having main function
-          );
+  private static final List<Class<?>> MAIN_CLASSES = Arrays.asList(End2EndLatency.class);
 
   private static String toString(List<Class<?>> mains) {
     return mains.stream().map(Class::getName).collect(Collectors.joining(","));
diff --git a/app/src/main/java/org/astraea/metrics/jmx/BeanObject.java b/app/src/main/java/org/astraea/metrics/jmx/BeanObject.java
new file mode 100644
index 00000000..a0a8984a
--- /dev/null
+++ b/app/src/main/java/org/astraea/metrics/jmx/BeanObject.java
@@ -0,0 +1,80 @@
+package org.astraea.metrics.jmx;
+
+import static java.util.Map.*;
+
+import java.util.*;
+import java.util.stream.Collectors;
+
+/** Snapshot of remote MBean value */
+public class BeanObject {
+  private final String domainName;
+  private final Map<String, String> properties;
+  private final Map<String, Object> attributes;
+
+  /**
+   * construct a {@link BeanObject}
+   *
+   * <p>Note that, for safety reason. Any null key/value entries may be discarded from the given
+   * properties & attributes map.
+   *
+   * @param domainName domain name of given Mbean snapshot
+   * @param properties properties of given Mbean snapshot
+   * @param attributes attribute and their value of given Mbean snapshot
+   */
+  public BeanObject(
+      String domainName, Map<String, String> properties, Map<String, Object> attributes) {
+    this.domainName = Objects.requireNonNull(domainName);
+
+    // copy properties, and remove null key or null value
+    Objects.requireNonNull(properties);
+    Map<String, String> propertyMap =
+        properties.entrySet().stream()
+            .filter(entry -> entry.getKey() != null && entry.getValue() != null)
+            .collect(Collectors.toMap(Entry::getKey, Entry::getValue));
+    this.properties = Collections.unmodifiableMap(propertyMap);
+
+    // copy attribute, and remove null key or null value
+    Objects.requireNonNull(attributes);
+    Map<String, Object> attributeMap =
+        attributes.entrySet().stream()
+            .filter(entry -> entry.getKey() != null && entry.getValue() != null)
+            .collect(Collectors.toMap(Entry::getKey, Entry::getValue));
+    this.attributes = Collections.unmodifiableMap(attributeMap);
+  }
+
+  public String domainName() {
+    return domainName;
+  }
+
+  public Map<String, String> getProperties() {
+    return properties;
+  }
+
+  public Map<String, Object> getAttributes() {
+    return attributes;
+  }
+
+  @Override
+  public String toString() {
+    String propertyList =
+        properties.entrySet().stream()
+            .map((entry -> entry.getKey() + "=" + entry.getValue()))
+            .collect(Collectors.joining(","));
+    return "[" + domainName + ":" + propertyList + "]\n" + attributes;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) return true;
+    if (o == null || getClass() != o.getClass()) return false;
+    BeanObject that = (BeanObject) o;
+    return domainName.equals(that.domainName)
+        && properties.equals(that.properties)
+        && attributes.equals(that.attributes);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(domainName, properties, attributes);
+  }
+}
diff --git a/app/src/main/java/org/astraea/metrics/jmx/BeanQuery.java b/app/src/main/java/org/astraea/metrics/jmx/BeanQuery.java
new file mode 100644
index 00000000..c17161aa
--- /dev/null
+++ b/app/src/main/java/org/astraea/metrics/jmx/BeanQuery.java
@@ -0,0 +1,205 @@
+package org.astraea.metrics.jmx;
+
+import java.util.*;
+import java.util.stream.Collectors;
+import javax.management.MalformedObjectNameException;
+import javax.management.ObjectName;
+
+/**
+ * MBean query class.
+ *
+ * <p>For the specific rule of query pattern, consider look into {@link ObjectName} Here is some
+ * code example to initialize a {@link BeanQuery}
+ *
+ * <pre>{@code
+ * // Query specific MBean from a JMX server:
+ * BeanQuery.builder("java.lang")
+ *       .property("type", "MemoryManager")
+ *       .property("name", "CodeCacheManager")
+ *       .build();
+ *
+ * // Query MBeans with specific property pattern from a JMX server:
+ * BeanQuery.builder("java.lang")
+ *       .property("type", "MemoryManager")
+ *       .property("name", "*")
+ *       .build();
+ *
+ * // Query all Mbeans from a JMX server:
+ * BeanQuery.all()
+ *
+ * // Query all Mbeans under specific domain name from a JMX server:
+ * BeanQuery.all("java.lang")
+ *
+ * // Query all Mbeans under specific domain name pattern from a JMX server:
+ * BeanQuery.all("java.*")
+ * }</pre>
+ */
+public class BeanQuery {
+
+  private final String domainName;
+  private final Map<String, String> properties;
+  private final ObjectName objectName;
+
+  /**
+   * Initialize a BeanQuery.
+   *
+   * @param domainName the target MBeans's domain name
+   * @param properties the target MBeans's properties
+   * @param usePropertyListPattern use property list pattern or not. If used, a ",*" or "*" string
+   *     will be appended to ObjectName.
+   */
+  public BeanQuery(
+      String domainName, Map<String, String> properties, boolean usePropertyListPattern) {
+    this.domainName = Objects.requireNonNull(domainName);
+    this.properties = Map.copyOf(Objects.requireNonNull(properties));
+    try {
+      if (usePropertyListPattern) {
+        String propertyList =
+            properties.entrySet().stream()
+                .map((entry -> String.format("%s=%s", entry.getKey(), entry.getValue())))
+                .collect(Collectors.joining(","));
+        StringBuilder sb = new StringBuilder();
+        sb.append(domainName);
+        sb.append(":");
+        sb.append(propertyList);
+        sb.append((properties.size() > 0) ? ",*" : "*");
+        this.objectName = ObjectName.getInstance(sb.toString());
+      } else {
+        Hashtable<String, String> ht = new Hashtable<>(this.properties);
+        this.objectName = ObjectName.getInstance(domainName, ht);
+      }
+    } catch (MalformedObjectNameException e) {
+      throw new IllegalArgumentException(e);
+    }
+  }
+
+  public String domainName() {
+    return domainName;
+  }
+
+  public Map<String, String> properties() {
+    return Map.copyOf(properties);
+  }
+
+  ObjectName objectName() {
+    return this.objectName;
+  }
+
+  static class BeanQueryBuilder {
+
+    private final String domainName;
+    private final Map<String, String> properties;
+    private boolean usePropertyListPattern;
+
+    BeanQueryBuilder(String domainName) {
+      this.domainName = domainName;
+      this.properties = new HashMap<>();
+      this.usePropertyListPattern = false;
+    }
+
+    BeanQueryBuilder(String domainName, Map<String, String> properties) {
+      this.domainName = domainName;
+      this.properties = new HashMap<>(properties);
+      this.usePropertyListPattern = false;
+    }
+
+    /**
+     * Apply new search property to the query being built.
+     *
+     * @param key the property key to match.
+     * @param value the property value to match.
+     * @see <a
+     *     href="https://docs.oracle.com/javase/7/docs/api/javax/management/ObjectName.html">ObjectName</a>
+     *     for how Oracle documentation describe property.
+     * @return the current {@link BeanQueryBuilder} instance with the new property applied.
+     */
+    public BeanQueryBuilder property(String key, String value) {
+      this.properties.put(key, value);
+      return this;
+    }
+
+    /**
+     * Apply Property List Pattern to the query being built.
+     *
+     * <p>By default, the built query has property list pattern disabled. With Property List
+     * Pattern, the query will match whose domain matches and that contains the same keys and
+     * associated values, as well as possibly other keys and values.
+     *
+     * @see <a
+     *     href="https://docs.oracle.com/javase/7/docs/api/javax/management/ObjectName.html">ObjectName</a>
+     *     for explanation of property list pattern from Oracle documentation.
+     * @return the current {@link BeanQueryBuilder} instance with property list pattern applied.
+     */
+    public BeanQueryBuilder usePropertyListPattern() {
+      this.usePropertyListPattern = true;
+      return this;
+    }
+
+    /**
+     * Build a {@link BeanQuery} object based on current builder state.
+     *
+     * @return a {@link BeanQuery} with specific MBeans domain name & properties, based on the
+     *     previous calling to {@link BeanQueryBuilder#property(String, String)}.
+     */
+    public BeanQuery build() {
+      return new BeanQuery(domainName, properties, usePropertyListPattern);
+    }
+  }
+
+  /**
+   * construct a {@link BeanQuery} that target all MBeans under every domain name
+   *
+   * @return a {@link BeanQuery} object that target all MBeans under every domain name
+   */
+  public static BeanQuery all() {
+    return new BeanQueryBuilder("*").usePropertyListPattern().build();
+  }
+
+  /**
+   * construct a {@link BeanQuery} that target all MBeans under specific domain name
+   *
+   * @param domainName the domain name to query
+   * @return a {@link BeanQuery} object that target all MBeans under specific domain name
+   */
+  public static BeanQuery all(String domainName) {
+    return new BeanQueryBuilder(domainName).usePropertyListPattern().build();
+  }
+
+  /**
+   * construct a {@link BeanQueryBuilder} that target specific domainName.
+   *
+   * <pre>{@code
+   * // A typical usage of BeanQuery#builder
+   * BeanQuery myQuery = BeanQuery.builder("java.lang")
+   *      .property("type", "Memory")
+   *      .build();
+   * }</pre>
+   *
+   * @param domainName the query target domain name for Builder
+   * @return a {@link BeanQueryBuilder} that can be used to construct a query against specific MBean
+   *     domain name
+   */
+  public static BeanQueryBuilder builder(String domainName) {
+    return new BeanQueryBuilder(domainName);
+  }
+
+  /**
+   * construct a {@link BeanQueryBuilder} that target specific domainName. With some properties
+   * given already.
+   *
+   * @param domainName the query target domain name for Builder
+   * @param properties a {@link Map} of property entries, used to initialize the builder
+   * @return a {@link BeanQueryBuilder} that can be used to construct a query against specific MBean
+   *     domain name
+   */
+  public static BeanQueryBuilder builder(String domainName, Map<String, String> properties) {
+    return new BeanQueryBuilder(domainName, properties);
+  }
+
+  static BeanQuery fromObjectName(ObjectName objectName) {
+    return new BeanQuery(
+        objectName.getDomain(),
+        new HashMap<>(objectName.getKeyPropertyList()),
+        objectName.isPropertyListPattern());
+  }
+}
diff --git a/app/src/main/java/org/astraea/metrics/jmx/MBeanClient.java b/app/src/main/java/org/astraea/metrics/jmx/MBeanClient.java
new file mode 100644
index 00000000..8bdd9f6b
--- /dev/null
+++ b/app/src/main/java/org/astraea/metrics/jmx/MBeanClient.java
@@ -0,0 +1,257 @@
+package org.astraea.metrics.jmx;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.*;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import javax.management.*;
+import javax.management.remote.*;
+
+/**
+ * A MBeanClient used to retrieve mbean value from remote Jmx server.
+ *
+ * <pre>{@code
+ * try(MBeanClient client = new MBeanClient(jmxConnectorServer.getAddress())) {
+ *   BeanObject bean = client.queryBean(BeanQuery.builder("java.lang")
+ *            .property("type", "MemoryManager")
+ *            .property("name", "CodeCacheManager")
+ *            .build());
+ *   System.out.println(bean.getAttributes());
+ * }</pre>
+ */
+public class MBeanClient implements AutoCloseable {
+
+  private final JMXServiceURL jmxServiceURL;
+  private final JMXConnector jmxConnector;
+  private final MBeanServerConnection mBeanServerConnection;
+  private boolean isClosed;
+
+  public MBeanClient(JMXServiceURL jmxServiceURL) throws IOException {
+    this.jmxServiceURL = Objects.requireNonNull(jmxServiceURL);
+    this.jmxConnector = JMXConnectorFactory.connect(jmxServiceURL);
+    this.isClosed = false;
+    this.mBeanServerConnection = jmxConnector.getMBeanServerConnection();
+  }
+
+  /**
+   * Fetch all attributes of target mbean.
+   *
+   * <p>Note that when exception is raised during the attribute fetching process, the exactly
+   * exception will be placed into the attribute field.
+   *
+   * @param beanQuery the non-pattern BeanQuery
+   * @return A {@link BeanObject} contain all attributes if target resolved successfully.
+   * @throws InstanceNotFoundException If the pattern target doesn't exists on remote mbean server.
+   */
+  public BeanObject queryBean(BeanQuery beanQuery) throws InstanceNotFoundException {
+    ensureConnected();
+    try {
+      // ask for MBeanInfo
+      MBeanInfo mBeanInfo = mBeanServerConnection.getMBeanInfo(beanQuery.objectName());
+
+      // create a list builder all available attributes name
+      List<String> attributeName =
+          Arrays.stream(mBeanInfo.getAttributes())
+              .map(MBeanFeatureInfo::getName)
+              .collect(Collectors.toList());
+
+      // query the result
+      return queryBean(beanQuery, attributeName);
+    } catch (ReflectionException | IntrospectionException e) {
+      throw new RuntimeException(e);
+    } catch (IOException e) {
+      throw new UncheckedIOException(e);
+    }
+  }
+
+  /**
+   * Fetch given attributes of target mbean
+   *
+   * <p>Note that when exception is raised during the attribute fetching process, the exactly
+   * exception will be placed into the attribute field.
+   *
+   * @param beanQuery the non-pattern BeanQuery
+   * @param attributeNameCollection a list of attribute you want to retrieve
+   * @return A {@link BeanObject} contain given specific attributes if target resolved successfully.
+   * @throws InstanceNotFoundException If the pattern target doesn't exists on remote mbean server.
+   */
+  public BeanObject queryBean(BeanQuery beanQuery, Collection<String> attributeNameCollection)
+      throws InstanceNotFoundException {
+    ensureConnected();
+    try {
+
+      // fetch attribute value from mbean server
+      String[] attributeNameArray = attributeNameCollection.toArray(new String[0]);
+      List<Attribute> attributeList =
+          mBeanServerConnection.getAttributes(beanQuery.objectName(), attributeNameArray).asList();
+
+      // collect attribute name & value into a map
+      Map<String, Object> attributes = new HashMap<>();
+      for (Attribute attribute : attributeList) {
+        attributes.put(attribute.getName(), attribute.getValue());
+      }
+
+      // according to the javadoc of MBeanServerConnection#getAttributes, the API will
+      // ignore any
+      // error occurring during the fetch process (for example, attribute not exists). Below code
+      // check for such condition and try to figure out what exactly the error is. put it into
+      // attributes return result.
+      Set<String> notResolvedAttributes =
+          Arrays.stream(attributeNameArray)
+              .filter(str -> !attributes.containsKey(str))
+              .collect(Collectors.toSet());
+      for (String attributeName : notResolvedAttributes) {
+        attributes.put(attributeName, fetchAttributeObjectOrException(beanQuery, attributeName));
+      }
+
+      // collect result, and build a new BeanObject as return result
+      return new BeanObject(beanQuery.domainName(), beanQuery.properties(), attributes);
+
+    } catch (ReflectionException e) {
+      throw new RuntimeException(e);
+    } catch (IOException e) {
+      throw new UncheckedIOException(e);
+    }
+  }
+
+  private Object fetchAttributeObjectOrException(BeanQuery beanQuery, String attributeName) {
+    // It is possible to trigger some unexpected runtime exception during the following call.
+    // For example, on my machine when I try to get attribute "BootClassPath" from
+    // "java.lang:type=Runtime".
+    // I will get a {@link java.lang.UnsupportedOperationException} indicates that "Boot class path
+    // mechanism is not supported". Those attribute actually exists, but I cannot retrieve those
+    // attribute value. Doing so I get that error.
+    //
+    // Instead of blinding that attribute from the library user, I decided to put the
+    // exception
+    // into their result.
+    try {
+      return mBeanServerConnection.getAttribute(beanQuery.objectName(), attributeName);
+    } catch (IOException e) {
+      throw new UncheckedIOException(e);
+    } catch (Exception e) {
+      return e;
+    }
+  }
+
+  /**
+   * Fetch all attributes of target mbean, return with {@link Optional} support
+   *
+   * <p>During the attribute retrieve process, if a {@link InstanceNotFoundException} is raised, an
+   * {@link Optional#empty()} will return.
+   *
+   * <p>Note that when exception is raised during the attribute fetching process, the exactly
+   * exception will be placed into the attribute field.
+   *
+   * @param beanQuery the non-pattern BeanQuery
+   * @return A {@link Optional<BeanObject>} of {@link BeanObject} contain all attributes if target
+   *     resolved successfully. If the pattern target doesn't exists on remote mbean server, then an
+   *     {@link Optional#empty()} returned.
+   */
+  public Optional<BeanObject> tryQueryBean(BeanQuery beanQuery) {
+    ensureConnected();
+    try {
+      return Optional.of(this.queryBean(beanQuery));
+    } catch (InstanceNotFoundException e) {
+      return Optional.empty();
+    }
+  }
+
+  /**
+   * Fetch a list of mbean attribute, return with {@link Optional} support
+   *
+   * <p>During the attribute retrieve process, if a {@link InstanceNotFoundException} is raised, an
+   * {@link Optional#empty()} will return.
+   *
+   * <p>Note that when exception is raised during the attribute fetching process, the exactly
+   * exception will be placed into the attribute field.
+   *
+   * @param beanQuery the non-pattern BeanQuery
+   * @param attributeNameCollection array of attribute names you want to retrieve
+   * @return A {@link Optional<BeanObject>} contain the exactly {@link BeanObject} with specific
+   *     attributes resolved if target mbeans resolved successfully. If the pattern target doesn't
+   *     exists on remote mbean server, then an {@link Optional#empty()} returned.
+   */
+  public Optional<BeanObject> tryQueryBean(
+      BeanQuery beanQuery, Collection<String> attributeNameCollection) {
+    ensureConnected();
+    try {
+      return Optional.of(this.queryBean(beanQuery, attributeNameCollection));
+    } catch (InstanceNotFoundException e) {
+      return Optional.empty();
+    }
+  }
+
+  /**
+   * Query mBeans by pattern.
+   *
+   * <p>Query mbeans by {@link ObjectName} pattern, the returned {@link BeanObject}s will contain
+   * all the available attributes
+   *
+   * <p>Note that when exception is raised during the attribute fetching process, the exactly
+   * exception will be placed into the attribute field.
+   *
+   * @param beanQuery the pattern to query
+   * @return A {@link Set} of {@link BeanObject}, all BeanObject has its own attributes resolved.
+   */
+  public Collection<BeanObject> queryBeans(BeanQuery beanQuery) {
+    ensureConnected();
+    try {
+
+      // query mbeans
+      Set<ObjectInstance> objectInstances =
+          mBeanServerConnection.queryMBeans(beanQuery.objectName(), null);
+
+      // transform result into a set of BeanQuery
+      Stream<BeanQuery> queries =
+          objectInstances.stream()
+              .map(ObjectInstance::getObjectName)
+              .map(BeanQuery::fromObjectName);
+
+      // execute query on each BeanQuery, return result as a set of BeanObject
+      Set<BeanObject> queryResult =
+          queries.map(this::tryQueryBean).flatMap(Optional::stream).collect(Collectors.toSet());
+
+      return queryResult;
+
+    } catch (IOException e) {
+      throw new UncheckedIOException(e);
+    }
+  }
+
+  /**
+   * Returns the list of domains in which any MBean is currently registered.
+   *
+   * <p>The order of strings within the returned array is not defined.
+   *
+   * @return a {@link List} of domain name {@link String}
+   */
+  public List<String> listDomains() {
+    try {
+      return Arrays.asList(mBeanServerConnection.getDomains());
+    } catch (IOException e) {
+      throw new UncheckedIOException(e);
+    }
+  }
+
+  /**
+   * The JMX URL of current MBeanClient instance
+   *
+   * @return the origin JMX URL used to initiate MBeanClient
+   */
+  public JMXServiceURL getAddress() {
+    return jmxServiceURL;
+  }
+
+  private void ensureConnected() {
+    if (isClosed) throw new IllegalStateException("MBean client is closed");
+  }
+
+  @Override
+  public void close() throws Exception {
+    ensureConnected();
+    this.isClosed = true;
+    this.jmxConnector.close();
+  }
+}
diff --git a/app/src/main/java/org/astraea/performance/latency/CloseableThread.java b/app/src/main/java/org/astraea/performance/latency/CloseableThread.java
new file mode 100644
index 00000000..5bb860b7
--- /dev/null
+++ b/app/src/main/java/org/astraea/performance/latency/CloseableThread.java
@@ -0,0 +1,58 @@
+package org.astraea.performance.latency;
+
+import java.io.Closeable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+abstract class CloseableThread implements Runnable, Closeable {
+  private final AtomicBoolean closed = new AtomicBoolean();
+  private final CountDownLatch closeLatch = new CountDownLatch(1);
+  private final boolean executeOnce;
+  private final AtomicLong threadId = new AtomicLong();
+
+  protected CloseableThread() {
+    this(false);
+  }
+
+  protected CloseableThread(boolean executeOnce) {
+    this.executeOnce = executeOnce;
+  }
+
+  @Override
+  public final void run() {
+    try {
+      threadId.set(Thread.currentThread().getId());
+      do {
+        execute();
+      } while (!closed.get() && !executeOnce);
+    } catch (InterruptedException e) {
+      // swallow
+    } finally {
+      try {
+        cleanup();
+      } finally {
+        closeLatch.countDown();
+      }
+    }
+  }
+
+  /** looped action. */
+  abstract void execute() throws InterruptedException;
+
+  /** final action when leaving loop. */
+  void cleanup() {}
+
+  @Override
+  public void close() {
+    if (threadId.get() == Thread.currentThread().getId()) {
+      throw new RuntimeException("Should not call close() in execute().");
+    }
+    closed.set(true);
+    try {
+      closeLatch.await();
+    } catch (InterruptedException e) {
+      // swallow
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/performance/latency/ComponentFactory.java b/app/src/main/java/org/astraea/performance/latency/ComponentFactory.java
new file mode 100644
index 00000000..fe0ac24b
--- /dev/null
+++ b/app/src/main/java/org/astraea/performance/latency/ComponentFactory.java
@@ -0,0 +1,58 @@
+package org.astraea.performance.latency;
+
+import java.util.Properties;
+import java.util.Set;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.ProducerConfig;
+
+interface ComponentFactory {
+
+  /**
+   * create a factory based on kafka cluster. All components created by this factory will send
+   * request to kafka to get responses.
+   *
+   * @param brokers kafka broker addresses
+   * @param topics to subscribe
+   * @return a factory based on kafka
+   */
+  static ComponentFactory fromKafka(String brokers, Set<String> topics) {
+
+    return new ComponentFactory() {
+      private final String groupId = "group-id-" + System.currentTimeMillis();
+
+      @Override
+      public Producer producer() {
+        var props = new Properties();
+        props.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
+        return Producer.fromKafka(props);
+      }
+
+      @Override
+      public Consumer createConsumer() {
+        var props = new Properties();
+        props.setProperty(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, brokers);
+        // all consumers are in same group, so there is no duplicate data in read workload.
+        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
+        return Consumer.fromKafka(props, topics);
+      }
+
+      @Override
+      public TopicAdmin createTopicAdmin() {
+        var props = new Properties();
+        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
+        return TopicAdmin.fromKafka(props);
+      }
+    };
+  }
+
+  /** @return a new producer. Please close it when you don't need it. */
+  Producer producer();
+
+  /** @return a new consumer. Please close it when you don't need it */
+  Consumer createConsumer();
+
+  /** @return a new topic admin. Please close it when you don't need it. */
+  TopicAdmin createTopicAdmin();
+}
diff --git a/app/src/main/java/org/astraea/performance/latency/Consumer.java b/app/src/main/java/org/astraea/performance/latency/Consumer.java
new file mode 100644
index 00000000..0bae7847
--- /dev/null
+++ b/app/src/main/java/org/astraea/performance/latency/Consumer.java
@@ -0,0 +1,45 @@
+package org.astraea.performance.latency;
+
+import java.io.Closeable;
+import java.time.Duration;
+import java.util.Properties;
+import java.util.Set;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+
+interface Consumer extends Closeable {
+  Duration POLL_TIMEOUT = Duration.ofMillis(500);
+
+  static Consumer fromKafka(Properties props, Set<String> topics) {
+    var kafkaConsumer =
+        new KafkaConsumer<>(props, new ByteArrayDeserializer(), new ByteArrayDeserializer());
+    kafkaConsumer.subscribe(topics);
+    return new Consumer() {
+
+      @Override
+      public ConsumerRecords<byte[], byte[]> poll() {
+        return kafkaConsumer.poll(POLL_TIMEOUT);
+      }
+
+      @Override
+      public void wakeup() {
+        kafkaConsumer.wakeup();
+      }
+
+      @Override
+      public void close() {
+        kafkaConsumer.close();
+      }
+    };
+  }
+
+  /** see {@link KafkaConsumer#poll(Duration)} */
+  ConsumerRecords<byte[], byte[]> poll();
+
+  /** see {@link KafkaConsumer#wakeup()} */
+  default void wakeup() {}
+
+  @Override
+  default void close() {}
+}
diff --git a/app/src/main/java/org/astraea/performance/latency/ConsumerThread.java b/app/src/main/java/org/astraea/performance/latency/ConsumerThread.java
new file mode 100644
index 00000000..2a04c04d
--- /dev/null
+++ b/app/src/main/java/org/astraea/performance/latency/ConsumerThread.java
@@ -0,0 +1,46 @@
+package org.astraea.performance.latency;
+
+import java.util.Objects;
+
+class ConsumerThread extends CloseableThread {
+
+  private final Consumer consumer;
+  private final DataManager dataManager;
+  private final MeterTracker tracker;
+
+  ConsumerThread(DataManager dataManager, MeterTracker tracker, Consumer consumer) {
+    this.dataManager = Objects.requireNonNull(dataManager);
+    this.consumer = Objects.requireNonNull(consumer);
+    this.tracker = Objects.requireNonNull(tracker);
+  }
+
+  @Override
+  public void execute() throws InterruptedException {
+    try {
+      var now = System.currentTimeMillis();
+      var records = consumer.poll();
+      records.forEach(
+          record -> {
+            var entry = dataManager.removeSendingRecord(record.key());
+            var latency = now - entry.getValue();
+            var produceRecord = entry.getKey();
+            if (!KafkaUtils.equal(produceRecord, record))
+              System.out.println("receive corrupt data!!!");
+            else tracker.record(record.serializedKeySize() + record.serializedValueSize(), latency);
+          });
+    } catch (org.apache.kafka.common.errors.WakeupException e) {
+      throw new InterruptedException(e.getMessage());
+    }
+  }
+
+  @Override
+  void cleanup() {
+    consumer.close();
+  }
+
+  @Override
+  public void close() {
+    consumer.wakeup();
+    super.close();
+  }
+}
diff --git a/app/src/main/java/org/astraea/performance/latency/DataManager.java b/app/src/main/java/org/astraea/performance/latency/DataManager.java
new file mode 100644
index 00000000..d23f4c2b
--- /dev/null
+++ b/app/src/main/java/org/astraea/performance/latency/DataManager.java
@@ -0,0 +1,102 @@
+package org.astraea.performance.latency;
+
+import java.util.*;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.header.Header;
+
+class DataManager {
+
+  private static final int NUMBER_OF_CACHED_VALUES = 100;
+  private static final int NUMBER_OF_CACHED_HEADERS = 100;
+  private static final int NUMBER_OF_SENDING_HEADERS = 3;
+
+  static DataManager of(String topic, int valueSize) {
+    return new DataManager(topic, valueSize, true);
+  }
+
+  static DataManager noConsumer(String topic, int valueSize) {
+    return new DataManager(topic, valueSize, false);
+  }
+
+  /** record key -> (record, timestamp_done) */
+  private final ConcurrentMap<byte[], Map.Entry<ProducerRecord<byte[], byte[]>, Long>>
+      sendingRecords = new ConcurrentSkipListMap<>(Arrays::compare);
+
+  private final String topic;
+  private final boolean hasConsumer;
+  private final List<byte[]> values;
+  private final List<Header> headers;
+
+  private final AtomicLong recordIndex = new AtomicLong(0);
+
+  final AtomicLong producerRecords = new AtomicLong(0);
+
+  private DataManager(String topic, int valueSize, boolean hasConsumer) {
+    this.topic = Objects.requireNonNull(topic);
+    this.values =
+        IntStream.range(0, NUMBER_OF_CACHED_VALUES)
+            .mapToObj(i -> randomString(valueSize).getBytes())
+            .collect(Collectors.toList());
+    this.headers =
+        IntStream.range(0, NUMBER_OF_CACHED_HEADERS)
+            .mapToObj(i -> KafkaUtils.header(randomString(5), randomString(5).getBytes()))
+            .collect(Collectors.toList());
+    this.hasConsumer = hasConsumer;
+  }
+
+  static String randomString(int length) {
+    return new Random()
+        .ints('a', 'z')
+        .limit(length)
+        .collect(StringBuilder::new, StringBuilder::appendCodePoint, StringBuilder::append)
+        .toString();
+  }
+
+  /**
+   * @return generate a new record with random data and specify topic name. The record consists of
+   *     key, value, topic and headers.
+   */
+  ProducerRecord<byte[], byte[]> producerRecord() {
+    return new ProducerRecord<>(
+        topic,
+        null,
+        (long) (Math.random() * System.currentTimeMillis()),
+        String.format("%020d", recordIndex.getAndIncrement()).getBytes(),
+        values.get((int) (Math.random() * values.size())),
+        IntStream.range(0, NUMBER_OF_SENDING_HEADERS)
+            .mapToObj(i -> headers.get((int) (Math.random() * headers.size())))
+            .collect(Collectors.toList()));
+  }
+
+  void sendingRecord(ProducerRecord<byte[], byte[]> record, long now) {
+    if (hasConsumer) {
+      var previous =
+          sendingRecords.put(record.key(), new AbstractMap.SimpleImmutableEntry<>(record, now));
+      if (previous != null) throw new RuntimeException("duplicate data!!!");
+    }
+    producerRecords.incrementAndGet();
+  }
+
+  /**
+   * get sending record
+   *
+   * @param key of completed record
+   * @return the completed record. Or NullPointerException if there is no related record.
+   */
+  Map.Entry<ProducerRecord<byte[], byte[]>, Long> removeSendingRecord(byte[] key) {
+    if (!hasConsumer)
+      throw new UnsupportedOperationException(
+          "removeSendingRecord is unsupported when there is no consumer");
+    return Objects.requireNonNull(sendingRecords.remove(key));
+  }
+
+  /** @return number of completed records */
+  long numberOfProducerRecords() {
+    return producerRecords.get();
+  }
+}
diff --git a/app/src/main/java/org/astraea/performance/latency/End2EndLatency.java b/app/src/main/java/org/astraea/performance/latency/End2EndLatency.java
new file mode 100644
index 00000000..ad2e422a
--- /dev/null
+++ b/app/src/main/java/org/astraea/performance/latency/End2EndLatency.java
@@ -0,0 +1,159 @@
+package org.astraea.performance.latency;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Objects;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+public class End2EndLatency {
+
+  static final String BROKERS_KEY = "--bootstrap.servers";
+  static final String TOPIC_KEY = "--topic";
+  static final String PRODUCERS_KEY = "--producers";
+  static final String CONSUMERS_KEY = "--consumers";
+  static final String DURATION_KEY = "--duration";
+  static final String VALUE_SIZE_KEY = "--valueSize";
+  static final String FLUSH_DURATION_KEY = "--flushDuration";
+
+  private static IllegalArgumentException exception(String key, Object incorrectValue) {
+    return new IllegalArgumentException(
+        "the value: " + incorrectValue + " is illegal. Please use another value for " + key);
+  }
+
+  static class Parameters {
+    final String brokers;
+    final String topic;
+    final int numberOfProducers;
+    final int numberOfConsumers;
+    final Duration duration;
+    final int valueSize;
+    final Duration flushDuration;
+
+    Parameters(
+        String brokers,
+        String topic,
+        int numberOfProducers,
+        int numberOfConsumers,
+        Duration duration,
+        int valueSize,
+        Duration flushDuration) {
+      this.brokers = Objects.requireNonNull(brokers);
+      this.topic = Objects.requireNonNull(topic);
+      this.numberOfProducers = numberOfProducers;
+      this.numberOfConsumers = numberOfConsumers;
+      this.duration = duration;
+      this.valueSize = valueSize;
+      this.flushDuration = flushDuration;
+
+      if (brokers.isBlank()) throw exception(BROKERS_KEY, brokers);
+      if (topic.isBlank()) throw exception(TOPIC_KEY, topic);
+      if (numberOfProducers <= 0) throw exception(PRODUCERS_KEY, numberOfProducers);
+      if (numberOfConsumers < 0) throw exception(CONSUMERS_KEY, numberOfConsumers);
+      if (duration.toMillis() <= 0) throw exception(DURATION_KEY, duration);
+      if (flushDuration.toMillis() <= 0) throw exception(FLUSH_DURATION_KEY, flushDuration);
+      if (valueSize <= 0) throw exception(VALUE_SIZE_KEY, valueSize);
+    }
+  }
+
+  static Parameters parameters(String[] args) {
+    // group arguments by key
+    var argMap = new HashMap<String, String>();
+    for (var i = 0; i <= args.length; i += 2) {
+      if (i + 1 >= args.length) break;
+      argMap.put(args[i], args[i + 1]);
+    }
+
+    // take values from legal keys. We just ignore the invalid configs.
+    return new Parameters(
+        argMap.getOrDefault(BROKERS_KEY, ""),
+        argMap.getOrDefault(TOPIC_KEY, "testLatency-" + System.currentTimeMillis()),
+        Integer.parseInt(argMap.getOrDefault(PRODUCERS_KEY, "1")),
+        Integer.parseInt(argMap.getOrDefault(CONSUMERS_KEY, "1")),
+        Duration.ofSeconds(Long.parseLong(argMap.getOrDefault(DURATION_KEY, "5"))),
+        Integer.parseInt(argMap.getOrDefault(VALUE_SIZE_KEY, "100")),
+        Duration.ofSeconds(Long.parseLong(argMap.getOrDefault(FLUSH_DURATION_KEY, "2"))));
+  }
+
+  public static void main(String[] args) throws Exception {
+    var parameters = parameters(args);
+
+    System.out.println("brokers: " + parameters.brokers);
+    System.out.println("topic: " + parameters.topic);
+    System.out.println("numberOfProducers: " + parameters.numberOfProducers);
+    System.out.println("numberOfConsumers: " + parameters.numberOfConsumers);
+    System.out.println("duration: " + parameters.duration.toSeconds() + " seconds");
+    System.out.println("value size: " + parameters.valueSize + " bytes");
+    System.out.println("flush duration: " + parameters.flushDuration.toSeconds() + " seconds");
+
+    try (var closeFlag =
+        execute(
+            ComponentFactory.fromKafka(parameters.brokers, Collections.singleton(parameters.topic)),
+            parameters)) {
+      TimeUnit.MILLISECONDS.sleep(parameters.duration.toMillis());
+    }
+  }
+
+  static AutoCloseable execute(ComponentFactory factory, Parameters parameters) throws Exception {
+    var consumerTracker = new MeterTracker("consumer latency");
+    var producerTracker = new MeterTracker("producer latency");
+    var dataManager =
+        parameters.numberOfConsumers <= 0
+            ? DataManager.noConsumer(parameters.topic, parameters.valueSize)
+            : DataManager.of(parameters.topic, parameters.valueSize);
+
+    // create producer threads
+    var producers =
+        IntStream.range(0, parameters.numberOfProducers)
+            .mapToObj(
+                i ->
+                    new ProducerThread(
+                        dataManager, producerTracker, factory.producer(), parameters.flushDuration))
+            .collect(Collectors.toList());
+
+    // create consumers threads
+    var consumers =
+        IntStream.range(0, parameters.numberOfConsumers)
+            .mapToObj(
+                i -> new ConsumerThread(dataManager, consumerTracker, factory.createConsumer()))
+            .collect(Collectors.toList());
+
+    // + 2 for latency trackers
+    var services =
+        Executors.newFixedThreadPool(
+            parameters.numberOfProducers + parameters.numberOfConsumers + 2);
+
+    AutoCloseable releaseAllObjects =
+        () -> {
+          producerTracker.close();
+          consumerTracker.close();
+          producers.forEach(ProducerThread::close);
+          consumers.forEach(ConsumerThread::close);
+          services.shutdownNow();
+          if (!services.awaitTermination(30, TimeUnit.SECONDS)) {
+            System.out.println("timeout to wait all threads");
+          }
+        };
+    try {
+      try (var topicAdmin = factory.createTopicAdmin()) {
+        // the number of partitions is equal to number of consumers. That make each consumer can
+        // consume a part of topic.
+        KafkaUtils.createTopicIfNotExist(
+            topicAdmin,
+            parameters.topic,
+            parameters.numberOfConsumers <= 0 ? 1 : parameters.numberOfConsumers);
+      }
+      consumers.forEach(services::execute);
+      producers.forEach(services::execute);
+      services.execute(consumerTracker);
+      services.execute(producerTracker);
+      return releaseAllObjects;
+    } catch (Exception e) {
+      releaseAllObjects.close();
+      throw e;
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/performance/latency/KafkaUtils.java b/app/src/main/java/org/astraea/performance/latency/KafkaUtils.java
new file mode 100644
index 00000000..6519bd73
--- /dev/null
+++ b/app/src/main/java/org/astraea/performance/latency/KafkaUtils.java
@@ -0,0 +1,55 @@
+package org.astraea.performance.latency;
+
+import java.util.*;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.internals.RecordHeader;
+
+final class KafkaUtils {
+
+  static Header header(String key, byte[] value) {
+    return new RecordHeader(key, value);
+  }
+
+  static boolean equal(Iterable<Header> lhs, Iterable<Header> rhs) {
+    Comparator<Header> comparator =
+        (o1, o2) -> {
+          var r = o1.key().compareTo(o2.key());
+          if (r != 0) return r;
+          else return Arrays.compare(o1.value(), o2.value());
+        };
+    var lhsList = new ArrayList<Header>();
+    lhs.forEach(lhsList::add);
+    var rhsList = new ArrayList<Header>();
+    rhs.forEach(rhsList::add);
+    lhsList.sort(comparator);
+    rhsList.sort(comparator);
+    return lhsList.equals(rhsList);
+  }
+
+  static boolean equal(
+      ProducerRecord<byte[], byte[]> producerRecord,
+      ConsumerRecord<byte[], byte[]> consumerRecord) {
+    if (!Arrays.equals(producerRecord.key(), consumerRecord.key())) return false;
+    if (!Arrays.equals(producerRecord.value(), consumerRecord.value())) return false;
+    if (!producerRecord.topic().equals(consumerRecord.topic())) return false;
+    if (!equal(producerRecord.headers(), consumerRecord.headers())) return false;
+    return producerRecord.timestamp() == null
+        || producerRecord.timestamp() == consumerRecord.timestamp();
+  }
+
+  static void createTopicIfNotExist(TopicAdmin adminClient, String name, int numberOfPartitions) {
+    try {
+      if (!adminClient.listTopics().contains(name))
+        adminClient.createTopics(
+            Collections.singletonList(
+                new NewTopic(name, Optional.of(numberOfPartitions), Optional.empty())));
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  private KafkaUtils() {}
+}
diff --git a/app/src/main/java/org/astraea/performance/latency/MeterTracker.java b/app/src/main/java/org/astraea/performance/latency/MeterTracker.java
new file mode 100644
index 00000000..86d3d0ae
--- /dev/null
+++ b/app/src/main/java/org/astraea/performance/latency/MeterTracker.java
@@ -0,0 +1,78 @@
+package org.astraea.performance.latency;
+
+import java.util.concurrent.TimeUnit;
+
+class MeterTracker extends CloseableThread {
+  private final long startTime = System.currentTimeMillis();
+  private final String name;
+  private long bytes = 0;
+  private long count = 0;
+  private double avgLatency = 0;
+  private long maxLatency = 0;
+
+  MeterTracker(String name) {
+    this.name = name;
+  }
+
+  synchronized void record(long bytes, long latency) {
+    maxLatency = Math.max(maxLatency, latency);
+    avgLatency = (avgLatency * count + latency) / (count + 1);
+    count += 1;
+    this.bytes += bytes;
+  }
+
+  synchronized long count() {
+    return count;
+  }
+
+  synchronized long bytes() {
+    return bytes;
+  }
+
+  synchronized double averageLatency() {
+    return avgLatency;
+  }
+
+  synchronized long maxLatency() {
+    return maxLatency;
+  }
+
+  synchronized double throughput() {
+    var second = (System.currentTimeMillis() - startTime) / 1000;
+    return ((double) bytes / second);
+  }
+
+  private static double round(double value) {
+    return Math.round(value * 100) / (double) 100;
+  }
+
+  @Override
+  public synchronized String toString() {
+    return " records: "
+        + count
+        + " throughput: "
+        + round(throughput() / 1024 / 1024)
+        + " MB/s"
+        + " avg_latency: "
+        + round(avgLatency)
+        + " ms"
+        + " max_latency: "
+        + maxLatency
+        + " ms";
+  }
+
+  @Override
+  void execute() {
+    try {
+      TimeUnit.SECONDS.sleep(2);
+      String output;
+      synchronized (this) {
+        if (count == 0) return;
+        else output = "[" + name + "] " + this;
+      }
+      System.out.println(output);
+    } catch (InterruptedException e) {
+      // swallow
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/performance/latency/Producer.java b/app/src/main/java/org/astraea/performance/latency/Producer.java
new file mode 100644
index 00000000..9e16de32
--- /dev/null
+++ b/app/src/main/java/org/astraea/performance/latency/Producer.java
@@ -0,0 +1,51 @@
+package org.astraea.performance.latency;
+
+import java.io.Closeable;
+import java.util.Properties;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+
+interface Producer extends Closeable {
+
+  static Producer fromKafka(Properties props) {
+    var kafkaProducer =
+        new KafkaProducer<>(props, new ByteArraySerializer(), new ByteArraySerializer());
+    return new Producer() {
+
+      @Override
+      public CompletionStage<RecordMetadata> send(ProducerRecord<byte[], byte[]> record) {
+        var f = new CompletableFuture<RecordMetadata>();
+        kafkaProducer.send(
+            record,
+            (r, e) -> {
+              if (e != null) f.completeExceptionally(e);
+              else f.complete(r);
+            });
+        return f;
+      }
+
+      @Override
+      public void flush() {
+        kafkaProducer.flush();
+      }
+
+      @Override
+      public void close() {
+        kafkaProducer.close();
+      }
+    };
+  }
+
+  /** see {@link KafkaProducer#send(ProducerRecord)} */
+  CompletionStage<RecordMetadata> send(ProducerRecord<byte[], byte[]> record);
+
+  /** see {@link KafkaProducer#flush()}} */
+  default void flush() {}
+
+  @Override
+  default void close() {}
+}
diff --git a/app/src/main/java/org/astraea/performance/latency/ProducerThread.java b/app/src/main/java/org/astraea/performance/latency/ProducerThread.java
new file mode 100644
index 00000000..f3eb6ac1
--- /dev/null
+++ b/app/src/main/java/org/astraea/performance/latency/ProducerThread.java
@@ -0,0 +1,47 @@
+package org.astraea.performance.latency;
+
+import java.time.Duration;
+import java.util.Objects;
+
+class ProducerThread extends CloseableThread {
+  private final Producer producer;
+  private final DataManager dataManager;
+  private final MeterTracker tracker;
+  private final Duration flushDuration;
+  private long lastSend = 0;
+
+  ProducerThread(
+      DataManager dataManager, MeterTracker tracker, Producer producer, Duration flushDuration) {
+    this.dataManager = Objects.requireNonNull(dataManager);
+    this.producer = Objects.requireNonNull(producer);
+    this.tracker = Objects.requireNonNull(tracker);
+    this.flushDuration = Objects.requireNonNull(flushDuration);
+  }
+
+  @Override
+  void execute() {
+    var data = dataManager.producerRecord();
+    var now = System.currentTimeMillis();
+    dataManager.sendingRecord(data, now);
+    producer
+        .send(data)
+        .whenComplete(
+            (r, e) -> {
+              if (e != null) tracker.record(0, System.currentTimeMillis() - now);
+              else
+                tracker.record(
+                    r.serializedKeySize() + r.serializedValueSize(),
+                    System.currentTimeMillis() - now);
+            });
+    if (lastSend <= 0) lastSend = now;
+    else if (lastSend + flushDuration.toMillis() < now) {
+      lastSend = now;
+      producer.flush();
+    }
+  }
+
+  @Override
+  void cleanup() {
+    producer.close();
+  }
+}
diff --git a/app/src/main/java/org/astraea/performance/latency/TopicAdmin.java b/app/src/main/java/org/astraea/performance/latency/TopicAdmin.java
new file mode 100644
index 00000000..f2aadd5e
--- /dev/null
+++ b/app/src/main/java/org/astraea/performance/latency/TopicAdmin.java
@@ -0,0 +1,50 @@
+package org.astraea.performance.latency;
+
+import java.io.Closeable;
+import java.util.Collection;
+import java.util.Properties;
+import java.util.Set;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.NewTopic;
+
+interface TopicAdmin extends Closeable {
+
+  static TopicAdmin fromKafka(Properties props) {
+    var adminClient = Admin.create(props);
+
+    return new TopicAdmin() {
+
+      @Override
+      public Set<String> listTopics() {
+        try {
+          return adminClient.listTopics().names().get();
+        } catch (Exception e) {
+          throw new RuntimeException(e);
+        }
+      }
+
+      @Override
+      public void createTopics(Collection<NewTopic> newTopics) {
+        try {
+          adminClient.createTopics(newTopics).all().get();
+        } catch (Exception e) {
+          throw new RuntimeException(e);
+        }
+      }
+
+      @Override
+      public void close() {
+        adminClient.close();
+      }
+    };
+  }
+
+  /** see {@link Admin#listTopics()} */
+  Set<String> listTopics();
+
+  /** see {@link Admin#createTopics(Collection)} */
+  void createTopics(Collection<NewTopic> newTopics);
+
+  @Override
+  default void close() {}
+}
diff --git a/docker/start_broker.sh b/docker/start_broker.sh
index 0e976775..b7de45f6 100755
--- a/docker/start_broker.sh
+++ b/docker/start_broker.sh
@@ -31,12 +31,14 @@ if [[ "$(which docker)" == "" ]]; then
   exit 2
 fi
 
+if [[ -z "$KAFKA_VERSION" ]]; then
+  KAFKA_VERSION=2.8.0
+fi
+
 USER=broker
-KAFKA_VERSION=2.8.0
-image_name=astraea/broker
+image_name=astraea/broker:$KAFKA_VERSION
 broker_id="$(($RANDOM % 1000))"
 address=$(getAddress)
-jvm_memory="-Xmx2G -Xms2G"
 broker_port="$(($(($RANDOM % 10000 )) + 10000))"
 broker_jmx_port="$(($(($RANDOM % 10000 )) + 10000))"
 jmx_opts="-Dcom.sun.management.jmxremote \
@@ -55,9 +57,6 @@ while [[ $# -gt 0 ]]; do
     showHelp
     exit 2
   fi
-  if [[ "$1" == "memory=*" ]]; then
-    jvm_memory="$(echo $1 | cut -d'=' -f 2)"
-  fi
   echo "$1" >> "$config_file"
   shift
 done
@@ -68,6 +67,9 @@ if [[ "$(cat $config_file | grep zookeeper.connect)" == "" ]]; then
   exit 2
 fi
 
+# set JVM heap
+KAFKA_HEAP="${KAFKA_HEAP:-"-Xmx2G -Xms2G"}"
+
 # listeners will be generated automatically
 if [[ "$(cat $config_file | grep listeners)" != "" ]]; then
   echo "you should not define listeners"
@@ -139,7 +141,7 @@ WORKDIR /home/$USER/kafka_2.13-${KAFKA_VERSION}
 Dockerfile
 
 docker run -d \
-  -e KAFKA_HEAP_OPTS="$jvm_memory" \
+  -e KAFKA_HEAP_OPTS="$KAFKA_HEAP" \
   -e KAFKA_JMX_OPTS="$jmx_opts" \
   -v $config_file:/tmp/broker.properties:ro \
   -p $broker_port:9092 \
diff --git a/docker/start_zk.sh b/docker/start_zookeeper.sh
similarity index 71%
rename from docker/start_zk.sh
rename to docker/start_zookeeper.sh
index 07479e47..c196806f 100755
--- a/docker/start_zk.sh
+++ b/docker/start_zookeeper.sh
@@ -20,9 +20,12 @@ if [[ "$(which docker)" == "" ]]; then
   exit 2
 fi
 
+if [[ -z "$ZOOKEEPER_VERSION" ]]; then
+  ZOOKEEPER_VERSION=3.6.3
+fi
+
 USER=zookeeper
-ZK_VERSION=3.6.3
-image_name=astraea/zookeeper
+image_name=astraea/zookeeper:$ZOOKEEPER_VERSION
 zk_port="$(($(($RANDOM % 10000 )) + 10000))"
 address=$(getAddress)
 
@@ -40,9 +43,9 @@ USER $USER
 
 # download zookeeper
 WORKDIR /home/$USER
-RUN wget https://dlcdn.apache.org/zookeeper/zookeeper-${ZK_VERSION}/apache-zookeeper-${ZK_VERSION}-bin.tar.gz
-RUN tar -zxvf apache-zookeeper-${ZK_VERSION}-bin.tar.gz
-WORKDIR /home/$USER/apache-zookeeper-${ZK_VERSION}-bin
+RUN wget https://archive.apache.org/dist/zookeeper/zookeeper-${ZOOKEEPER_VERSION}/apache-zookeeper-${ZOOKEEPER_VERSION}-bin.tar.gz
+RUN tar -zxvf apache-zookeeper-${ZOOKEEPER_VERSION}-bin.tar.gz
+WORKDIR /home/$USER/apache-zookeeper-${ZOOKEEPER_VERSION}-bin
 
 # create config file
 RUN echo "tickTime=2000" >> ./conf/zoo.cfg
@@ -52,7 +55,7 @@ Dockerfile
 
 docker run -d \
   -p $zk_port:2181 \
-  $image_name /home/$USER/apache-zookeeper-${ZK_VERSION}-bin/bin/zkServer.sh start-foreground
+  $image_name /home/$USER/apache-zookeeper-${ZOOKEEPER_VERSION}-bin/bin/zkServer.sh start-foreground
 
 echo "================================================="
 echo "run ./docker/start_broker.sh zookeeper.connect=$address:$zk_port to join kafka broker"


Test patch:
diff --git a/app/src/test/java/org/astraea/metrics/jmx/BeanObjectTest.java b/app/src/test/java/org/astraea/metrics/jmx/BeanObjectTest.java
new file mode 100644
index 00000000..8581eaa4
--- /dev/null
+++ b/app/src/test/java/org/astraea/metrics/jmx/BeanObjectTest.java
@@ -0,0 +1,50 @@
+package org.astraea.metrics.jmx;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.util.Map;
+import org.junit.jupiter.api.Test;
+
+class BeanObjectTest {
+
+  BeanObject createBeanObject() {
+    return new BeanObject(
+        "java.lang", Map.of("type", "Memory"), Map.of("HeapMemoryUsage", "content"));
+  }
+
+  @Test
+  void domainName() {
+    BeanObject beanObject = createBeanObject();
+
+    String domainName = beanObject.domainName();
+
+    assertEquals("java.lang", domainName);
+  }
+
+  @Test
+  void getPropertyView() {
+    BeanObject beanObject = createBeanObject();
+
+    Map<String, String> propertyView = beanObject.getProperties();
+
+    assertEquals(Map.of("type", "Memory"), propertyView);
+  }
+
+  @Test
+  void getAttributeView() {
+    BeanObject beanObject = createBeanObject();
+
+    Map<String, Object> result = beanObject.getAttributes();
+
+    assertEquals(Map.of("HeapMemoryUsage", "content"), result);
+  }
+
+  @Test
+  void testToString() {
+    BeanObject beanObject = createBeanObject();
+
+    String s = beanObject.toString();
+
+    assertTrue(s.contains(beanObject.domainName()));
+  }
+}
diff --git a/app/src/test/java/org/astraea/metrics/jmx/BeanQueryTest.java b/app/src/test/java/org/astraea/metrics/jmx/BeanQueryTest.java
new file mode 100644
index 00000000..b5c55c9f
--- /dev/null
+++ b/app/src/test/java/org/astraea/metrics/jmx/BeanQueryTest.java
@@ -0,0 +1,102 @@
+package org.astraea.metrics.jmx;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.util.Map;
+import javax.management.MalformedObjectNameException;
+import javax.management.ObjectName;
+import org.junit.jupiter.api.Test;
+
+class BeanQueryTest {
+
+  @Test
+  void domainName() {
+    BeanQuery beanQuery =
+        BeanQuery.builder("java.lang")
+            .property("type", "MemoryManager")
+            .property("name", "CodeCacheManager")
+            .build();
+
+    assertEquals("java.lang", beanQuery.domainName());
+  }
+
+  @Test
+  void propertyView() {
+    BeanQuery beanQuery =
+        BeanQuery.builder("java.lang")
+            .property("type", "MemoryManager")
+            .property("name", "CodeCacheManager")
+            .build();
+
+    assertEquals(
+        Map.of("type", "MemoryManager", "name", "CodeCacheManager"), beanQuery.properties());
+  }
+
+  @Test
+  void objectName() throws MalformedObjectNameException {
+    BeanQuery beanQuery =
+        BeanQuery.builder("java.lang")
+            .property("type", "MemoryManager")
+            .property("name", "CodeCacheManager")
+            .build();
+
+    assertEquals(
+        ObjectName.getInstance("java.lang:type=MemoryManager,name=CodeCacheManager"),
+        beanQuery.objectName());
+  }
+
+  @Test
+  void whereProperty() {
+    BeanQuery.BeanQueryBuilder beanQueryBuilder =
+        BeanQuery.builder("java.lang")
+            .property("type", "MemoryManager")
+            .property("name", "CodeCacheManager");
+
+    BeanQuery beanQuery0 = beanQueryBuilder.build();
+    BeanQuery beanQuery1 = beanQueryBuilder.property("hello", "world").build();
+
+    assertTrue(beanQuery1.properties().containsKey("hello"));
+    assertTrue(beanQuery1.properties().containsValue("world"));
+    assertFalse(beanQuery0.properties().containsKey("hello"));
+    assertFalse(beanQuery0.properties().containsValue("world"));
+  }
+
+  @Test
+  void of() throws MalformedObjectNameException {
+    // ObjectName version
+    BeanQuery beanQueryFromObjectName =
+        BeanQuery.fromObjectName(ObjectName.getInstance("java.lang:type=Memory"));
+    assertEquals("java.lang", beanQueryFromObjectName.domainName());
+    assertEquals(Map.of("type", "Memory"), beanQueryFromObjectName.properties());
+
+    // map version
+    BeanQuery beanQueryFromMap = BeanQuery.builder("java.lang", Map.of("type", "Memory")).build();
+    assertEquals("java.lang", beanQueryFromMap.domainName());
+    assertEquals(Map.of("type", "Memory"), beanQueryFromMap.properties());
+
+    // all under specific domain
+    BeanQuery beanQueryForDomain = BeanQuery.all("java.lang");
+    assertEquals(0, beanQueryForDomain.objectName().getKeyPropertyList().size());
+    assertEquals("java.lang", beanQueryForDomain.objectName().getDomain());
+
+    // all in JMX
+    BeanQuery beanQueryForAllDomain = BeanQuery.all();
+    assertEquals(0, beanQueryForAllDomain.objectName().getKeyPropertyList().size());
+    assertEquals("*", beanQueryForAllDomain.objectName().getDomain());
+
+    // usePropertyListPattern
+    BeanQuery withPattern =
+        BeanQuery.builder("java.lang")
+            .property("type", "MemoryManager")
+            .usePropertyListPattern()
+            .build();
+    assertTrue(withPattern.objectName().isPropertyListPattern());
+
+    // invalid query: query whole domain name without specify "property list pattern"
+    assertThrows(
+        IllegalArgumentException.class,
+        () -> {
+          BeanQuery.builder("java.lang").build();
+        });
+  }
+}
diff --git a/app/src/test/java/org/astraea/metrics/jmx/MBeanClientTest.java b/app/src/test/java/org/astraea/metrics/jmx/MBeanClientTest.java
new file mode 100644
index 00000000..c53b4ad9
--- /dev/null
+++ b/app/src/test/java/org/astraea/metrics/jmx/MBeanClientTest.java
@@ -0,0 +1,475 @@
+package org.astraea.metrics.jmx;
+
+import static java.util.stream.Collectors.*;
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.io.IOException;
+import java.lang.management.ManagementFactory;
+import java.util.*;
+import java.util.stream.IntStream;
+import javax.management.*;
+import javax.management.remote.*;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+class MBeanClientTest {
+
+  private MBeanServer mBeanServer;
+  private JMXConnectorServer jmxServer;
+  private Map<ObjectName, Object> registeredBeans = new HashMap<>();
+
+  private void register(ObjectName name, Object mBean) {
+    registeredBeans.put(name, mBean);
+    try {
+      mBeanServer.registerMBean(mBean, name);
+    } catch (InstanceAlreadyExistsException
+        | MBeanRegistrationException
+        | NotCompliantMBeanException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  private void clearRegisteredMBeans() {
+    registeredBeans.forEach(
+        (name, mbeans) -> {
+          try {
+            mBeanServer.unregisterMBean(name);
+          } catch (InstanceNotFoundException | MBeanRegistrationException e) {
+            throw new RuntimeException(e);
+          }
+        });
+    registeredBeans.clear();
+  }
+
+  @BeforeEach
+  void setUp() throws IOException {
+    JMXServiceURL serviceURL = new JMXServiceURL("service:jmx:rmi://127.0.0.1");
+
+    mBeanServer = ManagementFactory.getPlatformMBeanServer();
+
+    jmxServer = JMXConnectorServerFactory.newJMXConnectorServer(serviceURL, null, mBeanServer);
+    jmxServer.start();
+  }
+
+  @AfterEach
+  void tearDown() throws IOException {
+    jmxServer.stop();
+    clearRegisteredMBeans();
+    mBeanServer = null;
+  }
+
+  @Test
+  void testFetchAttributes() throws Exception {
+    // arrange
+    try (MBeanClient sut = new MBeanClient(jmxServer.getAddress())) {
+      BeanQuery beanQuery = BeanQuery.builder("java.lang").property("type", "Memory").build();
+
+      // act
+      BeanObject beanObject = sut.queryBean(beanQuery);
+
+      // assert
+      assertTrue(beanObject.getProperties().containsKey("type"));
+      assertTrue(beanObject.getAttributes().containsKey("HeapMemoryUsage"));
+      assertTrue(beanObject.getAttributes().containsKey("NonHeapMemoryUsage"));
+    }
+  }
+
+  @Test
+  void testFetchMbeanWithMultipleProperties() throws Exception {
+    // arrange
+    try (MBeanClient sut = new MBeanClient(jmxServer.getAddress())) {
+      BeanQuery query1 =
+          BeanQuery.builder(
+                  "java.lang", Map.of("type", "MemoryManager", "name", "CodeCacheManager"))
+              .build();
+      BeanQuery query2 =
+          BeanQuery.builder("java.lang")
+              .property("type", "MemoryManager")
+              .property("name", "CodeCacheManager")
+              .build();
+
+      // act
+      BeanObject beanObject1 = sut.queryBean(query1);
+      BeanObject beanObject2 = sut.queryBean(query2);
+
+      // assert
+      assertTrue(beanObject1.getProperties().containsKey("type"));
+      assertTrue(beanObject1.getProperties().containsKey("name"));
+      assertTrue(beanObject1.getAttributes().containsKey("MemoryPoolNames"));
+      assertTrue(beanObject1.getAttributes().containsKey("Name"));
+      assertTrue(beanObject1.getAttributes().containsKey("ObjectName"));
+      assertTrue(beanObject1.getAttributes().containsKey("Valid"));
+
+      assertTrue(beanObject2.getProperties().containsKey("type"));
+      assertTrue(beanObject2.getProperties().containsKey("name"));
+      assertTrue(beanObject2.getAttributes().containsKey("MemoryPoolNames"));
+      assertTrue(beanObject2.getAttributes().containsKey("Name"));
+      assertTrue(beanObject2.getAttributes().containsKey("ObjectName"));
+      assertTrue(beanObject2.getAttributes().containsKey("Valid"));
+    }
+  }
+
+  @Test
+  void testFetchSelectedAttributes() throws Exception {
+    // arrange
+    try (MBeanClient sut = new MBeanClient(jmxServer.getAddress())) {
+      BeanQuery beanQuery = BeanQuery.builder("java.lang").property("type", "Memory").build();
+      List<String> selectedAttribute = List.of("HeapMemoryUsage");
+
+      // act
+      BeanObject beanObject = sut.queryBean(beanQuery, selectedAttribute);
+
+      // assert
+      assertTrue(beanObject.getProperties().containsKey("type"));
+      assertTrue(beanObject.getAttributes().containsKey("HeapMemoryUsage"));
+      assertFalse(beanObject.getAttributes().containsKey("NonHeapMemoryUsage"));
+    }
+  }
+
+  @Test
+  void testTryFetchAttributes() throws Exception {
+    // arrange
+    try (MBeanClient sut = new MBeanClient(jmxServer.getAddress())) {
+      BeanQuery beanQuery = BeanQuery.builder("java.lang").property("type", "Memory").build();
+
+      // act
+      Optional<BeanObject> beanObject = sut.tryQueryBean(beanQuery);
+
+      // assert
+      assertTrue(beanObject.isPresent());
+      assertTrue(beanObject.get().getProperties().containsKey("type"));
+      assertTrue(beanObject.get().getAttributes().containsKey("HeapMemoryUsage"));
+      assertTrue(beanObject.get().getAttributes().containsKey("NonHeapMemoryUsage"));
+    }
+  }
+
+  @Test
+  void testTryFetchNonExistsMBean() throws Exception {
+    // arrange
+    try (MBeanClient sut = new MBeanClient(jmxServer.getAddress())) {
+      BeanQuery beanQuery = BeanQuery.builder("java.lang").property("type", "HelloWorld").build();
+
+      // act
+      Optional<BeanObject> beanObject = sut.tryQueryBean(beanQuery);
+
+      // assert
+      assertTrue(beanObject.isEmpty());
+    }
+  }
+
+  @Test
+  void testTryFetchSelectedAttributes() throws Exception {
+    // arrange
+    try (MBeanClient sut = new MBeanClient(jmxServer.getAddress())) {
+      BeanQuery beanQuery = BeanQuery.builder("java.lang").property("type", "Memory").build();
+      List<String> selectedAttribute = List.of("HeapMemoryUsage");
+
+      // act
+      Optional<BeanObject> beanObject = sut.tryQueryBean(beanQuery, selectedAttribute);
+
+      // assert
+      assertTrue(beanObject.isPresent());
+      assertTrue(beanObject.get().getProperties().containsKey("type"));
+      assertTrue(beanObject.get().getAttributes().containsKey("HeapMemoryUsage"));
+      assertFalse(beanObject.get().getAttributes().containsKey("NonHeapMemoryUsage"));
+    }
+  }
+
+  @Test
+  void testQueryBeans() throws Exception {
+    // arrange 1 query beans
+    try (MBeanClient sut = new MBeanClient(jmxServer.getAddress())) {
+      BeanQuery beanQuery = BeanQuery.builder("java.lang").property("type", "C*").build();
+
+      // act 1
+      Collection<BeanObject> beanObjects = sut.queryBeans(beanQuery);
+
+      // assert 1
+      assertEquals(2, beanObjects.size());
+      assertTrue(
+          beanObjects.stream().anyMatch(x -> x.getProperties().get("type").equals("ClassLoading")));
+      assertTrue(
+          beanObjects.stream().anyMatch(x -> x.getProperties().get("type").equals("Compilation")));
+
+      // arrange 2 look into ClassLoading content
+
+      // act
+      Optional<BeanObject> classLoading =
+          beanObjects.stream()
+              .filter(x -> x.getProperties().get("type").equals("ClassLoading"))
+              .findFirst();
+
+      // assert
+      assertTrue(classLoading.isPresent());
+      assertEquals("java.lang", classLoading.get().domainName());
+      assertEquals(5, classLoading.get().getAttributes().size());
+      assertTrue(classLoading.get().getAttributes().containsKey("LoadedClassCount"));
+      assertTrue(classLoading.get().getAttributes().containsKey("ObjectName"));
+      assertTrue(classLoading.get().getAttributes().containsKey("TotalLoadedClassCount"));
+      assertTrue(classLoading.get().getAttributes().containsKey("UnloadedClassCount"));
+      assertTrue(classLoading.get().getAttributes().containsKey("Verbose"));
+      assertTrue(classLoading.get().getAttributes().get("Verbose") instanceof Boolean);
+    }
+  }
+
+  @Test
+  void testQueryNonExistsBeans() throws Exception {
+    // arrange
+    try (MBeanClient sut = new MBeanClient(jmxServer.getAddress())) {
+      BeanQuery beanQuery = BeanQuery.builder("java.lang").property("type", "Something").build();
+
+      // act
+      Collection<BeanObject> beanObjects = sut.queryBeans(beanQuery);
+
+      // assert
+      assertEquals(0, beanObjects.size());
+    }
+  }
+
+  @Test
+  void testFetchNonExistsBeans() throws Exception {
+    // arrange
+    try (MBeanClient sut = new MBeanClient(jmxServer.getAddress())) {
+      BeanQuery beanQuery = BeanQuery.builder("java.lang").property("type", "Something").build();
+
+      // act assert
+      assertThrows(InstanceNotFoundException.class, () -> sut.queryBean(beanQuery));
+      assertThrows(
+          InstanceNotFoundException.class, () -> sut.queryBean(beanQuery, Collections.emptyList()));
+      assertDoesNotThrow(
+          () -> {
+            Optional<BeanObject> beanObject = sut.tryQueryBean(beanQuery);
+            assertTrue(beanObject.isEmpty());
+          });
+      assertDoesNotThrow(
+          () -> {
+            Optional<BeanObject> beanObject = sut.tryQueryBean(beanQuery, Collections.emptyList());
+            assertTrue(beanObject.isEmpty());
+          });
+    }
+  }
+
+  @Test
+  void testUseClosedClientWillThrowError() throws Exception {
+    // arrange
+    MBeanClient sut = new MBeanClient(jmxServer.getAddress());
+    BeanQuery query = BeanQuery.builder("java.lang").property("type", "Memory").build();
+
+    // act
+    sut.close();
+
+    // assert
+    assertThrows(IllegalStateException.class, () -> sut.queryBean(query));
+    assertThrows(IllegalStateException.class, () -> sut.queryBean(query, Collections.emptyList()));
+    assertThrows(IllegalStateException.class, () -> sut.tryQueryBean(query));
+    assertThrows(
+        IllegalStateException.class, () -> sut.tryQueryBean(query, Collections.emptyList()));
+    assertThrows(IllegalStateException.class, () -> sut.queryBeans(query));
+  }
+
+  @Test
+  void testCloseOnceMoreWillThrowError() throws Exception {
+    // arrange
+    MBeanClient sut = new MBeanClient(jmxServer.getAddress());
+
+    // act
+    sut.close();
+
+    // assert
+    assertThrows(IllegalStateException.class, sut::close);
+    assertThrows(IllegalStateException.class, sut::close);
+    assertThrows(IllegalStateException.class, sut::close);
+    assertThrows(IllegalStateException.class, sut::close);
+    assertThrows(IllegalStateException.class, sut::close);
+  }
+
+  @Test
+  void testGetAllMBeans() throws Exception {
+    // arrange
+    try (MBeanClient client = new MBeanClient(jmxServer.getAddress())) {
+
+      // act
+      Collection<BeanObject> beanObjects = client.queryBeans(BeanQuery.all());
+
+      // assert
+      assertTrue(beanObjects.stream().anyMatch(x -> x.domainName().equals("java.lang")));
+      assertTrue(beanObjects.stream().anyMatch(x -> x.domainName().equals("java.nio")));
+    }
+  }
+
+  @Test
+  void testGetAllMBeansUnderSpecificDomainName() throws Exception {
+    // arrange
+    try (MBeanClient client = new MBeanClient(jmxServer.getAddress())) {
+
+      // act
+      Collection<BeanObject> beanObjects = client.queryBeans(BeanQuery.all("java.lang"));
+
+      // assert
+      assertTrue(beanObjects.size() > 1);
+      assertTrue(beanObjects.stream().allMatch(x -> x.domainName().equals("java.lang")));
+    }
+  }
+
+  @Test
+  void testGetAllMBeansUnderSpecificDomainNamePattern() throws Exception {
+    // arrange
+    try (MBeanClient client = new MBeanClient(jmxServer.getAddress())) {
+
+      // act
+      Collection<BeanObject> beanObjects = client.queryBeans(BeanQuery.all("java.*"));
+
+      // assert
+      assertTrue(beanObjects.size() > 1);
+      assertTrue(beanObjects.stream().allMatch(x -> x.domainName().matches("java.*")));
+    }
+  }
+
+  @Test
+  void testUsePropertyListPattern() throws Exception {
+    // arrange
+    try (MBeanClient client = new MBeanClient(jmxServer.getAddress())) {
+      BeanQuery patternQuery =
+          BeanQuery.builder("java.lang").property("type", "*").usePropertyListPattern().build();
+
+      // act
+      Collection<BeanObject> beanObjects = client.queryBeans(patternQuery);
+
+      // assert
+      /*
+      It might be hard to understand what this test is testing for.
+      The keypoint is we are using BeanQueryBuilder#usePropertyListPattern()
+
+      Without it the query will be "java.lang:type=*"
+      And we only match the following
+      java.lang:{type=OperatingSystem}
+      java.lang:{type=Threading}
+      java.lang:{type=ClassLoading}
+      java.lang:{type=Compilation}
+      java.lang:{type=Memory}
+      java.lang:{type=Runtime}
+
+      With it the query will be "java.lang:type=*,*"
+      And we will match the following
+      java.lang:{type=MemoryPool, name=CodeHeap 'non-nmethods'}
+      java.lang:{type=GarbageCollector, name=G1 Young Generation}
+      java.lang:{type=Runtime}
+      java.lang:{type=OperatingSystem}
+      java.lang:{type=Threading}
+      java.lang:{type=MemoryPool, name=G1 Old Gen}
+      java.lang:{type=MemoryPool, name=CodeHeap 'profiled nmethods'}
+      java.lang:{type=MemoryPool, name=G1 Eden Space}
+      java.lang:{type=MemoryPool, name=Metaspace}
+      java.lang:{type=GarbageCollector, name=G1 Old Generation}
+      java.lang:{type=Memory}
+      java.lang:{type=MemoryPool, name=G1 Survivor Space}
+      java.lang:{type=Compilation}
+      java.lang:{type=MemoryManager, name=CodeCacheManager}
+      java.lang:{type=MemoryPool, name=CodeHeap 'non-profiled nmethods'}
+      java.lang:{type=MemoryManager, name=Metaspace Manager}
+      java.lang:{type=ClassLoading}
+      Notice how everything with "type" is match, even those with "name"
+      */
+      assertTrue(beanObjects.stream().anyMatch(x -> x.getProperties().containsKey("name")));
+    }
+  }
+
+  @Test
+  void testListDomains() throws Exception {
+    // arrange
+    try (MBeanClient client = new MBeanClient(jmxServer.getAddress())) {
+
+      // act
+      List<String> domains = client.listDomains();
+
+      // assert
+      assertTrue(domains.contains("java.lang"));
+      assertTrue(domains.contains("java.nio"));
+    }
+  }
+
+  @Test
+  void testGetAddress() throws Exception {
+    // arrange
+    try (MBeanClient client = new MBeanClient(jmxServer.getAddress())) {
+
+      // act
+      JMXServiceURL address = client.getAddress();
+
+      // assert
+      assertEquals(jmxServer.getAddress(), address);
+    }
+  }
+
+  @Test
+  void testCustomMBeanWith500Attributes() throws Exception {
+    // arrange
+    Map<String, Integer> mbeanPayload =
+        IntStream.range(0, 500).boxed().collect(toMap(i -> "attribute" + i, i -> i));
+    Object customMBean0 = Utility.createReadOnlyDynamicMBean(mbeanPayload);
+    ObjectName objectName0 = ObjectName.getInstance("com.example:type=test0");
+
+    register(objectName0, customMBean0);
+
+    try (MBeanClient client = new MBeanClient(jmxServer.getAddress())) {
+
+      // act
+      Collection<BeanObject> all =
+          client.queryBeans(BeanQuery.builder("com.example").property("type", "test*").build());
+
+      // assert
+      Map<String, Object> mergedCollect =
+          all.stream()
+              .flatMap(beanObject -> beanObject.getAttributes().entrySet().stream())
+              .collect(toMap(Map.Entry::getKey, Map.Entry::getValue));
+
+      assertEquals(500, mergedCollect.size());
+      assertEquals(mbeanPayload, mergedCollect);
+    }
+  }
+
+  @Test
+  void testWith100CustomMBeans() throws Exception {
+    // arrange
+    for (int i = 0; i < 100; i++) {
+      ObjectName objectName = ObjectName.getInstance("com.example:type=test" + i);
+      Object mbean = Utility.createReadOnlyDynamicMBean(Map.of("attribute", i));
+      register(objectName, mbean);
+    }
+
+    try (MBeanClient client = new MBeanClient(jmxServer.getAddress())) {
+
+      // act
+      Collection<BeanObject> all =
+          client.queryBeans(BeanQuery.builder("com.example").property("type", "test*").build());
+
+      // assert
+      List<Map.Entry<String, String>> properties =
+          all.stream()
+              .map(BeanObject::getProperties)
+              .map(Map::entrySet)
+              .flatMap(Collection::stream)
+              .collect(toList());
+      Set<String> propKeys = properties.stream().map(Map.Entry::getKey).collect(toSet());
+      Set<String> propValues = properties.stream().map(Map.Entry::getValue).collect(toSet());
+
+      List<Map.Entry<String, Object>> attributes =
+          all.stream()
+              .map(BeanObject::getAttributes)
+              .map(Map::entrySet)
+              .flatMap(Collection::stream)
+              .collect(toList());
+      Set<String> attrKeys = attributes.stream().map(Map.Entry::getKey).collect(toSet());
+      Set<Object> attrValues = attributes.stream().map(Map.Entry::getValue).collect(toSet());
+
+      assertEquals(100, all.size());
+      assertEquals(100, properties.size());
+      assertEquals(100, attributes.size());
+      assertEquals(Set.of("type"), propKeys);
+      assertEquals(Set.of("attribute"), attrKeys);
+      assertEquals(IntStream.range(0, 100).mapToObj(x -> "test" + x).collect(toSet()), propValues);
+      assertEquals(IntStream.range(0, 100).boxed().collect(toSet()), attrValues);
+    }
+  }
+}
diff --git a/app/src/test/java/org/astraea/metrics/jmx/Utility.java b/app/src/test/java/org/astraea/metrics/jmx/Utility.java
new file mode 100644
index 00000000..cccf1d91
--- /dev/null
+++ b/app/src/test/java/org/astraea/metrics/jmx/Utility.java
@@ -0,0 +1,87 @@
+package org.astraea.metrics.jmx;
+
+import java.util.Map;
+import javax.management.*;
+
+public class Utility {
+
+  static class DynamicMBean implements javax.management.DynamicMBean {
+
+    private final MBeanInfo mBeanInfo;
+    private final Map<String, ?> readonlyAttributes;
+
+    DynamicMBean(Map<String, ?> readonlyAttributes) {
+      this.readonlyAttributes = readonlyAttributes;
+      MBeanAttributeInfo[] attributeInfos =
+          readonlyAttributes.entrySet().stream()
+              .map(
+                  entry ->
+                      new MBeanAttributeInfo(
+                          entry.getKey(),
+                          entry.getValue().getClass().getName(),
+                          "",
+                          true,
+                          false,
+                          false))
+              .toArray(MBeanAttributeInfo[]::new);
+      this.mBeanInfo =
+          new MBeanInfo(
+              DynamicMBean.class.getName(),
+              "Readonly Dynamic MBean for Testing Purpose",
+              attributeInfos,
+              new MBeanConstructorInfo[0],
+              new MBeanOperationInfo[0],
+              new MBeanNotificationInfo[0]);
+    }
+
+    @Override
+    public Object getAttribute(String attributeName) throws AttributeNotFoundException {
+      if (readonlyAttributes.containsKey(attributeName))
+        return readonlyAttributes.get(attributeName);
+
+      throw new AttributeNotFoundException();
+    }
+
+    @Override
+    public void setAttribute(Attribute attribute) {
+      throw new RuntimeOperationsException(new IllegalArgumentException("readonly"));
+    }
+
+    @Override
+    public AttributeList getAttributes(String[] attributes) {
+      final AttributeList list = new AttributeList();
+      for (String attribute : attributes) {
+        if (readonlyAttributes.containsKey(attribute))
+          list.add(new Attribute(attribute, readonlyAttributes.get(attribute)));
+      }
+
+      return list;
+    }
+
+    @Override
+    public AttributeList setAttributes(AttributeList attributes) {
+      throw new RuntimeOperationsException(new IllegalArgumentException("readonly"));
+    }
+
+    @Override
+    public Object invoke(String actionName, Object[] params, String[] signature)
+        throws MBeanException, ReflectionException {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public MBeanInfo getMBeanInfo() {
+      return mBeanInfo;
+    }
+  }
+
+  /**
+   * Create a readonly dynamic MBeans
+   *
+   * @param attributes for each key/value pair. the key string represent the attribute name, and the
+   *     value represent the attribute value
+   */
+  public static DynamicMBean createReadOnlyDynamicMBean(Map<String, ?> attributes) {
+    return new DynamicMBean(attributes);
+  }
+}
diff --git a/app/src/test/java/org/astraea/performance/latency/CloseableThreadTest.java b/app/src/test/java/org/astraea/performance/latency/CloseableThreadTest.java
new file mode 100644
index 00000000..bcfae7a4
--- /dev/null
+++ b/app/src/test/java/org/astraea/performance/latency/CloseableThreadTest.java
@@ -0,0 +1,72 @@
+package org.astraea.performance.latency;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class CloseableThreadTest {
+
+  @Test
+  void testExecuteAndCleanup() throws InterruptedException {
+    var executeCount = new AtomicInteger();
+    var cleanupCount = new AtomicInteger();
+    var thread =
+        new CloseableThread() {
+
+          @Override
+          void execute() {
+            executeCount.incrementAndGet();
+          }
+
+          @Override
+          void cleanup() {
+            cleanupCount.incrementAndGet();
+          }
+        };
+
+    var service = Executors.newSingleThreadExecutor();
+    service.execute(thread);
+
+    TimeUnit.SECONDS.sleep(1);
+
+    thread.close();
+    service.shutdown();
+    Assertions.assertTrue(service.awaitTermination(10, TimeUnit.SECONDS));
+
+    Assertions.assertTrue(executeCount.get() > 0);
+    Assertions.assertEquals(1, cleanupCount.get());
+  }
+
+  @Test
+  void testThrowException() {
+    var thread =
+        new CloseableThread() {
+          @Override
+          void execute() {
+            close();
+          }
+        };
+
+    Assertions.assertEquals(
+        "Should not call close() in execute().",
+        Assertions.assertThrows(RuntimeException.class, thread::run).getMessage());
+  }
+
+  @Test
+  void testExecuteOnce() {
+    var executeCount = new AtomicInteger();
+    var thread =
+        new CloseableThread(true) {
+          @Override
+          void execute() {
+            executeCount.incrementAndGet();
+          }
+        };
+
+    thread.run();
+
+    Assertions.assertEquals(1, executeCount.get());
+  }
+}
diff --git a/app/src/test/java/org/astraea/performance/latency/ConsumerThreadTest.java b/app/src/test/java/org/astraea/performance/latency/ConsumerThreadTest.java
new file mode 100644
index 00000000..9d073489
--- /dev/null
+++ b/app/src/test/java/org/astraea/performance/latency/ConsumerThreadTest.java
@@ -0,0 +1,44 @@
+package org.astraea.performance.latency;
+
+import java.util.Collections;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.common.TopicPartition;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class ConsumerThreadTest {
+  private final String topic = "topic-" + System.currentTimeMillis();
+  private final DataManager dataManager = DataManager.of(topic, 10);
+
+  @Test
+  void testExecute() throws InterruptedException {
+    var count = new AtomicInteger(0);
+    var producerRecord = dataManager.producerRecord();
+    var consumerRecord = FakeComponentFactory.toConsumerRecord(producerRecord);
+    try (Consumer consumer =
+        () -> {
+          count.incrementAndGet();
+          return new ConsumerRecords<>(
+              Collections.singletonMap(
+                  new TopicPartition(topic, 1), Collections.singletonList(consumerRecord)));
+        }) {
+      var tracker = new MeterTracker("test consumer");
+      Assertions.assertEquals(0, tracker.maxLatency());
+      Assertions.assertEquals(0, tracker.averageLatency());
+
+      var thread = new ConsumerThread(dataManager, tracker, consumer);
+      dataManager.sendingRecord(producerRecord, System.currentTimeMillis());
+
+      // add the latency
+      TimeUnit.SECONDS.sleep(1);
+      thread.execute();
+      Assertions.assertEquals(1, count.get());
+      Assertions.assertEquals(1, dataManager.numberOfProducerRecords());
+      Assertions.assertEquals(1, tracker.count());
+      Assertions.assertNotEquals(0, tracker.maxLatency());
+      Assertions.assertNotEquals(0, tracker.averageLatency());
+    }
+  }
+}
diff --git a/app/src/test/java/org/astraea/performance/latency/DataManagerTest.java b/app/src/test/java/org/astraea/performance/latency/DataManagerTest.java
new file mode 100644
index 00000000..5e918301
--- /dev/null
+++ b/app/src/test/java/org/astraea/performance/latency/DataManagerTest.java
@@ -0,0 +1,42 @@
+package org.astraea.performance.latency;
+
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class DataManagerTest {
+  private final String topic = "topic-" + String.valueOf(System.currentTimeMillis());
+  private final DataManager dataManager = DataManager.of(topic, 10);
+
+  @Test
+  void testRandomString() {
+    var result =
+        IntStream.range(0, 10)
+            .mapToObj(i -> DataManager.randomString(10))
+            .collect(Collectors.toSet());
+    Assertions.assertEquals(10, result.size());
+    result.forEach(r -> Assertions.assertEquals(10, r.length()));
+  }
+
+  @Test
+  void testTakeRecord() {
+    var record = dataManager.producerRecord();
+    Assertions.assertEquals(topic, record.topic());
+    Assertions.assertNotNull(record.key());
+    Assertions.assertNotNull(record.value());
+    Assertions.assertNotNull(record.headers());
+  }
+
+  @Test
+  void testCompleteRecord() {
+    Assertions.assertEquals(0, dataManager.numberOfProducerRecords());
+    var record = dataManager.producerRecord();
+    dataManager.sendingRecord(record, System.currentTimeMillis());
+    Assertions.assertEquals(1, dataManager.numberOfProducerRecords());
+    Assertions.assertEquals(record, dataManager.removeSendingRecord(record.key()).getKey());
+
+    Assertions.assertThrows(
+        NullPointerException.class, () -> dataManager.removeSendingRecord("aa".getBytes()));
+  }
+}
diff --git a/app/src/test/java/org/astraea/performance/latency/End2EndLatencyTest.java b/app/src/test/java/org/astraea/performance/latency/End2EndLatencyTest.java
new file mode 100644
index 00000000..60d702bb
--- /dev/null
+++ b/app/src/test/java/org/astraea/performance/latency/End2EndLatencyTest.java
@@ -0,0 +1,136 @@
+package org.astraea.performance.latency;
+
+import java.time.Duration;
+import java.util.concurrent.*;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class End2EndLatencyTest {
+
+  @Test
+  void testIncorrectParameters() {
+    Assertions.assertThrows(
+        IllegalArgumentException.class,
+        () -> End2EndLatency.parameters(new String[] {End2EndLatency.BROKERS_KEY, ""}));
+    Assertions.assertThrows(
+        IllegalArgumentException.class,
+        () ->
+            End2EndLatency.parameters(
+                new String[] {
+                  End2EndLatency.BROKERS_KEY, "localhost:11111",
+                  End2EndLatency.TOPIC_KEY, ""
+                }));
+    Assertions.assertThrows(
+        IllegalArgumentException.class,
+        () ->
+            End2EndLatency.parameters(
+                new String[] {
+                  End2EndLatency.BROKERS_KEY, "localhost:11111",
+                  End2EndLatency.DURATION_KEY, "-1"
+                }));
+    Assertions.assertThrows(
+        IllegalArgumentException.class,
+        () ->
+            End2EndLatency.parameters(
+                new String[] {
+                  End2EndLatency.BROKERS_KEY, "localhost:11111",
+                  End2EndLatency.FLUSH_DURATION_KEY, "-1"
+                }));
+    Assertions.assertThrows(
+        IllegalArgumentException.class,
+        () ->
+            End2EndLatency.parameters(
+                new String[] {
+                  End2EndLatency.BROKERS_KEY, "localhost:11111",
+                  End2EndLatency.PRODUCERS_KEY, "-1"
+                }));
+    Assertions.assertThrows(
+        IllegalArgumentException.class,
+        () ->
+            End2EndLatency.parameters(
+                new String[] {
+                  End2EndLatency.BROKERS_KEY, "localhost:11111",
+                  End2EndLatency.PRODUCERS_KEY, "0"
+                }));
+    Assertions.assertThrows(
+        IllegalArgumentException.class,
+        () ->
+            End2EndLatency.parameters(
+                new String[] {
+                  End2EndLatency.BROKERS_KEY, "localhost:11111",
+                  End2EndLatency.CONSUMERS_KEY, "-1"
+                }));
+
+    Assertions.assertEquals(
+        0,
+        End2EndLatency.parameters(
+                new String[] {
+                  End2EndLatency.BROKERS_KEY, "localhost:11111",
+                  End2EndLatency.CONSUMERS_KEY, "0"
+                })
+            .numberOfConsumers);
+  }
+
+  @Test
+  void testParameters() {
+    var brokers = "broker00:12345";
+    var topic = "topic";
+    var numberOfProducers = 100;
+    var numberOfConsumers = 10;
+    var duration = Duration.ofSeconds(10);
+    var valueSize = 888;
+    var flushDuration = Duration.ofSeconds(3);
+    var parameters =
+        End2EndLatency.parameters(
+            new String[] {
+              End2EndLatency.BROKERS_KEY,
+              brokers,
+              End2EndLatency.CONSUMERS_KEY,
+              String.valueOf(numberOfConsumers),
+              End2EndLatency.DURATION_KEY,
+              String.valueOf(duration.toSeconds()),
+              End2EndLatency.PRODUCERS_KEY,
+              String.valueOf(numberOfProducers),
+              End2EndLatency.TOPIC_KEY,
+              topic,
+              End2EndLatency.VALUE_SIZE_KEY,
+              String.valueOf(valueSize),
+              End2EndLatency.FLUSH_DURATION_KEY,
+              String.valueOf(flushDuration.toSeconds())
+            });
+    Assertions.assertEquals(brokers, parameters.brokers);
+    Assertions.assertEquals(topic, parameters.topic);
+    Assertions.assertEquals(numberOfConsumers, parameters.numberOfConsumers);
+    Assertions.assertEquals(numberOfProducers, parameters.numberOfProducers);
+    Assertions.assertEquals(duration, parameters.duration);
+    Assertions.assertEquals(valueSize, parameters.valueSize);
+    Assertions.assertEquals(flushDuration, parameters.flushDuration);
+  }
+
+  @Test
+  void testExecute() throws Exception {
+    var factory = new FakeComponentFactory();
+    try (var r =
+        End2EndLatency.execute(
+            factory,
+            new End2EndLatency.Parameters(
+                "brokers", "topic", 1, 1, Duration.ofSeconds(1), 10, Duration.ofSeconds(1)))) {
+      TimeUnit.SECONDS.sleep(2);
+    }
+
+    // check producers count
+    Assertions.assertTrue(factory.producerSendCount.get() > 0);
+    Assertions.assertTrue(factory.producerFlushCount.get() > 0);
+    Assertions.assertEquals(1, factory.producerCloseCount.get());
+
+    // check consumer count
+    Assertions.assertTrue(factory.consumerPoolCount.get() > 0);
+    Assertions.assertEquals(1, factory.consumerWakeupCount.get());
+    Assertions.assertEquals(1, factory.consumerCloseCount.get());
+
+    // check admin topic count
+    Assertions.assertEquals(1, factory.topicAdminListCount.get());
+    Assertions.assertEquals(1, factory.topicAdminCloseCount.get());
+    Assertions.assertEquals(1, factory.topicAdminCreateCount.get());
+  }
+}
diff --git a/app/src/test/java/org/astraea/performance/latency/FakeComponentFactory.java b/app/src/test/java/org/astraea/performance/latency/FakeComponentFactory.java
new file mode 100644
index 00000000..ed66dd32
--- /dev/null
+++ b/app/src/test/java/org/astraea/performance/latency/FakeComponentFactory.java
@@ -0,0 +1,140 @@
+package org.astraea.performance.latency;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Set;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.WakeupException;
+import org.apache.kafka.common.header.internals.RecordHeaders;
+import org.apache.kafka.common.record.TimestampType;
+
+class FakeComponentFactory implements ComponentFactory {
+
+  static ConsumerRecord<byte[], byte[]> toConsumerRecord(
+      ProducerRecord<byte[], byte[]> producerRecord) {
+    return new ConsumerRecord<>(
+        producerRecord.topic(),
+        1,
+        1L,
+        producerRecord.timestamp() == null
+            ? System.currentTimeMillis()
+            : producerRecord.timestamp(),
+        TimestampType.CREATE_TIME,
+        1L,
+        producerRecord.key() == null ? 0 : producerRecord.key().length,
+        producerRecord.value() == null ? 0 : producerRecord.value().length,
+        producerRecord.key(),
+        producerRecord.value(),
+        new RecordHeaders(producerRecord.headers()));
+  }
+
+  final AtomicInteger producerSendCount = new AtomicInteger();
+  final AtomicInteger producerFlushCount = new AtomicInteger();
+  final AtomicInteger producerCloseCount = new AtomicInteger();
+
+  final AtomicInteger consumerPoolCount = new AtomicInteger();
+  final AtomicInteger consumerWakeupCount = new AtomicInteger();
+  final AtomicInteger consumerCloseCount = new AtomicInteger();
+
+  final AtomicInteger topicAdminListCount = new AtomicInteger();
+  final AtomicInteger topicAdminCreateCount = new AtomicInteger();
+  final AtomicInteger topicAdminCloseCount = new AtomicInteger();
+
+  /** save all records from producer */
+  final BlockingQueue<ProducerRecord<byte[], byte[]>> allRecords = new LinkedBlockingQueue<>();
+
+  /** save non-consumed records */
+  final BlockingQueue<ProducerRecord<byte[], byte[]>> records = new LinkedBlockingQueue<>();
+
+  @Override
+  public Producer producer() {
+    return new Producer() {
+
+      @Override
+      public CompletionStage<RecordMetadata> send(ProducerRecord<byte[], byte[]> record) {
+        producerSendCount.incrementAndGet();
+        allRecords.add(record);
+        records.add(record);
+        return CompletableFuture.completedFuture(
+            new RecordMetadata(
+                new TopicPartition(record.topic(), 1),
+                1L,
+                1L,
+                1L,
+                1L,
+                record.key() == null ? 0 : record.key().length,
+                record.value() == null ? 0 : record.value().length));
+      }
+
+      @Override
+      public void flush() {
+        producerFlushCount.incrementAndGet();
+      }
+
+      @Override
+      public void close() {
+        producerCloseCount.incrementAndGet();
+      }
+    };
+  }
+
+  @Override
+  public Consumer createConsumer() {
+    return new Consumer() {
+
+      @Override
+      public ConsumerRecords<byte[], byte[]> poll() {
+        consumerPoolCount.incrementAndGet();
+        try {
+          var producerRecord = records.poll(1, TimeUnit.SECONDS);
+          if (producerRecord == null) return new ConsumerRecords<>(Collections.emptyMap());
+          return new ConsumerRecords<>(
+              Collections.singletonMap(
+                  new TopicPartition(producerRecord.topic(), 1),
+                  Collections.singletonList(toConsumerRecord(producerRecord))));
+        } catch (InterruptedException e) {
+          throw new WakeupException();
+        }
+      }
+
+      @Override
+      public void wakeup() {
+        consumerWakeupCount.incrementAndGet();
+      }
+
+      @Override
+      public void close() {
+        consumerCloseCount.incrementAndGet();
+      }
+    };
+  }
+
+  @Override
+  public TopicAdmin createTopicAdmin() {
+    return new TopicAdmin() {
+
+      @Override
+      public Set<String> listTopics() {
+        topicAdminListCount.incrementAndGet();
+        return Collections.emptySet();
+      }
+
+      @Override
+      public void createTopics(Collection<NewTopic> newTopics) {
+        topicAdminCreateCount.incrementAndGet();
+      }
+
+      @Override
+      public void close() {
+        topicAdminCloseCount.incrementAndGet();
+      }
+    };
+  }
+}
diff --git a/app/src/test/java/org/astraea/performance/latency/KafkaUtilsTest.java b/app/src/test/java/org/astraea/performance/latency/KafkaUtilsTest.java
new file mode 100644
index 00000000..68b954e2
--- /dev/null
+++ b/app/src/test/java/org/astraea/performance/latency/KafkaUtilsTest.java
@@ -0,0 +1,91 @@
+package org.astraea.performance.latency;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class KafkaUtilsTest {
+
+  @Test
+  void testCreateTopic() {
+    var numberOfPartitions = 1000;
+    var existentTopic = "existentTopic";
+    var count = new AtomicInteger();
+    var topicAdmin =
+        new TopicAdmin() {
+
+          @Override
+          public Set<String> listTopics() {
+            return Collections.singleton(existentTopic);
+          }
+
+          @Override
+          public void createTopics(Collection<NewTopic> newTopics) {
+            count.incrementAndGet();
+            Assertions.assertEquals(1, newTopics.size());
+            Assertions.assertEquals(
+                numberOfPartitions, newTopics.iterator().next().numPartitions());
+          }
+        };
+    KafkaUtils.createTopicIfNotExist(topicAdmin, existentTopic, 10);
+    Assertions.assertEquals(0, count.get());
+
+    KafkaUtils.createTopicIfNotExist(topicAdmin, "nonexistent", numberOfPartitions);
+    Assertions.assertEquals(1, count.get());
+  }
+
+  @Test
+  void testCreateHeader() {
+    var header = KafkaUtils.header("a", "a".getBytes());
+    Assertions.assertEquals("a", header.key());
+    Assertions.assertArrayEquals("a".getBytes(), header.value());
+  }
+
+  @Test
+  void testEqualOfHeader() {
+    var headers0 =
+        IntStream.range(0, 5)
+            .mapToObj(i -> KafkaUtils.header(String.valueOf(i), String.valueOf(i).getBytes()))
+            .collect(Collectors.toList());
+    var headers1 =
+        IntStream.range(0, 5)
+            .mapToObj(i -> KafkaUtils.header(String.valueOf(i), String.valueOf(i).getBytes()))
+            .collect(Collectors.toList());
+
+    // reorder
+    IntStream.range(0, 10)
+        .forEach(
+            i -> {
+              Collections.shuffle(headers1);
+              Collections.shuffle(headers0);
+              Assertions.assertTrue(KafkaUtils.equal(headers0, headers0));
+              Assertions.assertTrue(KafkaUtils.equal(headers1, headers1));
+              Assertions.assertTrue(KafkaUtils.equal(headers0, headers1));
+              Assertions.assertTrue(KafkaUtils.equal(headers1, headers0));
+            });
+  }
+
+  @Test
+  void testEqualOfRecord() {
+    var topic = "topic";
+    var key = "key".getBytes();
+    var value = "value".getBytes();
+    var headers = Collections.singleton(KafkaUtils.header("a", "b".getBytes()));
+    var producerRecords =
+        Arrays.asList(
+            new ProducerRecord<>(topic, null, key, value, headers),
+            new ProducerRecord<>(topic, null, 100L, key, value, headers));
+    producerRecords.forEach(
+        record ->
+            Assertions.assertTrue(
+                KafkaUtils.equal(record, FakeComponentFactory.toConsumerRecord(record))));
+  }
+}
diff --git a/app/src/test/java/org/astraea/performance/latency/MeterTrackerTest.java b/app/src/test/java/org/astraea/performance/latency/MeterTrackerTest.java
new file mode 100644
index 00000000..235343cb
--- /dev/null
+++ b/app/src/test/java/org/astraea/performance/latency/MeterTrackerTest.java
@@ -0,0 +1,21 @@
+package org.astraea.performance.latency;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class MeterTrackerTest {
+
+  @Test
+  void testRecord() {
+    var tracker = new MeterTracker("test tracker");
+    Assertions.assertEquals(0, tracker.count());
+    Assertions.assertEquals(0, tracker.maxLatency());
+    Assertions.assertEquals(0, tracker.averageLatency());
+
+    tracker.record(10, 10);
+    Assertions.assertEquals(1, tracker.count());
+    Assertions.assertEquals(10, tracker.bytes());
+    Assertions.assertEquals(10, tracker.maxLatency());
+    Assertions.assertEquals(10, tracker.averageLatency());
+  }
+}
diff --git a/app/src/test/java/org/astraea/performance/latency/ProducerThreadTest.java b/app/src/test/java/org/astraea/performance/latency/ProducerThreadTest.java
new file mode 100644
index 00000000..a4cf4769
--- /dev/null
+++ b/app/src/test/java/org/astraea/performance/latency/ProducerThreadTest.java
@@ -0,0 +1,32 @@
+package org.astraea.performance.latency;
+
+import java.time.Duration;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.TopicPartition;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class ProducerThreadTest {
+  private final String topic = "topic-" + String.valueOf(System.currentTimeMillis());
+  private final DataManager dataManager = DataManager.of(topic, 10);
+
+  @Test
+  void testExecute() {
+    var count = new AtomicInteger(0);
+    try (Producer producer =
+        record -> {
+          count.incrementAndGet();
+          return CompletableFuture.completedFuture(
+              new RecordMetadata(new TopicPartition(record.topic(), 1), 1L, 1L, 1L, 1L, 1, 1));
+        }) {
+      var thread =
+          new ProducerThread(
+              dataManager, new MeterTracker("test producer"), producer, Duration.ZERO);
+      thread.execute();
+      Assertions.assertEquals(1, count.get());
+      Assertions.assertEquals(1, dataManager.numberOfProducerRecords());
+    }
+  }
+}