Repository: skiptests/astraea
Pull request #296 (2022-04-08 14:01:22)

Add created timestamp to all bean objects
新增`createdTimestamp`給所有bean objects，同時該timestamp將用作`BeanCollector`內部的時間淘汰選擇


Patch:
diff --git a/app/src/main/java/org/astraea/metrics/HasBeanObject.java b/app/src/main/java/org/astraea/metrics/HasBeanObject.java
index 324ab249..52252068 100644
--- a/app/src/main/java/org/astraea/metrics/HasBeanObject.java
+++ b/app/src/main/java/org/astraea/metrics/HasBeanObject.java
@@ -4,4 +4,8 @@ import org.astraea.metrics.jmx.BeanObject;
 
 public interface HasBeanObject {
   BeanObject beanObject();
+
+  default long createdTimestamp() {
+    return beanObject().createdTimestamp();
+  }
 }
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 36154ecd..5a52b83f 100644
--- a/app/src/main/java/org/astraea/metrics/collector/BeanCollector.java
+++ b/app/src/main/java/org/astraea/metrics/collector/BeanCollector.java
@@ -141,7 +141,11 @@ public class BeanCollector {
                       if (objects.size() + beans.size() <= numberOfObjectsPerNode) break;
                       objects.remove(t);
                     }
-                    long now = System.currentTimeMillis();
+                    long now =
+                        beans.stream()
+                            .mapToLong(HasBeanObject::createdTimestamp)
+                            .min()
+                            .orElse(System.currentTimeMillis());
                     for (var bean : beans) objects.put(now++, bean);
                   } finally {
                     node.lock.unlock();
diff --git a/app/src/main/java/org/astraea/metrics/jmx/BeanObject.java b/app/src/main/java/org/astraea/metrics/jmx/BeanObject.java
index 5628969c..b1a6e509 100644
--- a/app/src/main/java/org/astraea/metrics/jmx/BeanObject.java
+++ b/app/src/main/java/org/astraea/metrics/jmx/BeanObject.java
@@ -12,6 +12,7 @@ public class BeanObject {
   private final String domainName;
   private final Map<String, String> properties;
   private final Map<String, Object> attributes;
+  private final long createdTimestamp = System.currentTimeMillis();
 
   /**
    * construct a {@link BeanObject}
@@ -56,6 +57,10 @@ public class BeanObject {
     return attributes;
   }
 
+  public long createdTimestamp() {
+    return createdTimestamp;
+  }
+
   @Override
   public String toString() {
     String propertyList =


Test patch:
diff --git a/app/src/test/java/org/astraea/metrics/collector/BeanCollectorTest.java b/app/src/test/java/org/astraea/metrics/collector/BeanCollectorTest.java
index df7f3fd9..06d74bf6 100644
--- a/app/src/test/java/org/astraea/metrics/collector/BeanCollectorTest.java
+++ b/app/src/test/java/org/astraea/metrics/collector/BeanCollectorTest.java
@@ -13,13 +13,13 @@ import org.astraea.concurrent.Executor;
 import org.astraea.concurrent.State;
 import org.astraea.concurrent.ThreadPool;
 import org.astraea.metrics.HasBeanObject;
+import org.astraea.metrics.jmx.BeanObject;
 import org.astraea.metrics.jmx.MBeanClient;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
 
 public class BeanCollectorTest {
-  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;
@@ -32,6 +32,18 @@ public class BeanCollectorTest {
     }
   }
 
+  private static Executor executor(Runnable runnable) {
+    return () -> {
+      runnable.run();
+      return State.RUNNING;
+    };
+  }
+
+  private static HasBeanObject createBeanObject() {
+    var obj = new BeanObject("domain", Map.of(), Map.of());
+    return () -> obj;
+  }
+
   @Test
   void testAddress() {
     var collector = BeanCollector.builder().clientCreator(clientCreator).build();
@@ -53,7 +65,7 @@ public class BeanCollectorTest {
     receivers.forEach(
         r ->
             Assertions.assertThrows(
-                UnsupportedOperationException.class, () -> r.current().add(mbean)));
+                UnsupportedOperationException.class, () -> r.current().add(createBeanObject())));
   }
 
   @Test
@@ -70,7 +82,7 @@ public class BeanCollectorTest {
             .register()
             .host("unknown")
             .port(100)
-            .fetcher(client -> List.of(Mockito.mock(HasBeanObject.class)))
+            .fetcher(client -> List.of(createBeanObject()))
             .build();
 
     var c0 = receiver.current();
@@ -118,7 +130,7 @@ public class BeanCollectorTest {
                     .register()
                     .host("unknown")
                     .port(100)
-                    .fetcher(client -> List.of(mbean))
+                    .fetcher(client -> List.of(createBeanObject()))
                     .build();
             synchronized (receivers) {
               receivers.add(receiver);
@@ -200,7 +212,7 @@ public class BeanCollectorTest {
             .fetcher(
                 client -> {
                   count.incrementAndGet();
-                  return List.of(mbean);
+                  return List.of(createBeanObject());
                 })
             .build();
 
@@ -233,7 +245,7 @@ public class BeanCollectorTest {
                           .fetcher(
                               client -> {
                                 count.incrementAndGet();
-                                return List.of(mbean);
+                                return List.of(createBeanObject());
                               })
                           .build());
                 })
@@ -247,10 +259,26 @@ public class BeanCollectorTest {
     receivers.forEach(e -> Assertions.assertEquals(1, e.getKey().get()));
   }
 
-  private static Executor executor(Runnable runnable) {
-    return () -> {
-      runnable.run();
-      return State.RUNNING;
-    };
+  @Test
+  void testCreatedTimestamp() throws InterruptedException {
+    var collector =
+        BeanCollector.builder()
+            .interval(Duration.ofSeconds(1))
+            .clientCreator(clientCreator)
+            .build();
+    var obj = new BeanObject("domain", Map.of(), Map.of());
+    try (var receiver =
+        collector
+            .register()
+            .host("unknown")
+            .port(100)
+            .fetcher(client -> List.of(() -> obj))
+            .build()) {
+      // wait for updating cache
+      TimeUnit.SECONDS.sleep(1);
+      var objs = receiver.current();
+      Assertions.assertEquals(1, objs.size());
+      Assertions.assertEquals(obj.createdTimestamp(), objs.iterator().next().createdTimestamp());
+    }
   }
 }