risingwavelabs/risingwave · error · SinkError::Config

SinkError::Config(anyhow!(e))

Error message

SinkError::Config(anyhow!(e))

What it means

MongodbConfig::from_btreemap parses the sink's user properties into a MongodbConfig via a serde_json round-trip. If deserialization fails (missing required fields like 'url', or fields with wrong types), the serde error is wrapped in SinkError::Config.

Source

Thrown at src/connector/src/sink/mongodb.rs:169

    pub bulk_write_max_entries: usize,

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

crate::impl_sink_unknown_fields!(MongodbConfig);

impl EnforceSecret for MongodbConfig {
    fn enforce_one(prop: &str) -> crate::error::ConnectorResult<()> {
        MongodbCommon::enforce_one(prop)
    }
}

impl MongodbConfig {
    pub fn from_btreemap(properties: BTreeMap<String, String>) -> crate::sink::Result<Self> {
        let config =
            serde_json::from_value::<MongodbConfig>(serde_json::to_value(properties).unwrap())
                .map_err(|e| SinkError::Config(anyhow!(e)))?;
        if config.r#type != SINK_TYPE_APPEND_ONLY && config.r#type != SINK_TYPE_UPSERT {
            return Err(SinkError::Config(anyhow!(
                "`{}` must be {}, or {}",
                SINK_TYPE_OPTION,
                SINK_TYPE_APPEND_ONLY,
                SINK_TYPE_UPSERT
            )));
        }
        Ok(config)
    }
}

/// An async-drop style `Guard` for `mongodb::Client`. Use this guard to hold a client,
/// the `client::shutdown` is called in an async manner when the guard is dropped.
/// Please be aware this is a "best effort" style shutdown, which may not be successful if the
/// tokio runtime is in the process of terminating. However, the server-side resources will be
/// cleaned up eventually due to the session expiration.
/// see [this issue](https://github.com/mongodb/mongo-rust-driver/issues/719) for more information

View on GitHub (pinned to 6469eb736d)

Solutions

1. Ensure 'url' (mongodb://... connection string) and 'type' are present in the WITH options.
2. Check for typos in property names against the MongodbConfig struct fields.
3. Quote string values in the DDL so serde receives strings, not numbers/booleans.
4. Validate the connection string format: mongodb://user:pass@host:port/db.

Example fix

// before
CREATE SINK s FROM mv WITH (
  connector = 'mongodb',
  endpoint = 'mongodb://localhost:27017'  -- unknown/wrong field
);
// after
CREATE SINK s FROM mv WITH (
  connector = 'mongodb',
  url = 'mongodb://localhost:27017',
  type = 'append_only'
);

-- validate WITH options before CREATE SINK
-- required: connector='mongodb', url (string), type in ('append_only','upsert')
SELECT * FROM rw_catalog.rw_databases; -- or dry-run the config locally

// catch config parse failures at sink creation time and show the serde message
match MongodbConfig::from_btreemap(props) {
    Ok(cfg) => proceed(cfg),
    Err(e) => return Err(format!("invalid mongodb sink options: {e}")),
}

When it happens

Trigger: serde_json::from_value::<MongodbConfig> fails: the WITH properties map lacks required fields (e.g. 'url' or 'type') or a property's value type doesn't match the config struct (e.g. url is not a string).

Common situations: Typo in a required option name in the CREATE SINK WITH (...) clause, omitting 'url' or 'type' entirely, passing numeric/boolean values where strings are expected, or extra unknown properties when deny_unknown_fields is active.

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

• (serde_json deserialization error)
• `{}` must be {}, or {}
• {serde_json deserialization error for PulsarConfig from prop
• {serde_json deserialization error for RedisConfig from prope
• SinkError::Config(anyhow!(e))

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