risingwavelabs/risingwave · error · SinkError::Nats

SinkError::Nats(anyhow!(e))

Error message

SinkError::Nats(anyhow!(e))

What it means

`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.

Source

Thrown at src/connector/src/sink/nats.rs:161

        Ok(())
    }

    async fn new_log_sinker(&self, _writer_param: SinkWriterParam) -> Result<Self::LogSinker> {
        Ok(
            NatsSinkWriter::new(self.config.clone(), self.schema.clone())
                .await?
                .into_log_sinker(NATS_SEND_FUTURE_BUFFER_MAX_SIZE),
        )
    }
}

impl NatsSinkWriter {
    pub async fn new(config: NatsConfig, schema: Schema) -> Result<Self> {
        let client = config
            .common
            .build_client()
            .await
            .map_err(|e| SinkError::Nats(anyhow!(e)))?;
        let context = NatsCommon::build_context_from_client(&client);
        Ok::<_, SinkError>(Self {
            config: config.clone(),
            context,
            client,
            schema: schema.clone(),
            json_encoder: JsonEncoder::new(
                schema,
                None,
                DateHandlingMode::FromCe,
                TimestampHandlingMode::Milli,
                TimestamptzHandlingMode::UtcWithoutSuffix,
                TimeHandlingMode::Milli,
                JsonbHandlingMode::String,
            ),
        })
    }
}

View on GitHub (pinned to 6469eb736d)

Solutions

1. Verify the NATS `url` (scheme, host, port) in the sink options, e.g. `url='nats://localhost:4222'`
2. Confirm the NATS server is running and reachable from the RisingWave node (`nats -s nats://host:4222 server check` or telnet)
3. Provide credentials if the server requires auth (user/password/token options)
4. Match the TLS scheme to the server configuration

Example fix

// before
WITH (connector='nats', url='nats://localhost:4422');
// after
WITH (connector='nats', url='nats://localhost:4222');

let url = options.get("url").expect("nats url required");
let (host, port) = parse_host_port(url).expect("valid nats url");
tokio::net::TcpStream::connect((host.as_str(), port)).await.expect("nats server reachable");

fn is_valid_nats_url(u: &str) -> bool {
    matches!(u.split_once("://"), Some(("nats" | "tls", rest))) && !rest.is_empty()
}

When it happens

Trigger: 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`.

Common situations: NATS server not running or wrong port in the `url` option; credentials required by the server not provided; TLS scheme mismatch (nats:// vs tls://).

Understand the failure class

Background: ECONNREFUSED and "connection refused" / "could not connect to server" errors: what they mean and how to fix them — this error's family across 44 libraries.

Related errors

• Nats error: {0}
• Doris/Starrocks connect error: {0}
• SinkError::Mqtt(anyhow!(e))
• failed to parse response body
• Failed connection {:?},{:?}

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/1bb1c9cbda71dda2. Report an issue: GitHub.