{"record":{"id":"a58212d9224ba3c8","repo":"risingwavelabs/risingwave","slug":"must-specify-or","errorCode":null,"errorMessage":"Must specify '{}' or '{}'","messagePattern":"Must specify '(.+?)' or '(.+?)'","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/connector/src/parser/utils.rs","lineNumber":61,"sourceCode":"use risingwave_pb::plan_common::additional_column::ColumnType;\n\nuse crate::parser::{AccessError, AccessResult};\nuse crate::source::cdc::DebeziumCdcMeta;\n\n/// get kafka topic name\npub(super) fn get_kafka_topic(props: &BTreeMap<String, String>) -> ConnectorResult<&String> {\n    const KAFKA_TOPIC_KEY1: &str = \"kafka.topic\";\n    const KAFKA_TOPIC_KEY2: &str = \"topic\";\n\n    if let Some(topic) = props.get(KAFKA_TOPIC_KEY1) {\n        return Ok(topic);\n    }\n    if let Some(topic) = props.get(KAFKA_TOPIC_KEY2) {\n        return Ok(topic);\n    }\n\n    // config\n    bail!(\n        \"Must specify '{}' or '{}'\",\n        KAFKA_TOPIC_KEY1,\n        KAFKA_TOPIC_KEY2\n    )\n}\n\n/// download bytes from http(s) url\npub(super) async fn download_from_http(location: &Url) -> ConnectorResult<Bytes> {\n    let res = reqwest::get(location.clone())\n        .await\n        .with_context(|| format!(\"failed to make request to {location}\"))?\n        .error_for_status()\n        .with_context(|| format!(\"http request failed for {location}\"))?;\n\n    let bytes = res\n        .bytes()\n        .await\n        .with_context(|| format!(\"failed to read HTTP body of {location}\"))?;","sourceCodeStart":43,"sourceCodeEnd":79,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/parser/utils.rs#L43-L79","documentation":"get_kafka_topic requires the Kafka topic to be resolvable from connector properties, checked via two option keys (KAFKA_TOPIC_KEY1, e.g. 'kafka.topic', and KAFKA_TOPIC_KEY2, e.g. 'topic'). If neither key is present in the properties map, schema resolution (Confluent schema registry or JSON schema location) cannot proceed and this error is raised.","triggerScenarios":"SchemaLocation::Confluent or fetch_json_schema_and_map_to_columns is invoked with WITH options that lack both topic keys — e.g. schema.location/schema.registry configured but no 'kafka.topic'/'topic' option.","commonSituations":"Setting schema.registry but forgetting the topic option; using a wrong option name like 'kafka_topic'; building schema config programmatically and omitting the topic field.","solutions":["Add the topic to the WITH options: WITH (connector='kafka', topic='<name>', ...) or 'kafka.topic'.","Check exact accepted key names (KAFKA_TOPIC_KEY1/KEY2 in src/connector/src/parser/utils.rs) for typos.","If constructing options in code, populate the topic key before calling schema resolution."],"exampleFix":"// before\nWITH (connector = 'kafka', schema.registry = 'http://localhost:8081')\n// after\nWITH (connector = 'kafka', topic = 'my_topic', schema.registry = 'http://localhost:8081')","handlingStrategy":"validation","validationCode":"fn validate_kafka_options(opts: &std::collections::HashMap<String, String>) -> Result<(), String> {\n    if !opts.contains_key(\"topic\") && !opts.contains_key(\"kafka.topic\") {\n        return Err(\"must specify 'topic' or 'kafka.topic' alongside schema.registry/schema.location\".into());\n    }\n    Ok(())\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Always include topic in Kafka WITH options when using schema registry or schema.location.","Keep a canonical DDL template per connector to avoid missing keys.","Validate options with a dry-run create in dev."],"tags":["kafka","config","missing-option","schema-registry"],"backgroundTag":"missing-required-config-field","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}