{"record":{"id":"95a5f73cebefdeea","repo":"risingwavelabs/risingwave","slug":"must-specify-connector-in-with-clause-95a5f7","errorCode":null,"errorMessage":"Must specify 'connector' in WITH clause","messagePattern":"Must specify 'connector' in WITH clause","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/connector/src/source/base.rs","lineNumber":672,"sourceCode":"}\n\nimpl ConnectorProperties {\n    /// Creates typed source properties from the raw `WITH` properties.\n    ///\n    /// It checks the `connector` field, and them dispatches to the corresponding type's `try_from_btreemap` method.\n    ///\n    /// `deny_unknown_fields`: Since `WITH` options are persisted in meta, we do not deny unknown fields when restoring from\n    /// existing data to avoid breaking backwards compatibility. We only deny unknown fields when creating new sources.\n    pub fn extract(\n        with_properties: WithOptionsSecResolved,\n        deny_unknown_fields: bool,\n    ) -> Result<Self> {\n        let (options, secret_refs) = with_properties.into_parts();\n        let mut options_with_secret =\n            LocalSecretManager::global().fill_secrets(options, secret_refs)?;\n        let connector = options_with_secret\n            .remove(UPSTREAM_SOURCE_KEY)\n            .ok_or_else(|| anyhow!(\"Must specify 'connector' in WITH clause\"))?\n            .to_lowercase();\n        match_source_name_str!(\n            connector.as_str(),\n            PropType,\n            PropType::try_from_btreemap(options_with_secret, deny_unknown_fields)\n                .map(ConnectorProperties::from),\n            |other| bail!(\"connector '{}' is not supported\", other)\n        )\n    }\n\n    pub fn enforce_secret_source(\n        with_properties: &impl WithPropertiesExt,\n    ) -> crate::error::ConnectorResult<()> {\n        let connector = with_properties\n            .get_connector()\n            .ok_or_else(|| anyhow!(\"Must specify 'connector' in WITH clause\"))?\n            .to_lowercase();\n        let key_iter = with_properties.key_iter();","sourceCodeStart":654,"sourceCodeEnd":690,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/source/base.rs#L654-L690","documentation":"ConnectorProperties::extract requires the WITH clause of a CREATE SOURCE / SINK to contain the `connector` property (UPSTREAM_SOURCE_KEY), which names the connector type. The key is absent from the provided properties map, so the type cannot be determined and extraction fails.","triggerScenarios":"Calling ConnectorProperties::extract on a with-properties BTreeMap lacking the `connector` key; SQL like CREATE SOURCE ... WITH (topic='x') omitting connector='kafka'.","commonSituations":"Hand-building properties programmatically and forgetting the connector key; SQL that specifies only topic/broker options; renaming/refactor dropping the key.","solutions":["Add connector='<name>' to the WITH clause of the CREATE SOURCE/SINK statement","If constructing properties in code, insert UPSTREAM_SOURCE_KEY (\"connector\") into the options map before calling extract","Check the option name spelling (it must literally be `connector`)"],"exampleFix":"// before\nCREATE SOURCE s (...) WITH (topic='events');\n// after\nCREATE SOURCE s (...) WITH (connector='kafka', topic='events');","handlingStrategy":"validation","validationCode":"assert!(with_options.contains_key(\"connector\"), \"WITH clause must include connector='<name>'\");","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Always include connector='...' in CREATE SOURCE/SINK WITH clauses","When building property maps in code, insert the \"connector\" key first","Lint SQL templates for the required option"],"tags":["rust","connector","sql","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"}