{"record":{"id":"74183e99beacc31e","repo":"risingwavelabs/risingwave","slug":"startup-mode-must-be-earliest-latest-or-em","errorCode":null,"errorMessage":"`startup_mode` must be `earliest`, `latest`, or empty","messagePattern":"`startup_mode` must be `earliest`, `latest`, or empty","errorType":"validation","errorClass":"ConnectorError","httpStatus":null,"severity":"error","filePath":"src/connector/src/source/pulsar/enumerator/client.rs","lineNumber":73,"sourceCode":"            .build_client(\n                &properties.oauth,\n                &properties.aws_auth_props,\n                properties.operation_retry.to_pulsar_options(),\n            )\n            .await?;\n        let topic = properties.common.topic;\n        let parsed_topic = parse_topic(&topic)?;\n\n        let mut scan_start_offset = match properties\n            .scan_startup_mode\n            .map(|s| s.to_lowercase())\n            .as_deref()\n        {\n            Some(\"earliest\") => PulsarEnumeratorOffset::Earliest,\n            Some(\"latest\") => PulsarEnumeratorOffset::Latest,\n            None => PulsarEnumeratorOffset::Earliest,\n            _ => {\n                bail!(\"`startup_mode` must be `earliest`, `latest`, or empty\");\n            }\n        };\n\n        if let Some(s) = properties.time_offset {\n            let time_offset = s.parse::<i64>().map_err(|e| anyhow!(e))?;\n            scan_start_offset = PulsarEnumeratorOffset::Timestamp(time_offset)\n        }\n\n        Ok(PulsarSplitEnumerator {\n            client: pulsar,\n            topic: parsed_topic,\n            start_offset: scan_start_offset,\n        })\n    }\n\n    async fn list_splits(&mut self) -> ConnectorResult<Vec<PulsarSplit>> {\n        let offset = self.start_offset.clone();\n        // MessageId is only used when recovering from a State","sourceCodeStart":55,"sourceCodeEnd":91,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/source/pulsar/enumerator/client.rs#L55-L91","documentation":"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.","triggerScenarios":"Creating a Pulsar source with `startup_mode` set to something other than 'earliest', 'latest', or omitted — e.g. 'timestamp', 'EARLIEST', or 'beginning'.","commonSituations":"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).","solutions":["Set startup_mode to 'earliest' or 'latest', or remove it entirely (defaults to earliest).","For timestamp-based startup, set the `time_offset` property instead of startup_mode.","Ensure the value is lowercase with no surrounding whitespace or quotes issues in the WITH clause."],"exampleFix":"// before\nWITH (connector = 'pulsar', startup_mode = 'beginning')\n// after\nWITH (connector = 'pulsar', startup_mode = 'earliest')","handlingStrategy":"validation","validationCode":"match props.get(\"startup_mode\").map(String::as_str) {\n    None | Some(\"earliest\") | Some(\"latest\") => Ok(()),\n    Some(other) => Err(format!(\"unsupported startup_mode: {other}\")),\n}","typeGuard":"fn is_valid_pulsar_startup_mode(mode: Option<&str>) -> bool {\n    matches!(mode, None | Some(\"earliest\") | Some(\"latest\"))\n}","tryCatchPattern":"match source.create().await {\n    Err(e) if e.to_string().contains(\"startup_mode\") => eprintln!(\"set startup_mode to earliest or latest\"),\n    other => other,\n}","preventionTips":["Only use 'earliest'/'latest' for Pulsar startup_mode; leave unset to default to earliest.","Use time_offset for timestamp-based starts instead of inventing startup_mode values.","Ensure lowercase values without stray whitespace in WITH clauses."],"tags":["pulsar","config","validation","enum"],"backgroundTag":"invalid-enum-value","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"}