public void onRequestReceived(RpcRequest<?> request) {
                // Retain to avoid being released by FluxReceive
                request.retain();
                ChannelOperations<?, ?> conn = connection.getConnection();
                TracingContext ctx = request.getTracingContext();
                String nodeId = connection.getNodeId();
                requestExecutor.runRpcRequest(ctx, request, connection, nodeId)
                        .cast(Object.class)
                        .onErrorResume(RpcException.class, Mono::just)
                        .defaultIfEmpty(Null.INSTANCE)
                        .doOnNext(response -> {
                            if (conn.isDisposed()) {
                                try (TracingCloseableContext ignored = ctx.asCloseable()) {
                                    LOGGER.error(
                                            "Could not send the response ({}) to the disposed connection of the node: {}",
                                            response,
                                            nodeId);
                                }
                                return;
                            }
                            ByteBuf buf;
                            try {
                                buf = RpcFrameEncoder.INSTANCE.encode(request.getRequestId(),
                                        response);
                                buf.touch(response);
                            } catch (Exception e) {
                                try (TracingCloseableContext ignored = ctx.asCloseable()) {
                                    LOGGER.error("Failed to encode the response: {}", response, e);
                                }
                                if (response instanceof RpcException) {
                                    return;
                                }
                                // Try to fall back to the RpcException response
                                try {
                                    response = RpcException.get(RpcErrorCode.CODEC_FAILED_TO_ENCODE,
                                            ResponseStatusCode.SERVER_INTERNAL_ERROR,
                                            "Failed to encode the response: "
                                                    + response);
                                    buf = RpcFrameEncoder.INSTANCE.encode(request.getRequestId(),
                                            response);
                                } catch (Exception exception) {
                                    try (TracingCloseableContext ignored = ctx.asCloseable()) {
                                        LOGGER.error(
                                                "Failed to fall back to the RpcException since failing to encode the response: {}",
                                                response,
                                                exception);
                                    }
                                    return;
                                }
                            }
                            if (buf.refCnt() == 0) {
                                try (TracingCloseableContext ignored = ctx.asCloseable()) {
                                    LOGGER.error(
                                            "The buffer of response ({}) is released unexpectedly",
                                            response);
                                }
                                return;
                            }
                            // Duplicate the buffer to use an independent reader index
                            // because we don't want to modify the reader index of the original
                            // buffer
                            // if it is an unreleasable buffer internally, or it may be sent to
                            // multiple endpoints.
                            // Note that the content of the buffer is not copied, so "duplicate()"
                            // is efficient.
                            ByteBuf buffer = buf.duplicate();
                            conn.sendObject(buffer)
                                    .then()
                                    .subscribe(null, t -> {
                                        try (TracingCloseableContext ignored = ctx.asCloseable()) {
                                            LOGGER.error("Failed to send the response buffer: "
                                                    + buffer, t);
                                        }
                                    });
                        })
                        .contextWrite(context -> context.put(TracingContext.CTX_KEY_NAME, ctx))
                        .subscribe();
            }
@implNote 1. We don't need to close the endpoint after removing them because they
          should be closed when "onConnectionClosed" invoked.
          <p>
          2. We only need to try to get the node ID from the connection because if
          its member exists, the connection should have bound with the node ID too.
turms-server-common/src/main/java/im/turms/server/common/infra/cluster/service/rpc/RpcService.java