risingwavelabs/risingwave · error · SinkError::Config

{serde_json deserialization error for PulsarConfig from prop

Error message

{serde_json deserialization error for PulsarConfig from properties}

What it means

This error wraps a serde_json deserialization failure that occurred while converting the sink's WITH-clause properties (a BTreeMap<String, String>) into a PulsarConfig via from_btreemap. The properties map is serialized to a JSON value and then deserialized against the PulsarConfig struct, so any property that is missing a required field, has the wrong type, or carries an unknown/invalid value surfaces here as a SinkError::Config. The library throws it because the sink cannot be constructed without a fully valid configuration.

Source

Thrown at src/connector/src/sink/pulsar.rs:202

    pub producer_properties: PulsarPropertiesProducer,

    #[serde(flatten)]
    pub unknown_fields: std::collections::HashMap<String, String>,
}

crate::impl_sink_unknown_fields!(PulsarConfig);

impl EnforceSecret for PulsarConfig {
    fn enforce_one(prop: &str) -> crate::error::ConnectorResult<()> {
        PulsarCommon::enforce_one(prop)?;
        AwsAuthProps::enforce_one(prop)?;
        Ok(())
    }
}
impl PulsarConfig {
    pub fn from_btreemap(values: BTreeMap<String, String>) -> Result<Self> {
        let config = serde_json::from_value::<PulsarConfig>(serde_json::to_value(values).unwrap())
            .map_err(|e| SinkError::Config(anyhow!(e)))?;

        Ok(config)
    }
}

#[derive(Debug)]
pub struct PulsarSink {
    pub config: PulsarConfig,
    schema: Schema,
    downstream_pk: Vec<usize>,
    format_desc: SinkFormatDesc,
    db_name: String,
    sink_from_name: String,
}

impl EnforceSecret for PulsarSink {
    fn enforce_secret<'a>(
        prop_iter: impl Iterator<Item = &'a str>,

View on GitHub (pinned to 6469eb736d)

Solutions

1. Check the SQL CREATE SINK WITH options and ensure every required PulsarConfig field is present and correctly spelled (e.g. service.url, topic).
2. Verify each option's value type matches PulsarConfig (strings, integers, booleans) — the serde error message names the offending field.
3. Remove unrelated/unknown WITH options that are not part of PulsarConfig or disable deny_unknown_fields expectations by removing stale keys.
4. Test the properties map locally with serde_json::from_value::<PulsarConfig>(serde_json::to_value(map).unwrap()) to see the exact serde error before wiring the sink.
5. Compare against the documented example in the RisingWave docs for the Pulsar sink.

Example fix

// before
CREATE SINK s FROM mv WITH (
  connector = 'pulsar',
  service_url = 'pulsar://localhost:6650',
  topic = 't'
);
// after
CREATE SINK s FROM mv WITH (
  connector = 'pulsar',
  service.url = 'pulsar://localhost:6650',
  topic = 't'
);

// Validate sink props before from_btreemap
fn validate_pulsar_props(props: &BTreeMap<String, String>) -> Result<(), String> {
    let required = ["service.url", "topic"];
    for k in required {
        if !props.contains_key(k) {
            return Err(format!("missing required pulsar sink option: {k}"));
        }
    }
    Ok(())
}

fn has_required_keys(props: &BTreeMap<String, String>, keys: &[&str]) -> bool {
    keys.iter().all(|k| props.contains_key(*k))
}

match PulsarConfig::from_btreemap(props) {
    Ok(cfg) => /* proceed */,
    Err(SinkError::Config(e)) => log::error!("invalid pulsar sink properties: {e:#}"),
    Err(e) => return Err(e),
}

When it happens

Trigger: Calling PulsarConfig::from_btreemap (or creating a Pulsar sink) with a properties map where a required PulsarConfig field (e.g. service.url, topic) is absent, or a field has a value that fails JSON type/enum parsing (e.g. numeric-seed=false given a non-boolean-like string, unparseable integer fields).

Common situations: Typos in sink WITH options when running CREATE SINK; passing extra unrelated properties; quoting/escaping mistakes in the SQL DDL so values do not deserialize; upgrading RisingWave and having renamed config fields.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

• missing FORMAT ... ENCODE ...
• {serde_json deserialization error for RedisConfig from prope
• SinkError::Config(anyhow!(e))
• serde_json deserialization error of ClickHouseConfig from pr
• `{}` must be {}, or {}

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