{"record":{"id":"c16cc92e6ba61782","repo":"risingwavelabs/risingwave","slug":"sinkerror-config-anyhow-e-c16cc9","errorCode":null,"errorMessage":"SinkError::Config(anyhow!(e))","messagePattern":"SinkError::Config\\(anyhow!\\(e\\)\\)","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/nats.rs","lineNumber":102,"sourceCode":"pub struct NatsSinkWriter {\n    pub config: NatsConfig,\n    context: Context,\n    /// Hold the client Arc to keep it alive. This allows the shared client cache to reuse\n    /// the connection while we're still using it.\n    #[expect(dead_code)]\n    client: Arc<async_nats::Client>,\n    #[expect(dead_code)]\n    schema: Schema,\n    json_encoder: JsonEncoder,\n}\n\npub type NatsSinkDeliveryFuture = impl TryFuture<Ok = (), Error = SinkError> + Unpin + 'static;\n\n/// Basic data types for use with the nats interface\nimpl NatsConfig {\n    pub fn from_btreemap(values: BTreeMap<String, String>) -> Result<Self> {\n        let config = serde_json::from_value::<NatsConfig>(serde_json::to_value(values).unwrap())\n            .map_err(|e| SinkError::Config(anyhow!(e)))?;\n        if config.r#type != SINK_TYPE_APPEND_ONLY {\n            Err(SinkError::Config(anyhow!(\n                \"NATS sink only supports append-only mode\"\n            )))\n        } else {\n            Ok(config)\n        }\n    }\n}\n\nimpl TryFrom<SinkParam> for NatsSink {\n    type Error = SinkError;\n\n    fn try_from(param: SinkParam) -> std::result::Result<Self, Self::Error> {\n        let schema = param.schema();\n        let config = NatsConfig::from_btreemap(param.properties)?;\n        Ok(Self {\n            config,","sourceCodeStart":84,"sourceCodeEnd":120,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/nats.rs#L84-L120","documentation":"`NatsConfig::from_btreemap` first deserializes the WITH options map into a `NatsConfig` via serde_json; any deserialization failure (unknown/missing/mistyped fields) becomes `SinkError::Config`. It means the NATS sink options could not be parsed into a valid configuration.","triggerScenarios":"Calling `NatsConfig::from_btreemap(values)` when the BTreeMap of WITH options contains invalid keys, missing required fields (e.g. url, subject), or wrong-typed values that serde cannot deserialize into NatsConfig.","commonSituations":"Typos in WITH option names; missing `url` or `subject`; passing unsupported options (unless unknown-field validation is bypassed); quoting/format mistakes in CREATE SINK options.","solutions":["Compare your WITH options against the expected NatsConfig fields (url, subject, connect_mode, etc.) and fix names/types","Ensure required options like `url` are present and well-formed","Remove or correct unsupported option keys","Note the exact serde message in the error — it names the offending field"],"exampleFix":"// before\nWITH (connector='nats', nats_url='nats://localhost:4222');\n// after\nWITH (connector='nats', url='nats://localhost:4222', subject='test');","handlingStrategy":"validation","validationCode":"// validate nats options before creating the sink\nlet required = [\"url\", \"subject\"];\nfor k in required { assert!(options.contains_key(k), \"missing option: {}\", k); }\nassert!(options[\"url\"].starts_with(\"nats://\") || options[\"url\"].starts_with(\"tls://\"));\nassert!(options.get(\"type\").map_or(true, |t| t == \"append-only\"));","typeGuard":null,"tryCatchPattern":"match NatsConfig::from_btreemap(opts) {\n    Ok(cfg) => cfg,\n    Err(SinkError::Config(e)) => return Err(format!(\"bad nats options: {e}\")),\n    Err(e) => return Err(format!(\"nats sink error: {e}\")),\n}","preventionTips":["Compare WITH option keys against documented NatsConfig fields","Always provide url and subject","Read the serde detail message which names the bad field"],"tags":["nats","sink","config","serde"],"backgroundTag":"config-type-mismatch","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"}