risingwavelabs/risingwave · error · SinkError

{e}

Error message

{e}

What it means

`WebhdfsConfig::from_btreemap` converts the sink WITH-options into `WebhdfsConfig` via serde JSON round-trip; deserialization failures are converted into this Config error with the serde message embedded.

Source

Thrown at src/connector/src/sink/file_sink/webhdfs.rs:87

pub struct WebhdfsSink;

impl UnknownFields for WebhdfsConfig {
    fn unknown_fields(&self) -> HashMap<String, String> {
        self.unknown_fields.clone()
    }
}

crate::impl_sink_unknown_fields!(WebhdfsConfig);

impl OpendalSinkBackend for WebhdfsSink {
    type Properties = WebhdfsConfig;

    const SINK_NAME: &'static str = WEBHDFS_SINK;

    fn from_btreemap(btree_map: BTreeMap<String, String>) -> Result<Self::Properties> {
        let config =
            serde_json::from_value::<WebhdfsConfig>(serde_json::to_value(btree_map).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)
    }

    fn new_operator(properties: WebhdfsConfig) -> Result<Operator> {
        FileSink::<WebhdfsSink>::new_webhdfs_sink(properties)
    }

    fn get_path(properties: Self::Properties) -> String {
        properties.common.path
    }

View on GitHub (pinned to 6469eb736d)

Solutions

1. Read the serde message to identify the bad option and correct its value
2. Compare each option against the WebhdfsConfig field types
3. Test the WebHDFS endpoint URL separately (curl to NameNode) to rule out malformed URLs

Example fix

// before
WITH (connector='webhdfs', webhdfs.host='hdfs://namenode:9870')
// after
WITH (connector='webhdfs', webhdfs.host='namenode', webdfs_port=9870)

// Validate WebHDFS options pre-flight
const webhdfsOpts = { 'webhdfs.host': 'namenode', path: 'out/' };
for (const [k, v] of Object.entries(webhdfsOpts)) {
  if (typeof v !== 'string' || v.trim() === '') throw new Error(`invalid ${k}`);
}

When it happens

Trigger: `CREATE SINK ... WITH (connector='webhdfs', ...)` where an option fails serde deserialization into WebhdfsConfig — e.g. malformed `webhdfs.host`, bad boolean/int strings, or unexpected characters in URL-ish fields.

Common situations: Wrong HDFS host/port formats, shell quoting stripping or adding characters, config struct evolution making previously valid properties invalid.

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 of ClickHouseConfig from pr
• {e}
• `{}` must be {}, or {}
• {e}
• SinkError::Config(anyhow!(e))

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