risingwavelabs/risingwave · error · SinkError::Config

{e}

Error message

{e}

What it means

GooglePubSubConfig::from_btreemap converts the sink's string properties map into a JSON object and deserializes it into GooglePubSubConfig. Any field with an unexpected type or unknown/invalid value makes serde fail, and the raw serde error message is surfaced as this SinkError::Config.

Source

Thrown at src/connector/src/sink/google_pubsub.rs:116

    #[serde(rename = "pubsub.credentials")]
    pub credentials: Option<String>,

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

crate::impl_sink_unknown_fields!(GooglePubSubConfig);

impl EnforceSecret for GooglePubSubConfig {
    const ENFORCE_SECRET_PROPERTIES: phf::Set<&'static str> = phf::phf_set! {
        "pubsub.credentials",
    };
}

impl GooglePubSubConfig {
    fn from_btreemap(values: BTreeMap<String, String>) -> Result<Self> {
        serde_json::from_value::<GooglePubSubConfig>(serde_json::to_value(values).unwrap())
            .map_err(|e| SinkError::Config(anyhow!(e)))
    }
}

#[derive(Clone, Debug)]
pub struct GooglePubSubSink {
    pub config: GooglePubSubConfig,
    is_append_only: bool,

    schema: Schema,
    pk_indices: Vec<usize>,
    format_desc: SinkFormatDesc,
    db_name: String,
    sink_from_name: String,
}

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

View on GitHub (pinned to 6469eb736d)

Solutions

1. Read the embedded serde message (the `{e}`) to see which field failed and fix that option name/value
2. Compare your WITH options against the GooglePubSubConfig struct fields in src/connector/src/sink/google_pubsub.rs
3. Ensure all option values are quoted strings in the sink definition

Example fix

// before
WITH (connector='google_pubsub', emulator_host=123)
// after
WITH (connector='google_pubsub', pubsub.emulator_host='localhost:8085')

fn validate_pubsub_props(props: &[(String, String)]) -> Result<(), String> {
    let allowed = ["pubsub.emulator_host", "pubsub.credentials", "pubsub.topic", "pubsub.endpoint"];
    for (k, v) in props {
        if !allowed.contains(&k.as_str()) {
            return Err(format!("unknown pubsub option: {k}"));
        }
        if v.is_empty() {
            return Err(format!("empty value for pubsub option: {k}"));
        }
    }
    Ok(())
}

When it happens

Trigger: Creating a Google Pub/Sub sink whose WITH options include a mistyped field (e.g. `endpoint` given a non-string, or an option name that doesn't exist on GooglePubSubConfig via deny_unknown_fields), so serde_json::from_value fails.

Common situations: Typos in pubsub option names like `pubsub.emulator_host` or `pubsub.credentials`; passing numbers/booleans where strings are expected; stale option names after a config struct change.

Related errors

• serde_json deserialization error of ClickHouseConfig from pr
• {e}
• {e}
• Configure at least one of `pubsub.emulator_host` and `pubsub
• missing FORMAT ... ENCODE ...

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