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.

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)

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

• unused FORMAT ENCODE option: key.message='{name}'
• failed to match the magic byte 0
• failed to read the 4-byte schema ID
• confluent schema registry error {error_code}: {message}
• missing FORMAT ... ENCODE ...

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