public <T> Mono<T> runRpcRequest(
            TracingContext tracingContext,
            RpcRequest<T> rpcRequest,
            @Nullable TurmsConnection connection,
            String fromNodeId) {
        rpcRequest.touch(rpcRequest);
        try {
            tracingContext.updateThreadContext();
            rpcRequest.init(context, connection, fromNodeId);
            Mono<T> result;
            // It is the responsibility of the implementation of call() or callAsync()
            // to release the request once if the request has a bound buffer
            if (rpcRequest.isAsync()) {
                result = rpcRequest.callAsync();
            } else {
                T data = rpcRequest.call();
                result = data == null
                        ? Mono.empty()
                        : Mono.just(data);
            }
            // TODO: slow log
            return result.onErrorMap(e -> e instanceof RpcException
                    ? e
                    : RpcException.get(RpcErrorCode.FAILED_TO_RUN_RPC,
                            ResponseStatusCode.SERVER_INTERNAL_ERROR,
                            e.toString(),
                            e))
                    .doFinally(signalType -> tracingContext.clearThreadContext());
        } catch (RpcException e) {
            rpcRequest.release();
            return Mono.error(e);
        } catch (ResponseException e) {
            rpcRequest.release();
            return Mono.error(
                    RpcException.get(RpcErrorCode.FAILED_TO_RUN_RPC, e.getCode(), e.getReason()));
        } catch (Exception e) {
            rpcRequest.release();
            return Mono.error(RpcException.get(RpcErrorCode.FAILED_TO_RUN_RPC,
                    ResponseStatusCode.SERVER_INTERNAL_ERROR,
                    e.toString(),
                    e));
        } finally {
            tracingContext.clearThreadContext();
        }
    }
@return {@link reactor.core.publisher.MonoEmpty} if the result of RPC request is null
@implNote 1. We record request time/response here because the RPC request may run on the
          local machine 2.The method itself will call {@link RpcRequest#release()}
turms-server-common/src/main/java/im/turms/server/common/infra/cluster/service/rpc/RpcRequestExecutor.java