public <T> Mono<T> requestResponse(RpcRequest<T> request) {
        // TODO: Provide richer load balancing strategies
        List<Member> otherMembers = getOtherActiveConnectedMembersToRespond(request);
        int size = otherMembers.size();
        if (size == 0) {
            request.release();
            return Mono.error(RpcException.get(RpcErrorCode.MEMBER_NOT_FOUND,
                    ResponseStatusCode.SERVER_UNAVAILABLE));
        }
        // use System.currentTimeMillis() instead of "RandomUtil.nextPositiveInt()" for better
        // performance
        int index = (int) (System.currentTimeMillis() % size);
        Member member = otherMembers.get(index);
        // fast path
        if (!member.getStatus()
                .isHealthy()) {
            // slow path
            int retry = 0;
            boolean isUnhealthy;
            do {
                retry++;
                index = (index + retry) % size;
                member = otherMembers.get(index);
                isUnhealthy = !member.getStatus()
                        .isHealthy();
            } while (isUnhealthy && retry < size);
            if (isUnhealthy) {
                request.release();
                return Mono.error(RpcException.get(RpcErrorCode.HEALTHY_MEMBER_NOT_FOUND,
                        ResponseStatusCode.SERVER_UNAVAILABLE));
            }
        }
        String memberNodeId = member.getNodeId();
        RpcEndpoint client;
        try {
            client = getOrCreateEndpoint(memberNodeId);
        } catch (Exception e) {
            request.release();
            return Mono.error(e);
        }
        // Retain to invoke requestResponse() again if an error occurs
        request.retain();
        return requestResponse(client, request, defaultRequestTimeoutDuration)
                .onErrorResume(throwable -> {
                    if (ThrowableUtil.isDisconnectedClientError(throwable)) {
                        for (Member newMember : getOtherActiveConnectedMembersToRespond(request)) {
                            String newMemberId = newMember.getNodeId();
                            if (!newMemberId.equals(memberNodeId)
                                    && newMember.getStatus()
                                            .isHealthy()) {
                                return requestResponse(newMemberId,
                                        request,
                                        defaultRequestTimeoutDuration);
                            }
                        }
                    }
                    // No need to translate the error because it should have been translated
                    return Mono.error(throwable);
                })
                .doFinally(signal -> request.release());
    }
@return 1. an empty publisher if the peer responds with a null value; 2. a non-empty
        publisher if the peer responds with a non-null value; 3. error for other cases (e.g.
        no peer exists).
turms-server-common/src/main/java/im/turms/server/common/infra/cluster/service/rpc/RpcService.java