{"record":{"id":"c2e8025204f55790","repo":"risingwavelabs/risingwave","slug":"stream-not-found-set-allow-create-stream-to","errorCode":null,"errorMessage":"stream {} not found, set `allow_create_stream` to true to create a stream","messagePattern":"stream (.+?) not found, set `allow_create_stream` to true to create a stream","errorType":"validation","errorClass":"ConnectorError","httpStatus":null,"severity":"error","filePath":"src/connector/src/connector_common/common.rs","lineNumber":1228,"sourceCode":"        &self,\n        jetstream: jetstream::Context,\n        stream_str: String,\n    ) -> ConnectorResult<jetstream::stream::Stream> {\n        let subjects: Vec<String> = self.subject.split(',').map(|s| s.to_owned()).collect();\n\n        // In `SourceEnumerator`, we may create a stream\n        // In `SourceReader`, the desired stream MUST exist\n        if let Ok(mut stream_instance) = jetstream.get_stream(&stream_str).await {\n            tracing::info!(\n                \"load existing nats stream ({:?}) with config {:?}\",\n                stream_str,\n                stream_instance.info().await?\n            );\n            return Ok(stream_instance);\n        }\n\n        if !self.allow_create_stream {\n            return Err(anyhow!(\n                \"stream {} not found, set `allow_create_stream` to true to create a stream\",\n                stream_str\n            )\n            .into());\n        }\n\n        let mut config = jetstream::stream::Config {\n            name: stream_str.clone(),\n            max_bytes: 1000000,\n            subjects,\n            ..Default::default()\n        };\n        if let Some(v) = self.max_bytes {\n            config.max_bytes = v;\n        }\n        if let Some(v) = self.max_messages {\n            config.max_messages = v;\n        }","sourceCodeStart":1210,"sourceCodeEnd":1246,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/connector_common/common.rs#L1210-L1246","documentation":"Thrown by `build_or_get_stream` in the NATS JetStream connector. When a NAT JetStream stream with the requested name cannot be fetched, and the source options have not set `allow_create_stream` to true, the connector refuses to auto-create the stream and returns this error. It protects users from unintentionally creating streams on the NATS server.","triggerScenarios":"Calling build_or_get_stream (e.g. from a SourceReader) where `jetstream.get_stream(stream_str)` fails because the stream does not exist on the NATS server, and the option `allow_create_stream` is not enabled in the source properties.","commonSituations":"Typo in the stream name in the CREATE SOURCE/SINK options; the stream was deleted or exists on a different NATS server/JetStream account; running a reader against a stream the enumerator was supposed to create but did not; forgetting the `allow_create_stream=true` option on a fresh NATS setup.","solutions":["Set the source option `allow_create_stream = true` so the connector may create the stream.","Verify the stream exists on the NATS server (e.g. `nats stream info <name>`) and fix any typo in the stream name option.","Point the connector at the correct NATS URL/account where the stream actually lives.","Pre-create the JetStream stream manually with the expected name and subjects."],"exampleFix":"// before\nCREATE SOURCE nats_src WITH (connector='nats', stream='orders', subject='orders.*');\n// after\nCREATE SOURCE nats_src WITH (connector='nats', stream='orders', subject='orders.*', allow_create_stream=true);","handlingStrategy":"validation","validationCode":"// before creating the source, check the stream exists or the flag is set\nlet stream_exists = jetstream.get_stream(&stream_name).await.is_ok();\nif !stream_exists && options.get(\"allow_create_stream\").map(|v| v != \"true\").unwrap_or(true) {\n    return Err(format!(\"stream {} not found; set allow_create_stream=true or create it first\", stream_name));\n}","typeGuard":null,"tryCatchPattern":"match source.create().await {\n    Err(e) if e.to_string().contains(\"set `allow_create_stream` to true\") => {\n        // surface a hint: either enable the flag or pre-create the JetStream stream\n    }\n    Err(e) => return Err(e),\n    Ok(src) => Ok(src),\n}","preventionTips":["Always set allow_create_stream=true when the stream may not pre-exist","Verify stream names with `nats stream ls` before creating sources","Use the same NATS URL/account for planning and reading"],"tags":["nats","jetstream","stream-not-found","config"],"backgroundTag":"resource-not-found","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"}