{"record":{"id":"155c84cad870f6cd","repo":"risingwavelabs/risingwave","slug":"serde-json-deserialization-error-for-pulsarconfig","errorCode":null,"errorMessage":"{serde_json deserialization error for PulsarConfig from properties}","messagePattern":"\\{serde_json deserialization error for PulsarConfig from properties\\}","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/pulsar.rs","lineNumber":202,"sourceCode":"    pub producer_properties: PulsarPropertiesProducer,\n\n    #[serde(flatten)]\n    pub unknown_fields: std::collections::HashMap<String, String>,\n}\n\ncrate::impl_sink_unknown_fields!(PulsarConfig);\n\nimpl EnforceSecret for PulsarConfig {\n    fn enforce_one(prop: &str) -> crate::error::ConnectorResult<()> {\n        PulsarCommon::enforce_one(prop)?;\n        AwsAuthProps::enforce_one(prop)?;\n        Ok(())\n    }\n}\nimpl PulsarConfig {\n    pub fn from_btreemap(values: BTreeMap<String, String>) -> Result<Self> {\n        let config = serde_json::from_value::<PulsarConfig>(serde_json::to_value(values).unwrap())\n            .map_err(|e| SinkError::Config(anyhow!(e)))?;\n\n        Ok(config)\n    }\n}\n\n#[derive(Debug)]\npub struct PulsarSink {\n    pub config: PulsarConfig,\n    schema: Schema,\n    downstream_pk: Vec<usize>,\n    format_desc: SinkFormatDesc,\n    db_name: String,\n    sink_from_name: String,\n}\n\nimpl EnforceSecret for PulsarSink {\n    fn enforce_secret<'a>(\n        prop_iter: impl Iterator<Item = &'a str>,","sourceCodeStart":184,"sourceCodeEnd":220,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/pulsar.rs#L184-L220","documentation":"This error wraps a serde_json deserialization failure that occurred while converting the sink's WITH-clause properties (a BTreeMap<String, String>) into a PulsarConfig via from_btreemap. The properties map is serialized to a JSON value and then deserialized against the PulsarConfig struct, so any property that is missing a required field, has the wrong type, or carries an unknown/invalid value surfaces here as a SinkError::Config. The library throws it because the sink cannot be constructed without a fully valid configuration.","triggerScenarios":"Calling PulsarConfig::from_btreemap (or creating a Pulsar sink) with a properties map where a required PulsarConfig field (e.g. service.url, topic) is absent, or a field has a value that fails JSON type/enum parsing (e.g. numeric-seed=false given a non-boolean-like string, unparseable integer fields).","commonSituations":"Typos in sink WITH options when running CREATE SINK; passing extra unrelated properties; quoting/escaping mistakes in the SQL DDL so values do not deserialize; upgrading RisingWave and having renamed config fields.","solutions":["Check the SQL CREATE SINK WITH options and ensure every required PulsarConfig field is present and correctly spelled (e.g. service.url, topic).","Verify each option's value type matches PulsarConfig (strings, integers, booleans) — the serde error message names the offending field.","Remove unrelated/unknown WITH options that are not part of PulsarConfig or disable deny_unknown_fields expectations by removing stale keys.","Test the properties map locally with serde_json::from_value::<PulsarConfig>(serde_json::to_value(map).unwrap()) to see the exact serde error before wiring the sink.","Compare against the documented example in the RisingWave docs for the Pulsar sink."],"exampleFix":"// before\nCREATE SINK s FROM mv WITH (\n  connector = 'pulsar',\n  service_url = 'pulsar://localhost:6650',\n  topic = 't'\n);\n// after\nCREATE SINK s FROM mv WITH (\n  connector = 'pulsar',\n  service.url = 'pulsar://localhost:6650',\n  topic = 't'\n);","handlingStrategy":"validation","validationCode":"// Validate sink props before from_btreemap\nfn validate_pulsar_props(props: &BTreeMap<String, String>) -> Result<(), String> {\n    let required = [\"service.url\", \"topic\"];\n    for k in required {\n        if !props.contains_key(k) {\n            return Err(format!(\"missing required pulsar sink option: {k}\"));\n        }\n    }\n    Ok(())\n}","typeGuard":"fn has_required_keys(props: &BTreeMap<String, String>, keys: &[&str]) -> bool {\n    keys.iter().all(|k| props.contains_key(*k))\n}","tryCatchPattern":"match PulsarConfig::from_btreemap(props) {\n    Ok(cfg) => /* proceed */,\n    Err(SinkError::Config(e)) => log::error!(\"invalid pulsar sink properties: {e:#}\"),\n    Err(e) => return Err(e),\n}","preventionTips":["Always include service.url and topic in WITH options","Keep option names in sync with PulsarConfig fields","Validate the DDL against docs before running","Test sink creation on a dev cluster first"],"tags":["rust","serde","pulsar","sink","config"],"backgroundTag":"schema-validation-failed","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}