Repository: skiptests/astraea
Pull request #150 (2021-12-11 18:52:38)

Introduce new BeanCollector which works without extra threads
這是考量#138後重新實作的版本，跟原本的比有幾個特色：

1. 內部不再用執行緒來處理，簡化模型
2. 因為沒有執行緒，因此不存在一定要關閉的議題（方便singleton)
3. 註冊一個節點後會回傳`Receiver`，之後就從該物件拿`HasBeanObject`，如此`BeanCollector`就不需要傳來傳去


Patch:
diff --git a/app/src/main/java/org/astraea/concurrent/ThreadPool.java b/app/src/main/java/org/astraea/concurrent/ThreadPool.java
index 2cd4a2dd..94ade8d0 100644
--- a/app/src/main/java/org/astraea/concurrent/ThreadPool.java
+++ b/app/src/main/java/org/astraea/concurrent/ThreadPool.java
@@ -56,6 +56,19 @@ public interface ThreadPool extends AutoCloseable {
 
     private Builder() {}
 
+    public Builder runnable(Runnable runnable) {
+      return executor(
+          () -> {
+            runnable.run();
+            return Executor.State.RUNNING;
+          });
+    }
+
+    public Builder runnables(Collection<Runnable> runnables) {
+      runnables.forEach(this::runnable);
+      return this;
+    }
+
     public Builder executor(Executor executor) {
       return executors(List.of(executor));
     }
diff --git a/app/src/main/java/org/astraea/metrics/BeanCollector.java b/app/src/main/java/org/astraea/metrics/BeanCollector.java
deleted file mode 100644
index 92680f75..00000000
--- a/app/src/main/java/org/astraea/metrics/BeanCollector.java
+++ /dev/null
@@ -1,280 +0,0 @@
-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.concurrent.atomic.AtomicBoolean;
-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(
-                        threadIndex ->
-                            (ThreadPool.Executor)
-                                () -> {
-                                  nodes.forEach(NodeImpl::updateObjects);
-                                  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;
-    final AtomicBoolean updating = new AtomicBoolean(false);
-
-    NodeImpl(MBeanClient client, int numberOfObjects) {
-      this.client = client;
-      this.numberOfObjects = numberOfObjects;
-    }
-
-    void updateObjects() {
-      // don't process one node by two threads
-      if (updating.compareAndSet(false, true)) {
-        try {
-          allGetters.forEach(
-              (name, getters) -> {
-                var getter = getters.getLast();
-                if (getter != null) {
-                  if (objects.size() >= numberOfObjects) objects.poll();
-                  objects.offer(getter.apply(client));
-                }
-              });
-        } finally {
-          updating.set(false);
-        }
-      }
-    }
-
-    @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/collector/BeanCollector.java b/app/src/main/java/org/astraea/metrics/collector/BeanCollector.java
new file mode 100644
index 00000000..9e0bcccd
--- /dev/null
+++ b/app/src/main/java/org/astraea/metrics/collector/BeanCollector.java
@@ -0,0 +1,208 @@
+package org.astraea.metrics.collector;
+
+import java.time.Duration;
+import java.util.*;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.BiFunction;
+import java.util.function.Function;
+import java.util.function.Supplier;
+import org.astraea.Utils;
+import org.astraea.metrics.HasBeanObject;
+import org.astraea.metrics.jmx.MBeanClient;
+
+public class BeanCollector {
+
+  public static Builder builder() {
+    return new Builder();
+  }
+
+  public static class Builder {
+    private BiFunction<String, Integer, MBeanClient> clientCreator = MBeanClient::jndi;
+    private Duration interval = Duration.ofSeconds(3);
+    private int numberOfObjectsPerNode = 300;
+
+    private Builder() {}
+
+    public Builder clientCreator(BiFunction<String, Integer, MBeanClient> clientCreator) {
+      this.clientCreator = Objects.requireNonNull(clientCreator);
+      return this;
+    }
+
+    public Builder interval(Duration interval) {
+      this.interval = Objects.requireNonNull(interval);
+      return this;
+    }
+
+    public Builder numberOfObjectsPerNode(int numberOfObjectsPerNode) {
+      this.numberOfObjectsPerNode = Utils.requirePositive(numberOfObjectsPerNode);
+      return this;
+    }
+
+    public BeanCollector build() {
+      return new BeanCollector(clientCreator, interval, numberOfObjectsPerNode);
+    }
+  }
+
+  private final BiFunction<String, Integer, MBeanClient> clientCreator;
+  private final Duration interval;
+  private final int numberOfObjectsPerNode;
+
+  // visible for testing
+  final ConcurrentMap<Node, NodeClient> clients = new ConcurrentSkipListMap<>();
+
+  private BeanCollector(
+      BiFunction<String, Integer, MBeanClient> clientCreator,
+      Duration interval,
+      int numberOfObjectsPerNode) {
+    this.clientCreator = clientCreator;
+    this.interval = interval;
+    this.numberOfObjectsPerNode = numberOfObjectsPerNode;
+  }
+
+  public Register register() {
+    return new Register() {
+      private String host;
+      private int port = -1;
+      private Function<MBeanClient, HasBeanObject> getter;
+
+      @Override
+      public Register host(String host) {
+        this.host = Objects.requireNonNull(host);
+        return this;
+      }
+
+      @Override
+      public Register port(int port) {
+        this.port = Utils.requirePositive(port);
+        return this;
+      }
+
+      @Override
+      public Register metricsGetter(Function<MBeanClient, HasBeanObject> getter) {
+        this.getter = Objects.requireNonNull(getter);
+        return this;
+      }
+
+      @Override
+      public Receiver build() {
+        var node = new Node(host, port);
+        var client = clients.computeIfAbsent(node, ignored -> new NodeClient());
+        return client.add(
+            new Receiver() {
+              @Override
+              public String host() {
+                return host;
+              }
+
+              @Override
+              public int port() {
+                return port;
+              }
+
+              @Override
+              public List<HasBeanObject> current() {
+                client.tryUpdate(
+                    interval,
+                    numberOfObjectsPerNode,
+                    getter,
+                    () -> clientCreator.apply(host, port));
+                return client.current();
+              }
+
+              @Override
+              public void close() {
+                client.remove(this);
+              }
+            });
+      }
+    };
+  }
+
+  // visible for testing
+  static class NodeClient {
+    private final Set<Receiver> receivers = new HashSet<>();
+    private final ReadWriteLock lock = new ReentrantReadWriteLock();
+    // visible for testing
+    MBeanClient mBeanClient;
+    private final Map<Long, HasBeanObject> objects = new HashMap<>();
+
+    Receiver add(Receiver receiver) {
+      lock.writeLock().lock();
+      try {
+        receivers.add(receiver);
+      } finally {
+        lock.writeLock().unlock();
+      }
+      return receiver;
+    }
+
+    void remove(Receiver receiver) {
+      lock.writeLock().lock();
+      try {
+        receivers.remove(receiver);
+        if (receivers.isEmpty()) Utils.close(mBeanClient);
+        mBeanClient = null;
+      } finally {
+        lock.writeLock().unlock();
+      }
+    }
+
+    private void removeOldest() {
+      objects.keySet().stream().min((Long::compare)).ifPresent(objects::remove);
+    }
+
+    List<HasBeanObject> current() {
+      lock.readLock().lock();
+      try {
+        return List.copyOf(objects.values());
+      } finally {
+        lock.readLock().unlock();
+      }
+    }
+
+    void tryUpdate(
+        Duration interval,
+        int numberOfObjectsPerNode,
+        Function<MBeanClient, HasBeanObject> beanGetter,
+        Supplier<MBeanClient> clientCreator) {
+
+      if (lock.writeLock().tryLock()) {
+        try {
+          var needUpdate =
+              objects.keySet().stream()
+                  .max((Long::compare))
+                  .map(last -> last + interval.toMillis() <= System.currentTimeMillis())
+                  .orElse(true);
+          if (needUpdate) {
+            if (mBeanClient == null) mBeanClient = clientCreator.get();
+            if (objects.size() >= numberOfObjectsPerNode) removeOldest();
+            objects.put(System.currentTimeMillis(), beanGetter.apply(mBeanClient));
+          }
+
+        } finally {
+          lock.writeLock().unlock();
+        }
+      }
+    }
+  }
+
+  private static final class Node implements Comparable<Node> {
+    public final String host;
+    public final int port;
+
+    Node(String host, int port) {
+      this.host = host;
+      this.port = port;
+    }
+
+    @Override
+    public int compareTo(Node other) {
+      var result = host.compareTo(other.host);
+      if (result != 0) return result;
+      return Integer.compare(port, other.port);
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/metrics/collector/Receiver.java b/app/src/main/java/org/astraea/metrics/collector/Receiver.java
new file mode 100644
index 00000000..ccaa103e
--- /dev/null
+++ b/app/src/main/java/org/astraea/metrics/collector/Receiver.java
@@ -0,0 +1,27 @@
+package org.astraea.metrics.collector;
+
+import java.util.List;
+import org.astraea.metrics.HasBeanObject;
+
+/**
+ * This Receiver is used to request mbeans. It must be closed. Otherwise, the true connection will
+ * get leaked.
+ */
+public interface Receiver extends AutoCloseable {
+
+  /** @return host of jmx server */
+  String host();
+
+  /** @return port of jmx server */
+  int port();
+
+  /**
+   * This method may request the latest mbeans if the current mbeans are out-of-date.
+   *
+   * @return current mbeans.
+   */
+  List<HasBeanObject> current();
+
+  @Override
+  void close();
+}
diff --git a/app/src/main/java/org/astraea/metrics/collector/Register.java b/app/src/main/java/org/astraea/metrics/collector/Register.java
new file mode 100644
index 00000000..5b27b604
--- /dev/null
+++ b/app/src/main/java/org/astraea/metrics/collector/Register.java
@@ -0,0 +1,29 @@
+package org.astraea.metrics.collector;
+
+import java.util.function.Function;
+import org.astraea.metrics.HasBeanObject;
+import org.astraea.metrics.jmx.MBeanClient;
+
+public interface Register {
+
+  /**
+   * @param host of jmx server
+   * @return this register
+   */
+  Register host(String host);
+
+  /**
+   * @param port of jmx server
+   * @return this register
+   */
+  Register port(int port);
+
+  /**
+   * @param getter to get metrics from MBeanClient
+   * @return this register
+   */
+  Register metricsGetter(Function<MBeanClient, HasBeanObject> getter);
+
+  /** @return create a Receiver */
+  Receiver build();
+}


Test patch:
diff --git a/app/src/test/java/org/astraea/metrics/BeanCollectorTest.java b/app/src/test/java/org/astraea/metrics/BeanCollectorTest.java
deleted file mode 100644
index 674ec055..00000000
--- a/app/src/test/java/org/astraea/metrics/BeanCollectorTest.java
+++ /dev/null
@@ -1,228 +0,0 @@
-package org.astraea.metrics;
-
-import java.time.Duration;
-import java.util.concurrent.ConcurrentSkipListSet;
-import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
-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(),
-                      String.format(
-                          "all objects: %s all nodes: %s current: %s",
-                          collector.numberOfObjects(),
-                          collector.nodes().stream()
-                              .map(n -> n.host() + ":" + n.port())
-                              .collect(Collectors.joining(",")),
-                          node.host() + ":" + node.port())));
-
-      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/metrics/collector/TestBeanCollector.java b/app/src/test/java/org/astraea/metrics/collector/TestBeanCollector.java
new file mode 100644
index 00000000..b3159a3a
--- /dev/null
+++ b/app/src/test/java/org/astraea/metrics/collector/TestBeanCollector.java
@@ -0,0 +1,208 @@
+package org.astraea.metrics.collector;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.BiFunction;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.astraea.concurrent.ThreadPool;
+import org.astraea.metrics.HasBeanObject;
+import org.astraea.metrics.jmx.MBeanClient;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+public class TestBeanCollector {
+  private final HasBeanObject mbean = Mockito.mock(HasBeanObject.class);
+  private final MBeanClient mbeanClient = Mockito.mock(MBeanClient.class);
+  private final BiFunction<String, Integer, MBeanClient> clientCreator =
+      (host, port) -> mbeanClient;
+
+  private static void sleep(int seconds) {
+    try {
+      TimeUnit.SECONDS.sleep(seconds);
+    } catch (InterruptedException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  @Test
+  void testAddress() {
+    var collector = BeanCollector.builder().clientCreator(clientCreator).build();
+    var receiver =
+        collector
+            .register()
+            .host("unknown")
+            .port(100)
+            .metricsGetter(client -> Mockito.mock(HasBeanObject.class))
+            .build();
+    Assertions.assertEquals("unknown", receiver.host());
+    Assertions.assertEquals(100, receiver.port());
+  }
+
+  @Test
+  void theImmutableCurrent() {
+    var collector = BeanCollector.builder().clientCreator(clientCreator).build();
+    var receivers = receivers(collector);
+    receivers.forEach(
+        r ->
+            Assertions.assertThrows(
+                UnsupportedOperationException.class, () -> r.current().add(mbean)));
+  }
+
+  @Test
+  void testNumberOfObjectsPerNode() {
+    var collector =
+        BeanCollector.builder()
+            .numberOfObjectsPerNode(2)
+            .interval(Duration.ofMillis(1))
+            .clientCreator(clientCreator)
+            .build();
+
+    var receiver =
+        collector
+            .register()
+            .host("unknown")
+            .port(100)
+            .metricsGetter(client -> Mockito.mock(HasBeanObject.class))
+            .build();
+
+    var c0 = receiver.current();
+    Assertions.assertEquals(1, c0.size());
+    var firstObject = c0.get(0);
+    sleep(1);
+
+    var c1 = receiver.current();
+    Assertions.assertEquals(2, c1.size());
+    var secondObject = c1.stream().filter(o -> o != firstObject).findFirst().get();
+    sleep(1);
+
+    var c2 = receiver.current();
+    Assertions.assertEquals(2, c2.size());
+    // the oldest element should be removed
+    Assertions.assertFalse(c2.contains(firstObject));
+    Assertions.assertTrue(c2.contains(secondObject));
+  }
+
+  @Test
+  void testBeanCollectorBuilder() {
+    Assertions.assertThrows(
+        NullPointerException.class, () -> BeanCollector.builder().interval(null));
+    Assertions.assertThrows(
+        NullPointerException.class, () -> BeanCollector.builder().clientCreator(null));
+    Assertions.assertThrows(
+        IllegalArgumentException.class, () -> BeanCollector.builder().numberOfObjectsPerNode(-1));
+  }
+
+  @Test
+  void testRegister() {
+    var collector = BeanCollector.builder().clientCreator(clientCreator).build();
+    Assertions.assertThrows(NullPointerException.class, () -> collector.register().host(null));
+    Assertions.assertThrows(IllegalArgumentException.class, () -> collector.register().port(-1));
+    Assertions.assertThrows(
+        NullPointerException.class, () -> collector.register().metricsGetter(null));
+  }
+
+  private List<Receiver> receivers(BeanCollector collector) {
+    var receivers = new ArrayList<Receiver>();
+    Runnable runnable =
+        () -> {
+          try {
+            var receiver =
+                collector
+                    .register()
+                    .host("unknown")
+                    .port(100)
+                    .metricsGetter(client -> mbean)
+                    .build();
+            synchronized (receivers) {
+              receivers.add(receiver);
+            }
+          } finally {
+            sleep(1);
+          }
+        };
+
+    try (var pool =
+        ThreadPool.builder()
+            .runnables(IntStream.range(0, 3).mapToObj(i -> runnable).collect(Collectors.toList()))
+            .build()) {
+      sleep(1);
+    }
+    return receivers;
+  }
+
+  @Test
+  void testCloseReceiver() {
+    var collector = BeanCollector.builder().clientCreator(clientCreator).build();
+
+    var receivers = receivers(collector);
+    receivers.forEach(Receiver::current);
+
+    Assertions.assertEquals(1, collector.clients.size());
+    Assertions.assertEquals(
+        mbeanClient, collector.clients.entrySet().iterator().next().getValue().mBeanClient);
+
+    receivers.forEach(Receiver::close);
+    Assertions.assertEquals(1, collector.clients.size());
+    Assertions.assertNull(collector.clients.entrySet().iterator().next().getValue().mBeanClient);
+  }
+
+  @Test
+  void testMultiThreadsUpdate() {
+
+    var collector =
+        BeanCollector.builder()
+            .interval(Duration.ofSeconds(100))
+            .clientCreator(clientCreator)
+            .build();
+
+    var receivers = receivers(collector);
+
+    try (var pool =
+        ThreadPool.builder()
+            .runnables(
+                receivers.stream()
+                    .map(receiver -> ((Runnable) receiver::current))
+                    .collect(Collectors.toList()))
+            .build()) {
+      sleep(3);
+    }
+    receivers.forEach(r -> Assertions.assertEquals(1, r.current().size()));
+  }
+
+  @Test
+  void testLargeInterval() {
+    testInterval(Duration.ofSeconds(100), List.of(1, 1, 1));
+  }
+
+  @Test
+  void testSmallInterval() {
+    testInterval(Duration.ofMillis(100), List.of(1, 2, 3));
+  }
+
+  private void testInterval(Duration interval, List<Integer> expectedSizes) {
+    var collector = BeanCollector.builder().interval(interval).clientCreator(clientCreator).build();
+    var count = new AtomicInteger();
+    var receiver =
+        collector
+            .register()
+            .host("unknown")
+            .port(100)
+            .metricsGetter(
+                c -> {
+                  count.incrementAndGet();
+                  return mbean;
+                })
+            .build();
+
+    for (var expect : expectedSizes) {
+      Assertions.assertEquals(expect, receiver.current().size());
+      Assertions.assertEquals(expect, count.get());
+      sleep(1);
+    }
+  }
+}