private <T> Mono<Map<String, T>> requestResponsesAsMap(
            List<Member> members,
            @NotNull RpcRequest<T> request,
            @NotNull Duration timeout,
            boolean rejectIfMissingAnyConnection) {
        try {
            assertCurrentNodeIsAllowedToSend(request);
        } catch (Exception e) {
            return Mono.error(e);
        }
        if (members.isEmpty()) {
            return Mono.error(RpcException.get(RpcErrorCode.MEMBER_NOT_FOUND,
                    ResponseStatusCode.SERVER_UNAVAILABLE));
        }
        if (rejectIfMissingAnyConnection
                && !discoveryService.getLocalNodeStatusManager()
                        .getLocalMember()
                        .getStatus()
                        .isActive()) {
            return Mono.error(RpcException.get(RpcErrorCode.CONNECTION_NOT_FOUND,
                    ResponseStatusCode.SERVER_UNAVAILABLE,
                    "Some connections have not been established"));
        }
        return Mono.deferContextual(context -> {
            addTraceIdToRequestFromContext(context, request);
            ByteBuf requestBody = codecService.serializeWithoutCodecId(request);
            int size = members.size();
            List<Mono<Pair<String, T>>> results = new ArrayList<>(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<Pair<String, 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()
                        .map(payload -> Pair.of(memberId, payload));
                results.add(responseMono);
            }
            Flux<Pair<String, T>> resultFlux = Flux.merge(results);
            Tag tag = request.tag();
            if (tag != null) {
                resultFlux = resultFlux.tag(tag.getKey(), tag.getValue());
            }
            return resultFlux.timeout(timeout, HashedWheelScheduler.getDaemon())
                    .collectMap(Pair::first, Pair::second, CollectorUtil.toMap(size))
                    .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