public <T> Mono<T> requestResponse(
            String memberNodeId,
            RpcRequest<T> request,
            Duration timeout,
            @Nullable TurmsConnection connection) {
        try {
            if (discoveryService.getLocalNodeStatusManager()
                    .isLocalNodeId(memberNodeId)) {
                return requestExecutor.runRpcRequest(new TracingContext(
                        request.getTracingContext()), request, null, memberNodeId);
            }
            RpcEndpoint endpoint = getOrCreateEndpoint(memberNodeId, connection);
            return requestResponse0(endpoint, request, timeout);
        } catch (Exception e) {
            request.release();
            return Mono.error(e);
        }
    }
@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