private <T> Mono<T> requestResponse0(
            RpcEndpoint endpoint,
            RpcRequest<T> request,
            @Nullable Duration timeout) {
        try {
            assertCurrentNodeIsAllowedToSend(request);
        } catch (Exception e) {
            request.release();
            return Mono.error(e);
        }
        if (timeout == null) {
            timeout = defaultRequestTimeoutDuration;
        }
        Mono<T> mono = Mono.deferContextual(context -> {
            addTraceIdToRequestFromContext(context, request);
            ByteBuf requestBody;
            try {
                requestBody = codecService.serializeWithoutCodecId(request);
            } catch (Exception e) {
                request.release();
                return Mono.error(new SerializationException(
                        "Failed to encode the request: "
                                + request,
                        e));
            }
            return endpoint.sendRequest(request, requestBody);
        })
                .timeout(timeout, HashedWheelScheduler.getDaemon())
                .name(METRICS_NAME_RPC_REQUEST)
                .tag(METRICS_TAG_REQUEST_NAME, request.name())
                .tag(METRICS_TAG_REQUEST_TARGET_NODE_ID, endpoint.getNodeId());
        Tag tag = request.tag();
        if (tag != null) {
            mono = mono.tag(tag.getKey(), tag.getValue());
        }
        return mono.metrics()
                .onErrorMap(t -> mapThrowable(t, request));
    }
@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