risingwavelabs/risingwave · error · anyhow::Error (ConnectorResult)

Must specify 'connector' in WITH clause

Error message

Must specify 'connector' in WITH clause

What it means

`enforce_secret_sink` resolves the connector name from WITH properties so it can enforce secret handling for sensitive options; `props.get_connector()` returning None means the DDL omitted `connector`, so enforcement cannot proceed and it fails with this message.

Source

Thrown at src/connector/src/sink/mod.rs:482

            sink_name: sink_catalog.name,
            properties: properties_with_secret,
            columns,
            downstream_pk: sink_catalog.downstream_pk,
            sink_type: sink_catalog.sink_type,
            ignore_delete: sink_catalog.ignore_delete,
            format_desc: format_desc_with_secret,
            db_name: sink_catalog.db_name,
            sink_from_name: sink_catalog.sink_from_name,
        })
    }
}

pub fn enforce_secret_sink(props: &impl WithPropertiesExt) -> ConnectorResult<()> {
    use crate::enforce_secret::EnforceSecret;

    let connector = props
        .get_connector()
        .ok_or_else(|| anyhow!("Must specify 'connector' in WITH clause"))?;
    let key_iter = props.key_iter();
    match_sink_name_str!(
        connector.as_str(),
        PropType,
        PropType::enforce_secret(key_iter),
        |other| bail!("connector '{}' is not supported", other)
    )
}

pub static GLOBAL_SINK_METRICS: LazyLock<SinkMetrics> =
    LazyLock::new(|| SinkMetrics::new(&GLOBAL_METRICS_REGISTRY));

#[derive(Clone)]
pub struct SinkMetrics {
    pub sink_commit_duration: LabelGuardedHistogramVec,
    pub connector_sink_rows_received: LabelGuardedIntCounterVec,

    // Log store writer metrics

View on GitHub (pinned to 6469eb736d)

Solutions

1. Add `connector = '<type>'` to the sink's WITH clause
2. Validate the WITH clause before submitting the DDL
3. Check `get_connector`/WithPropertiesExt mapping if the key exists under a different name

Example fix

// before
CREATE SINK s FROM mv WITH (kafka.brokers = '...');
// after
CREATE SINK s FROM mv WITH (connector = 'kafka', kafka.brokers = '...');

if props.get("connector").is_none() {
    return Err(anyhow!("Must specify 'connector' in WITH clause"));
}

enforce_secret_sink(&props).map_err(|e| {
    if e.to_string().contains("Must specify 'connector'") {
        anyhow!("sink plan rejected: add connector='<type>' to WITH clause")
    } else { e }
})?;

When it happens

Trigger: Calling `enforce_secret_sink` (via `gen_sink_plan`) on a sink whose WITH clause has no `connector` key.

Common situations: CREATE SINK without `connector='...'` in WITH; secret-bearing options passed while the connector key was stripped by plan building.

Understand the failure class

Background: "is required", "must be set", "missing required field": configuration validation errors across open-source libraries — this error's family across 36 libraries.

Related errors

• `{}` must be {}, or {}
• Primary key not defined for upsert doris sink (please define
• Can't get fe host from url
• please set only one of the 'index_column' or 'index' propert
• please ensure the data type of {} is varchar.

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