{"record":{"id":"ba233b243fbbd059","repo":"risingwavelabs/risingwave","slug":"unsupported-sink-connector","errorCode":null,"errorMessage":"unsupported sink connector {}","messagePattern":"unsupported sink connector (.+?)","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/mod.rs","lineNumber":272,"sourceCode":"/// same downstream primary key instead of compacting them into one final-state update within a\n/// barrier. Upstream changes under the same stream key may still be compacted earlier.\npub const SINK_USER_PRESERVE_ROW_LEVEL_CHANGES: &str = \"preserve_row_level_changes\";\n\n/// Return whether the configured sink uses exactly-once commit state.\n///\n/// Connector dispatch is centralized here, while each [`Sink`] implementation owns the\n/// interpretation and default of its properties.\npub fn sink_is_exactly_once(properties: &BTreeMap<String, String>) -> Result<bool> {\n    let sink_type = properties\n        .get(CONNECTOR_TYPE_KEY)\n        .ok_or_else(|| SinkError::Config(anyhow!(\"missing config: {}\", CONNECTOR_TYPE_KEY)))?\n        .to_lowercase();\n\n    match_sink_name_str!(\n        sink_type.as_str(),\n        SinkType,\n        SinkType::is_exactly_once(properties),\n        |other| Err(SinkError::Config(anyhow!(\n            \"unsupported sink connector {}\",\n            other\n        )))\n    )\n}\n\npub trait UnknownFields {\n    /// Unrecognized fields in the `WITH` clause.\n    fn unknown_fields(&self) -> HashMap<String, String>;\n}\n\nimpl UnknownFields for () {\n    fn unknown_fields(&self) -> HashMap<String, String> {\n        HashMap::new()\n    }\n}\n\nimpl UnknownFields for HashMap<String, String> {","sourceCodeStart":254,"sourceCodeEnd":290,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/mod.rs#L254-L290","documentation":"`sink_is_exactly_once` dispatches the lowercased connector name through `match_sink_name_str!`; if the name is not a known sink connector, the fallback arm returns this Config error. It means the connector string is recognized syntactically but not a supported sink type.","triggerScenarios":"Calling `sink_is_exactly_once` (or `prepare_replace_sink` on it) with `connector` set to an unknown/typo'd name, e.g. `connector='kafak'` or a source-only connector name.","commonSituations":"Typo in DDL WITH clause; using a connector available for sources but not sinks; connector feature disabled at compile time or in the deployed version.","solutions":["Correct the connector name to a supported sink connector (kafka, iceberg, jdbc, etc.)","Run `SHOW SINKS`/check docs for the list of supported sink connectors in this version","If the connector should be supported, ensure the corresponding feature flag is enabled"],"exampleFix":"// before\nWITH (connector = 'kafak')\n// after\nWITH (connector = 'kafka')","handlingStrategy":"validation","validationCode":"const SUPPORTED_SINKS: &[&str] = &[\"kafka\", \"iceberg\", \"jdbc\", /* ... */];\nlet conn = properties.get(\"connector\").map(|s| s.to_lowercase()).unwrap_or_default();\nif !SUPPORTED_SINKS.contains(&conn.as_str()) {\n    return Err(anyhow!(\"unsupported sink connector {conn}\"));\n}","typeGuard":null,"tryCatchPattern":"match sink_is_exactly_once(&props) {\n    Ok(v) => v,\n    Err(e) if e.to_string().contains(\"unsupported sink connector\") => {\n        bail!(\"check connector name against supported sink list; got: {e}\");\n    }\n    Err(e) => return Err(e.into()),\n}","preventionTips":["Copy connector names from the official docs, not from memory","Distinguish source-only connectors from sink-capable ones","Keep a checked-in list of valid sink connector names for CI validation"],"tags":["rust","config","sink","connector"],"backgroundTag":"unsupported-config-value","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"}