risingwavelabs/risingwave · error · SinkError::Config

`{}` must be {}, or {}

Error message

`{}` must be {}, or {}

What it means

Config validation in MongodbConfig::from_btreemap: an enumerated with-clause option (such as the sink `type`) was given a value outside the small allowed set; the message lists the accepted alternatives. Fires when serde-deserialized properties contain a value serde accepted but the explicit check rejects.

Source

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

    #[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
struct ClientGuard {
    _tx: tokio::sync::oneshot::Sender<()>,

View on GitHub (pinned to 6469eb736d)

Solutions

1. Set type = 'append_only' for insert-only sinks or type = 'upsert' for upsert sinks, exactly as spelled.
2. Check the RisingWave MongoDB sink docs for the accepted type values in your version.
3. Avoid hyphenated variants like 'append-only'; the constants use underscores.
4. If the type came from a generated/templated DDL, review the template's WITH block.

Example fix

// before
WITH (connector = 'mongodb', url = 'mongodb://...', type = 'append-only')
// after
WITH (connector = 'mongodb', url = 'mongodb://...', type = 'append_only')

// validate the type value before creating the sink
const VALID: [&str; 2] = ["append_only", "upsert"];
assert!(VALID.contains(&sink_type.as_str()), "type must be append_only or upsert, got: {sink_type}");

// surface the accepted values in the failure path
match sink_type.as_str() {
    "append_only" | "upsert" => proceed(),
    other => return Err(format!("invalid type '{other}'; use append_only or upsert")),
}

When it happens

Trigger: config.r#type is neither SINK_TYPE_APPEND_ONLY ('append_only') nor SINK_TYPE_UPSERT ('upsert') — i.e. the user supplied a different 'type' string (or an unexpected default) in the sink WITH options.

Common situations: Typo like 'append-only' or 'upsert-only', copying options from another connector with different type names, or forgetting 'type' when the serde default yields an invalid value.

Understand the failure class

Background: Invalid enum value errors: "Unknown type", "Invalid scope", "must be one of" — when a string is not on the library's allowed list — this error's family across 23 libraries.

Related errors

• 'copy-on-write' mode is not supported for append-only iceber
• SinkError::Config(anyhow!(e))
• Invalid value `{value}` for `{entry}`
• unrecognized configs: {:?}
• Invalid value for Bounded strategy: must be positive integer

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