{"record":{"id":"55a4888b636a69b5","repo":"risingwavelabs/risingwave","slug":"sink-fragment-not-found-for-sink-id","errorCode":null,"errorMessage":"sink fragment not found for sink id {}","messagePattern":"sink fragment not found for sink id (.+?)","errorType":"exception","errorClass":"MetaError","httpStatus":null,"severity":"error","filePath":"src/meta/src/controller/fragment.rs","lineNumber":2016,"sourceCode":"    ) -> MetaResult<Vec<UpstreamSinkInfo>>\n    where\n        C: ConnectionTrait,\n    {\n        let incoming_sinks = Sink::find()\n            .filter(sink::Column::TargetTable.eq(target_table.id))\n            .all(txn)\n            .await?;\n\n        let sink_ids = incoming_sinks.iter().map(|s| s.sink_id).collect_vec();\n        let sink_fragment_ids = get_sink_fragment_by_ids(txn, sink_ids).await?;\n\n        let mut upstream_sink_infos = Vec::with_capacity(incoming_sinks.len());\n        for sink in &incoming_sinks {\n            let sink_fragment_id =\n                sink_fragment_ids\n                    .get(&sink.sink_id)\n                    .cloned()\n                    .ok_or(anyhow::anyhow!(\n                        \"sink fragment not found for sink id {}\",\n                        sink.sink_id\n                    ))?;\n            let upstream_info = build_upstream_sink_info(\n                sink.sink_id,\n                sink.original_target_columns\n                    .as_ref()\n                    .map(|cols| cols.to_protobuf())\n                    .unwrap_or_default(),\n                sink_fragment_id,\n                target_table,\n                target_fragment_id,\n            )?;\n            upstream_sink_infos.push(upstream_info);\n        }\n\n        Ok(upstream_sink_infos)\n    }","sourceCodeStart":1998,"sourceCodeEnd":2034,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/controller/fragment.rs#L1998-L2034","documentation":"When building upstream sink info for a sink (during sink into-table / sink upstream resolution), the code maps each incoming sink's sink_id to its fragment id via sink_fragment_ids; a missing mapping returns 'sink fragment not found for sink id {}'. The sink's fragment was never registered or was removed while its downstream sink still references it.","triggerScenarios":"Creating a sink that reads from a table with sinks, when sink_fragment_ids (built from the table's fragment metadata) lacks an entry for one of the incoming sink_ids — e.g. the sink fragment was not yet persisted, or the sink was dropped concurrently.","commonSituations":"Chained sinks (sink into table, then sink from that table) created in quick succession; concurrent DROP SINK while creating the downstream sink; metadata from a failed/partial sink creation.","solutions":["Confirm the upstream sink exists and is fully created (SHOW SINKS) before creating the dependent sink","Retry the sink creation; a concurrent DROP SINK race usually resolves on retry","Check the table_fragments metadata for the sink fragment id mapping in the meta DB","If reproducible without concurrency, file a RisingWave issue with the sink id and job id"],"exampleFix":"// before\nlet sink_fragment_id = sink_fragment_ids.get(&sink.sink_id).cloned().ok_or(anyhow!(\"sink fragment not found for sink id {}\", sink.sink_id))?;\n// after\nlet sink_fragment_id = sink_fragment_ids.get(&sink.sink_id)\n    .cloned()\n    .ok_or_else(|| MetaError::catalog(rw_error::catalog::error::SinkNotFound::new(sink.sink_id).into()))?;","handlingStrategy":"validation","validationCode":"// before creating a chained sink, verify the upstream sink fragment exists\nif !sink_fragment_ids.contains_key(&upstream_sink_id) { return Err(sink_not_ready(upstream_sink_id)); }","typeGuard":"fn sink_fragment_registered(map: &HashMap<SinkId, FragmentId>, id: &SinkId) -> bool { map.contains_key(id) }","tryCatchPattern":"match create_sink(params).await { Err(e) if e.to_string().contains(\"sink fragment not found\") => retry_with_backoff(params), other => other }","preventionTips":["Wait for the upstream sink to be fully created before building dependent sinks","Serialize chained sink DDL to avoid drop/create races","Verify sink fragment mappings in table_fragments after creation"],"tags":["sink","fragment","not-found","metadata"],"backgroundTag":"record-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"}