risingwavelabs/risingwave · error

Must specify ' ' or

Error message

Must specify '{}' or '{}'

What it means

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.

Solutions

1. Add the topic to the WITH options: WITH (connector='kafka', topic='<name>', ...) or 'kafka.topic'.
2. Check exact accepted key names (KAFKA_TOPIC_KEY1/KEY2 in src/connector/src/parser/utils.rs) for typos.
3. If constructing options in code, populate the topic key before calling schema resolution.

Example fix

// before
WITH (connector = 'kafka', schema.registry = 'http://localhost:8081')
// after
WITH (connector = 'kafka', topic = 'my_topic', schema.registry = 'http://localhost:8081')

fn validate_kafka_options(opts: &std::collections::HashMap<String, String>) -> Result<(), String> {
    if !opts.contains_key("topic") && !opts.contains_key("kafka.topic") {
        return Err("must specify 'topic' or 'kafka.topic' alongside schema.registry/schema.location".into());
    }
    Ok(())
}

When it happens

Trigger: 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.

Common situations: 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.

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

• bytes.per.second expect usize
• cannot connect to kafka broker
• confluent schema registry error
• {e}
• expect a string or a json array for privatelink.endpoint…

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

Appendix: source

Thrown at src/connector/src/parser/utils.rs:61

use risingwave_pb::plan_common::additional_column::ColumnType;

use crate::parser::{AccessError, AccessResult};
use crate::source::cdc::DebeziumCdcMeta;

/// get kafka topic name
pub(super) fn get_kafka_topic(props: &BTreeMap<String, String>) -> ConnectorResult<&String> {
    const KAFKA_TOPIC_KEY1: &str = "kafka.topic";
    const KAFKA_TOPIC_KEY2: &str = "topic";

    if let Some(topic) = props.get(KAFKA_TOPIC_KEY1) {
        return Ok(topic);
    }
    if let Some(topic) = props.get(KAFKA_TOPIC_KEY2) {
        return Ok(topic);
    }

    // config
    bail!(
        "Must specify '{}' or '{}'",
        KAFKA_TOPIC_KEY1,
        KAFKA_TOPIC_KEY2
    )
}

/// download bytes from http(s) url
pub(super) async fn download_from_http(location: &Url) -> ConnectorResult<Bytes> {
    let res = reqwest::get(location.clone())
        .await
        .with_context(|| format!("failed to make request to {location}"))?
        .error_for_status()
        .with_context(|| format!("http request failed for {location}"))?;

    let bytes = res
        .bytes()
        .await
        .with_context(|| format!("failed to read HTTP body of {location}"))?;

View on GitHub (pinned to 6469eb736d)