risingwavelabs/risingwave · error · anyhow::Error

unsupported source type, only kafka sources are supported

Error message

unsupported source type, only kafka sources are supported

What it means

Guard in risectl validate_source: after reading the `connector` property, only `kafka` is accepted; any other connector value makes the standalone source-validation command fail, since risectl's validation path is implemented exclusively for Kafka sources.

Solutions

1. Use connector = 'kafka' in the WITH clause, or create the source through the SQL frontend instead of this risectl path
2. Fix the connector value spelling/casing to exactly 'kafka'
3. Upgrade risectl/risingwave if you need other connector types

Example fix

// before
WITH (connector = 'kinesis', stream = 's')
// after
WITH (connector = 'kafka', topic = 't')

match with_props.get("connector").map(|v| v.as_str()) {
    Some("kafka") => {},
    other => return Err(anyhow!("connector {:?} unsupported here", other)),
}

When it happens

Trigger: WITH clause sets connector to a non-kafka value (e.g. 'kinesis', 'pulsar', 'nexmark', or a typo like 'Kafka') and the source is validated via this risectl path.

Common situations: Trying to validate a newer connector type through an older risectl build; typos or case mismatches in the connector name.

Related errors

• exactly one of job_id or fragment_id must be provided
• missing 'connector' in with clause
• No offsets provided to inject
• No properties provided to update
• duplicate worker id in plan, prev ->

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

Appendix: source

Thrown at src/ctl/src/cmd_impl/meta/connection.rs:69

}

pub async fn validate_source(context: &CtlContext, props: String) -> anyhow::Result<()> {
    let with_props: HashMap<String, String> =
        serde_json::from_str::<HashMap<String, Value>>(props.as_str())
            .expect("error parsing with props json")
            .into_iter()
            .map(|(key, val)| match val {
                Value::String(s) => (key, s),
                _ => (key, val.to_string()),
            })
            .collect();
    let source_type = match with_props
        .get("connector")
        .expect("missing 'connector' in with clause")
        .as_str()
    {
        "kafka" => Ok(SourceType::Kafka),
        _ => Err(anyhow!(
            "unsupported source type, only kafka sources are supported"
        )),
    }?;
    let meta_client = context.meta_client().await?;
    let resp = meta_client
        .rw_cloud_validate_source(source_type, with_props)
        .await?;
    if !resp.ok {
        eprintln!("{}", serde_json::to_string(&resp).unwrap());
        std::process::exit(1);
    }
    Ok(())
}

View on GitHub (pinned to 6469eb736d)