Repository: skiptests/astraea
Pull request #154 (2021-12-17 08:11:21)

The receiver should keep independent objects
from https://github.com/skiptests/astraea/pull/138#discussion_r771094107

現在不同的receiver會維護各自的objects，而非跟其他receivers共用


Patch:
diff --git a/app/src/main/java/org/astraea/metrics/collector/BeanCollector.java b/app/src/main/java/org/astraea/metrics/collector/BeanCollector.java
index 9e0bcccd..84adaca4 100644
--- a/app/src/main/java/org/astraea/metrics/collector/BeanCollector.java
+++ b/app/src/main/java/org/astraea/metrics/collector/BeanCollector.java
@@ -4,11 +4,10 @@ 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.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 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;
@@ -51,7 +50,7 @@ public class BeanCollector {
   private final int numberOfObjectsPerNode;
 
   // visible for testing
-  final ConcurrentMap<Node, NodeClient> clients = new ConcurrentSkipListMap<>();
+  final ConcurrentMap<String, Node> nodes = new ConcurrentSkipListMap<>();
 
   private BeanCollector(
       BiFunction<String, Integer, MBeanClient> clientCreator,
@@ -88,10 +87,12 @@ public class BeanCollector {
 
       @Override
       public Receiver build() {
-        var node = new Node(host, port);
-        var client = clients.computeIfAbsent(node, ignored -> new NodeClient());
-        return client.add(
+        var nodeKey = host + ":" + port;
+        var node = nodes.computeIfAbsent(nodeKey, ignored -> new Node(host, port));
+        var receiver =
             new Receiver() {
+              private final Map<Long, HasBeanObject> objects = new HashMap<>();
+
               @Override
               public String host() {
                 return host;
@@ -104,92 +105,62 @@ public class BeanCollector {
 
               @Override
               public List<HasBeanObject> current() {
-                client.tryUpdate(
-                    interval,
-                    numberOfObjectsPerNode,
-                    getter,
-                    () -> clientCreator.apply(host, port));
-                return client.current();
+                tryUpdate();
+                return List.copyOf(objects.values());
               }
 
               @Override
               public void close() {
-                client.remove(this);
+                node.lock.lock();
+                try {
+                  node.receivers.remove(this);
+                  if (node.receivers.isEmpty()) Utils.close(node.mBeanClient);
+                  node.mBeanClient = null;
+                } finally {
+                  node.lock.unlock();
+                }
               }
-            });
-      }
-    };
-  }
 
-  // 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) {
+              private void tryUpdate() {
+                if (node.lock.tryLock()) {
+                  try {
+                    var needUpdate =
+                        objects.keySet().stream()
+                            .max((Long::compare))
+                            .map(last -> last + interval.toMillis() <= System.currentTimeMillis())
+                            .orElse(true);
+                    if (needUpdate) {
+                      if (node.mBeanClient == null)
+                        node.mBeanClient = clientCreator.apply(host, port);
+                      if (objects.size() >= numberOfObjectsPerNode)
+                        objects.keySet().stream().min((Long::compare)).ifPresent(objects::remove);
+                      objects.put(System.currentTimeMillis(), getter.apply(node.mBeanClient));
+                    }
+                  } finally {
+                    node.lock.unlock();
+                  }
+                }
+              }
+            };
 
-      if (lock.writeLock().tryLock()) {
+        // add receiver
+        node.lock.lock();
         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));
-          }
-
+          node.receivers.add(receiver);
         } finally {
-          lock.writeLock().unlock();
+          node.lock.unlock();
         }
+        return receiver;
       }
-    }
+    };
   }
 
-  private static final class Node implements Comparable<Node> {
+  // visible for testing
+  static final class Node {
+    private final Set<Receiver> receivers = new HashSet<>();
+    private final Lock lock = new ReentrantLock();
+    // visible for testing
+    MBeanClient mBeanClient;
     public final String host;
     public final int port;
 
@@ -197,12 +168,5 @@ public class BeanCollector {
       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);
-    }
   }
 }


Test patch:
diff --git a/app/src/test/java/org/astraea/metrics/collector/TestBeanCollector.java b/app/src/test/java/org/astraea/metrics/collector/TestBeanCollector.java
index b3159a3a..1e5589ea 100644
--- a/app/src/test/java/org/astraea/metrics/collector/TestBeanCollector.java
+++ b/app/src/test/java/org/astraea/metrics/collector/TestBeanCollector.java
@@ -3,6 +3,7 @@ package org.astraea.metrics.collector;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.BiFunction;
@@ -142,13 +143,13 @@ public class TestBeanCollector {
     var receivers = receivers(collector);
     receivers.forEach(Receiver::current);
 
-    Assertions.assertEquals(1, collector.clients.size());
+    Assertions.assertEquals(1, collector.nodes.size());
     Assertions.assertEquals(
-        mbeanClient, collector.clients.entrySet().iterator().next().getValue().mBeanClient);
+        mbeanClient, collector.nodes.entrySet().iterator().next().getValue().mBeanClient);
 
     receivers.forEach(Receiver::close);
-    Assertions.assertEquals(1, collector.clients.size());
-    Assertions.assertNull(collector.clients.entrySet().iterator().next().getValue().mBeanClient);
+    Assertions.assertEquals(1, collector.nodes.size());
+    Assertions.assertNull(collector.nodes.entrySet().iterator().next().getValue().mBeanClient);
   }
 
   @Test
@@ -205,4 +206,40 @@ public class TestBeanCollector {
       sleep(1);
     }
   }
+
+  @Test
+  void testMultiplesReceiversData() {
+    var collector =
+        BeanCollector.builder()
+            .interval(Duration.ofSeconds(1))
+            .clientCreator(clientCreator)
+            .build();
+
+    var receivers =
+        IntStream.range(0, 2)
+            .mapToObj(
+                i -> {
+                  var count = new AtomicInteger();
+                  return Map.entry(
+                      count,
+                      collector
+                          .register()
+                          .host("unknown")
+                          .port(100)
+                          .metricsGetter(
+                              c -> {
+                                count.incrementAndGet();
+                                return mbean;
+                              })
+                          .build());
+                })
+            .collect(Collectors.toList());
+
+    sleep(1);
+    receivers.forEach(e -> Assertions.assertEquals(0, e.getKey().get()));
+
+    sleep(1);
+    receivers.forEach(e -> Assertions.assertEquals(1, e.getValue().current().size()));
+    receivers.forEach(e -> Assertions.assertEquals(1, e.getKey().get()));
+  }
 }