{"record":{"id":"41c88be7a60e158e","repo":"risingwavelabs/risingwave","slug":"wait-for-pk-index-sink-commit-epoch-before-seeding","errorCode":null,"errorMessage":"wait for pk-index sink commit epoch before seeding merger staging","messagePattern":"wait for pk-index sink commit epoch before seeding merger staging","errorType":"exception","errorClass":"SinkError","httpStatus":null,"severity":"error","filePath":"src/stream/src/executor/iceberg_with_pk_index/position_delete_handler_impl.rs","lineNumber":257,"sourceCode":"            std::mem::replace(&mut self.inner, HandlerInner::Unseeded)\n        {\n            handle.abort();\n        }\n\n        let config = self.config.clone();\n        let actor_id = self.actor_id;\n        let vnode_bitmap = self.vnode_bitmap.clone();\n        let sink_id = self.sink_id;\n        let meta_client = self.meta_client.clone();\n        self.inner = HandlerInner::Seeding(tokio::spawn(async move {\n            let result: SinkResult<SeededState> = async move {\n                // 1. Block until meta has committed through `wait_epoch`; get the committed snapshot\n                //    lower bound.\n                let expected_snapshot = meta_client\n                    .wait_iceberg_pk_index_sink_epoch(sink_id, wait_epoch)\n                    .await\n                    .map_err(|e| {\n                        SinkError::Iceberg(anyhow!(e).context(\n                            \"wait for pk-index sink commit epoch before seeding merger staging\",\n                        ))\n                    })?;\n\n                // 2. Load the table, retrying until the catalog reflects at least `expected_snapshot`.\n                let table = load_table_at_least(&config, expected_snapshot).await?;\n\n                // 3. Derive per-table state + seed staging (shard-filtered).\n                let location_generator = DefaultLocationGenerator::new(table.metadata())?;\n                let uuid_suffix = Uuid::now_v7();\n                let puffin_file_name_generator = DefaultFileNameGenerator::new(\n                    actor_id.to_string(),\n                    Some(format!(\"delvec-{}\", uuid_suffix)),\n                    DataFileFormat::Puffin,\n                );\n                let parquet_file_name_generator = DefaultFileNameGenerator::new(\n                    actor_id.to_string(),\n                    Some(format!(\"pos-del-{}\", uuid_suffix)),","sourceCodeStart":239,"sourceCodeEnd":275,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/executor/iceberg_with_pk_index/position_delete_handler_impl.rs#L239-L275","documentation":"start_seed must first wait for the pk-index sink to commit through the requested epoch via meta_client.wait_iceberg_pk_index_sink_epoch before it can seed the merger's staging area from the committed Iceberg snapshot. When that meta-client wait fails, the error is wrapped with this context to indicate the merger cannot determine its expected snapshot lower bound. Seeding from a not-yet-committed epoch would produce an incomplete or corrupt staging state.","triggerScenarios":"Raised in start_seed when wait_iceberg_pk_index_sink_epoch(sink_id, wait_epoch) returns an error — e.g. the sink never commits the requested epoch (stalled/failed sink), the sink_id is unknown to the meta service, or the RPC to meta fails/times out.","commonSituations":"Pk-index sink stuck or failed so its commit epoch never advances; sink removed/reconfigured while the merger still waits on it; meta service unavailability; using an old checkpoint after a version upgrade where the sink no longer exists.","solutions":["Check the pk-index sink's health and confirm it is committing epochs (sink logs/metrics).","Verify the sink_id recorded by the merger still exists in the meta service.","Check meta service connectivity and RPC error details in the wrapped cause.","Restart/recover the merger after the sink has advanced past wait_epoch.","If the sink is permanently gone, resync/rebuild the merger state or recreate the pipeline."],"exampleFix":null,"handlingStrategy":"retry","validationCode":"// operator pre-check: confirm the sink has committed up to the epoch\nlet sink_epoch = meta_client.get_iceberg_pk_index_sink_epoch(sink_id).await?;\nif sink_epoch < wait_epoch {\n    eprintln!(\"sink at epoch {sink_epoch} has not reached {wait_epoch}; defer seeding\");\n}","typeGuard":null,"tryCatchPattern":"match start_seed(...).await {\n    Err(SinkError::Iceberg(e)) if e.root_cause().to_string().contains(\"wait for pk-index sink commit epoch\") => {\n        warn!(\"sink epoch not committed yet; will retry after sink advances\");\n        // schedule retry with backoff\n    }\n    other => other?,\n}","preventionTips":["Monitor the pk-index sink's commit epoch progress and alert on stalls.","Verify sink_id validity after job restarts or config changes.","Check meta service health before starting the merger.","Avoid recreating sinks with new ids while mergers hold old sink ids."],"tags":["iceberg","epoch","meta-service","rpc","streaming-sink"],"backgroundTag":"database-query-failed","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}