{"record":{"id":"1bb1c9cbda71dda2","repo":"risingwavelabs/risingwave","slug":"sinkerror-nats-anyhow-e","errorCode":null,"errorMessage":"SinkError::Nats(anyhow!(e))","messagePattern":"SinkError::Nats\\(anyhow!\\(e\\)\\)","errorType":"exception","errorClass":"SinkError::Nats","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/nats.rs","lineNumber":161,"sourceCode":"        Ok(())\n    }\n\n    async fn new_log_sinker(&self, _writer_param: SinkWriterParam) -> Result<Self::LogSinker> {\n        Ok(\n            NatsSinkWriter::new(self.config.clone(), self.schema.clone())\n                .await?\n                .into_log_sinker(NATS_SEND_FUTURE_BUFFER_MAX_SIZE),\n        )\n    }\n}\n\nimpl NatsSinkWriter {\n    pub async fn new(config: NatsConfig, schema: Schema) -> Result<Self> {\n        let client = config\n            .common\n            .build_client()\n            .await\n            .map_err(|e| SinkError::Nats(anyhow!(e)))?;\n        let context = NatsCommon::build_context_from_client(&client);\n        Ok::<_, SinkError>(Self {\n            config: config.clone(),\n            context,\n            client,\n            schema: schema.clone(),\n            json_encoder: JsonEncoder::new(\n                schema,\n                None,\n                DateHandlingMode::FromCe,\n                TimestampHandlingMode::Milli,\n                TimestamptzHandlingMode::UtcWithoutSuffix,\n                TimeHandlingMode::Milli,\n                JsonbHandlingMode::String,\n            ),\n        })\n    }\n}","sourceCodeStart":143,"sourceCodeEnd":179,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/nats.rs#L143-L179","documentation":"`NatsSinkWriter::new` wraps failures from `build_client().await` into `SinkError::Nats` while constructing the sink writer. It means the NATS client could not connect/initialize, so the sink writer cannot be created.","triggerScenarios":"Calling `NatsSinkWriter::new(config, schema)` when the NATS server is unreachable, the URL is wrong, authentication fails, or TLS settings are invalid during `NatsCommon::build_client`.","commonSituations":"NATS server not running or wrong port in the `url` option; credentials required by the server not provided; TLS scheme mismatch (nats:// vs tls://).","solutions":["Verify the NATS `url` (scheme, host, port) in the sink options, e.g. `url='nats://localhost:4222'`","Confirm the NATS server is running and reachable from the RisingWave node (`nats -s nats://host:4222 server check` or telnet)","Provide credentials if the server requires auth (user/password/token options)","Match the TLS scheme to the server configuration"],"exampleFix":"// before\nWITH (connector='nats', url='nats://localhost:4422');\n// after\nWITH (connector='nats', url='nats://localhost:4222');","handlingStrategy":"validation","validationCode":"let url = options.get(\"url\").expect(\"nats url required\");\nlet (host, port) = parse_host_port(url).expect(\"valid nats url\");\ntokio::net::TcpStream::connect((host.as_str(), port)).await.expect(\"nats server reachable\");","typeGuard":"fn is_valid_nats_url(u: &str) -> bool {\n    matches!(u.split_once(\"://\"), Some((\"nats\" | \"tls\", rest))) && !rest.is_empty()\n}","tryCatchPattern":null,"preventionTips":["Confirm the NATS server is running and the port is correct (default 4222)","Provide credentials when the server requires auth","Match tls:// vs nats:// to the server's TLS config"],"tags":["nats","sink","network","connection"],"backgroundTag":"connection-refused","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"}