public <T> Flux<T> requestResponsesFromOtherMembers(
            List<Member> members,
            @NotNull RpcRequest<T> request,
            @NotNull Duration timeout,
            boolean rejectIfMissingAnyConnection) {
        try {
            assertCurrentNodeIsAllowedToSend(request);
        } catch (Exception e) {
            return Flux.error(e);
        }
        if (members.isEmpty()) {
            return Flux.error(RpcException.get(RpcErrorCode.MEMBER_NOT_FOUND,
                    ResponseStatusCode.SERVER_UNAVAILABLE));
        }
        if (rejectIfMissingAnyConnection && !connectionService.isHasConnectedToAllMembers()) {
            return Flux.error(RpcException.get(RpcErrorCode.CONNECTION_NOT_FOUND,
                    ResponseStatusCode.SERVER_UNAVAILABLE,
                    "Not all connections are established"));
        }
        return Flux.deferContextual(context -> {
            addTraceIdToRequestFromContext(context, request);
            ByteBuf requestBody = codecService.serializeWithoutCodecId(request);
            List<Mono<T>> results = new ArrayList<>(members.size());
            for (Member member : members) {
                String memberId = member.getNodeId();
                RpcEndpoint client;
                try {
                    client = getOrCreateEndpoint(memberId);
                } catch (Exception e) {
                    results.add(Mono.error(e));
                    continue;
                }
                requestBody.retain();
                Mono<T> responseMono = client.sendRequest(request, requestBody)
                        .name(METRICS_NAME_RPC_REQUEST)
                        .tag(METRICS_TAG_REQUEST_NAME, request.name())
                        .tag(METRICS_TAG_REQUEST_TARGET_NODE_ID, memberId)
                        .metrics();
                results.add(responseMono);
            }
            Flux<T> responseFlux = Flux.merge(results);
            Tag tag = request.tag();
            if (tag != null) {
                responseFlux = responseFlux.tag(tag.getKey(), tag.getValue());
            }
            return responseFlux.timeout(timeout, HashedWheelScheduler.getDaemon())
                    .onErrorMap(t -> mapThrowable(t, request))
                    .doFinally(signal -> requestBody.release());
        });
    }
@return 1. a non-empty publisher that publishes an empty map if all peers respond with an
        empty payload; 2. a non-empty publisher that publishes a non-empty map if any peer
        responds with an non-empty valid payload; 3. error for other cases (e.g. no peer
        exists). The key is the member node ID, and the value is the response
turms-server-common/src/main/java/im/turms/server/common/infra/cluster/service/rpc/RpcService.java