private void listenMembersChangeEvent() {
        // Because the information of members changes frequently due to lastHeartbeatDate,
        // use DEFAULT instead of UPDATE_LOOKUP to reduce the overhead of data
        sharedConfigService.subscribe(Member.class, FullDocument.DEFAULT)
                .doOnNext(event -> {
                    Member changedMember = event.getFullDocument();
                    String clusterId = changedMember == null
                            ? ChangeStreamUtil.getStringFromId(event.getDocumentKey(),
                                    Member.Key.Fields.clusterId)
                            : changedMember.getClusterId();
                    String nodeId = ChangeStreamUtil.getStringFromId(event.getDocumentKey(),
                            Member.Key.Fields.nodeId);
                    if (!clusterId.equals(localNodeStatusManager.getLocalMember()
                            .getClusterId())) {
                        return;
                    }
                    switch (event.getOperationType()) {
                        case INSERT, REPLACE -> onMemberAddedOrReplaced(changedMember);
                        case UPDATE -> onMemberUpdated(nodeId, event.getUpdateDescription());
                        case DELETE -> {
                            Member deletedMember = allKnownMembers.remove(nodeId);
                            if (deletedMember == null) {
                                return;
                            }
                            LOGGER.info("A member has been deleted: "
                                    + deletedMember);
                            updateOtherActiveConnectedMemberList(false, deletedMember);
                            // Note that we assume that there is no the case:
                            // a node is running but has just been unregistered in the registry
                            // because the node may lose the connection with the registry and TTL
                            // has passed.
                            // During the time, another node with the SAME node ID registers itself.
                            // If the lost node recovers again, there is a potential bug.
                            if (nodeId.equals(localNodeStatusManager.getLocalMember()
                                    .getNodeId())) {
                                localNodeStatusManager.setLocalNodeRegistered(false);
                                if (!localNodeStatusManager.isClosing()) {
                                    // Ignore the error because the node may have been registered by
                                    // its heartbeat timer
                                    localNodeStatusManager.registerLocalNodeAsMember(true)
                                            .subscribe();
                                }
                            }
                        }
                        case INVALIDATE -> {
                            Iterator<Map.Entry<String, Member>> iterator =
                                    allKnownMembers.entrySet()
                                            .iterator();
                            while (iterator.hasNext()) {
                                Member deletedMember = iterator.next()
                                        .getValue();
                                LOGGER.info("A member has been deleted: "
                                        + deletedMember);
                                updateOtherActiveConnectedMemberList(false, deletedMember);
                                iterator.remove();
                            }
                        }
                        default -> LOGGER.fatal("Detected an illegal operation on the collection \""
                                + StringUtil
                                        .upperCamelToLowerCamelLatin1(Member.class.getSimpleName())
                                + "\" in the change stream event: {}", event);
                    }
                    updateActiveMembers(allKnownMembers.values());
                    connectionService.updateHasConnectedToAllMembers(allKnownMembers.keySet());
                })
                .onErrorContinue((throwable, o) -> LOGGER.error(
                        "Caught an error while processing the change stream event of Member: {}",
                        o,
                        throwable))
                .subscribe();
    }
@return a positive or negative number, and never 0 to ensure that the order is consistent in
        every node.
turms-server-common/src/main/java/im/turms/server/common/infra/cluster/service/discovery/DiscoveryService.java