{"record":{"id":"0fff1adff7836a8f","repo":"risingwavelabs/risingwave","slug":"source-not-found-in-source-manager","errorCode":null,"errorMessage":"source {} not found in source manager","messagePattern":"source (.+?) not found in source manager","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/stream/source_manager.rs","lineNumber":601,"sourceCode":"            // Force a tick to re-discover splits\n            tracing::info!(\n                %source_id,\n                \"Triggering split re-discovery via force_tick\"\n            );\n            handle.force_tick().await.with_context(|| {\n                format!(\n                    \"failed to force tick for source {} after split reset\",\n                    source_id\n                )\n            })?;\n\n            tracing::info!(\n                %source_id,\n                \"Split reset completed - new splits will be assigned on next tick\"\n            );\n            Ok(())\n        } else {\n            Err(anyhow::anyhow!(\"source {} not found in source manager\", source_id).into())\n        }\n    }\n\n    /// Validate split offsets before injecting them.\n    /// Returns `Ok(applied_split_ids)` if all validations pass, otherwise returns an error.\n    ///\n    /// Validations performed:\n    /// 1. Source exists in source manager\n    /// 2. All requested split IDs exist in the source's current splits (runtime assignment)\n    pub async fn validate_inject_source_offsets(\n        &self,\n        source_id: SourceId,\n        split_offsets: &HashMap<String, String>,\n    ) -> MetaResult<Vec<String>> {\n        let (fragment_ids, env) = {\n            let core = self.core.lock().await;\n\n            // Check if source exists","sourceCodeStart":583,"sourceCodeEnd":619,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/stream/source_manager.rs#L583-L619","documentation":"reset_source_splits looks up the given source_id in the in-memory SourceManager; this error fires when the source is not currently tracked (its entry was absent when the reset branch checked). Split reset only applies to sources the manager actively manages.","triggerScenarios":"Calling the split-reset path (debug/internal API) for a source_id that is not registered — source already dropped, never created on this meta node, or removed during recovery.","commonSituations":"Operator/tooling resetting splits for a stale source id after a DROP SOURCE; pointing a test utility at the wrong meta node; source removed between lookup and reset.","solutions":["Confirm the source_id exists (SELECT from rw_sources / catalog)","Re-check you are connected to the correct cluster/meta node","Recreate or refresh the source so the manager registers it, then reset splits"],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":"-- verify the source exists before resetting splits\nSELECT id FROM rw_catalog.rw_sources WHERE id = <source_id>;","typeGuard":null,"tryCatchPattern":"match manager.reset_source_splits(source_id).await {\n    Err(e) if e.to_string().contains(\"not found in source manager\") => {\n        tracing::warn!(\"source {} already gone; skipping reset\", source_id);\n    }\n    r => r?,\n}","preventionTips":["Refresh source ids from the catalog before reset operations","Handle DROP SOURCE races in automation that resets splits"],"tags":["rust","meta","source-manager","splits"],"backgroundTag":"entity-not-found","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"}