{"record":{"id":"4e824bac6ca57d64","repo":"risingwavelabs/risingwave","slug":"serde-de-serialization-error-for-kinesissinkconfi","errorCode":null,"errorMessage":"serde (de)serialization error for KinesisSinkConfig: {e}","messagePattern":"serde \\(de\\)serialization error for KinesisSinkConfig: (.+?)","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/kinesis.rs","lineNumber":161,"sourceCode":"\n    #[serde(flatten)]\n    pub unknown_fields: std::collections::HashMap<String, String>,\n}\n\ncrate::impl_sink_unknown_fields!(KinesisSinkConfig);\n\nimpl EnforceSecret for KinesisSinkConfig {\n    fn enforce_one(prop: &str) -> crate::error::ConnectorResult<()> {\n        KinesisCommon::enforce_one(prop)?;\n        Ok(())\n    }\n}\n\nimpl KinesisSinkConfig {\n    pub fn from_btreemap(properties: BTreeMap<String, String>) -> Result<Self> {\n        let config =\n            serde_json::from_value::<KinesisSinkConfig>(serde_json::to_value(properties).unwrap())\n                .map_err(|e| SinkError::Config(anyhow!(e)))?;\n        Ok(config)\n    }\n}\n\npub struct KinesisSinkWriter {\n    pub config: KinesisSinkConfig,\n    formatter: SinkFormatterImpl,\n    client: KinesisClient,\n}\n\nstruct KinesisSinkPayloadWriter {\n    client: KinesisClient,\n    entries: Vec<(PutRecordsRequestEntry, usize)>,\n    stream_name: String,\n}\n\nimpl KinesisSinkWriter {\n    pub async fn new(","sourceCodeStart":143,"sourceCodeEnd":179,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/kinesis.rs#L143-L179","documentation":"KinesisSinkConfig::from_btreemap serializes the WITH-option BTreeMap to JSON and deserializes it into KinesisSinkConfig with serde. Any option name/type the struct does not expect (or an unexpected value type after the conversion) causes a serde error that is wrapped as a SinkError::Config.","triggerScenarios":"Calling `KinesisSinkConfig::from_btreemap(properties)` where the map contains an unknown field for KinesisSinkConfig or a value that fails deserialization into the expected typed field.","commonSituations":"Typos in WITH options (e.g. `steam` instead of `stream`); passing options belonging to another connector; providing non-string-shaped values that serde cannot map.","solutions":["Check the option names against KinesisSinkConfig fields and fix typos","Remove options not supported by the Kinesis connector","Enable/inspect serde's error message which names the offending field and fix it in the CREATE SINK statement"],"exampleFix":"// before\nWITH ( connector='kinesis', stream_name='s1' )   // wrong field name\n// after\nWITH ( connector='kinesis', stream='s1' )","handlingStrategy":"validation","validationCode":"// validate WITH options against KinesisSinkConfig fields first\nlet expected = [\"connector\",\"stream\",\"aws.region\",\"endpoint\",\"primary_key\"];\nlet unknown: Vec<_> = props.keys().filter(|k| !expected.contains(&k.as_str())).collect();\nif !unknown.is_empty() { return Err(format!(\"unknown options: {:?}\", unknown)); }","typeGuard":null,"tryCatchPattern":"match KinesisSinkConfig::from_btreemap(props) {\n  Ok(c) => c,\n  Err(e) => { log::error!(\"invalid kinesis sink options: {e}\"); return Err(e); }\n}","preventionTips":["Copy option names from RisingWave's Kinesis sink documentation","Test the CREATE SINK statement in a dev environment first","Read the serde error message — it names the exact offending field"],"tags":["kinesis","serde","config","json"],"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-14T16:17:12.679Z"}