risingwavelabs/risingwave · error · ConnectorError

`startup_mode` must be `earliest`, `latest`, or empty

Error message

`startup_mode` must be `earliest`, `latest`, or empty

What it means

The Pulsar enumerator accepts only 'earliest' or 'latest' (or unset, defaulting to earliest) for the startup_mode property. Any other string makes the enumerator construction fail since the offset semantics would be undefined.

Source

Thrown at src/connector/src/source/pulsar/enumerator/client.rs:73

            .build_client(
                &properties.oauth,
                &properties.aws_auth_props,
                properties.operation_retry.to_pulsar_options(),
            )
            .await?;
        let topic = properties.common.topic;
        let parsed_topic = parse_topic(&topic)?;

        let mut scan_start_offset = match properties
            .scan_startup_mode
            .map(|s| s.to_lowercase())
            .as_deref()
        {
            Some("earliest") => PulsarEnumeratorOffset::Earliest,
            Some("latest") => PulsarEnumeratorOffset::Latest,
            None => PulsarEnumeratorOffset::Earliest,
            _ => {
                bail!("`startup_mode` must be `earliest`, `latest`, or empty");
            }
        };

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

        Ok(PulsarSplitEnumerator {
            client: pulsar,
            topic: parsed_topic,
            start_offset: scan_start_offset,
        })
    }

    async fn list_splits(&mut self) -> ConnectorResult<Vec<PulsarSplit>> {
        let offset = self.start_offset.clone();
        // MessageId is only used when recovering from a State

View on GitHub (pinned to 6469eb736d)

Solutions

1. Set startup_mode to 'earliest' or 'latest', or remove it entirely (defaults to earliest).
2. For timestamp-based startup, set the `time_offset` property instead of startup_mode.
3. Ensure the value is lowercase with no surrounding whitespace or quotes issues in the WITH clause.

Example fix

// before
WITH (connector = 'pulsar', startup_mode = 'beginning')
// after
WITH (connector = 'pulsar', startup_mode = 'earliest')

match props.get("startup_mode").map(String::as_str) {
    None | Some("earliest") | Some("latest") => Ok(()),
    Some(other) => Err(format!("unsupported startup_mode: {other}")),
}

fn is_valid_pulsar_startup_mode(mode: Option<&str>) -> bool {
    matches!(mode, None | Some("earliest") | Some("latest"))
}

match source.create().await {
    Err(e) if e.to_string().contains("startup_mode") => eprintln!("set startup_mode to earliest or latest"),
    other => other,
}

When it happens

Trigger: Creating a Pulsar source with `startup_mode` set to something other than 'earliest', 'latest', or omitted — e.g. 'timestamp', 'EARLIEST', or 'beginning'.

Common situations: Reusing Kafka/NATS startup config keys on Pulsar; uppercase values; expecting timestamp-based startup to work via startup_mode (Pulsar uses a separate time_offset field instead).

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

• NATS connect mode must be one of `user_and_password`, `crede
• invalid scan.startup.mode, accept earliest/latest/timestamp
• unrecognized configs: {:?}
• Invalid value for Bounded strategy: must be positive integer
• Invalid value for Ratio strategy: must be between 0.0 and 1.

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