Merge remote-tracking branch 'origin/main'

Author: zhangShunLin

base-cluster/src/main/java/com/baidu/bifromq/basecluster/AgentHost.java

@@ -41,6 +41,8 @@
 import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
 import com.google.protobuf.ByteString;
+import io.micrometer.core.instrument.Metrics;
+import io.micrometer.core.instrument.binder.jvm.ExecutorServiceMetrics;
 import io.reactivex.rxjava3.core.Observable;
 import io.reactivex.rxjava3.core.Scheduler;
 import io.reactivex.rxjava3.disposables.CompositeDisposable;
@@ -55,24 +57,20 @@
 
 @Slf4j
 final class AgentHost implements IAgentHost {
-    private enum State {
-        INIT, STARTING, STARTED, STOPPING, SHUTDOWN
-    }
-
     private final AtomicReference<State> state = new AtomicReference<>(State.INIT);
     private final AgentHostOptions options;
     private final ICRDTStore store;
     private final ITransport transport;
     private final IMessenger messenger;
     private final IHostMemberList memberList;
+    private final Scheduler hostScheduler;
     private final MemberSelector memberSelector;
     private final AutoHealer healer;
     private final AutoSeeder seeder;
     private final AutoDropper deadDropper;
     private final CompositeDisposable disposables = new CompositeDisposable();
     private final IHostAddressResolver hostAddressResolver;
     private final String[] tags;
-
     AgentHost(ITransport transport, IHostAddressResolver resolver, AgentHostOptions options) {
         checkArgument(!Strings.isNullOrEmpty(options.addr()) && !"0.0.0.0".equals(options.addr()),
             "Invalid bind address");
@@ -84,21 +82,22 @@ private enum State {
             .maxHealthScore(options.awarenessMaxMultiplier())
             .retransmitMultiplier(options.retransmitMultiplier())
             .spreadPeriod(options.gossipPeriod());
-        Scheduler scheduler = Schedulers.from(newSingleThreadScheduledExecutor(
-            EnvProvider.INSTANCE.newThreadFactory("agent-host-scheduler", true)));
+        hostScheduler = Schedulers.from(ExecutorServiceMetrics.monitor(Metrics.globalRegistry,
+            newSingleThreadScheduledExecutor(EnvProvider.INSTANCE.newThreadFactory("agent-host-scheduler", true)),
+            "agent-host-scheduler"));
         this.store = ICRDTStore.newInstance(options.crdtStoreOptions());
         this.messenger = Messenger.builder()
             .transport(transport)
             .opts(messengerOptions)
-            .scheduler(scheduler)
+            .scheduler(hostScheduler)
             .build();
         hostAddressResolver = resolver;
         this.store.start(messenger.receive()
             .filter(m -> m.value().message.hasCrdtStoreMessage())
             .map(m -> m.value().message.getCrdtStoreMessage()));
         tags = new String[] {"local", options.addr() + ":" + messenger.bindAddress().getPort()};
         this.memberList = new HostMemberList(options.addr(), messenger.bindAddress().getPort(),
-            messenger, scheduler, store, hostAddressResolver, tags);
+            messenger, hostScheduler, store, hostAddressResolver, tags);
         IFailureDetector failureDetector = FailureDetector.builder()
             .local(new IProbingTarget() {
                 @Override
@@ -116,15 +115,15 @@ public InetSocketAddress addr() {
             .indirectProbes(options.indirectProbes())
             .worstHealthScore(options.awarenessMaxMultiplier())
             .messenger(messenger)
-            .scheduler(scheduler)
+            .scheduler(hostScheduler)
             .build();
-        healer = new AutoHealer(messenger, scheduler, memberList, hostAddressResolver, options.autoHealingTimeout(),
+        healer = new AutoHealer(messenger, hostScheduler, memberList, hostAddressResolver, options.autoHealingTimeout(),
             options.autoHealingInterval(), tags);
-        seeder = new AutoSeeder(messenger, scheduler, memberList, hostAddressResolver, options.joinTimeout(),
+        seeder = new AutoSeeder(messenger, hostScheduler, memberList, hostAddressResolver, options.joinTimeout(),
             Duration.ofSeconds(options.joinRetryInSec()), tags);
-        deadDropper = new AutoDropper(messenger, scheduler, memberList, failureDetector, hostAddressResolver,
+        deadDropper = new AutoDropper(messenger, hostScheduler, memberList, failureDetector, hostAddressResolver,
             options.suspicionMultiplier(), options.suspicionMaxTimeoutMultiplier(), tags);
-        memberSelector = new MemberSelector(memberList, scheduler, hostAddressResolver);
+        memberSelector = new MemberSelector(memberList, hostScheduler, hostAddressResolver);
         disposables.add(store.storeMessages().subscribe(this::sendCRDTStoreMessage));
 
         start();
@@ -184,6 +183,7 @@ public void close() {
                 .whenComplete((v, e) -> {
                     memberSelector.stop();
                     disposables.dispose();
+                    hostScheduler.shutdown();
                     state.set(State.SHUTDOWN);
                 }).join();
             log.debug("AgentHost stopped");
@@ -198,7 +198,6 @@ private void start() {
         }
     }
 
-
     private void sendCRDTStoreMessage(CRDTStoreMessage storeMsg) {
         ClusterMessage msg = ClusterMessage.newBuilder().setCrdtStoreMessage(storeMsg).build();
         try {
@@ -213,4 +212,9 @@ private void sendCRDTStoreMessage(CRDTStoreMessage storeMsg) {
             log.error("Target Host[{}] not found:\n{}", storeMsg.getReceiver(), storeMsg);
         }
     }
+
+
+    private enum State {
+        INIT, STARTING, STARTED, STOPPING, SHUTDOWN
+    }
 }

base-cluster/src/main/java/com/baidu/bifromq/basecluster/memberlist/HostMemberList.java

@@ -66,16 +66,11 @@
 
 @Slf4j
 public class HostMemberList implements IHostMemberList {
-    private enum State {
-        JOINED, QUITTING, QUITED
-    }
-
     private final AtomicReference<State> state = new AtomicReference<>(State.JOINED);
     private final IMessenger messenger;
     private final Scheduler scheduler;
     private final ICRDTStore store;
     private final IHostAddressResolver addressResolver;
-
     private final BehaviorSubject<Map<HostEndpoint, HostMember>> membershipSubject = BehaviorSubject.createDefault(
         new ConcurrentHashMap<>());
     private final Map<String, Agent> agentMap = new ConcurrentHashMap<>();
@@ -84,7 +79,6 @@ private enum State {
     private final MetricManager metricManager;
     private final String[] tags;
     private volatile HostMember local;
-
     public HostMemberList(String bindAddr,
                           int port,
                           IMessenger messenger,
@@ -180,36 +174,36 @@ private InetSocketAddress getMemberAddress(HostEndpoint endpoint) {
     @Override
     public CompletableFuture<Void> stop() {
         if (state.compareAndSet(State.JOINED, State.QUITTING)) {
-            synchronized (this) {
-                return CompletableFuture.allOf(agentMap.values().stream()
-                        .map(Agent::quit).toArray(CompletableFuture[]::new))
-                    .exceptionally(e -> null)
-                    .thenCompose(v -> {
+            return CompletableFuture.allOf(agentMap.values().stream()
+                    .map(Agent::quit).toArray(CompletableFuture[]::new))
+                .exceptionally(e -> null)
+                .thenCompose(v -> {
+                    synchronized (this) {
                         disposables.dispose();
                         // remove self from alive host list
                         removeMember(local.getEndpoint(), local.getIncarnation());
                         // delete from crdt and wait for
                         return hostListCRDT.execute(
                                 ORMapOperation.remove(local.getEndpoint().toByteString()).of(mvreg))
                             .exceptionally(e -> null);
-                    })
-                    .thenCompose(v1 -> {
-                        ClusterMessage quit = ClusterMessage.newBuilder()
-                            .setQuit(Quit.newBuilder()
-                                .setEndpoint(local.getEndpoint())
-                                .setIncarnation(local.getIncarnation())
-                                .build())
-                            .build();
-                        return messenger.spread(quit)
-                            .handle((v, e) -> null);
-                    })
-                    .thenCompose(v -> store.stopHosting(hostListCRDT.id()))
-                    .whenComplete((v, e) -> {
-                        membershipSubject.onComplete();
-                        metricManager.close();
-                        state.set(State.QUITED);
-                    });
-            }
+                    }
+                })
+                .thenCompose(v1 -> {
+                    ClusterMessage quit = ClusterMessage.newBuilder()
+                        .setQuit(Quit.newBuilder()
+                            .setEndpoint(local.getEndpoint())
+                            .setIncarnation(local.getIncarnation())
+                            .build())
+                        .build();
+                    return messenger.spread(quit)
+                        .handle((v, e) -> null);
+                })
+                .thenCompose(v -> store.stopHosting(hostListCRDT.id()))
+                .whenComplete((v, e) -> {
+                    membershipSubject.onComplete();
+                    metricManager.close();
+                    state.set(State.QUITED);
+                });
         } else if (state.get() == State.QUITTING) {
             return CompletableFuture.failedFuture(new IllegalStateException("quit has started"));
         } else {
@@ -425,6 +419,10 @@ private void checkState() {
         Preconditions.checkState(state.get() == State.JOINED);
     }
 
+    private enum State {
+        JOINED, QUITTING, QUITED
+    }
+
     private class MetricManager {
 
         private final Set<Meter> meters = new HashSet<>();

base-cluster/src/main/java/com/baidu/bifromq/basecluster/memberlist/agent/AgentMember.java

@@ -18,6 +18,7 @@
 import com.baidu.bifromq.basecluster.agent.proto.AgentMemberAddr;
 import com.baidu.bifromq.basecluster.agent.proto.AgentMemberMetadata;
 import com.baidu.bifromq.basecluster.agent.proto.AgentMessage;
+import com.baidu.bifromq.basecluster.agent.proto.AgentMessageEnvelope;
 import com.baidu.bifromq.basecrdt.core.api.IORMap;
 import com.baidu.bifromq.basecrdt.core.api.MVRegOperation;
 import com.baidu.bifromq.basecrdt.core.api.ORMapOperation;
@@ -41,7 +42,6 @@ class AgentMember implements IAgentMember {
     private final AgentMemberAddr localAddr;
     private final IORMap agentCRDT;
     private final IAgentMessenger messenger;
-    private final Scheduler scheduler;
     private final Supplier<Set<AgentMemberAddr>> memberAddresses;
     private final PublishSubject<AgentMessage> agentMessageSubject = PublishSubject.create();
     private final CompositeDisposable disposables = new CompositeDisposable();
@@ -57,7 +57,6 @@ class AgentMember implements IAgentMember {
         this.localAddr = memberAddr;
         this.agentCRDT = agentCRDT;
         this.messenger = messenger;
-        this.scheduler = scheduler;
         this.memberAddresses = memberAddresses;
         metadata = AgentMemberMetadata.newBuilder().setHlc(HLC.INST.get()).build();
         updateCRDT();
@@ -66,7 +65,7 @@ class AgentMember implements IAgentMember {
             .subscribe(this::updateCRDT));
         disposables.add(messenger.receive()
             .filter(msg -> msg.getReceiver().equals(localAddr))
-            .map(msg -> msg.getMessage())
+            .map(AgentMessageEnvelope::getMessage)
             .observeOn(scheduler)
             .subscribe(agentMessageSubject::onNext));
     }
@@ -95,10 +94,9 @@ public AgentMemberAddr address() {
     public CompletableFuture<Void> broadcast(ByteString message, boolean reliable) {
         return throwsWhenDestroyed(() -> {
             AgentMessage agentMessage = AgentMessage.newBuilder().setSender(localAddr).setPayload(message).build();
-            CompletableFuture<Void>[] sendFutures = memberAddresses.get().stream()
+            return CompletableFuture.allOf(memberAddresses.get().stream()
                 .map(memberAddr -> messenger.send(agentMessage, memberAddr, reliable))
-                .toArray(CompletableFuture[]::new);
-            return CompletableFuture.allOf(sendFutures).exceptionally(e -> null);
+                .toArray(CompletableFuture[]::new)).exceptionally(e -> null);
         });
     }
 
@@ -132,7 +130,7 @@ public CompletableFuture<Void> multicast(String targetMemberName, ByteString mes
     private void updateCRDT(long ts) {
         skipRunWhenDestroyed(() -> {
             Optional<AgentMemberMetadata> metaOnCRDT = CRDTUtil.getAgentMemberMetadata(agentCRDT, localAddr);
-            if (!metaOnCRDT.isPresent() || !metaOnCRDT.get().equals(metadata)) {
+            if (metaOnCRDT.isEmpty() || !metaOnCRDT.get().equals(metadata)) {
                 updateCRDT();
             }
         });

base-env/base-env-provider/pom.xml

@@ -12,8 +12,8 @@
   ~ See the License for the specific language governing permissions and limitations under the License.
   -->
 
-<project xmlns="http://maven.apache.org/POM/4.0.0"
-         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xmlns="http://maven.apache.org/POM/4.0.0"
          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
     <parent>
@@ -40,6 +40,10 @@
             <groupId>com.google.guava</groupId>
             <artifactId>guava</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.google.protobuf</groupId>
+            <artifactId>protobuf-java</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-api</artifactId>