Repository: skiptests/astraea
Pull request #123 (2021-11-25 09:07:18)

Implmenet thread-safe mbean collector
這是用來示範如何將partitioner中最需要統一管理（因為效能和連線數的考量）的mbean集中化


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/Utils.java b/app/src/main/java/org/astraea/Utils.java
index 94e003db..eaf3c062 100644
--- a/app/src/main/java/org/astraea/Utils.java
+++ b/app/src/main/java/org/astraea/Utils.java
@@ -94,5 +94,11 @@ public final class Utils {
     throw new RuntimeException("Timeout to wait procedure");
   }
 
+  public static int requirePositive(int value) {
+    if (value <= 0)
+      throw new IllegalArgumentException("the value: " + value + " must be bigger than zero");
+    return value;
+  }
+
   private Utils() {}
 }
diff --git a/app/src/main/java/org/astraea/concurrent/ThreadPool.java b/app/src/main/java/org/astraea/concurrent/ThreadPool.java
index 4f270678..2cd4a2dd 100644
--- a/app/src/main/java/org/astraea/concurrent/ThreadPool.java
+++ b/app/src/main/java/org/astraea/concurrent/ThreadPool.java
@@ -18,6 +18,11 @@ public interface ThreadPool extends AutoCloseable {
   /** wait all executors to be done. */
   void waitAll();
 
+  boolean isClosed();
+
+  /** @return the number of threads */
+  int size();
+
   static Builder builder() {
     return new Builder();
   }
@@ -85,7 +90,7 @@ public interface ThreadPool extends AutoCloseable {
       return new ThreadPool() {
         @Override
         public void close() {
-          service.shutdown();
+          service.shutdownNow();
           closed.set(true);
           executors.forEach(Executor::wakeup);
           waitAll();
@@ -99,6 +104,16 @@ public interface ThreadPool extends AutoCloseable {
             // swallow
           }
         }
+
+        @Override
+        public boolean isClosed() {
+          return closed.get();
+        }
+
+        @Override
+        public int size() {
+          return executors.size();
+        }
       };
     }
   }
