Repository: skiptests/astraea
Pull request #106 (2021-11-12 05:48:02)

Random record and producing duration setting
Implement functions in #26. 
In this pull request, changes are
* Add record producing duration.
* Random record size.
* Random record content.
* Change default producing time and records. By default, it runs for about 1 hour. And produce 1,000,000,000 many records.
* Add unit test of random payload generator.
* Change consumer completed rate to ratio of number of consumed and number of produced

chia7712: @chinghongfang 那個百分比是不是要修一下？如果是依照時間來停止的話
chinghongfang: @chia7712 好的，現在的做法是雙限制，任一條件到達，就會停止。
可改成輸出百分比較高的(時間%, 數量%)，但目前想到可能會發生 consumer完成度 > producer完成度。
我再思考如何解決。
chinghongfang: > 但目前想到可能會發生 consumer完成度 > producer完成度。

想過以後發現，並不會出現 "consumer完成度 > producer完成度" 的情況。

但是有一個問題，依時間來顯示百分比，會不會讓人以為 consumer"完全跟上" producer 的速度？
chia7712: > 依時間來顯示百分比，會不會讓人以為 consumer"完全跟上" producer 的速度？

或許兩者（producer and consumer）應該要用不同邏輯，對producer而言，應該要傳到時間結束為止。而對consumer而言，則應該收到所有這次producers送出去的資料為止
chinghongfang: 關於 "何時結束" producing/consuming 應該要統一地方做判斷，不把判斷分散各地，可以使程式更清楚明瞭。
目前打算把判斷放在`DataManager`內。
chinghongfang: 這次主要更新
* 記錄producer是否完成
* 更改consumer thread停止條件
* 更改API，改成指定"records數量"或"持續時間"
* 避免重複的運算（`metrics`, `manager`重複紀錄record數量）
* `Metrics` 實作`Biconsumer<>`

Patch:
diff --git a/README.md b/README.md
index 344c8709..fdde5578 100644
--- a/README.md
+++ b/README.md
@@ -115,9 +115,12 @@ Run the benchmark from source
 4. --replicas: topic config when creating new topic. Default: 1
 5. --consumers: the number of consumers (threads). Default: 1
 6. --producers: the number of producers (threads). Default: 1
-7. --records: the total number of records sent by the producers. Default: 1000
-8. --record.size: the record size in byte. Default: 1024 byte
-9. --prop.file: the path to property file.
+7. --run.until: the total number of records sent by the producers. Default: 1000records
+8. --record.size: the (bound of) record size in byte. Default: 1024 byte
+9. --fixed.size: the flag to let all records have the same size
+10. --prop.file: the path to property file.
+11. --partitioner: the partitioner to use in producers
+12. --jmx.servers: the jmx server addresses of the brokers 
 
 ---
 
diff --git a/app/src/main/java/org/astraea/metrics/MetricExplorer.java b/app/src/main/java/org/astraea/metrics/MetricExplorer.java
index 26ac4cd1..3e96fcaf 100644
--- a/app/src/main/java/org/astraea/metrics/MetricExplorer.java
+++ b/app/src/main/java/org/astraea/metrics/MetricExplorer.java
@@ -162,10 +162,10 @@ public class MetricExplorer {
         .collect(Collectors.toList());
   }
 
-  public static void main(String[] args) throws Exception {
+  public static void main(String[] args) {
     var arguments = ArgumentUtil.parseArgument(new Argument(), args);
 
-    try (MBeanClient mBeanClient = new MBeanClient(arguments.jmxServer)) {
+    try (var mBeanClient = MBeanClient.of(arguments.jmxServer)) {
       execute(mBeanClient, arguments);
     }
   }
diff --git a/app/src/main/java/org/astraea/metrics/jmx/MBeanClient.java b/app/src/main/java/org/astraea/metrics/jmx/MBeanClient.java
index 4dbd1b1f..ec674d5d 100644
--- a/app/src/main/java/org/astraea/metrics/jmx/MBeanClient.java
+++ b/app/src/main/java/org/astraea/metrics/jmx/MBeanClient.java
@@ -1,11 +1,11 @@
 package org.astraea.metrics.jmx;
 
-import java.io.IOException;
-import java.io.UncheckedIOException;
-import java.util.*;
-import java.util.stream.Collectors;
-import javax.management.*;
-import javax.management.remote.*;
+import java.net.MalformedURLException;
+import java.util.Collection;
+import java.util.List;
+import java.util.Set;
+import javax.management.ObjectName;
+import javax.management.remote.JMXServiceURL;
 
 /**
  * A MBeanClient used to retrieve mbean value from remote Jmx server.
@@ -19,51 +19,37 @@ import javax.management.remote.*;
  *   System.out.println(bean.getAttributes());
  * }</pre>
  */
-public class MBeanClient implements AutoCloseable {
+public interface MBeanClient extends AutoCloseable {
 
-  private final JMXServiceURL jmxServiceURL;
-  private final JMXConnector jmxConnector;
-  private final MBeanServerConnection mBeanServerConnection;
-  private boolean isClosed;
+  /**
+   * @param host the address of jmx server
+   * @param port the port of jmx server
+   * @return a mbean client using JNDI to lookup metrics.
+   */
+  static MBeanClient jndi(String host, int port) {
+    try {
+      return of(
+          new JMXServiceURL(
+              String.format("service:jmx:rmi:///jndi/rmi://%s:%s/jmxrmi", host, port)));
+    } catch (MalformedURLException e) {
+      throw new IllegalArgumentException(e);
+    }
+  }
 
-  public MBeanClient(JMXServiceURL jmxServiceURL) throws IOException {
-    this.jmxServiceURL = Objects.requireNonNull(jmxServiceURL);
-    this.jmxConnector = JMXConnectorFactory.connect(jmxServiceURL);
-    this.isClosed = false;
-    this.mBeanServerConnection = jmxConnector.getMBeanServerConnection();
+  static MBeanClient of(JMXServiceURL url) {
+    return new MBeanClientImpl(url);
   }
 
   /**
    * Fetch all attributes of target mbean.
    *
-   * <p>Note that when exception is raised during the attribute fetching process, the exactly
+   * <p>Note that when exception is raised during the attribute fetching process, the exact
    * 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.
    */
-  public BeanObject queryBean(BeanQuery beanQuery) {
-    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);
-    } catch (InstanceNotFoundException e) {
-      throw new NoSuchElementException(e.getMessage());
-    }
-  }
+  BeanObject queryBean(BeanQuery beanQuery);
 
   /**
    * Fetch given attributes of target mbean
@@ -75,65 +61,7 @@ public class MBeanClient implements AutoCloseable {
    * @param attributeNameCollection a list of attribute you want to retrieve
    * @return A {@link BeanObject} contain given specific attributes if target resolved successfully.
    */
-  public BeanObject queryBean(BeanQuery beanQuery, Collection<String> attributeNameCollection) {
-    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);
-    } catch (InstanceNotFoundException e) {
-      throw new NoSuchElementException(e.getMessage());
-    }
-  }
-
-  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;
-    }
-  }
+  BeanObject queryBean(BeanQuery beanQuery, Collection<String> attributeNameCollection);
 
   /**
    * Query mBeans by pattern.
@@ -147,19 +75,7 @@ public class MBeanClient implements AutoCloseable {
    * @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 {
-      return mBeanServerConnection.queryMBeans(beanQuery.objectName(), null).stream()
-          .map(ObjectInstance::getObjectName)
-          .map(BeanQuery::fromObjectName)
-          .map(this::queryBean)
-          .collect(Collectors.toSet());
-
-    } catch (IOException e) {
-      throw new UncheckedIOException(e);
-    }
-  }
+  Collection<BeanObject> queryBeans(BeanQuery beanQuery);
 
   /**
    * Returns the list of domains in which any MBean is currently registered.
@@ -168,31 +84,14 @@ public class MBeanClient implements AutoCloseable {
    *
    * @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);
-    }
-  }
+  List<String> listDomains();
 
-  /**
-   * The JMX URL of current MBeanClient instance
-   *
-   * @return the origin JMX URL used to initiate MBeanClient
-   */
-  public JMXServiceURL getAddress() {
-    return jmxServiceURL;
-  }
+  /** @return the host address of jmx server */
+  String host();
 
