{"record":{"id":"66bd807dc5a2d34e","repo":"risingwavelabs/risingwave","slug":"iceberg-pk-index-sink-coordinator-for-sink-is-n","errorCode":null,"errorMessage":"iceberg pk-index sink coordinator for sink {} is not registered","messagePattern":"iceberg pk-index sink coordinator for sink (.+?) is not registered","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/manager/iceberg_pk_index_sink/manager.rs","lineNumber":182,"sourceCode":"            }\n        }\n    }\n\n    /// Drop every coordinator. Used at recovery time.\n    pub fn reset(&self) {\n        let mut coordinators = self.inner.coordinators.write();\n        coordinators.clear();\n        self.inner.committed_epochs.clear();\n    }\n\n    fn coordinator(&self, sink_id: SinkId) -> anyhow::Result<CoordinatorRef> {\n        self.inner\n            .coordinators\n            .read()\n            .get(&sink_id)\n            .map(|(_pg, coord)| coord.clone())\n            .ok_or_else(|| {\n                anyhow!(\n                    \"iceberg pk-index sink coordinator for sink {} is not registered\",\n                    sink_id\n                )\n            })\n    }\n\n    fn partial_graph_of(&self, sink_id: SinkId) -> anyhow::Result<PartialGraphId> {\n        self.inner\n            .coordinators\n            .read()\n            .get(&sink_id)\n            .map(|(pg, _coord)| *pg)\n            .ok_or_else(|| {\n                anyhow!(\n                    \"iceberg pk-index sink coordinator for sink {} is not registered\",\n                    sink_id\n                )\n            })","sourceCodeStart":164,"sourceCodeEnd":200,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/manager/iceberg_pk_index_sink/manager.rs#L164-L200","documentation":"IcebergPkIndexSinkManager::coordinator looks up the in-memory coordinator registered for a given sink id. If the sink id has no registered coordinator (never registered, already dropped, or lost after a meta restart), it returns this anyhow error. It surfaces on the pre_commit_epoch, commit_epoch and wait_epoch RPC paths.","triggerScenarios":"pre_commit_epoch/commit_epoch/wait_epoch invoked with a sink_id absent from self.inner.coordinators — e.g. the RPC hit a meta node that did not host the coordinator, the sink was dropped, or the id is stale after a meta failover.","commonSituations":"Meta node failover/restart losing in-memory coordinator state while workers still reference the sink; sink deleted while a commit RPC is in flight; non-leader meta node receiving the RPC.","solutions":["Check whether the sink still exists in the catalog; if deleted, stop sending commits for it.","Re-create/re-register the coordinator by restarting the sink worker so the manager rebuilds its entry, then retry.","Verify meta single-active/leader routing so commit RPCs reach the meta node that owns the coordinator.","If the coordinator was lost on meta restart, report persistent loss as a durability bug and recreate the sink."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":"// caller-side check before committing\nif !manager_has_coordinator(sink_id) {\n    return Err(anyhow!(\"sink {} coordinator missing; re-register first\", sink_id));\n}","typeGuard":null,"tryCatchPattern":"match manager.coordinator(sink_id) {\n    Ok(coord) => coord.commit_epoch(epoch).await?,\n    Err(e) if e.to_string().contains(\"not registered\") => {\n        // coordinator lost (failover/deleted sink): re-create or skip\n        re_register_sink(sink_id).await?;\n    }\n    Err(e) => return Err(e),\n}","preventionTips":["Always route sink commit RPCs to the meta leader that owns the coordinator.","Check sink existence in the catalog before committing.","After meta restarts, wait for worker re-registration before issuing commits.","Track coordinator registration/unregistration in logs for auditing."],"tags":["iceberg","meta","state"],"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-14T16:17:12.679Z"}