risingwavelabs/risingwave · error · ErrorCode::ConnectorError

connector error: {0}

Error message

connector error: {0}

What it means

`ErrorCode::ConnectorError` wraps a type-erased `BoxedError` originating from a source connector (Kafka, Kinesis, Pulsar, S3, etc.). RisingWave surfaces connector-internal failures under this uniform prefix while preserving the underlying error as `#[source]`.

Source

Thrown at src/frontend/src/error.rs:52

/// entire RisingWave project.
// TODO(error-handling): this is migrated from the `common` crate, and there could
// be some further refactoring to do:
// - Some variants are never constructed.
// - Some variants store a type-erased `BoxedError` to resolve the reverse dependency.
//   It's not necessary anymore as the error type is now defined at the top-level.
#[derive(Error, thiserror_ext::ReportDebug, thiserror_ext::Box, thiserror_ext::Macro)]
#[thiserror_ext(newtype(name = RwError, backtrace), macro(path = "crate::error"))]
pub enum ErrorCode {
    #[error("internal error: {0}")]
    InternalError(String),
    // TODO: unify with the above
    #[error(transparent)]
    Uncategorized(
        #[from]
        #[backtrace]
        anyhow::Error,
    ),
    #[error("connector error: {0}")]
    ConnectorError(
        #[source]
        #[backtrace]
        BoxedError,
    ),
    #[error(transparent)]
    NotImplemented(#[from] NotImplemented),
    // Tips: Use this only if it's intended to reject the query
    #[error("Not supported: {0}\nHINT: {1}")]
    NotSupported(String, String),
    #[error(transparent)]
    NoFunction(#[from] NoFunction),
    #[error(transparent)]
    IoError(#[from] std::io::Error),
    #[error("Storage error: {0}")]
    StorageError(
        #[backtrace]
        #[source]

View on GitHub (pinned to 6469eb736d)

Solutions

1. Inspect the chained `#[source]` error for the connector-specific cause
2. Verify connector options in the WITH clause (hosts, ports, credentials, topic/region)
3. Test connectivity from the RisingWave host to the external service (kafka CLI, aws cli)
4. Check authentication credentials and IAM policies
5. Retry creation after fixing; for runtime failures check the connector's own dashboard/logs

Example fix

// before
CREATE SOURCE s (...) WITH (connector='kafka', properties.bootstrap.server='localhost:9092');
// after (reachable broker + auth)
CREATE SOURCE s (...) WITH (connector='kafka', properties.bootstrap.server='broker:9092', properties.sasl.mechanism='SCRAM-SHA-256', properties.security.protocol='SASL_SSL');

// pre-check connector reachability before CREATE SOURCE
nc -zv broker 9092
# kafka: kcat -b broker:9092 -L

fn is_connector_error(e: &RwError) -> bool { matches!(e.get_code(), ErrorCode::ConnectorError(_)) }

match err.get_code() { ErrorCode::ConnectorError(inner) => inspect_source_chain(inner), _ => propagate }

When it happens

Trigger: Creating or running a `CREATE SOURCE`/`CREATE TABLE ... WITH (connector = ...)` or a sink: connector initialization failures (bad broker address, auth failure), or runtime read/write failures inside the source executor.

Common situations: Misconfigured Kafka bootstrap servers or SASL credentials, missing IAM permissions for Kinesis/S3, network egress blocked from the RisingWave instance, connector schema/topic mismatch.

Related errors

• Nats connection status is not connected, current status is {
• Sink error: sink_id={1}, error: {0}
• additional column is not supported for connector {}, accepta
• additional column type {} is not supported for connector {},
• unsupported encoding for Upsert

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