risingwavelabs/risingwave · error · StreamExecutorError

RpcError

Error message

RpcError

What it means

This is the RpcError variant of StreamExecutorError, which transparently wraps an RpcError from risingwave_rpc_client. Stream executors throw it when a gRPC call to another service (meta, compute, or another worker) fails. Since the variant uses #[error(transparent)] and #[from], the underlying RPC error's Display and source are passed through unchanged.

Source

Thrown at src/stream/src/executor/error.rs:78

    ),

    // TODO: remove this after state table is fully used
    #[error("Serialize/deserialize error: {0}")]
    SerdeError(
        #[source]
        #[backtrace]
        BoxedError,
    ),

    #[error("Sink error: sink_id={1}, error: {0}")]
    SinkError(
        #[source]
        #[backtrace]
        SinkError,
        SinkId,
    ),

    #[error(transparent)]
    RpcError(
        #[from]
        #[backtrace]
        RpcError,
    ),

    #[error("Channel closed: {0}")]
    ChannelClosed(String),

    #[error(transparent)]
    ExchangeChannelClosed(
        #[from]
        #[backtrace]
        ExchangeChannelClosed,
    ),

    #[error("Failed to align barrier: expected `{0:?}` but got `{1:?}`")]
    AlignBarrier(Box<Barrier>, Box<Barrier>),

View on GitHub (pinned to 6469eb736d)

Solutions

1. Check connectivity between the compute node and the meta service (address, port, DNS).
2. Inspect the wrapped RpcError source (via its `source()`) for the concrete gRPC status (Unavailable, DeadlineExceeded, etc.).
3. Retry the operation if the failure was transient; the stream framework typically restarts the actor.
4. Verify service versions are compatible (proto mismatch can cause RPC failures).

Example fix

// before
let resp = rpc_client.do_something().await?; // RpcError propagates as executor error
// after
let resp = rpc_client.do_something().await.inspect_err(|e| {
    tracing::warn!(error = ?e, "rpc failed, will be retried by actor restart");
})?;

// before issuing RPC: verify endpoint reachable
async fn ensure_rpc_ready(addr: &str) -> anyhow::Result<()> {
    tokio::net::TcpStream::connect(addr).await
        .map_err(|e| anyhow::anyhow!("rpc endpoint {addr} unreachable: {e}"))?;
    Ok(())
}

fn as_rpc_error(e: &StreamExecutorError) -> Option<&risingwave_rpc_client::error::RpcError> {
    use risingwave_rpc_client::error::RpcError;
    e.0.downcast_ref::<RpcError>()
}

match result {
    Err(e) if matches!(e.variant_name(), "RpcError") => {
        // inspect inner RpcError, apply backoff and retry
    }
    other => other.map(|_| ())?,
}

When it happens

Trigger: Any executor path that performs an RPC via risingwave_rpc_client and uses `?` on a Result whose error is RpcError, e.g. remote sink/table lookups or meta node communication during streaming. The `#[from]` impl converts the RpcError automatically.

Common situations: Meta node down or unreachable; network partitions between compute nodes; RPC deadlines exceeded under load; service restarting during deployment.

Related errors

• RpcError
• bigquery insert error: end of resp stream
• Cannot open client to compute node {addr:?}
• failed to send request to {} {:?}
• serverless backfill controller returned error

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/60eb3a9fecdba1a3. Report an issue: GitHub.