risingwavelabs/risingwave · error · MetaError

Sink error: {0}

Error message

Sink error: {0}

What it means

This error wraps a SinkError (from the sink/connector sink subsystem) into the meta node's unified MetaError. It is raised when meta-node code that manages sinks (creation, drops, streaming job scheduling, sink coordinator) propagates a failure that originated in a sink implementation (e.g. a connector sink writing to an external system). The `#[from]` attribute means any SinkError is automatically converted into MetaError::Sink via `?`.

Source

Thrown at src/meta/src/error.rs:117

    #[error("System parameters error: {0}")]
    SystemParams(String),

    #[error("Session parameters error: {0}")]
    SessionConfig(
        #[from]
        #[backtrace]
        SessionConfigError,
    ),

    #[error(transparent)]
    Connector(
        #[from]
        #[backtrace]
        ConnectorError,
    ),

    #[error("Sink error: {0}")]
    Sink(
        #[from]
        #[backtrace]
        SinkError,
    ),

    #[error(transparent)]
    Internal(
        #[from]
        #[backtrace]
        anyhow::Error,
    ),

    // Indicates that recovery was triggered manually.
    #[error("adhoc recovery triggered")]
    AdhocRecovery,

    #[error("Integrity check failed")]

View on GitHub (pinned to 6469eb736d)

Solutions

1. Inspect the wrapped SinkError (source of this error) to find the underlying connector failure
2. Verify the sink connection options (bootstrap servers, URL, credentials) in the CREATE SINK statement
3. Check network connectivity from meta/compute nodes to the sink target
4. Recreate or restart the sink after fixing the external system
5. Check meta node logs for the full backtrace attached to this variant

Example fix

// before: blindly propagating
let result = sink_manager.write(msg)?;
// after: match on the specific sink failure
match sink_manager.write(msg) {
    Err(MetaError::Sink(e)) => { tracing::error!("sink write failed: {e}"); retry_or_pause(e) }
    r => r,
}

// Rust: check sink options before CREATE SINK
async fn ensure_sink_target_reachable(opts: &SinkOptions) -> Result<(), MetaError> {
    connector::test_sink_connectivity(opts).await
        .map_err(|e| MetaError::Sink(SinkError::from(e)))
}

fn is_sink_error(e: &MetaError) -> Option<&SinkError> {
    match e { MetaError::Sink(s) => Some(s), _ => None }
}

match meta_result {
    Err(MetaError::Sink(sink_err)) => {
        tracing::error!("sink failed: {sink_err:#}");
        // inspect sink_err.source() for the connector cause, then retry/backoff
    }
    other => other?,
}

When it happens

Trigger: Any code path in the meta service that calls sink-related APIs and propagates a SinkError with `?` or From conversion: creating/starting a sink via the sink manager, sink coordinator writing to the external target, or sink schema validation failing during sink creation.

Common situations: External sink target (Kafka, Elasticsearch, Redis, etc.) is down or misconfigured; sink credentials are wrong; target schema changed so the sink can no longer write; network partition between meta/compute nodes and the sink system.

Related errors

• Unknown sink connector: {sink_name}
• Field '{field}' is not allowed to be altered on the fly for
• unsupported sink connector {}
• connector '{}' is not supported
• Only Es sink supports struct, got {:?}: {:?}

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