risingwavelabs/risingwave · error · SinkError::Config

NATS sink only supports append-only mode

Error message

NATS sink only supports append-only mode

What it means

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

Source

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

    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,
            schema,
            is_append_only: param.sink_type.is_append_only(),

View on GitHub (pinned to 6469eb736d)

Solutions

1. Set `type='append-only'` in the sink WITH options
2. If you need upsert semantics, choose a sink connector that supports upsert (e.g. JDBC/Postgres, Redis)
3. Ensure the upstream data is append-only or use a version/append-only materialized view

Example fix

// before
WITH (connector='nats', url='...', type='upsert');
// after
WITH (connector='nats', url='...', type='append-only');

let sink_type = options.get("type").map(String::as_str).unwrap_or("append-only");
assert_eq!(sink_type, "append-only", "NATS sink supports only append-only");

if let Err(SinkError::Config(e)) = NatsConfig::from_btreemap(opts.clone()) {
    if e.to_string().contains("append-only") { fix_type_option(opts); }
}

When it happens

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

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

Related errors

• SinkError::Config(anyhow!(e))
• NATS connect mode `user_and_password` requires both `user` a
• NATS connect mode `credential` requires both `nkey` and `jwt
• NATS connect mode must be one of `user_and_password`, `crede
• stream {} not found, set `allow_create_stream` to true to cr

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