{"record":{"id":"571b884c809ec862","repo":"risingwavelabs/risingwave","slug":"scan-startup-timestamp-millis-is-required-571b88","errorCode":null,"errorMessage":"scan.startup.timestamp.millis is required","messagePattern":"scan\\.startup\\.timestamp\\.millis is required","errorType":"validation","errorClass":"ConnectorError","httpStatus":null,"severity":"error","filePath":"src/connector/src/source/nats/source/reader.rs","lineNumber":74,"sourceCode":"        parser_config: ParserConfig,\n        source_ctx: SourceContextRef,\n        _columns: Option<Vec<Column>>,\n    ) -> Result<Self> {\n        // We guarantee the split num always align with parallelism\n        assert_eq!(splits.len(), 1);\n        let split = splits.into_iter().next().unwrap();\n        let split_id = split.split_id;\n        let start_position = match &split.start_sequence {\n            NatsOffset::None => match &properties.scan_startup_mode {\n                None => NatsOffset::Earliest,\n                Some(mode) => match mode.as_str() {\n                    \"latest\" => NatsOffset::Latest,\n                    \"earliest\" => NatsOffset::Earliest,\n                    \"timestamp\" | \"timestamp_millis\" /* backward-compat */ => {\n                        if let Some(ts) = &properties.start_timestamp_millis {\n                            NatsOffset::Timestamp(*ts)\n                        } else {\n                            bail!(\"scan.startup.timestamp.millis is required\");\n                        }\n                    }\n                    _ => {\n                        bail!(\"invalid scan.startup.mode, accept earliest/latest/timestamp\")\n                    }\n                },\n            },\n            // We have record on this Nats Split, contains the last seen offset (seq id) or reply subject\n            // We do not use the seq id as start position anymore,\n            // but just let the reader load from durable consumer on broker.\n            start_position => start_position.to_owned(),\n        };\n\n        let mut config = consumer::pull::Config {\n            ..Default::default()\n        };\n        properties.set_config(&mut config)?;\n","sourceCodeStart":56,"sourceCodeEnd":92,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/source/nats/source/reader.rs#L56-L92","documentation":"The NATS source reader requires a startup timestamp when scan.startup.mode is set to 'timestamp' (or the legacy 'timestamp_millis'), but the property scan.startup.timestamp.millis was not provided. Without it there is no point in time from which to begin consuming, so the connector rejects the configuration.","triggerScenarios":"Creating a NATS source with `scan.startup.mode = 'timestamp'` (or 'timestamp_millis') while omitting `scan.startup.timestamp.millis` in the WITH clause.","commonSituations":"Copy-pasting a source DDL with timestamp startup mode from an example that had the millis field; renaming the property but forgetting to add it; assuming the current time is used as a default.","solutions":["Add `scan.startup.timestamp.millis = '<epoch_millis>'` to the source WITH clause.","If you want to consume from the beginning instead, set `scan.startup.mode = 'earliest'` and drop the timestamp mode.","Use `scan.startup.mode = 'latest'` to consume only new messages if no historical point is needed."],"exampleFix":"// before\nWITH (connector = 'nats', scan.startup.mode = 'timestamp')\n// after\nWITH (connector = 'nats', scan.startup.mode = 'timestamp', scan.startup.timestamp.millis = '1700000000000')","handlingStrategy":"validation","validationCode":"if mode == \"timestamp\" && !props.contains_key(\"scan.startup.timestamp.millis\") {\n    return Err(\"scan.startup.mode='timestamp' requires scan.startup.timestamp.millis\");\n}","typeGuard":"fn require_timestamp_millis(mode: &str, millis: Option<&str>) -> Option<u64> {\n    match mode {\n        \"timestamp\" | \"timestamp_millis\" => millis?.parse::<u64>().ok(),\n        _ => None,\n    }\n}","tryCatchPattern":"match source.create().await {\n    Err(e) if e.to_string().contains(\"scan.startup.timestamp.millis is required\") => fix_and_recreate_source(),\n    other => other,\n}","preventionTips":["Always pair scan.startup.mode='timestamp' with scan.startup.timestamp.millis in source templates.","Validate the full WITH clause before issuing CREATE SOURCE.","Use 'earliest'/'latest' modes when a specific point in time is not required."],"tags":["nats","config","source","validation"],"backgroundTag":"missing-required-config-field","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"}