Repository: skiptests/astraea (java)

Issue:
[SERIALIZATION] `ClusterBean` Serialization
未來 https://github.com/skiptests/astraea/issues/1693 會有序列化/反序列化 `ClusterBean` 的需求。

這個 Issue 用來討論該如何實作這個需求。

目前最主要的問題是 `ClusterBean` 內部的變數 `Map<Integer, Collection<HasBeanObject>>`，這個變數中有用到 `HasBeanObject`，這個物件目前是一個 Interface，專案內有許多 `HasBeanObject` 是透過 lambda 而非 class 的方式來建立，由於 lambda 建立的 class 是匿名的，我們很難在別的 JVM 上重建這個匿名的 type。

Pull request description:
[METRICS] Implementation of fixed receiver
Resolve #1710 

這個 PR 實作一個新的 Metric Store receiver `Receiver#fixed(Map<Integer, Collection<BeanObject>)`

這個 Receiver 只會在第一次呼叫 `receive` 時回傳建立時提供的 Beans，然後就不再回傳任何新的 Beans。後面這個會和 #1769 的序列化功能搭配，從檔案系統將 BeanObjects 還原回 `Map<Integer, Collection<BeanObject>`，然後把這個東西當做輸入給 `Receiver` 的輸出使用，而後從 `MetricStore#clusterBean` 取得最後的 `ClusterBean` 物件。

這個大致上是後面 benchmark 的 `ClusterBean` 還原方法。

Patch:
diff --git a/common/src/main/java/org/astraea/common/metrics/collector/MetricStore.java b/common/src/main/java/org/astraea/common/metrics/collector/MetricStore.java
index d4e17d6e..67ac8fc0 100644
--- a/common/src/main/java/org/astraea/common/metrics/collector/MetricStore.java
+++ b/common/src/main/java/org/astraea/common/metrics/collector/MetricStore.java
@@ -82,6 +82,22 @@ public interface MetricStore extends AutoCloseable {
       return LocalSenderReceiver.of();
     }
 
+    static Receiver fixed(Map<Integer, Collection<BeanObject>> beans) {
+      return new Receiver() {
+        private final AtomicBoolean done = new AtomicBoolean(false);
+
+        @Override
+        public Map<Integer, Collection<BeanObject>> receive(Duration timeout) {
+          return done.compareAndSet(false, true) ? beans : Map.of();
+        }
+
+        @Override
+        public void close() {
+          done.set(true);
+        }
+      };
+    }
+
     /**
      * Using an embedded fetcher build the receiver. The fetcher will keep fetching beans
      * background, and it pushes all beans to store internally.


Test patch:
diff --git a/common/src/test/java/org/astraea/common/metrics/collector/FixedMetricsTest.java b/common/src/test/java/org/astraea/common/metrics/collector/FixedMetricsTest.java
new file mode 100644
index 00000000..f9b2bc15
--- /dev/null
+++ b/common/src/test/java/org/astraea/common/metrics/collector/FixedMetricsTest.java
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.astraea.common.metrics.collector;
+
+import java.time.Duration;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.astraea.common.metrics.BeanObject;
+import org.astraea.common.metrics.BeanQuery;
+import org.astraea.common.metrics.HasBeanObject;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class FixedMetricsTest {
+
+  @Test
+  void testFixedReceiverOnce() {
+    var beans =
+        Map.of(1, (Collection<BeanObject>) List.of(new BeanObject("domain", Map.of(), Map.of())));
+    try (var receiver = MetricStore.Receiver.fixed(beans)) {
+      Assertions.assertEquals(beans, receiver.receive(Duration.ZERO));
+      Assertions.assertEquals(Map.of(), receiver.receive(Duration.ZERO));
+      Assertions.assertEquals(Map.of(), receiver.receive(Duration.ZERO));
+      Assertions.assertEquals(Map.of(), receiver.receive(Duration.ZERO));
+    }
+
+    var receiver = MetricStore.Receiver.fixed(beans);
+    receiver.close();
+    Assertions.assertEquals(Map.of(), receiver.receive(Duration.ZERO));
+  }
+
+  @Test
+  void testFixedReceiver() {
+    interface MyBeanObject extends HasBeanObject {}
+
+    var beans =
+        Map.of(
+            1,
+            (Collection<BeanObject>)
+                List.of(new BeanObject("domain", Map.of("Hello", "World"), Map.of())));
+    var sensor =
+        (MetricSensor)
+            (client, cb) ->
+                List.of(
+                    (MyBeanObject)
+                        () ->
+                            client.bean(
+                                BeanQuery.builder()
+                                    .domainName("domain")
+                                    .property("Hello", "World")
+                                    .build()));
+
+    try (var store =
+        MetricStore.builder()
+            .receivers(List.of(MetricStore.Receiver.fixed(beans)))
+            .beanExpiration(Duration.ofDays(3))
+            .sensorsSupplier(() -> Map.of(sensor, (a, x) -> {}))
+            .build()) {
+      store.wait(cb -> cb.all().containsKey(1), Duration.ofSeconds(1));
+      Assertions.assertEquals(Set.of(1), store.clusterBean().all().keySet());
+      Assertions.assertEquals(1, store.clusterBean().all().get(1).size());
+      Assertions.assertInstanceOf(
+          MyBeanObject.class, store.clusterBean().all().get(1).iterator().next());
+    }
+  }
+}