{"record":{"id":"9f86b3faf2ecccc4","repo":"risingwavelabs/risingwave","slug":"nats-sink-only-supports-append-only-mode","errorCode":null,"errorMessage":"NATS sink only supports append-only mode","messagePattern":"NATS sink only supports append-only mode","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/nats.rs","lineNumber":104,"sourceCode":"    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,\n            schema,\n            is_append_only: param.sink_type.is_append_only(),","sourceCodeStart":86,"sourceCodeEnd":122,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/nats.rs#L86-L122","documentation":"`NatsConfig::from_btreemap` rejects NATS sink configurations whose `type` option is not `append-only`, because the NATS sink only supports appending rows without updates/deletes. The configured sink type is invalid for this connector.","triggerScenarios":"Creating a NATS sink with `type='upsert'` (or any value other than `append-only`) in the WITH options, which fails during config parsing in `from_btreemap`.","commonSituations":"Copy-pasting a sink definition from a connector that supports upsert (e.g. Postgres/JDBC) onto NATS; forgetting to change the default type when the source materialized view receives updates/deletes.","solutions":["Set `type='append-only'` in the sink WITH options","If you need upsert semantics, choose a sink connector that supports upsert (e.g. JDBC/Postgres, Redis)","Ensure the upstream data is append-only or use a version/append-only materialized view"],"exampleFix":"// before\nWITH (connector='nats', url='...', type='upsert');\n// after\nWITH (connector='nats', url='...', type='append-only');","handlingStrategy":"validation","validationCode":"let sink_type = options.get(\"type\").map(String::as_str).unwrap_or(\"append-only\");\nassert_eq!(sink_type, \"append-only\", \"NATS sink supports only append-only\");","typeGuard":null,"tryCatchPattern":"if let Err(SinkError::Config(e)) = NatsConfig::from_btreemap(opts.clone()) {\n    if e.to_string().contains(\"append-only\") { fix_type_option(opts); }\n}","preventionTips":["Never set type='upsert' for NATS sinks","Use NATS only for append-only streams","If upsert is needed, switch to a connector that supports it"],"tags":["nats","sink","config","unsupported-mode"],"backgroundTag":"unsupported-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"}