risingwavelabs/risingwave · error · ConnectorError

Unknown sink connector

Error message

Unknown sink connector: {sink_name}

What it means

check_sink_allow_alter_on_fly_fields could not map sink_name to a registered sink configuration type, so no allow_alter_on_fly allowlist exists. The sink connector name is unrecognized for on-the-fly field alteration (note: JDBC sinks are special-cased via CONNECTION_ALLOW_ALTER_ON_FLY_FIELDS).

Solutions

  1. Correct the sink connector name spelling in the WITH clause
  2. Verify supported sink connectors for your RW version
  3. Ensure the sink type is compiled/registered (check build features or enterprise edition)

Example fix

// before
CREATE SINK sk FROM t WITH (connector = 'redis_s', ...);
// after
CREATE SINK sk FROM t WITH (connector = 'redis', ...);
Defensive patterns

Strategy: validation

Validate before calling

const KNOWN_SINKS: &[&str] = &["kafka","iceberg","jdbc","redis","deltalake" /*...*/];
if !KNOWN_SINKS.contains(&sink_name) {
    return Err(format!("unknown sink connector: {sink_name}"));
}

Try / catch

match res {
    Err(e) if e.to_string().starts_with("Unknown sink connector") => {
        // show list of valid sink connectors to the user
    }
    other => other?,
}

Prevention

When it happens

Trigger: ALTER SINK (alter on the fly) where the sink's connector name fails sink_name_to_config_type_name: typo, unsupported sink, or a sink removed in this build.

Common situations: Typos like connector='deltlake' instead of 'deltalake'; using a sink not available in the deployment (e.g. enterprise-only); upgrading RW and connector names changing.

Understand the failure class

Background: Invalid enum value errors: "Unknown type", "Invalid scope", "must be one of" — when a string is not on the library's allowed list — this error's family across 23 libraries.

Related errors

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

Appendix: source

Thrown at src/connector/src/allow_alter_on_fly_fields.rs:530

}

/// Checks if all given fields are allowed to be altered on the fly for the specified sink connector.
/// Returns Ok(()) if all fields are allowed, otherwise returns a `ConnectorError`.
pub fn check_sink_allow_alter_on_fly_fields(
    sink_name: &str,
    fields: &[String],
) -> crate::error::ConnectorResult<()> {
    // TODO(#24846): JDBC sink currently uses `()` as sink config type in `for_all_sinks!`,
    // so it cannot have an isolated key in `SINK_ALLOW_ALTER_ON_FLY_FIELDS`.
    // Reuse the JDBC entry in `CONNECTION_ALLOW_ALTER_ON_FLY_FIELDS` for now.
    // TODO(#24846): remove this special case after JDBC sink has a dedicated config type
    // and allow-alter fields are generated directly into `SINK_ALLOW_ALTER_ON_FLY_FIELDS`.
    let allowed_fields = if sink_name == JdbcSink::SINK_NAME {
        CONNECTION_ALLOW_ALTER_ON_FLY_FIELDS.get(JdbcSink::SINK_NAME)
    } else {
        // Convert sink name to the type name key
        let Some(type_name) = sink_properties::sink_name_to_config_type_name(sink_name) else {
            return Err(ConnectorError::from(anyhow::anyhow!(
                "Unknown sink connector: {sink_name}"
            )));
        };
        SINK_ALLOW_ALTER_ON_FLY_FIELDS.get(type_name)
    };
    let Some(allowed_fields) = allowed_fields else {
        return Err(ConnectorError::from(anyhow::anyhow!(
            "No allow_alter_on_fly fields registered for sink: {sink_name}"
        )));
    };
    for field in fields {
        if !allowed_fields.contains(field) {
            return Err(ConnectorError::from(anyhow::anyhow!(
                "Field '{field}' is not allowed to be altered on the fly for sink: {sink_name}"
            )));
        }
    }
    Ok(())

View on GitHub (pinned to 6469eb736d)