-  private void ensureConnected() {
-    if (isClosed) throw new IllegalStateException("MBean client is closed");
-  }
+  /** @return the port listened by jmx server */
+  int port();
 
   @Override
-  public void close() throws Exception {
-    ensureConnected();
-    this.isClosed = true;
-    this.jmxConnector.close();
-  }
+  void close();
 }
diff --git a/app/src/main/java/org/astraea/metrics/jmx/MBeanClientImpl.java b/app/src/main/java/org/astraea/metrics/jmx/MBeanClientImpl.java
new file mode 100644
index 00000000..ac088fc1
--- /dev/null
+++ b/app/src/main/java/org/astraea/metrics/jmx/MBeanClientImpl.java
@@ -0,0 +1,168 @@
+package org.astraea.metrics.jmx;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.*;
+import java.util.stream.Collectors;
+import javax.management.*;
+import javax.management.remote.*;
+import org.astraea.Utils;
+
+class MBeanClientImpl implements MBeanClient {
+
+  private final JMXServiceURL jmxServiceURL;
+  private final JMXConnector jmxConnector;
+  private final MBeanServerConnection mBeanServerConnection;
+  private boolean isClosed;
+
+  MBeanClientImpl(JMXServiceURL jmxServiceURL) {
+    try {
+      this.jmxServiceURL = Objects.requireNonNull(jmxServiceURL);
+      this.jmxConnector = JMXConnectorFactory.connect(jmxServiceURL);
+      this.isClosed = false;
+      this.mBeanServerConnection = jmxConnector.getMBeanServerConnection();
+    } catch (IOException e) {
+      throw new UncheckedIOException(e);
+    }
+  }
+
+  @Override
+  public BeanObject queryBean(BeanQuery beanQuery) {
+    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);
+    } catch (InstanceNotFoundException e) {
+      throw new NoSuchElementException(e.getMessage());
+    }
+  }
+
+  @Override
+  public BeanObject queryBean(BeanQuery beanQuery, Collection<String> attributeNameCollection) {
+    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);
+    } catch (InstanceNotFoundException e) {
+      throw new NoSuchElementException(e.getMessage());
+    }
+  }
+
+  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;
+    }
+  }
+
+  @Override
+  public Collection<BeanObject> queryBeans(BeanQuery beanQuery) {
+    ensureConnected();
+    try {
+      return mBeanServerConnection.queryMBeans(beanQuery.objectName(), null).stream()
+          .map(ObjectInstance::getObjectName)
+          .map(BeanQuery::fromObjectName)
+          .map(this::queryBean)
+          .collect(Collectors.toSet());
+
+    } catch (IOException e) {
+      throw new UncheckedIOException(e);
+    }
+  }
+
+  @Override
+  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;
+  }
+
+  @Override
+  public String host() {
+    return jmxServiceURL.getHost();
+  }
+
+  @Override
+  public int port() {
+    return jmxServiceURL.getPort();
+  }
+
+  private void ensureConnected() {
+    if (isClosed) throw new IllegalStateException("MBean client is closed");
+  }
+
+  @Override
+  public void close() {
+    if (!isClosed) {
+      isClosed = true;
+      Utils.close(jmxConnector);
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClient.java b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClient.java
index 009d5229..5afe5c2c 100644
--- a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClient.java
+++ b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeLoadClient.java
@@ -5,8 +5,8 @@ import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.concurrent.TimeUnit;
+import org.astraea.Utils;
 import org.astraea.concurrent.ThreadPool;
-import org.astraea.metrics.jmx.MBeanClient;
 
 public class NodeLoadClient implements ThreadPool.Executor {
 
@@ -42,12 +42,7 @@ public class NodeLoadClient implements ThreadPool.Executor {
   public void close() {
     for (NodeMetadata nodeMetadata : nodeMetadataCollection) {
       NodeMetrics nodeMetrics = nodeMetadata.getNodeMetrics();
-      MBeanClient mBeanClient = nodeMetrics.getKafkaMetricClient();
-      try {
-        mBeanClient.close();
-      } catch (Exception e) {
-        e.printStackTrace();
-      }
+      Utils.close(nodeMetrics.getKafkaMetricClient());
     }
   }
 
diff --git a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeMetrics.java b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeMetrics.java
index 92c857cd..df3c1e3b 100644
--- a/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeMetrics.java
+++ b/app/src/main/java/org/astraea/partitioner/nodeLoadMetric/NodeMetrics.java
@@ -28,7 +28,7 @@ public class NodeMetrics {
     if (Pattern.compile("^service:").matcher(address).find())
       serviceURL = new JMXServiceURL(address);
     else serviceURL = new JMXServiceURL(createJmxUrl(address));
-    mBeanClient = new MBeanClient(serviceURL);
+    mBeanClient = MBeanClient.of(serviceURL);
     metricsValues = new HashMap();
   }
 
diff --git a/app/src/main/java/org/astraea/performance/Manager.java b/app/src/main/java/org/astraea/performance/Manager.java
new file mode 100644
index 00000000..0d737a60
--- /dev/null
+++ b/app/src/main/java/org/astraea/performance/Manager.java
@@ -0,0 +1,104 @@
+package org.astraea.performance;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.Random;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Thread safe This class is used for managing the start/end of the producer/consumer threads.
+ * Producer can start producing until all consumers get assignment. Consumers can stop after all
+ * producers are closed and all records are consumed.
+ */
+public class Manager {
+  private final ExeTime exeTime;
+  private final boolean fixedSize;
+  private final int size;
+  private final CountDownLatch getAssignment;
+  private final AtomicInteger producerClosed;
+  private final Random rand = new Random();
+
+  private final List<Metrics> producerMetrics, consumerMetrics;
+  private final long start = System.currentTimeMillis();
+  private final AtomicLong payloadNum = new AtomicLong(0);
+
+  /**
+   * Used to manage producing/consuming.
+   *
+   * @param argument
+   *     <ol>
+   *       <li>"fixedSize" for setting whether the size of the record is fixed
+   *       <li>"recordSize" for setting (the bound of) the size of the record
+   *       <li>"consumers" for number of consumers to wait on getting assignment
+   *       <li>"producers" for number of producers to wait on stop producing
+   *       <li>"exeTime" for determining whether the producers/consumers are completed
+   *     </ol>
+   *
+   * @param producerMetrics for counting the number of records have been produced
+   * @param consumerMetrics for counting the number of records have been consumed
+   */
+  public Manager(
+      Performance.Argument argument, List<Metrics> producerMetrics, List<Metrics> consumerMetrics) {
+    this.fixedSize = argument.fixedSize;
+    this.size = argument.recordSize;
+    this.getAssignment = new CountDownLatch(argument.consumers);
+    this.producerClosed = new AtomicInteger(argument.producers);
+    this.producerMetrics = producerMetrics;
+    this.consumerMetrics = consumerMetrics;
+    this.exeTime = argument.exeTime;
+  }
+
+  /**
+   * Generate random byte array in random/fixed length.
+   *
+   * @return random byte array. When no payload should be generated, Optional.empty() is generated.
+   *     Whether the payload should be generated is determined in construct time. "Number of
+   *     records" and "execution time" are considered.
+   */
+  public Optional<byte[]> payload() {
+    if (exeTime.percentage(payloadNum.getAndIncrement(), System.currentTimeMillis() - start)
+        >= 100D) return Optional.empty();
+
+    byte[] payload = (this.fixedSize) ? new byte[size] : new byte[rand.nextInt(size) + 1];
+    rand.nextBytes(payload);
+    return Optional.of(payload);
+  }
+
+  public long producedRecords() {
+    return producerMetrics.stream().mapToLong(Metrics::num).sum();
+  }
+
+  public long consumedRecords() {
+    return consumerMetrics.stream().mapToLong(Metrics::num).sum();
+  }
+
+  /** Used in "consumerRebalanceListener" callback. To */
+  public void countDownGetAssignment() {
+    this.getAssignment.countDown();
+  }
+
+  /** Called after producer is closed. For informing consumers there will be no new records */
+  public void producerClosed() {
+    this.producerClosed.decrementAndGet();
+  }
+
+  public void awaitPartitionAssignment() throws InterruptedException {
+    getAssignment.await();
+  }
+
+  public ExeTime exeTime() {
+    return exeTime;
+  }
+
+  /** Check if all producer are closed. */
+  public boolean producedDone() {
+    return producerClosed.get() == 0;
+  }
+
+  /** Check if we should keep consuming record. */
+  public boolean consumedDone() {
+    return producedDone() && consumedRecords() >= producedRecords();
+  }
+}
diff --git a/app/src/main/java/org/astraea/performance/Metrics.java b/app/src/main/java/org/astraea/performance/Metrics.java
index b8e3e637..ec279c4f 100644
--- a/app/src/main/java/org/astraea/performance/Metrics.java
+++ b/app/src/main/java/org/astraea/performance/Metrics.java
@@ -1,7 +1,9 @@
 package org.astraea.performance;
 
+import java.util.function.BiConsumer;
+
 /** Used to record statistics. This is thread safe. */
-public class Metrics {
+public class Metrics implements BiConsumer<Long, Long> {
   private double avgLatency;
   private long num;
   private long max;
@@ -17,7 +19,8 @@ public class Metrics {
   }
 
   /** Simultaneously add latency and bytes. */
-  public synchronized void put(long latency, long bytes) {
+  @Override
+  public synchronized void accept(Long latency, Long bytes) {
     ++num;
     putLatency(latency);
     addBytes(bytes);
diff --git a/app/src/main/java/org/astraea/performance/Performance.java b/app/src/main/java/org/astraea/performance/Performance.java
index 563d2206..4af2460e 100644
--- a/app/src/main/java/org/astraea/performance/Performance.java
+++ b/app/src/main/java/org/astraea/performance/Performance.java
@@ -7,9 +7,9 @@ import java.io.IOException;
 import java.time.Duration;
 import java.util.Map;
 import java.util.Set;
-import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BiConsumer;
+import java.util.function.BiFunction;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
@@ -77,11 +77,9 @@ public class Performance {
             .mapToObj(i -> new Metrics())
             .collect(Collectors.toUnmodifiableList());
 
-    var consumerRecords = new AtomicLong(param.records);
-    var producerRecords = new AtomicLong(param.records);
-    var tracker = new Tracker(producerMetrics, consumerMetrics, param.records);
+    var manager = new Manager(param, producerMetrics, consumerMetrics);
+    var tracker = new Tracker(producerMetrics, consumerMetrics, manager);
     var groupId = "groupId-" + System.currentTimeMillis();
-    var getAssignment = new CountDownLatch(param.consumers);
     try (ThreadPool threadPool =
         ThreadPool.builder()
             .executors(
@@ -94,10 +92,11 @@ public class Performance {
                                     .topics(Set.of(param.topic))
                                     .groupId(groupId)
                                     .configs(param.props())
-                                    .consumerRebalanceListener(ignore -> getAssignment.countDown())
+                                    .consumerRebalanceListener(
+                                        ignore -> manager.countDownGetAssignment())
                                     .build(),
                                 consumerMetrics.get(i),
-                                consumerRecords))
+                                manager))
                     .collect(Collectors.toUnmodifiableList()))
             .executors(
                 IntStream.range(0, param.producers)
@@ -110,8 +109,7 @@ public class Performance {
                                     .build(),
                                 param,
                                 producerMetrics.get(i),
-                                producerRecords,
-                                getAssignment))
+                                manager))
                     .collect(Collectors.toUnmodifiableList()))
             .executor(tracker)
             .build()) {
@@ -120,7 +118,7 @@ public class Performance {
   }
 
   static ThreadPool.Executor consumerExecutor(
-      Consumer<byte[], byte[]> consumer, Metrics metrics, AtomicLong records) {
+      Consumer<byte[], byte[]> consumer, BiConsumer<Long, Long> observer, Manager manager) {
     return new ThreadPool.Executor() {
       @Override
       public State execute() {
@@ -130,12 +128,12 @@ public class Performance {
               .forEach(
                   record -> {
                     // 記錄端到端延時, 記錄輸入byte(沒有算入header和timestamp)
-                    metrics.put(
+                    observer.accept(
                         System.currentTimeMillis() - record.timestamp(),
-                        record.serializedKeySize() + record.serializedValueSize());
-                    records.decrementAndGet();
+                        (long) record.serializedKeySize() + record.serializedValueSize());
                   });
-          return records.get() <= 0 ? State.DONE : State.RUNNING;
+          // Consumer reached the record upperbound or consumed all the record producer produced.
+          return manager.consumedDone() ? State.DONE : State.RUNNING;
         } catch (WakeupException ignore) {
           // Stop polling and being ready to clean up
           return State.DONE;
@@ -157,32 +155,37 @@ public class Performance {
   static ThreadPool.Executor producerExecutor(
       Producer<byte[], byte[]> producer,
       Argument param,
-      Metrics metrics,
-      AtomicLong records,
-      CountDownLatch getAssignment) {
-    byte[] payload = new byte[param.recordSize];
+      BiConsumer<Long, Long> observer,
+      Manager manager) {
     return new ThreadPool.Executor() {
       @Override
       public State execute() throws InterruptedException {
         // Wait for all consumers get assignment.
-        getAssignment.await();
-        var currentRecords = records.getAndDecrement();
-        if (currentRecords <= 0) return State.DONE;
+        manager.awaitPartitionAssignment();
+
+        var payload = manager.payload();
+        if (payload.isEmpty()) return State.DONE;
+
         long start = System.currentTimeMillis();
         producer
             .sender()
             .topic(param.topic)
-            .value(payload)
+            .value(payload.get())
             .timestamp(start)
             .run()
             .whenComplete(
-                (m, e) -> metrics.put(System.currentTimeMillis() - start, payload.length));
+                (m, e) ->
+                    observer.accept(System.currentTimeMillis() - start, m.serializedValueSize()));
         return State.RUNNING;
       }
 
       @Override
       public void close() {
-        producer.close();
+        try {
+          producer.close();
+        } finally {
+          manager.producerClosed();
+        }
       }
     };
   }
@@ -221,10 +224,19 @@ public class Performance {
     int consumers = 1;
 
     @Parameter(
-        names = {"--records"},
-        description = "Integer: number of records to send",
-        validateWith = ArgumentUtil.NonNegativeLong.class)
-    long records = 1000;
+        names = {"--run.until"},
+        description =
+            "Run until number of records are produced and consumed or until duration meets."
+                + " The duration formats accepted are (a number) + (a time unit)."
+                + " The time units can be \"days\", \"day\", \"h\", \"m\", \"s, \"ms\","
+                + " \"us\", \"ns\"",
+        converter = ExeTimeArgument.class)
+    ExeTime exeTime = ExeTime.of("1000records");
+
+    @Parameter(
+        names = {"--fixed.size"},
+        description = "boolean: send fixed size records if this flag is set")
+    boolean fixedSize = false;
 
     @Parameter(
         names = {"--record.size"},
@@ -277,4 +289,43 @@ public class Performance {
       }
     }
   }
+
+  static class ExeTimeArgument implements IStringConverter<ExeTime> {
+    @Override
+    public ExeTime convert(String value) {
+      return ExeTime.of(value);
+    }
+  }
+}
+
+/**
+ * Two kind of running modes. One runs for a duration of time. The other runs for a number of
+ * records.
+ */
+interface ExeTime {
+
+  double percentage(long records, long elapsedTime);
+
+  static ExeTime of(String exeTime) {
+    if (exeTime.endsWith("records")) {
+      final long records = Long.parseLong(exeTime.replace("records", ""));
+      return ExeTime.of((completeRecords, ignore) -> 100D * completeRecords / records, exeTime);
+    }
+    final Duration duration = new ArgumentUtil.DurationConverter().convert(exeTime);
+    return ExeTime.of((ignore, elapsedTime) -> 100D * elapsedTime / duration.toMillis(), exeTime);
+  }
+
+  static ExeTime of(BiFunction<Long, Long, Double> function, String toString) {
+    return new ExeTime() {
+      @Override
+      public double percentage(long records, long duration) {
+        return function.apply(records, duration);
+      }
+
+      @Override
+      public String toString() {
+        return toString;
+      }
+    };
+  }
 }
diff --git a/app/src/main/java/org/astraea/performance/Tracker.java b/app/src/main/java/org/astraea/performance/Tracker.java
index e524dd25..e0055d77 100644
--- a/app/src/main/java/org/astraea/performance/Tracker.java
+++ b/app/src/main/java/org/astraea/performance/Tracker.java
@@ -6,17 +6,17 @@ import java.util.Collections;
 import java.util.List;
 import org.astraea.concurrent.ThreadPool;
 
-/** Print out the given metrics. Run until `close()` is called. */
+/** Print out the given metrics. */
 public class Tracker implements ThreadPool.Executor {
   private final List<Metrics> producerData;
   private final List<Metrics> consumerData;
-  private final long records;
-  private long start = 0L;
+  private final Manager manager;
+  long start = 0L;
 
-  public Tracker(List<Metrics> producerData, List<Metrics> consumerData, long records) {
+  public Tracker(List<Metrics> producerData, List<Metrics> consumerData, Manager manager) {
     this.producerData = producerData;
     this.consumerData = consumerData;
-    this.records = records;
+    this.manager = manager;
   }
 
   @Override
@@ -28,7 +28,7 @@ public class Tracker implements ThreadPool.Executor {
   }
 
   private Duration duration() {
-    if (start == 0L) start = System.currentTimeMillis();
+    if (start == 0L) start = System.currentTimeMillis() - Duration.ofSeconds(1).toMillis();
     return Duration.ofMillis(System.currentTimeMillis() - start);
   }
 
@@ -40,9 +40,14 @@ public class Tracker implements ThreadPool.Executor {
 
   private boolean logProducers() {
     var result = result(producerData);
+
     if (result.completedRecords == 0) return false;
 
     var duration = duration();
+    // Print completion rate (by number of records) or (by time)
+    var percentage =
+        Math.min(100D, manager.exeTime().percentage(result.completedRecords, duration.toMillis()));
+
     System.out.println(
         "Time: "
             + duration.toHoursPart()
@@ -51,8 +56,7 @@ public class Tracker implements ThreadPool.Executor {
             + "min "
             + duration.toSecondsPart()
             + "sec");
-    System.out.printf(
-        "producers完成度: %.2f%%%n", ((double) result.completedRecords * 100.0 / (double) records));
+    System.out.printf("producers完成度: %.2f%%%n", percentage);
     System.out.printf("  輸出%.3fMB/second%n", result.averageBytes(duration));
     System.out.println("  發送max latency: " + result.maxLatency + "ms");
     System.out.println("  發送mim latency: " + result.minLatency + "ms");
@@ -63,7 +67,7 @@ public class Tracker implements ThreadPool.Executor {
           "  producer[%d]的發送average latency: %.3fms%n", i, result.averageLatencies.get(i));
     }
     System.out.println("\n");
-    return result.completedRecords >= records;
+    return percentage >= 100D;
   }
 
   private boolean logConsumers() {
@@ -73,8 +77,9 @@ public class Tracker implements ThreadPool.Executor {
     if (result.completedRecords == 0) return false;
     var duration = duration();
 
-    System.out.printf(
-        "consumer完成度: %.2f%%%n", ((double) result.completedRecords * 100.0 / (double) records));
+    // Print out percentage of (consumed records) and (produced records)
+    var percentage = result.completedRecords * 100D / manager.producedRecords();
+    System.out.printf("consumer完成度: %.2f%%%n", percentage);
     System.out.printf("  輸入%.3fMB/second%n", result.averageBytes(duration));
     System.out.println("  端到端max latency: " + result.maxLatency + "ms");
     System.out.println("  端到端mim latency: " + result.minLatency + "ms");
@@ -85,7 +90,8 @@ public class Tracker implements ThreadPool.Executor {
           "  consumer[%d]的端到端average latency: %.3fms%n", i, result.averageLatencies.get(i));
     }
     System.out.println("\n");
-    return result.completedRecords >= records;
+    // Target number of records consumed OR consumed all that produced
+    return manager.producedDone() && percentage >= 100D;
   }
 
   private static Result result(List<Metrics> metrics) {
diff --git a/app/src/main/java/org/astraea/topic/Builder.java b/app/src/main/java/org/astraea/topic/Builder.java
index abc0f6f9..acfbb513 100644
--- a/app/src/main/java/org/astraea/topic/Builder.java
+++ b/app/src/main/java/org/astraea/topic/Builder.java
@@ -1,6 +1,7 @@
 package org.astraea.topic;
 
 import java.util.*;
+import java.util.function.BiFunction;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 import org.apache.kafka.clients.admin.*;
@@ -213,33 +214,18 @@ public class Builder {
       var replicaInfos =
           Utils.handleException(() -> admin.describeLogDirs(brokerIds()).allDescriptions().get());
 
-      var replicaLags =
-          replicaInfos.entrySet().stream()
-              .collect(
-                  Collectors.toMap(
-                      Map.Entry::getKey,
-                      e ->
-                          e.getValue().values().stream()
-                              .flatMap(
-                                  logDirDescription ->
-                                      logDirDescription.replicaInfos().entrySet().stream())
-                              .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))));
-
-      var replicaPaths =
-          replicaInfos.entrySet().stream()
-              .collect(
-                  Collectors.toMap(
-                      Map.Entry::getKey,
-                      e ->
-                          e.getValue().entrySet().stream()
-                              .flatMap(
-                                  logDirDescription ->
-                                      logDirDescription.getValue().replicaInfos().keySet().stream()
-                                          .map(
-                                              topicPartition ->
-                                                  Map.entry(
-                                                      topicPartition, logDirDescription.getKey())))
-                              .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))));
+      BiFunction<Integer, TopicPartition, List<Map.Entry<String, ReplicaInfo>>> findReplicas =
+          (id, partition) ->
+              replicaInfos.getOrDefault(id, Map.of()).entrySet().stream()
+                  .flatMap(
+                      entry -> {
+                        var path = entry.getKey();
+                        var replicas = entry.getValue().replicaInfos();
+                        return replicas.entrySet().stream()
+                            .filter(e -> e.getKey().equals(partition))
+                            .map(e -> Map.entry(path, e.getValue()));
+                      })
+                  .collect(Collectors.toList());
 
       return Utils.handleException(
           () ->
@@ -248,44 +234,34 @@ public class Builder {
                       e ->
                           e.getValue().partitions().stream()
                               .map(
-                                  topicPartitionInfo ->
-                                      Map.entry(
-                                          new TopicPartition(
-                                              e.getKey(), topicPartitionInfo.partition()),
-                                          topicPartitionInfo.replicas().stream()
-                                              .map(
-                                                  node -> {
-                                                    var replicaInfo =
-                                                        replicaLags
-                                                            .getOrDefault(node.id(), Map.of())
-                                                            .get(
-                                                                new TopicPartition(
-                                                                    e.getKey(),
-                                                                    topicPartitionInfo
-                                                                        .partition()));
-                                                    return new Replica(
-                                                        node.id(),
-                                                        replicaInfo == null
-                                                            ? -1
-                                                            : replicaInfo.offsetLag(),
-                                                        replicaInfo == null
-                                                            ? -1
-                                                            : replicaInfo.size(),
-                                                        topicPartitionInfo.leader().id()
-                                                            == node.id(),
-                                                        topicPartitionInfo.isr().contains(node),
-                                                        replicaInfo != null
-                                                            && replicaInfo.isFuture(),
-                                                        replicaPaths
-                                                            .get(node.id())
-                                                            .get(
-                                                                new TopicPartition(
-                                                                    e.getKey(),
+                                  topicPartitionInfo -> {
+                                    var partition =
+                                        new TopicPartition(
+                                            e.getKey(), topicPartitionInfo.partition());
+                                    return Map.entry(
+                                        partition,
+                                        topicPartitionInfo.replicas().stream()
+                                            .flatMap(
+                                                node ->
+                                                    findReplicas
+                                                        .apply(node.id(), partition)
+                                                        .stream()
+                                                        .map(
+                                                            entry ->
+                                                                new Replica(
+                                                                    node.id(),
+                                                                    entry.getValue().offsetLag(),
+                                                                    entry.getValue().size(),
+                                                                    topicPartitionInfo.leader().id()
+                                                                        == node.id(),
                                                                     topicPartitionInfo
-                                                                        .partition())));
-                                                  })
-                                              .sorted(Comparator.comparing(Replica::broker))
-                                              .collect(Collectors.toList()))))
+                                                                        .isr()
+                                                                        .contains(node),
+                                                                    entry.getValue().isFuture(),
+                                                                    entry.getKey())))
+                                            .sorted(Comparator.comparing(Replica::broker))
+                                            .collect(Collectors.toList()));
+                                  }))
                   .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)));
     }
 


Test patch:
diff --git a/app/src/test/java/org/astraea/metrics/jmx/MBeanClientTest.java b/app/src/test/java/org/astraea/metrics/jmx/MBeanClientTest.java
index 3628d685..eab5cdff 100644
--- a/app/src/test/java/org/astraea/metrics/jmx/MBeanClientTest.java
+++ b/app/src/test/java/org/astraea/metrics/jmx/MBeanClientTest.java
@@ -17,7 +17,7 @@ class MBeanClientTest {
 
   private MBeanServer mBeanServer;
   private JMXConnectorServer jmxServer;
-  private Map<ObjectName, Object> registeredBeans = new HashMap<>();
+  private final Map<ObjectName, Object> registeredBeans = new HashMap<>();
 
   private void register(ObjectName name, Object mBean) {
     registeredBeans.put(name, mBean);
@@ -60,13 +60,13 @@ class MBeanClientTest {
   }
 
   @Test
-  void testFetchAttributes() throws Exception {
+  void testFetchAttributes() {
     // arrange
-    try (MBeanClient sut = new MBeanClient(jmxServer.getAddress())) {
+    try (var client = MBeanClient.of(jmxServer.getAddress())) {
       BeanQuery beanQuery = BeanQuery.builder("java.lang").property("type", "Memory").build();
 
       // act
-      BeanObject beanObject = sut.queryBean(beanQuery);
+      BeanObject beanObject = client.queryBean(beanQuery);
 
       // assert
       assertTrue(beanObject.getProperties().containsKey("type"));
@@ -76,9 +76,9 @@ class MBeanClientTest {
   }
 
   @Test
-  void testFetchMbeanWithMultipleProperties() throws Exception {
+  void testFetchMbeanWithMultipleProperties() {
     // arrange
-    try (MBeanClient sut = new MBeanClient(jmxServer.getAddress())) {
+    try (var client = MBeanClient.of(jmxServer.getAddress())) {
       BeanQuery query1 =
           BeanQuery.builder(
                   "java.lang", Map.of("type", "MemoryManager", "name", "CodeCacheManager"))
@@ -90,8 +90,8 @@ class MBeanClientTest {
               .build();
 
       // act
-      BeanObject beanObject1 = sut.queryBean(query1);
-      BeanObject beanObject2 = sut.queryBean(query2);
+      BeanObject beanObject1 = client.queryBean(query1);
+      BeanObject beanObject2 = client.queryBean(query2);
 
       // assert
       assertTrue(beanObject1.getProperties().containsKey("type"));
@@ -111,14 +111,14 @@ class MBeanClientTest {
   }
 
   @Test
-  void testFetchSelectedAttributes() throws Exception {
+  void testFetchSelectedAttributes() {
     // arrange
-    try (MBeanClient sut = new MBeanClient(jmxServer.getAddress())) {
+    try (var client = MBeanClient.of(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);
+      BeanObject beanObject = client.queryBean(beanQuery, selectedAttribute);
 
       // assert
       assertTrue(beanObject.getProperties().containsKey("type"));
@@ -128,13 +128,13 @@ class MBeanClientTest {
   }
 
   @Test
-  void testQueryBeans() throws Exception {
+  void testQueryBeans() {
     // arrange 1 query beans
-    try (MBeanClient sut = new MBeanClient(jmxServer.getAddress())) {
+    try (var client = MBeanClient.of(jmxServer.getAddress())) {
       BeanQuery beanQuery = BeanQuery.builder("java.lang").property("type", "C*").build();
 
       // act 1
-      Collection<BeanObject> beanObjects = sut.queryBeans(beanQuery);
+      Collection<BeanObject> beanObjects = client.queryBeans(beanQuery);
 
       // assert 1
       assertEquals(2, beanObjects.size());
@@ -165,13 +165,13 @@ class MBeanClientTest {
   }
 
   @Test
-  void testQueryNonExistsBeans() throws Exception {
+  void testQueryNonExistsBeans() {
     // arrange
-    try (MBeanClient sut = new MBeanClient(jmxServer.getAddress())) {
+    try (var client = MBeanClient.of(jmxServer.getAddress())) {
       BeanQuery beanQuery = BeanQuery.builder("java.lang").property("type", "Something").build();
 
       // act
-      Collection<BeanObject> beanObjects = sut.queryBeans(beanQuery);
+      Collection<BeanObject> beanObjects = client.queryBeans(beanQuery);
 
       // assert
       assertEquals(0, beanObjects.size());
@@ -179,53 +179,54 @@ class MBeanClientTest {
   }
 
   @Test
-  void testFetchNonExistsBeans() throws Exception {
+  void testFetchNonExistsBeans() {
     // arrange
-    try (MBeanClient sut = new MBeanClient(jmxServer.getAddress())) {
+    try (var client = MBeanClient.of(jmxServer.getAddress())) {
       BeanQuery beanQuery = BeanQuery.builder("java.lang").property("type", "Something").build();
 
       // act assert
-      assertThrows(NoSuchElementException.class, () -> sut.queryBean(beanQuery));
+      assertThrows(NoSuchElementException.class, () -> client.queryBean(beanQuery));
       assertThrows(
-          NoSuchElementException.class, () -> sut.queryBean(beanQuery, Collections.emptyList()));
+          NoSuchElementException.class, () -> client.queryBean(beanQuery, Collections.emptyList()));
     }
   }
 
   @Test
-  void testUseClosedClientWillThrowError() throws Exception {
+  void testUseClosedClientWillThrowError() {
     // arrange
-    MBeanClient sut = new MBeanClient(jmxServer.getAddress());
+    var client = MBeanClient.of(jmxServer.getAddress());
     BeanQuery query = BeanQuery.builder("java.lang").property("type", "Memory").build();
 
     // act
-    sut.close();
+    client.close();
 
     // assert
-    assertThrows(IllegalStateException.class, () -> sut.queryBean(query));
-    assertThrows(IllegalStateException.class, () -> sut.queryBean(query, Collections.emptyList()));
-    assertThrows(IllegalStateException.class, () -> sut.queryBeans(query));
+    assertThrows(IllegalStateException.class, () -> client.queryBean(query));
+    assertThrows(
+        IllegalStateException.class, () -> client.queryBean(query, Collections.emptyList()));
+    assertThrows(IllegalStateException.class, () -> client.queryBeans(query));
   }
 
   @Test
-  void testCloseOnceMoreWillThrowError() throws Exception {
+  void testCloseOnceMore() {
     // arrange
-    MBeanClient sut = new MBeanClient(jmxServer.getAddress());
+    var client = MBeanClient.of(jmxServer.getAddress());
 
     // act
-    sut.close();
+    client.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);
+    client.close();
+    client.close();
+    client.close();
+    client.close();
+    client.close();
   }
 
   @Test
-  void testGetAllMBeans() throws Exception {
+  void testGetAllMBeans() {
     // arrange
-    try (MBeanClient client = new MBeanClient(jmxServer.getAddress())) {
+    try (var client = MBeanClient.of(jmxServer.getAddress())) {
 
       // act
       Collection<BeanObject> beanObjects = client.queryBeans(BeanQuery.all());
@@ -237,9 +238,9 @@ class MBeanClientTest {
   }
 
   @Test
-  void testGetAllMBeansUnderSpecificDomainName() throws Exception {
+  void testGetAllMBeansUnderSpecificDomainName() {
     // arrange
-    try (MBeanClient client = new MBeanClient(jmxServer.getAddress())) {
+    try (var client = MBeanClient.of(jmxServer.getAddress())) {
 
       // act
       Collection<BeanObject> beanObjects = client.queryBeans(BeanQuery.all("java.lang"));
@@ -251,9 +252,9 @@ class MBeanClientTest {
   }
 
   @Test
-  void testGetAllMBeansUnderSpecificDomainNamePattern() throws Exception {
+  void testGetAllMBeansUnderSpecificDomainNamePattern() {
     // arrange
-    try (MBeanClient client = new MBeanClient(jmxServer.getAddress())) {
+    try (var client = MBeanClient.of(jmxServer.getAddress())) {
 
       // act
       Collection<BeanObject> beanObjects = client.queryBeans(BeanQuery.all("java.*"));
@@ -265,9 +266,9 @@ class MBeanClientTest {
   }
 
   @Test
-  void testUsePropertyListPattern() throws Exception {
+  void testUsePropertyListPattern() {
     // arrange
-    try (MBeanClient client = new MBeanClient(jmxServer.getAddress())) {
+    try (var client = MBeanClient.of(jmxServer.getAddress())) {
       BeanQuery patternQuery =
           BeanQuery.builder("java.lang").property("type", "*").usePropertyListPattern().build();
 
@@ -314,9 +315,9 @@ class MBeanClientTest {
   }
 
   @Test
-  void testListDomains() throws Exception {
+  void testListDomains() {
     // arrange
-    try (MBeanClient client = new MBeanClient(jmxServer.getAddress())) {
+    try (var client = MBeanClient.of(jmxServer.getAddress())) {
 
       // act
       List<String> domains = client.listDomains();
@@ -328,15 +329,12 @@ class MBeanClientTest {
   }
 
   @Test
-  void testGetAddress() throws Exception {
+  void testHostAndPort() {
     // arrange
-    try (MBeanClient client = new MBeanClient(jmxServer.getAddress())) {
+    try (var client = MBeanClient.of(jmxServer.getAddress())) {
 
-      // act
-      JMXServiceURL address = client.getAddress();
-
-      // assert
-      assertEquals(jmxServer.getAddress(), address);
+      assertEquals(jmxServer.getAddress().getHost(), client.host());
+      assertEquals(jmxServer.getAddress().getPort(), client.port());
     }
   }
 
@@ -350,7 +348,7 @@ class MBeanClientTest {
 
     register(objectName0, customMBean0);
 
-    try (MBeanClient client = new MBeanClient(jmxServer.getAddress())) {
+    try (var client = MBeanClient.of(jmxServer.getAddress())) {
 
       // act
       Collection<BeanObject> all =
@@ -376,7 +374,7 @@ class MBeanClientTest {
       register(objectName, mbean);
     }
 
-    try (MBeanClient client = new MBeanClient(jmxServer.getAddress())) {
+    try (var client = MBeanClient.of(jmxServer.getAddress())) {
 
       // act
       Collection<BeanObject> all =
diff --git a/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricsTest.java b/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricsTest.java
index 794dd3ad..a64c9370 100644
--- a/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricsTest.java
+++ b/app/src/test/java/org/astraea/metrics/kafka/KafkaMetricsTest.java
@@ -41,7 +41,7 @@ class KafkaMetricsTest extends RequireBrokerCluster {
     jmxServer = JMXConnectorServerFactory.newJMXConnectorServer(serviceURL, null, mBeanServer);
     jmxServer.start();
 
-    mBeanClient = new MBeanClient(jmxServer.getAddress());
+    mBeanClient = MBeanClient.of(jmxServer.getAddress());
   }
 
   @AfterEach
diff --git a/app/src/test/java/org/astraea/performance/ManagerTest.java b/app/src/test/java/org/astraea/performance/ManagerTest.java
new file mode 100644
index 00000000..29c0bbf8
--- /dev/null
+++ b/app/src/test/java/org/astraea/performance/ManagerTest.java
@@ -0,0 +1,55 @@
+package org.astraea.performance;
+
+import java.util.Arrays;
+import java.util.List;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class ManagerTest {
+  @Test
+  void testReturnNull() throws InterruptedException {
+    var argument = new Performance.Argument();
+    var producerMetric = new Metrics();
+    argument.exeTime = ExeTime.of("1records");
+    var manager = new Manager(argument, List.of(producerMetric), List.of());
+
+    Assertions.assertTrue(manager.payload().isPresent());
+    producerMetric.accept(1L, 1L);
+    Assertions.assertTrue(manager.payload().isEmpty());
+
+    producerMetric = new Metrics();
+    argument.exeTime = ExeTime.of("50ms");
+    manager = new Manager(argument, List.of(producerMetric), List.of());
+
+    Assertions.assertTrue(manager.payload().isPresent());
+    producerMetric.accept(1L, 1L);
+    Thread.sleep(50);
+    Assertions.assertTrue(manager.payload().isEmpty());
+  }
+
+  @Test
+  void testRandomSize() {
+    var argument = new Performance.Argument();
+    argument.exeTime = ExeTime.of("3records");
+    argument.recordSize = 102400;
+    var dataManager = new Manager(argument, List.of(), List.of());
+    boolean sameSize = dataManager.payload().get().length == dataManager.payload().get().length;
+
+    // Assertion failed with probability 1/102400 ~ 0.001%
+    Assertions.assertFalse(sameSize);
+
+    Assertions.assertTrue(dataManager.payload().get().length <= 102400);
+  }
+
+  @Test
+  void testRandomContent() {
+    var argument = new Performance.Argument();
+    argument.exeTime = ExeTime.of("2records");
+    argument.recordSize = 102400;
+    var manager = new Manager(argument, List.of(), List.of());
+    boolean same = Arrays.equals(manager.payload().get(), manager.payload().get());
+
+    // Assertion failed with probability < 1/102400
+    Assertions.assertFalse(same);
+  }
+}
diff --git a/app/src/test/java/org/astraea/performance/MetricsTest.java b/app/src/test/java/org/astraea/performance/MetricsTest.java
index 103f1769..ac9799da 100644
--- a/app/src/test/java/org/astraea/performance/MetricsTest.java
+++ b/app/src/test/java/org/astraea/performance/MetricsTest.java
@@ -18,7 +18,7 @@ public class MetricsTest {
     for (int i = 0; i < num; ++i) {
       long next = rand.nextInt();
       avg += ((double) next - avg) / (i + 1);
-      metrics.put(next, 0);
+      metrics.accept(next, 0L);
     }
 
     Assertions.assertEquals(avg, metrics.avgLatency());
@@ -29,7 +29,7 @@ public class MetricsTest {
     var metrics = new Metrics();
 
     Assertions.assertEquals(0, metrics.bytes());
-    metrics.put(0, 1000);
+    metrics.accept(0L, 1000L);
     Assertions.assertEquals(1000, metrics.bytes());
   }
 }
diff --git a/app/src/test/java/org/astraea/performance/PerformanceTest.java b/app/src/test/java/org/astraea/performance/PerformanceTest.java
index 33b9b39c..a2e50993 100644
--- a/app/src/test/java/org/astraea/performance/PerformanceTest.java
+++ b/app/src/test/java/org/astraea/performance/PerformanceTest.java
@@ -1,9 +1,8 @@
 package org.astraea.performance;
 
+import java.util.List;
 import java.util.Set;
-import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.atomic.AtomicLong;
 import org.astraea.Utils;
 import org.astraea.concurrent.ThreadPool;
 import org.astraea.consumer.Consumer;
@@ -19,13 +18,14 @@ public class PerformanceTest extends RequireBrokerCluster {
     var metrics = new Metrics();
     var param = new Performance.Argument();
     param.topic = "testProducerExecutor-" + System.currentTimeMillis();
+    param.fixedSize = true;
+    param.consumers = 0;
     try (ThreadPool.Executor executor =
         Performance.producerExecutor(
             Producer.builder().brokers(bootstrapServers()).build(),
             param,
             metrics,
-            new AtomicLong(10),
-            new CountDownLatch(0))) {
+            new Manager(param, List.of(), List.of()))) {
       executor.execute();
 
       Utils.waitFor(() -> metrics.num() == 1);
@@ -37,11 +37,13 @@ public class PerformanceTest extends RequireBrokerCluster {
   void testConsumerExecutor() throws InterruptedException, ExecutionException {
     Metrics metrics = new Metrics();
     var topicName = "testConsumerExecutor-" + System.currentTimeMillis();
+    var param = new Performance.Argument();
+    param.fixedSize = true;
     try (ThreadPool.Executor executor =
         Performance.consumerExecutor(
             Consumer.builder().topics(Set.of(topicName)).brokers(bootstrapServers()).build(),
             metrics,
-            new AtomicLong(10))) {
+            new Manager(param, List.of(), List.of()))) {
       executor.execute();
 
       Assertions.assertEquals(0, metrics.num());
diff --git a/app/src/test/java/org/astraea/performance/TrackerTest.java b/app/src/test/java/org/astraea/performance/TrackerTest.java
index fcc3aa14..e725e321 100644
--- a/app/src/test/java/org/astraea/performance/TrackerTest.java
+++ b/app/src/test/java/org/astraea/performance/TrackerTest.java
@@ -11,20 +11,43 @@ public class TrackerTest {
     var producerData = List.of(new Metrics());
     var consumerData = List.of(new Metrics());
     List<Metrics> empty = List.of();
-    int records = 1;
+    var argument = new Performance.Argument();
+    argument.exeTime = ExeTime.of("1records");
 
-    try (Tracker tracker = new Tracker(producerData, consumerData, records)) {
+    var manager = new Manager(argument, producerData, consumerData);
+    try (Tracker tracker = new Tracker(producerData, consumerData, manager)) {
       Assertions.assertEquals(ThreadPool.Executor.State.RUNNING, tracker.execute());
-      producerData.get(0).put(1, 1);
-      consumerData.get(0).put(1, 1);
+      producerData.get(0).accept(1L, 1L);
+      consumerData.get(0).accept(1L, 1L);
+      manager.producerClosed();
       Assertions.assertEquals(ThreadPool.Executor.State.DONE, tracker.execute());
     }
 
     // Zero consumer
     producerData = List.of(new Metrics());
-    try (Tracker tracker = new Tracker(producerData, empty, records)) {
+    manager = new Manager(argument, producerData, empty);
+    try (Tracker tracker = new Tracker(producerData, empty, manager)) {
       Assertions.assertEquals(ThreadPool.Executor.State.RUNNING, tracker.execute());
-      producerData.get(0).put(1, 1);
+      producerData.get(0).accept(1L, 1L);
+      manager.producerClosed();
+      Assertions.assertEquals(ThreadPool.Executor.State.DONE, tracker.execute());
+    }
+
+    // Stop by duration time out
+    argument.exeTime = ExeTime.of("2s");
+    producerData = List.of(new Metrics());
+    consumerData = List.of(new Metrics());
+    manager = new Manager(argument, producerData, consumerData);
+    try (Tracker tracker = new Tracker(producerData, consumerData, manager)) {
+      tracker.start = System.currentTimeMillis();
+      Assertions.assertEquals(ThreadPool.Executor.State.RUNNING, tracker.execute());
+
+      // Mock record producing
+      producerData.get(0).accept(1L, 1L);
+      consumerData.get(0).accept(1L, 1L);
+      Thread.sleep(2000);
+      manager.producerClosed();
+
       Assertions.assertEquals(ThreadPool.Executor.State.DONE, tracker.execute());
     }
   }
diff --git a/app/src/test/java/org/astraea/service/RequireJmxServer.java b/app/src/test/java/org/astraea/service/RequireJmxServer.java
index c449d724..80f538de 100644
--- a/app/src/test/java/org/astraea/service/RequireJmxServer.java
+++ b/app/src/test/java/org/astraea/service/RequireJmxServer.java
@@ -1,6 +1,10 @@
 package org.astraea.service;
 
+import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.lang.management.ManagementFactory;
+import java.net.ServerSocket;
+import java.rmi.registry.LocateRegistry;
 import javax.management.remote.JMXConnectorServer;
 import javax.management.remote.JMXConnectorServerFactory;
 import javax.management.remote.JMXServiceURL;
@@ -9,7 +13,7 @@ import org.junit.jupiter.api.AfterAll;
 
 public abstract class RequireJmxServer {
 
-  private static final JMXConnectorServer JMX_CONNECTOR_SERVER = jmxConnectorServer();
+  private static final JMXConnectorServer JMX_CONNECTOR_SERVER = jmxConnectorServer(freePort());
 
   protected static JMXServiceURL jmxServiceURL() {
     return JMX_CONNECTOR_SERVER.getAddress();
@@ -25,16 +29,31 @@ public abstract class RequireJmxServer {
    *
    * @return JMXConnectorServer
    */
-  private static JMXConnectorServer jmxConnectorServer() {
+  private static JMXConnectorServer jmxConnectorServer(int port) {
     try {
-      var serviceURL = new JMXServiceURL("service:jmx:rmi://127.0.0.1");
+      LocateRegistry.createRegistry(port);
       var mBeanServer = ManagementFactory.getPlatformMBeanServer();
       var jmxServer =
-          JMXConnectorServerFactory.newJMXConnectorServer(serviceURL, null, mBeanServer);
+          JMXConnectorServerFactory.newJMXConnectorServer(
+              // we usually use JNDI to get metrics in production, so the embedded server should use
+              // JNDI too.
+              new JMXServiceURL(
+                  String.format(
+                      "service:jmx:rmi://127.0.0.1:%s/jndi/rmi://127.0.0.1:%s/jmxrmi", port, port)),
+              null,
+              mBeanServer);
       jmxServer.start();
       return jmxServer;
     } catch (Exception e) {
       throw new RuntimeException(e);
     }
   }
+
+  private static int freePort() {
+    try (var server = new ServerSocket(0)) {
+      return server.getLocalPort();
+    } catch (IOException e) {
+      throw new UncheckedIOException(e);
+    }
+  }
 }
diff --git a/app/src/test/java/org/astraea/service/RequireJmxServerTest.java b/app/src/test/java/org/astraea/service/RequireJmxServerTest.java
index ca8b69c0..926d5b04 100644
--- a/app/src/test/java/org/astraea/service/RequireJmxServerTest.java
+++ b/app/src/test/java/org/astraea/service/RequireJmxServerTest.java
@@ -9,8 +9,13 @@ import org.junit.jupiter.api.Test;
 public class RequireJmxServerTest extends RequireBrokerCluster {
 
   @Test
-  void testQueryBeans() throws Exception {
-    try (var client = new MBeanClient(jmxServiceURL())) {
+  void testQueryBeans() {
+    testQueryBeans(MBeanClient.jndi(jmxServiceURL().getHost(), jmxServiceURL().getPort()));
+    testQueryBeans(MBeanClient.of(jmxServiceURL()));
+  }
+
+  private void testQueryBeans(MBeanClient client) {
+    try (client) {
       var result = client.queryBeans(BeanQuery.all());
       Assertions.assertFalse(result.isEmpty());
     }
@@ -18,7 +23,12 @@ public class RequireJmxServerTest extends RequireBrokerCluster {
 
   @Test
   void testMemory() throws Exception {
-    try (var client = new MBeanClient(jmxServiceURL())) {
+    testMemory(MBeanClient.jndi(jmxServiceURL().getHost(), jmxServiceURL().getPort()));
+    testMemory(MBeanClient.of(jmxServiceURL()));
+  }
+
+  private void testMemory(MBeanClient client) {
+    try (client) {
       var memory = KafkaMetrics.Host.jvmMemory(client);
       Assertions.assertNotEquals(0, memory.heapMemoryUsage().getMax());
     }