risingwavelabs/risingwave · error

properties `scan_startup_mode` only supports earliest and la

Error message

properties `scan_startup_mode` only supports earliest and latest or leaving it empty

What it means

Raised while building the librdkafka consumer config in the Kafka split enumerator when the `scan_startup_mode` property is set to anything other than `earliest` or `latest` (or left unset). The match on the startup mode has no arm for other values, so source startup fails with a config error.

Source

Thrown at src/connector/src/source/kafka/enumerator.rs:199

        if let Some(log_level) = read_kafka_log_level() {
            config.set_log_level(log_level);
        }
        properties.connection.set_security_properties(&mut config);
        properties.set_client(&mut config);
        // The meta-side split enumerator does not export librdkafka native stats, so disable
        // periodic statistics callbacks here even if the source properties enable them for
        // compute-side readers.
        config.set("statistics.interval.ms", "0");
        let mut scan_start_offset = match properties
            .scan_startup_mode
            .as_ref()
            .map(|s| s.to_lowercase())
            .as_deref()
        {
            Some("earliest") => KafkaEnumeratorOffset::Earliest,
            Some("latest") => KafkaEnumeratorOffset::Latest,
            None => KafkaEnumeratorOffset::Earliest,
            _ => bail!(
                "properties `scan_startup_mode` only supports earliest and latest or leaving it empty"
            ),
        };

        if let Some(s) = &properties.time_offset {
            let time_offset = s.parse::<i64>().map_err(|e| anyhow!(e))?;
            scan_start_offset = KafkaEnumeratorOffset::Timestamp(time_offset)
        }

        let mut client: Option<Arc<KafkaConsumer>> = None;
        SHARED_KAFKA_CONSUMER
            .entry_by_ref(&properties.connection)
            .and_try_compute_with::<_, _, ConnectorError>(|maybe_entry| async {
                if let Some(entry) = maybe_entry {
                    let entry_value = entry.into_value();
                    if let Some(client_) = entry_value.upgrade() {
                        // return if the client is already built
                        tracing::info!("reuse existing kafka client for {}", broker_address);

View on GitHub (pinned to 6469eb736d)

Solutions

1. Set `scan_startup_mode='earliest'` or `scan_startup_mode='latest'` in the WITH options
2. Remove the option entirely to use the default (earliest)
3. Fix typos/casing — value is lowercased automatically, but extra characters or wrong words are rejected

Example fix

// before
WITH (connector='kafka', scan_startup_mode='beginning')
// after
WITH (connector='kafka', scan_startup_mode='earliest')

match scan_startup_mode.as_deref().map(|s| s.to_lowercase()).as_deref() {
    Some("earliest") | Some("latest") | None => {},
    v => return Err(format!("invalid scan_startup_mode: {:?}")),
}

When it happens

Trigger: Creating a Kafka source/MV with WITH option `scan_startup_mode` set to anything other than 'earliest' or 'latest' (case-insensitive), e.g. 'beginning', 'end', 'EARLIEST ' with typo/extra whitespace is not trimmed — 'earliest '/'beginning' still fail after lowercase.

Common situations: Copy-pasted configs using other systems' terminology ('beginning', 'offset_reset', 'none'), typo like 'earlist', or trailing whitespace/quotes in the option value.

Understand the failure class

Background: Invalid enum value errors: "Unknown type", "Invalid scope", "must be one of" — when a string is not on the library's allowed list — this error's family across 23 libraries.

Related errors

• schemas.enable is expected to be `true` or `false`, got {s}
• serde (de)serialization error for KafkaConfig: {e}
• primary key not defined for {:?} kafka sink (please define i
• expected JSON in the form {{"host": "endpoint url"}}, but go
• unrecognized configs: {:?}

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