{"record":{"id":"099b40c62fa6ea56","repo":"risingwavelabs/risingwave","slug":"sink-not-found","errorCode":null,"errorMessage":"Sink not found: {}","messagePattern":"Sink not found: (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/manager/iceberg_compaction/mod.rs","lineNumber":116,"sourceCode":"                    manifest_rewrite_sink_ids: HashSet::default(),\n                    manual_compaction_waiters: HashMap::default(),\n                })),\n                metadata_manager,\n                iceberg_compactor_manager,\n                compactor_streams_change_tx,\n                metrics,\n            }),\n            compactor_streams_change_rx,\n        )\n    }\n\n    async fn get_sink_param(&self, sink_id: SinkId) -> MetaResult<SinkParam> {\n        let prost_sink_catalog = self\n            .metadata_manager\n            .catalog_controller\n            .get_sink_by_id(sink_id)\n            .await?\n            .ok_or_else(|| anyhow!(\"Sink not found: {}\", sink_id))?;\n        let sink_catalog = SinkCatalog::from(prost_sink_catalog);\n        let param = SinkParam::try_from_sink_catalog(sink_catalog)?;\n        Ok(param)\n    }\n\n    async fn load_iceberg_config(&self, sink_id: SinkId) -> MetaResult<IcebergConfig> {\n        let sink_param = self.get_sink_param(sink_id).await?;\n        let iceberg_config = IcebergConfig::from_btreemap(sink_param.properties)?;\n        Ok(iceberg_config)\n    }\n\n    /// Clear the iceberg maintenance state of the sink aborted by\n    /// `try_abort_creating_streaming_job`, if any.\n    pub fn clear_maintenance_for_aborted_job(&self, abort_result: &AbortCreatingJobResult) {\n        for &sink_id in &abort_result.aborted_sink_ids {\n            self.clear_iceberg_maintenance_by_sink_id(sink_id);\n        }\n    }","sourceCodeStart":98,"sourceCodeEnd":134,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/manager/iceberg_compaction/mod.rs#L98-L134","documentation":"get_sink_param loads the sink catalog row by id via the catalog controller. When no sink exists with the given id, it fails with 'Sink not found: {id}' before Iceberg compaction configuration can be loaded for it.","triggerScenarios":"load_iceberg_config -> get_sink_param(sink_id) where get_sink_by_id returns None — the sink was dropped, never existed, or the id is stale (e.g. referenced from a leftover compaction/gc queue).","commonSituations":"Sink dropped while background GC/compaction still holds its id; typo'd sink id in a manual CALL; meta catalog restored from a backup lacking the sink; concurrent DROP SINK racing with a scheduled compaction.","solutions":["Confirm the sink exists: SELECT ... FROM rw_catalog.rw_sinks WHERE sink_id = <id>.","If the sink was dropped, ignore/清理 the stale queue entry; GC should skip missing sinks.","If calling manually, use the correct sink id (not table id or sink name).","Check for DROP SINK racing with scheduled compaction; re-trigger after catalog settles."],"exampleFix":"// before: wrong id passed (name instead of id)\nCALL rw_iceberg_compaction('my_iceberg_sink');\n// after: resolve id first\nSELECT sink_id FROM rw_catalog.rw_sinks WHERE name = 'my_iceberg_sink';\nCALL rw_iceberg_compaction(<sink_id>);","handlingStrategy":"validation","validationCode":"-- SQL: resolve and verify the sink id first\nSELECT sink_id FROM rw_catalog.rw_sinks WHERE sink_id = $1 OR name = $2;","typeGuard":null,"tryCatchPattern":"match load_iceberg_config(sink_id).await {\n    Err(e) if e.to_string().starts_with(\"Sink not found\") => {\n        // stale id: refresh sink registry and skip this sink in the batch\n        skip_sink(sink_id);\n    }\n    other => other?,\n}","preventionTips":["Look up sink_id from rw_catalog.rw_sinks by name before manual calls.","Purge queued GC/compaction entries when a sink is dropped.","Serialize DROP SINK with compaction scheduling to avoid stale references."],"tags":["rust","meta-service","iceberg","catalog","entity-lookup"],"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"}