diff --git a/app/src/main/java/org/astraea/metrics/BeanCollector.java b/app/src/main/java/org/astraea/metrics/BeanCollector.java
new file mode 100644
index 00000000..4b9de205
--- /dev/null
+++ b/app/src/main/java/org/astraea/metrics/BeanCollector.java
@@ -0,0 +1,278 @@
+package org.astraea.metrics;
+
+import java.time.Duration;
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.function.Function;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.astraea.Utils;
+import org.astraea.concurrent.ThreadPool;
+import org.astraea.metrics.jmx.MBeanClient;
+
+/**
+ * this class is used to manage multiples jmx connections. Normally, we want to get different
+ * metrics from the same jmx server, but we hate the high cost caused by multiples connections.
+ * Hence, this class keeps the single connection for each jmx server, and you can register the
+ * `getter` the fetch various mbean objects through same connection.
+ */
+public class BeanCollector implements AutoCloseable {
+
+  public static Builder builder() {
+    return new Builder();
+  }
+
+  static class Builder {
+    private int numberOfThreads = 2;
+    private Duration interval = Duration.ofSeconds(1);
+    private int numberOfObjectsPerNode = 300;
+
+    private Builder() {}
+
+    public Builder numberOfThreads(int numberOfThreads) {
+      this.numberOfThreads = numberOfThreads;
+      return this;
+    }
+
+    public Builder interval(Duration interval) {
+      this.interval = interval;
+      return this;
+    }
+
+    public Builder numberOfObjectsPerNode(int numberOfObjectsPerNode) {
+      this.numberOfObjectsPerNode = numberOfObjectsPerNode;
+      return this;
+    }
+
+    public BeanCollector build() {
+      return new BeanCollector(numberOfThreads, interval, numberOfObjectsPerNode);
+    }
+  }
+
+  private final Queue<NodeImpl> nodes = new ConcurrentLinkedQueue<>();
+  private final ThreadPool pool;
+  private final int numberOfObjectsPerNode;
+  private final Object notification = new Object();
+
+  private BeanCollector(int numberOfThreads, Duration interval, int numberOfObjectsPerNode) {
+    this.numberOfObjectsPerNode = numberOfObjectsPerNode;
+    this.pool =
+        ThreadPool.builder()
+            .executors(
+                IntStream.range(0, numberOfThreads)
+                    .mapToObj(
+                        i ->
+                            (ThreadPool.Executor)
+                                () -> {
+                                  var node = nodes.poll();
+                                  if (node != null)
+                                    try {
+                                      node.updateObjects();
+                                    } finally {
+                                      nodes.add(node);
+                                    }
+
+                                  synchronized (notification) {
+                                    notification.wait(interval.toMillis());
+                                  }
+                                  return ThreadPool.Executor.State.RUNNING;
+                                })
+                    .collect(Collectors.toList()))
+            .build();
+  }
+
+  /** @return the monitored host/port */
+  public List<Node> nodes() {
+    return nodes.stream().map(n -> (Node) n).collect(Collectors.toList());
+  }
+
+  /**
+   * @param host target host
+   * @param port target port
+   * @return the objects from target host/port
+   */
+  public List<HasBeanObject> objects(String host, int port) {
+    return nodes.stream()
+        .filter(n -> n.host().equals(host) && n.port() == port)
+        .flatMap(n -> n.objects.stream())
+        .collect(Collectors.toList());
+  }
+
+  public Map<Node, List<HasBeanObject>> objects() {
+    return nodes.stream()
+        .collect(Collectors.toMap(Function.identity(), node -> new ArrayList<>(node.objects)));
+  }
+
+  /** @return the number of all objects */
+  public int numberOfObjects() {
+    return nodes.stream().mapToInt(node -> node.objects.size()).sum();
+  }
+
+  /** @return the number of all getters */
+  int numberOfGetters() {
+    return nodes.stream()
+        .mapToInt(node -> node.allGetters.values().stream().mapToInt(Deque::size).sum())
+        .sum();
+  }
+
+  /** wake up all threads to update mbean objects */
+  public void requestToUpdate() {
+    synchronized (notification) {
+      notification.notifyAll();
+    }
+  }
+
+  @Override
+  public void close() {
+    pool.close();
+    pool.waitAll();
+
+    // close all nodes
+    while (!nodes.isEmpty()) {
+      var node = nodes.poll();
+      if (node != null) node.close();
+    }
+  }
+
+  /**
+   * @return Register is used to store your getter which can fetch mbean objects from jmx
+   *     connection.
+   */
+  public Register register() {
+    return new Register() {
+      private String host;
+      private int port = -1;
+      private Supplier<MBeanClient> supplier;
+      private String getterName;
+      private Function<MBeanClient, HasBeanObject> getter;
+
+      @Override
+      public Register host(String host) {
+        this.host = host;
+        return this;
+      }
+
+      @Override
+      public Register port(int port) {
+        this.port = port;
+        return this;
+      }
+
+      @Override
+      public Register clientSupplier(Supplier<MBeanClient> supplier) {
+        this.supplier = supplier;
+        return this;
+      }
+
+      @Override
+      public Register metricsGetter(String name, Function<MBeanClient, HasBeanObject> getter) {
+        this.getterName = name;
+        this.getter = getter;
+        return this;
+      }
+
+      @Override
+      public Unregister build() {
+        if (pool.isClosed()) throw new RuntimeException("this is closed!!!");
+        var finalHost = Objects.requireNonNull(host);
+        var finalPort = Utils.requirePositive(port);
+        Supplier<MBeanClient> finalSupplier =
+            supplier == null ? () -> MBeanClient.jndi(finalHost, finalPort) : supplier;
+        var finalGetter = Objects.requireNonNull(getter);
+        var finalGetterName = getterName == null ? finalGetter.toString() : getterName;
+        var node =
+            nodes.stream()
+                .filter(n -> n.host().equals(finalHost) && n.port() == finalPort)
+                .findFirst()
+                .orElseGet(
+                    () -> {
+                      var n = new NodeImpl(finalSupplier.get(), numberOfObjectsPerNode);
+                      nodes.add(n);
+                      return n;
+                    });
+        node.allGetters
+            .computeIfAbsent(finalGetterName, ignore -> new ConcurrentLinkedDeque<>())
+            .add(getter);
+        return () -> {
+          var getters = node.allGetters.get(finalGetterName);
+          if (getters != null) getters.remove(getter);
+        };
+      }
+    };
+  }
+
+  interface Register {
+    Register host(String host);
+
+    Register port(int port);
+
+    Register clientSupplier(Supplier<MBeanClient> supplier);
+
+    default Register metricsGetter(Function<MBeanClient, HasBeanObject> getter) {
+      return metricsGetter(null, getter);
+    }
+
+    /**
+     * @param name of getter. This must be unique. The getters having same name will be in same
+     *     group and only the latest one gets work.
+     * @param getter getter
+     * @return this register
+     */
+    Register metricsGetter(String name, Function<MBeanClient, HasBeanObject> getter);
+
+    Unregister build();
+  }
+
+  interface Unregister {
+    /** remove the getter */
+    void removeGetter();
+  }
+
+  interface Node {
+    String host();
+
+    int port();
+  }
+
+  private static class NodeImpl implements AutoCloseable, Node {
+    final MBeanClient client;
+    final Map<String, Deque<Function<MBeanClient, HasBeanObject>>> allGetters =
+        new ConcurrentHashMap<>();
+    final Queue<HasBeanObject> objects = new ConcurrentLinkedQueue<>();
+    final int numberOfObjects;
+
+    NodeImpl(MBeanClient client, int numberOfObjects) {
+      this.client = client;
+      this.numberOfObjects = numberOfObjects;
+    }
+
+    void updateObjects() {
+      allGetters.forEach(
+          (name, getters) -> {
+            var getter = getters.getLast();
+            if (getter != null) {
+              if (objects.size() >= numberOfObjects) objects.poll();
+              objects.offer(getter.apply(client));
+            }
+          });
+    }
+
+    @Override
+    public void close() {
+      Utils.close(client);
+    }
+
+    @Override
+    public String host() {
+      return client.host();
+    }
+
+    @Override
+    public int port() {
+      return client.port();
+    }
+  }
+}
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 ec674d5d..91ad9654 100644
--- a/app/src/main/java/org/astraea/metrics/jmx/MBeanClient.java
+++ b/app/src/main/java/org/astraea/metrics/jmx/MBeanClient.java
@@ -30,7 +30,8 @@ public interface MBeanClient extends AutoCloseable {
     try {
       return of(
           new JMXServiceURL(
-              String.format("service:jmx:rmi:///jndi/rmi://%s:%s/jmxrmi", host, port)));
+              String.format(
+                  "service:jmx:rmi://%s:%s/jndi/rmi://%s:%s/jmxrmi", host, port, host, port)));
     } catch (MalformedURLException e) {
       throw new IllegalArgumentException(e);
     }
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) {


Test patch:
diff --git a/app/src/test/java/org/astraea/concurrent/ThreadPoolTest.java b/app/src/test/java/org/astraea/concurrent/ThreadPoolTest.java
index 57301bc0..c973347c 100644
--- a/app/src/test/java/org/astraea/concurrent/ThreadPoolTest.java
+++ b/app/src/test/java/org/astraea/concurrent/ThreadPoolTest.java
@@ -4,6 +4,7 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
 
 public class ThreadPoolTest {
 
@@ -47,4 +48,19 @@ public class ThreadPoolTest {
       pool.waitAll();
     }
   }
+
+  @Timeout(10)
+  @Test
+  void testInterrupt() {
+    var pool =
+        ThreadPool.builder()
+            .executor(
+                () -> {
+                  TimeUnit.SECONDS.sleep(1000);
+                  return ThreadPool.Executor.State.DONE;
+                })
+            .build();
+    pool.close();
+    Assertions.assertTrue(pool.isClosed());
+  }
 }
