{"record":{"id":"2c5c155aa16ff834","repo":"risingwavelabs/risingwave","slug":"must-specify-connector-in-with-clause","errorCode":null,"errorMessage":"Must specify 'connector' in WITH clause","messagePattern":"Must specify 'connector' in WITH clause","errorType":"validation","errorClass":"anyhow::Error (ConnectorResult)","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/mod.rs","lineNumber":482,"sourceCode":"            sink_name: sink_catalog.name,\n            properties: properties_with_secret,\n            columns,\n            downstream_pk: sink_catalog.downstream_pk,\n            sink_type: sink_catalog.sink_type,\n            ignore_delete: sink_catalog.ignore_delete,\n            format_desc: format_desc_with_secret,\n            db_name: sink_catalog.db_name,\n            sink_from_name: sink_catalog.sink_from_name,\n        })\n    }\n}\n\npub fn enforce_secret_sink(props: &impl WithPropertiesExt) -> ConnectorResult<()> {\n    use crate::enforce_secret::EnforceSecret;\n\n    let connector = props\n        .get_connector()\n        .ok_or_else(|| anyhow!(\"Must specify 'connector' in WITH clause\"))?;\n    let key_iter = props.key_iter();\n    match_sink_name_str!(\n        connector.as_str(),\n        PropType,\n        PropType::enforce_secret(key_iter),\n        |other| bail!(\"connector '{}' is not supported\", other)\n    )\n}\n\npub static GLOBAL_SINK_METRICS: LazyLock<SinkMetrics> =\n    LazyLock::new(|| SinkMetrics::new(&GLOBAL_METRICS_REGISTRY));\n\n#[derive(Clone)]\npub struct SinkMetrics {\n    pub sink_commit_duration: LabelGuardedHistogramVec,\n    pub connector_sink_rows_received: LabelGuardedIntCounterVec,\n\n    // Log store writer metrics","sourceCodeStart":464,"sourceCodeEnd":500,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/mod.rs#L464-L500","documentation":"`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.","triggerScenarios":"Calling `enforce_secret_sink` (via `gen_sink_plan`) on a sink whose WITH clause has no `connector` key.","commonSituations":"CREATE SINK without `connector='...'` in WITH; secret-bearing options passed while the connector key was stripped by plan building.","solutions":["Add `connector = '<type>'` to the sink's WITH clause","Validate the WITH clause before submitting the DDL","Check `get_connector`/WithPropertiesExt mapping if the key exists under a different name"],"exampleFix":"// before\nCREATE SINK s FROM mv WITH (kafka.brokers = '...');\n// after\nCREATE SINK s FROM mv WITH (connector = 'kafka', kafka.brokers = '...');","handlingStrategy":"validation","validationCode":"if props.get(\"connector\").is_none() {\n    return Err(anyhow!(\"Must specify 'connector' in WITH clause\"));\n}","typeGuard":null,"tryCatchPattern":"enforce_secret_sink(&props).map_err(|e| {\n    if e.to_string().contains(\"Must specify 'connector'\") {\n        anyhow!(\"sink plan rejected: add connector='<type>' to WITH clause\")\n    } else { e }\n})?;","preventionTips":["Always set connector in sink WITH clauses when secrets are involved","Validate DDL properties client-side before CREATE SINK","Ensure plan serialization preserves the connector key"],"tags":["rust","config","sink","with-clause"],"backgroundTag":"missing-required-config-field","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}