{"record":{"id":"ff7a4f343440e93e","repo":"risingwavelabs/risingwave","slug":"no-state-table-id-in-sink","errorCode":null,"errorMessage":"no state table id in sink: {}","messagePattern":"no state table id in sink: (.+?)","errorType":"exception","errorClass":"anyhow::Error","httpStatus":null,"severity":"error","filePath":"src/meta/src/manager/sink_coordination/manager.rs","lineNumber":90,"sourceCode":"\n#[derive(Clone)]\npub struct SinkCoordinatorManager {\n    request_tx: mpsc::Sender<ManagerRequest>,\n}\nfn new_committed_epoch_subscriber(\n    hummock_manager: HummockManagerRef,\n    metadata_manager: MetadataManager,\n) -> SinkCommittedEpochSubscriber {\n    Arc::new(move |sink_id| {\n        let hummock_manager = hummock_manager.clone();\n        let metadata_manager = metadata_manager.clone();\n        async move {\n            let state_table_ids = metadata_manager\n                .get_sink_state_table_ids(sink_id)\n                .await\n                .map_err(SinkError::from)?;\n            let Some(table_id) = state_table_ids.first() else {\n                return Err(anyhow!(\"no state table id in sink: {}\", sink_id).into());\n            };\n            hummock_manager\n                .subscribe_table_committed_epoch(*table_id)\n                .await\n                .map_err(SinkError::from)\n        }\n        .boxed()\n    })\n}\n\nimpl SinkCoordinatorManager {\n    pub fn start_worker(\n        db: DatabaseConnection,\n        hummock_manager: HummockManagerRef,\n        metadata_manager: MetadataManager,\n        iceberg_compact_stat_sender: UnboundedSender<IcebergSinkCompactionUpdate>,\n        await_tree_reg: await_tree::Registry,\n    ) -> (Self, (JoinHandle<()>, Sender<()>)) {","sourceCodeStart":72,"sourceCodeEnd":108,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/manager/sink_coordination/manager.rs#L72-L108","documentation":"When a committed-epoch subscriber is created for a sink, the meta node looks up the sink's state table IDs and requires at least one to subscribe its committed epoch in Hummock. A sink with no state table cannot be coordinated for commit epochs, so the subscriber task returns this error and the sink coordination fails to start.","triggerScenarios":"`start_worker` spawns `new_committed_epoch_subscriber`; `get_sink_state_table_ids(sink_id)` returns an empty list for the given sink id.","commonSituations":"Sink created without a state table (corrupt/incomplete catalog state, e.g. interrupted DDL or migration from an older format); querying a sink id that was dropped concurrently; append-only sinks of a legacy version that lacked state tables.","solutions":["Inspect the sink catalog (`rw_catalog.rw_sinks` / meta metadata) to confirm the sink has an associated state (internal) table; recreate the sink if the state table is missing.","If the sink id belongs to a dropped sink, guard against coordinating already-removed sinks and clean up stale coordination state.","Ensure DDL completes atomically — re-run the sink creation if a previous `CREATE SINK` failed midway.","Upgrade if migrating from a version where some sink types had no state table."],"exampleFix":"// before\nlet Some(table_id) = state_table_ids.first() else {\n    return Err(anyhow!(\"no state table id in sink: {}\", sink_id).into());\n};\n// after\nlet Some(table_id) = state_table_ids.first() else {\n    return Err(SinkError::from(anyhow!(\n        \"no state table id in sink: {}; sink may be corrupted or dropped, recreate the sink\",\n        sink_id\n    )));\n};","handlingStrategy":"validation","validationCode":"// Before starting sink coordination, verify the sink has a state table\nlet state_table_ids = metadata_manager.get_sink_state_table_ids(sink_id).await?;\nif state_table_ids.is_empty() {\n    return Err(anyhow!(\"sink {} has no state table; recreate the sink\", sink_id));\n}","typeGuard":null,"tryCatchPattern":"match new_committed_epoch_subscriber(sink_id).await {\n    Err(e) if e.to_string().contains(\"no state table id\") => {\n        tracing::error!(\"sink {} is missing its state table; check catalog or recreate\", sink_id);\n    }\n    other => { /* normal handling */ }\n}","preventionTips":["Verify CREATE SINK completed fully before starting coordination","Clean up coordination state for dropped sinks","Check the catalog for internal tables after failed DDL or migrations"],"tags":["rust","meta","sink","catalog","state-table"],"backgroundTag":"empty-result-set","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"}