{"record":{"id":"5f3ab6fd3abc6409","repo":"risingwavelabs/risingwave","slug":"upstream-fragment-not-found-in-initial-upstream","errorCode":null,"errorMessage":"upstream fragment {} not found in initial upstream actors","messagePattern":"upstream fragment (.+?) not found in initial upstream actors","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/stream/src/executor/upstream_sink_union.rs","lineNumber":136,"sourceCode":"pub struct UpstreamFragmentInfo {\n    pub upstream_fragment_id: FragmentId,\n    pub upstream_actors: Vec<PbActorInfo>,\n    pub merge_schema: Schema,\n    pub project_exprs: Vec<NonStrictExpression>,\n}\n\nimpl UpstreamFragmentInfo {\n    pub fn new(\n        upstream_fragment_id: FragmentId,\n        initial_upstream_actors: &HashMap<FragmentId, UpstreamActors>,\n        sink_output_schema: &[PbField],\n        project_exprs: &[PbExprNode],\n        error_report: impl EvalErrorReport + 'static,\n    ) -> StreamResult<Self> {\n        let actors = initial_upstream_actors\n            .get(&upstream_fragment_id)\n            .ok_or_else(|| {\n                anyhow::anyhow!(\n                    \"upstream fragment {} not found in initial upstream actors\",\n                    upstream_fragment_id\n                )\n            })?;\n        let merge_schema = sink_output_schema.iter().map(Field::from).collect();\n        let project_exprs = project_exprs\n            .iter()\n            .map(|e| build_non_strict_from_prost(e, error_report.clone()))\n            .try_collect()\n            .map_err(|err| anyhow::anyhow!(err))?;\n        Ok(Self {\n            upstream_fragment_id,\n            upstream_actors: actors.actors.clone(),\n            merge_schema,\n            project_exprs,\n        })\n    }\n}","sourceCodeStart":118,"sourceCodeEnd":154,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/executor/upstream_sink_union.rs#L118-L154","documentation":"Thrown in `UpstreamSinkUnion::new` when the `upstream_fragment_id` supplied by the meta service has no entry in the `initial_upstream_actors` map. The executor cannot build its merge inputs without knowing which upstream actors to read from. This indicates the fragment passed in was never provisioned in the actor's upstream fragment list, i.e. an inconsistency between the stream plan and the actor deployment info.","triggerScenarios":"Calling `UpstreamSinkUnion::new` with an `upstream_fragment_id` that is absent from the `initial_upstream_actors` HashMap handed to the constructor, typically during actor creation from a StreamFragmentGraph / actor deployment where the sink's upstream fragment was not recorded.","commonSituations":"Meta-node bugs where sink upstream fragments are pruned or not serialized into the actor proto; a stale or partially upgraded cluster where new executor code expects fragments older dispatch code never populated; hand-modified or corrupted fragment graph during recovery.","solutions":["Check that the sink node's upstream fragment IDs in the stream plan are present in the actor's `initial_upstream_actors` map (proto field) before dispatch.","Upgrade meta and stream nodes together; a version skew can cause the producer of `initial_upstream_actors` to omit new fragment types.","Re-create the affected materialization/sink so the fragment graph is regenerated and redeployed.","If reproducible, file a bug with the fragment graph: this is an internal plan-consistency violation."],"exampleFix":"// before\nlet actors = initial_upstream_actors.get(&upstream_fragment_id).ok_or_else(...)?;\n// after\n// Ensure the fragment id exists before constructing:\nassert!(\n    initial_upstream_actors.contains_key(&upstream_fragment_id),\n    \"upstream fragment {upstream_fragment_id} missing from initial_upstream_actors\"\n);","handlingStrategy":"try-catch","validationCode":"// before dispatching the actor\nassert!(\n    initial_upstream_actors.contains_key(&upstream_fragment_id),\n    \"upstream fragment {upstream_fragment_id} missing\"\n);","typeGuard":"fn fragment_exists(map: &HashMap<u32, _>, id: u32) -> bool { map.contains_key(&id) }","tryCatchPattern":"// Rust: this is an anyhow error surfaced as StreamResult; handle at actor-build level\nmatch UpstreamSinkUnion::new(...) {\n    Ok(e) => e,\n    Err(err) if err.to_string().contains(\"not found in initial upstream actors\") => {\n        // fail actor creation and report plan inconsistency to meta\n        return Err(err);\n    }\n}","preventionTips":["Keep meta and stream compute versions aligned","Validate the fragment graph (sink upstream fragments present in actor deployment) before dispatch","Integration-test sink creation after frontend planner changes"],"tags":["streaming","actor-deployment","internal-invariant"],"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-23T08:17:48.524Z"}