risingwavelabs/risingwave · error

missing 'connector' in with clause

Error message

missing 'connector' in with clause

What it means

validate_source parses a CREATE SOURCE's WITH properties and reads the 'connector' key to determine the source type. This expect() panic means the WITH clause passed to the risectl/meta connection validation had no 'connector' entry at all, so the source type cannot be classified.

Solutions

1. Add "connector" = "<type>" (e.g. 'kafka') to the WITH clause before invoking validation
2. Verify the props JSON passed to the command is the full source WITH clause, not a subset
3. Replace the CLI path that builds props so it always includes the connector field

Example fix

// before
WITH (topic = 'events')
// after
WITH (connector = 'kafka', topic = 'events')

if !with_props.contains_key("connector") {
    return Err(anyhow!("with clause must include 'connector'"));
}

When it happens

Trigger: Calling the source-validation command path with a WITH props map that lacks the 'connector' key (e.g. an empty or hand-built props map omitting 'connector' => 'kafka').

Common situations: Hand-crafting a with-clause JSON for risectl testing; a config migration or CLI wrapper that drops connector properties; copy-pasted props from another system that uses a different key name.

Understand the failure class

Background: "is required", "must be set", "missing required field": configuration validation errors across open-source libraries — this error's family across 36 libraries.

Related errors

• exactly one of job_id or fragment_id must be provided
• No offsets provided to inject
• No properties provided to update
• unsupported source type, only kafka sources are supported
• duplicate worker id in plan, prev ->

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

Appendix: source

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

            }
        );
    }
    Ok(())
}

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)