risingwavelabs/risingwave · error · SinkError::Config

serde (de)serialization error for KafkaConfig: {e}

Error message

serde (de)serialization error for KafkaConfig: {e}

What it means

KafkaConfig::from_btreemap serializes the BTreeMap of properties to JSON and deserializes it into the KafkaConfig struct. Any serde failure (unknown fields, type mismatches like non-numeric values for numeric fields) is surfaced as this SinkError::Config error, meaning one or more user-supplied sink properties are invalid for KafkaConfig.

Source

Thrown at src/connector/src/sink/kafka.rs:274

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

crate::impl_sink_unknown_fields!(KafkaConfig);

impl EnforceSecret for KafkaConfig {
    fn enforce_one(prop: &str) -> crate::error::ConnectorResult<()> {
        KafkaConnectionProps::enforce_one(prop)?;
        AwsAuthProps::enforce_one(prop)?;
        Ok(())
    }
}

impl KafkaConfig {
    pub fn from_btreemap(values: BTreeMap<String, String>) -> Result<Self> {
        let config = serde_json::from_value::<KafkaConfig>(serde_json::to_value(values).unwrap())
            .map_err(|e| SinkError::Config(anyhow!(e)))?;

        Ok(config)
    }

    pub(crate) fn set_client(&self, c: &mut rdkafka::ClientConfig) {
        self.rdkafka_properties_common.set_client(c);
        self.rdkafka_properties_producer.set_client(c);
    }
}

impl From<KafkaConfig> for KafkaProperties {
    fn from(val: KafkaConfig) -> Self {
        KafkaProperties {
            bytes_per_second: None,
            max_num_messages: None,
            scan_startup_mode: None,
            time_offset: None,
            upsert: None,

View on GitHub (pinned to 6469eb736d)

Solutions

1. Print the underlying serde error (it names the offending field) and fix that property in the sink's WITH options
2. Cast values to the expected types (integers without quotes for numeric fields)
3. Check the KafkaConfig struct definition for exact field names and types
4. Drop unknown/unsupported properties from the WITH clause

Example fix

// before
CREATE SINK s FROM t WITH (connector='kafka', properties.bootstrap.server='b:9092');
// after
CREATE SINK s FROM t WITH (connector='kafka', properties.bootstrap.server='b:9092', type='append only');

fn validate_kafka_props(props: &BTreeMap<String, String>) -> Result<()> {
    KafkaConfig::from_btreemap(props.clone())?;
    Ok(())
} // run before CREATE SINK takes effect

match KafkaConfig::from_btreemap(props) {
    Ok(cfg) => cfg,
    Err(e) => { log::error!("invalid kafka properties: {e}"); return Err(e); }
}

When it happens

Trigger: CREATE SINK ... WITH (...) options converted to a BTreeMap then mapped into KafkaConfig via serde_json, where a value has the wrong type (e.g. string where a number/bool is expected) or an unknown field is present under a struct that disallows unknown keys.

Common situations: Typo'd property names in the WITH clause; passing quoted numbers like "'9092'" where an integer is expected; wrong property grouped under rdkafka_properties vs typed fields; RisingWave version changes to KafkaConfig field types.

Understand the failure class

Background: "Invalid value" and "allowed values are" config errors: what your library rejected and how to fix it — this error's family across 41 libraries.

Related errors

• schemas.enable is expected to be `true` or `false`, got {s}
• primary key not defined for {:?} kafka sink (please define i
• properties `scan_startup_mode` only supports earliest and la
• expected JSON in the form {{"host": "endpoint url"}}, but go
• unrecognized configs: {:?}

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