{"record":{"id":"69d2c12d4e34a617","repo":"risingwavelabs/risingwave","slug":"serde-de-serialization-error-for-kafkaconfig-e","errorCode":null,"errorMessage":"serde (de)serialization error for KafkaConfig: {e}","messagePattern":"serde \\(de\\)serialization error for KafkaConfig: (.+?)","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/kafka.rs","lineNumber":274,"sourceCode":"\n    #[serde(flatten)]\n    pub unknown_fields: std::collections::HashMap<String, String>,\n}\n\ncrate::impl_sink_unknown_fields!(KafkaConfig);\n\nimpl EnforceSecret for KafkaConfig {\n    fn enforce_one(prop: &str) -> crate::error::ConnectorResult<()> {\n        KafkaConnectionProps::enforce_one(prop)?;\n        AwsAuthProps::enforce_one(prop)?;\n        Ok(())\n    }\n}\n\nimpl KafkaConfig {\n    pub fn from_btreemap(values: BTreeMap<String, String>) -> Result<Self> {\n        let config = serde_json::from_value::<KafkaConfig>(serde_json::to_value(values).unwrap())\n            .map_err(|e| SinkError::Config(anyhow!(e)))?;\n\n        Ok(config)\n    }\n\n    pub(crate) fn set_client(&self, c: &mut rdkafka::ClientConfig) {\n        self.rdkafka_properties_common.set_client(c);\n        self.rdkafka_properties_producer.set_client(c);\n    }\n}\n\nimpl From<KafkaConfig> for KafkaProperties {\n    fn from(val: KafkaConfig) -> Self {\n        KafkaProperties {\n            bytes_per_second: None,\n            max_num_messages: None,\n            scan_startup_mode: None,\n            time_offset: None,\n            upsert: None,","sourceCodeStart":256,"sourceCodeEnd":292,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/kafka.rs#L256-L292","documentation":"KafkaConfig::from_btreemap serializes the BTreeMap of properties to JSON and deserializes it into the KafkaConfig struct. Any serde failure (unknown fields, type mismatches like non-numeric values for numeric fields) is surfaced as this SinkError::Config error, meaning one or more user-supplied sink properties are invalid for KafkaConfig.","triggerScenarios":"CREATE SINK ... WITH (...) options converted to a BTreeMap then mapped into KafkaConfig via serde_json, where a value has the wrong type (e.g. string where a number/bool is expected) or an unknown field is present under a struct that disallows unknown keys.","commonSituations":"Typo'd property names in the WITH clause; passing quoted numbers like \"'9092'\" where an integer is expected; wrong property grouped under rdkafka_properties vs typed fields; RisingWave version changes to KafkaConfig field types.","solutions":["Print the underlying serde error (it names the offending field) and fix that property in the sink's WITH options","Cast values to the expected types (integers without quotes for numeric fields)","Check the KafkaConfig struct definition for exact field names and types","Drop unknown/unsupported properties from the WITH clause"],"exampleFix":"// before\nCREATE SINK s FROM t WITH (connector='kafka', properties.bootstrap.server='b:9092');\n// after\nCREATE SINK s FROM t WITH (connector='kafka', properties.bootstrap.server='b:9092', type='append only');","handlingStrategy":"validation","validationCode":"fn validate_kafka_props(props: &BTreeMap<String, String>) -> Result<()> {\n    KafkaConfig::from_btreemap(props.clone())?;\n    Ok(())\n} // run before CREATE SINK takes effect","typeGuard":null,"tryCatchPattern":"match KafkaConfig::from_btreemap(props) {\n    Ok(cfg) => cfg,\n    Err(e) => { log::error!(\"invalid kafka properties: {e}\"); return Err(e); }\n}","preventionTips":["Validate WITH properties against KafkaConfig fields before creating the sink","Use unquoted numeric values for ports/retries/timeouts","Keep property names in sync with the KafkaConfig struct for your RisingWave version"],"tags":["kafka","config","serde","validation"],"backgroundTag":"invalid-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"}