risingwavelabs/risingwave · error · SinkError::Config

SinkError::Config(anyhow!(e))

Error message

SinkError::Config(anyhow!(e))

What it means

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

Source

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

pub struct NatsSinkWriter {
    pub config: NatsConfig,
    context: Context,
    /// Hold the client Arc to keep it alive. This allows the shared client cache to reuse
    /// the connection while we're still using it.
    #[expect(dead_code)]
    client: Arc<async_nats::Client>,
    #[expect(dead_code)]
    schema: Schema,
    json_encoder: JsonEncoder,
}

pub type NatsSinkDeliveryFuture = impl TryFuture<Ok = (), Error = SinkError> + Unpin + 'static;

/// Basic data types for use with the nats interface
impl NatsConfig {
    pub fn from_btreemap(values: BTreeMap<String, String>) -> Result<Self> {
        let config = serde_json::from_value::<NatsConfig>(serde_json::to_value(values).unwrap())
            .map_err(|e| SinkError::Config(anyhow!(e)))?;
        if config.r#type != SINK_TYPE_APPEND_ONLY {
            Err(SinkError::Config(anyhow!(
                "NATS sink only supports append-only mode"
            )))
        } else {
            Ok(config)
        }
    }
}

impl TryFrom<SinkParam> for NatsSink {
    type Error = SinkError;

    fn try_from(param: SinkParam) -> std::result::Result<Self, Self::Error> {
        let schema = param.schema();
        let config = NatsConfig::from_btreemap(param.properties)?;
        Ok(Self {
            config,

View on GitHub (pinned to 6469eb736d)

Solutions

1. Compare your WITH options against the expected NatsConfig fields (url, subject, connect_mode, etc.) and fix names/types
2. Ensure required options like `url` are present and well-formed
3. Remove or correct unsupported option keys
4. Note the exact serde message in the error — it names the offending field

Example fix

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

// validate nats options before creating the sink
let required = ["url", "subject"];
for k in required { assert!(options.contains_key(k), "missing option: {}", k); }
assert!(options["url"].starts_with("nats://") || options["url"].starts_with("tls://"));
assert!(options.get("type").map_or(true, |t| t == "append-only"));

match NatsConfig::from_btreemap(opts) {
    Ok(cfg) => cfg,
    Err(SinkError::Config(e)) => return Err(format!("bad nats options: {e}")),
    Err(e) => return Err(format!("nats sink error: {e}")),
}

When it happens

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

Common situations: 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.

Related errors

• serde_json deserialization error of ClickHouseConfig from pr
• {e}
• {e}
• {e}
• NATS sink only supports append-only mode

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