{"record":{"id":"e9ef5ce4add22019","repo":"risingwavelabs/risingwave","slug":"unsupported-source","errorCode":null,"errorMessage":"Unsupported source: {:?}","messagePattern":"Unsupported source: (.+?)","errorType":"exception","errorClass":"ConnectorError","httpStatus":null,"severity":"error","filePath":"src/connector/src/source/reader/reader.rs","lineNumber":119,"sourceCode":"                    &prop.s3_properties,\n                    prop.assume_role,\n                    prop.fs_common.compression_format,\n                )?;\n                Ok(build_opendal_fs_list_stream(lister, list_interval_sec))\n            }\n            ConnectorProperties::Azblob(prop) => {\n                list_interval_sec = get_list_interval_sec(prop.fs_common.refresh_interval_sec);\n                let lister: OpendalEnumerator<OpendalAzblob> =\n                    OpendalEnumerator::new_azblob_source(*prop)?;\n                Ok(build_opendal_fs_list_stream(lister, list_interval_sec))\n            }\n            ConnectorProperties::PosixFs(prop) => {\n                list_interval_sec = get_list_interval_sec(prop.fs_common.refresh_interval_sec);\n                let lister: OpendalEnumerator<OpendalPosixFs> =\n                    OpendalEnumerator::new_posix_fs_source(*prop)?;\n                Ok(build_opendal_fs_list_stream(lister, list_interval_sec))\n            }\n            other => bail!(\"Unsupported source: {:?}\", other),\n        }\n    }\n\n    /// Refer to `WaitCheckpointWorker` for more details.\n    pub async fn create_wait_checkpoint_task(&self) -> ConnectorResult<Option<WaitCheckpointTask>> {\n        Ok(match &self.config {\n            ConnectorProperties::PostgresCdc(_) => Some(WaitCheckpointTask::CommitCdcOffset(None)),\n            ConnectorProperties::GooglePubsub(prop) => Some(WaitCheckpointTask::AckPubsubMessage(\n                prop.subscription_client().await?,\n                vec![],\n            )),\n            ConnectorProperties::Nats(prop) => {\n                match prop.nats_properties_consumer.get_ack_policy()? {\n                    a @ AckPolicy::Explicit | a @ AckPolicy::All => {\n                        Some(WaitCheckpointTask::AckNatsJetStream(\n                            prop.common.build_context().await?,\n                            vec![],\n                            a,","sourceCodeStart":101,"sourceCodeEnd":137,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/source/reader/reader.rs#L101-L137","documentation":"The connector enumerator factory only recognizes a fixed set of source configs; get_source_list matches on the ConnectorProperties variant and bails with 'Unsupported source' for any variant that has no list-stream implementation. This indicates the caller constructed a source config the enumerator cannot enumerate.","triggerScenarios":"Calling get_source_list on a ConnectorConnector/properties enum holding a source type without enumerator support (e.g. a CDC or push-based source passed into the FS enumeration path), typically due to a wiring/config mistake rather than user input.","commonSituations":"Adding a new connector without updating the match in reader.rs, or misconfiguring a source so the wrong ConnectorProperties variant is built; also binary/feature builds where some enumerators are compiled out.","solutions":["Check which source type was configured and use one supported by the list/enumerator path (S3, GCS, POSIX FS, etc.)","Update the match statement in get_source_list to handle the new variant if you are adding a connector","Rebuild with the relevant `source-*` cargo feature enabled so the variant has an implementation"],"exampleFix":"// before\nother => bail!(\"Unsupported source: {:?}\", other),\n// after\nConnectorProperties::S3(prop) => { /* build S3 list stream */ }\nother => bail!(\"Unsupported source: {:?}\", other),","handlingStrategy":"type-guard","validationCode":"const supportedListSources = ['s3','gcs','posix_fs','opendal'];\nfunction validateEnumerableSource(sourceType) {\n  if (!supportedListSources.includes(sourceType)) throw new Error(`source ${sourceType} has no list/enumerator support`);\n}","typeGuard":"const isEnumerableProps = (p) => ['S3','Gcs','PosixFs','Opendal'].some(v => p.variant === v);","tryCatchPattern":"match get_source_list(props) { Err(e) if String(e).contains(\"Unsupported source\") => pick_supported_enumerator_or_static_assignment(), r => r }","preventionTips":["Only wire sources with enumerator implementations into the list-based discovery path","When adding a connector, update every match in reader.rs","Check that the binary build includes the enumerator implementations for the configured source"],"tags":["connector","unsupported","enum"],"backgroundTag":"unsupported-enum-value","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}