risingwavelabs/risingwave · error · SinkError::Config

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

Error message

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

What it means

KinesisSinkConfig::from_btreemap serializes the WITH-option BTreeMap to JSON and deserializes it into KinesisSinkConfig with serde. Any option name/type the struct does not expect (or an unexpected value type after the conversion) causes a serde error that is wrapped as a SinkError::Config.

Source

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

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

crate::impl_sink_unknown_fields!(KinesisSinkConfig);

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

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

pub struct KinesisSinkWriter {
    pub config: KinesisSinkConfig,
    formatter: SinkFormatterImpl,
    client: KinesisClient,
}

struct KinesisSinkPayloadWriter {
    client: KinesisClient,
    entries: Vec<(PutRecordsRequestEntry, usize)>,
    stream_name: String,
}

impl KinesisSinkWriter {
    pub async fn new(

View on GitHub (pinned to 6469eb736d)

Solutions

  1. Check the option names against KinesisSinkConfig fields and fix typos
  2. Remove options not supported by the Kinesis connector
  3. Enable/inspect serde's error message which names the offending field and fix it in the CREATE SINK statement

Example fix

// before
WITH ( connector='kinesis', stream_name='s1' )   // wrong field name
// after
WITH ( connector='kinesis', stream='s1' )
Defensive patterns

Strategy: validation

Validate before calling

// validate WITH options against KinesisSinkConfig fields first
let expected = ["connector","stream","aws.region","endpoint","primary_key"];
let unknown: Vec<_> = props.keys().filter(|k| !expected.contains(&k.as_str())).collect();
if !unknown.is_empty() { return Err(format!("unknown options: {:?}", unknown)); }

Try / catch

match KinesisSinkConfig::from_btreemap(props) {
  Ok(c) => c,
  Err(e) => { log::error!("invalid kinesis sink options: {e}"); return Err(e); }
}

Prevention

When it happens

Trigger: Calling `KinesisSinkConfig::from_btreemap(properties)` where the map contains an unknown field for KinesisSinkConfig or a value that fails deserialization into the expected typed field.

Common situations: Typos in WITH options (e.g. `steam` instead of `stream`); passing options belonging to another connector; providing non-string-shaped values that serde cannot map.

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

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