diff --git a/app/src/test/java/org/astraea/metrics/BeanCollectorTest.java b/app/src/test/java/org/astraea/metrics/BeanCollectorTest.java
new file mode 100644
index 00000000..5b4f6c5e
--- /dev/null
+++ b/app/src/test/java/org/astraea/metrics/BeanCollectorTest.java
@@ -0,0 +1,219 @@
+package org.astraea.metrics;
+
+import java.time.Duration;
+import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.TimeUnit;
+import org.astraea.Utils;
+import org.astraea.metrics.java.JvmMemory;
+import org.astraea.metrics.jmx.MBeanClient;
+import org.astraea.metrics.kafka.KafkaMetrics;
+import org.astraea.service.RequireBrokerCluster;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+public class BeanCollectorTest extends RequireBrokerCluster {
+
+  private static MBeanClient mock(String host, int port) {
+    var client = Mockito.mock(MBeanClient.class);
+    Mockito.when(client.host()).thenReturn(host);
+    Mockito.when(client.port()).thenReturn(port);
+    return client;
+  }
+
+  @Test
+  void testMultiplesNodes() {
+    var bean = Mockito.mock(HasBeanObject.class);
+    try (var collector = BeanCollector.builder().build()) {
+      collector
+          .register()
+          .host("aa")
+          .port(22222)
+          .clientSupplier(() -> mock("aa", 22222))
+          .metricsGetter(client -> bean)
+          .build();
+
+      Assertions.assertEquals(1, collector.nodes().size());
+
+      collector
+          .register()
+          .host("aa")
+          .port(33333)
+          .clientSupplier(() -> mock("aa", 33333))
+          .metricsGetter(client -> bean)
+          .build();
+
+      Assertions.assertEquals(2, collector.nodes().size());
+
+      // add it again
+      collector
+          .register()
+          .host("aa")
+          .port(22222)
+          .clientSupplier(() -> mock("aa", 22222))
+          .metricsGetter(client -> bean)
+          .build();
+      Assertions.assertEquals(2, collector.nodes().size());
+    }
+  }
+
+  @Test
+  void testNumberOfThreads() {
+    var numberOfThreads = 3;
+    var client = mock("host", 11111);
+    var bean = Mockito.mock(HasBeanObject.class);
+    try (var collector = BeanCollector.builder().numberOfThreads(numberOfThreads).build()) {
+      var threads = new ConcurrentSkipListSet<String>();
+      collector
+          .register()
+          .host("this")
+          .port(1234)
+          .clientSupplier(() -> client)
+          .metricsGetter(
+              ignore -> {
+                threads.add(Thread.currentThread().getName());
+                return bean;
+              })
+          .build();
+      Utils.waitFor(() -> threads.size() == numberOfThreads);
+    }
+  }
+
+  @Test
+  void testNumberOfObjectsPerNode() {
+    var numberOfObjectsPerNode = 10;
+    var client = mock("host", 11111);
+    var bean = Mockito.mock(HasBeanObject.class);
+    try (var collector =
+        BeanCollector.builder()
+            .interval(Duration.ofMillis(100))
+            .numberOfObjectsPerNode(numberOfObjectsPerNode)
+            .build()) {
+      collector
+          .register()
+          .host("this")
+          .port(1234)
+          .clientSupplier(() -> client)
+          .metricsGetter(ignore -> bean)
+          .build();
+      Utils.waitFor(() -> collector.numberOfObjects() >= numberOfObjectsPerNode);
+    }
+  }
+
+  @Test
+  void testInterval() throws Exception {
+    var client = mock("host", 11111);
+    var bean = Mockito.mock(HasBeanObject.class);
+    try (var collector = BeanCollector.builder().interval(Duration.ofSeconds(1000)).build()) {
+      collector
+          .register()
+          .host("this")
+          .port(1234)
+          .clientSupplier(() -> client)
+          .metricsGetter(ignore -> bean)
+          .build();
+
+      collector.requestToUpdate();
+      TimeUnit.SECONDS.sleep(3);
+      var current = collector.numberOfObjects();
+
+      collector.requestToUpdate();
+      Utils.waitFor(() -> collector.numberOfObjects() > current);
+    }
+  }
+
+  @Test
+  void testReleaseClient() {
+    var client = mock("host", 11111);
+    var bean = Mockito.mock(HasBeanObject.class);
+    try (var collector = BeanCollector.builder().build()) {
+      var object1 =
+          collector
+              .register()
+              .host("this")
+              .port(1234)
+              .clientSupplier(() -> client)
+              .metricsGetter(ignore -> bean)
+              .build();
+
+      Assertions.assertEquals(1, collector.numberOfGetters());
+
+      var object2 =
+          collector
+              .register()
+              .host("this")
+              .port(1234)
+              .clientSupplier(() -> client)
+              .metricsGetter(ignore -> bean)
+              .build();
+
+      Assertions.assertEquals(2, collector.numberOfGetters());
+
+      object1.removeGetter();
+      Assertions.assertEquals(1, collector.numberOfGetters());
+
+      object2.removeGetter();
+      Assertions.assertEquals(0, collector.numberOfGetters());
+    }
+  }
+
+  @Test
+  void testAddNode() {
+    try (var collector = BeanCollector.builder().build()) {
+      var releaseGetter =
+          collector
+              .register()
+              .host(jmxServiceURL().getHost())
+              .port(jmxServiceURL().getPort())
+              .metricsGetter(KafkaMetrics.Host::jvmMemory)
+              .build();
+      Utils.waitFor(() -> collector.numberOfObjects() > 0);
+      collector
+          .objects()
+          .values()
+          .forEach(
+              all -> all.forEach(object -> Assertions.assertTrue(object instanceof JvmMemory)));
+
+      collector
+          .nodes()
+          .forEach(
+              node ->
+                  Assertions.assertNotEquals(
+                      0, collector.objects(node.host(), node.port()).size()));
+
+      collector
+          .nodes()
+          .forEach(node -> Assertions.assertEquals(node.host(), jmxServiceURL().getHost()));
+
+      collector
+          .nodes()
+          .forEach(node -> Assertions.assertEquals(node.port(), jmxServiceURL().getPort()));
+
+      releaseGetter.removeGetter();
+      Assertions.assertEquals(0, collector.numberOfGetters());
+    }
+  }
+
+  @Test
+  void testAddDuplicateClient() {
+    try (var collector = BeanCollector.builder().build()) {
+      collector
+          .register()
+          .host(jmxServiceURL().getHost())
+          .port(jmxServiceURL().getPort())
+          .metricsGetter(KafkaMetrics.Host::jvmMemory)
+          .build();
+      collector
+          .register()
+          .host(jmxServiceURL().getHost())
+          .port(jmxServiceURL().getPort())
+          .clientSupplier(
+              () -> {
+                throw new RuntimeException("the client should be existent!!!");
+              })
+          .metricsGetter(KafkaMetrics.Host::jvmMemory)
+          .build();
+      Assertions.assertEquals(1, collector.nodes().size());
+    }
+  }
+}
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());
     }
   }