{"record":{"id":"aa653b11bc541eb6","repo":"risingwavelabs/risingwave","slug":"pk-index-merger-flush-before-start-seed","errorCode":null,"errorMessage":"pk-index merger flush before start_seed","messagePattern":"pk-index merger flush before start_seed","errorType":"exception","errorClass":"SinkError","httpStatus":null,"severity":"error","filePath":"src/stream/src/executor/iceberg_with_pk_index/position_delete_handler_impl.rs","lineNumber":139,"sourceCode":"\n    /// Await the background seed, transitioning `Seeding -> Ready`. Errors if called before\n    /// `start_seed`.\n    async fn await_ready(&mut self) -> SinkResult<&mut SeededState> {\n        if matches!(self.inner, HandlerInner::Seeding(_)) {\n            // Take the handle out before awaiting so an errored seed leaves `Unseeded` (a clean error\n            // on any later flush) rather than a `Seeding` holding a consumed `JoinHandle` that would\n            // panic if re-polled.\n            let HandlerInner::Seeding(handle) =\n                std::mem::replace(&mut self.inner, HandlerInner::Unseeded)\n            else {\n                unreachable!(\"guarded by matches! above\");\n            };\n            let state = handle.await.map_err(|e| SinkError::Iceberg(anyhow!(e)))??;\n            self.inner = HandlerInner::Ready(state);\n        }\n        match &mut self.inner {\n            HandlerInner::Ready(state) => Ok(state),\n            HandlerInner::Unseeded => Err(SinkError::Iceberg(anyhow!(\n                \"pk-index merger flush before start_seed\"\n            ))),\n            HandlerInner::Seeding(_) => unreachable!(\"just transitioned out of Seeding\"),\n        }\n    }\n\n    async fn flush_inner(&mut self) -> SinkResult<Option<SinkMetadata>> {\n        if self.pending.is_empty() {\n            return Ok(None);\n        }\n\n        let config = self.config.clone();\n        let pending = std::mem::take(&mut self.pending);\n        let state = self.await_ready().await?;\n\n        let mut delete_files = Vec::with_capacity(pending.len());\n        let mut overwrite_files = Vec::new();\n","sourceCodeStart":121,"sourceCodeEnd":157,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/executor/iceberg_with_pk_index/position_delete_handler_impl.rs#L121-L157","documentation":"await_ready resolves the position-delete handler's inner state; it only exists once the handler transitions out of Unseeded via start_seed. This error means a flush was attempted while the merger's staging was never seeded, i.e. the flush path ran before start_seed populated HandlerInner::Ready. It is an internal lifecycle-ordering violation: flush_inner should never be called in Unseeded state.","triggerScenarios":"Raised in await_ready (called by flush_inner) when self.inner is HandlerInner::Unseeded — a flush request arrives before start_seed has waited for the sink commit epoch and loaded/seeded the table.","commonSituations":"A code path invoking flush during initial bootstrap before the merger finished waiting for the sink's commit epoch; recovery/replay invoking flush earlier than expected; a race where flush is triggered by backpressure before seeding completes.","solutions":["Ensure start_seed is invoked before any flush_inner call in the handler's lifecycle.","Check logs for a failed/aborted start_seed (e.g. the epoch-wait error) that left the handler Unseeded.","If flushing must be optional pre-seed, guard the caller to skip flush when inner is Unseeded.","Report as a bug with the executor's message sequence if the normal barrier flow triggers it."],"exampleFix":"// before\nif matches!(self.inner, HandlerInner::Unseeded) {\n    self.await_ready().await?;\n}\n// after\nif matches!(self.inner, HandlerInner::Unseeded) {\n    return Ok(()); // nothing to flush before seeding\n}","handlingStrategy":"validation","validationCode":"// before flushing, confirm the handler is seeded\nif matches!(handler.state(), HandlerState::Unseeded) {\n    // skip flush or trigger start_seed first\n}","typeGuard":"fn is_seeded(inner: &HandlerInner) -> bool {\n    matches!(inner, HandlerInner::Ready(_) | HandlerInner::Seeding(_))\n}","tryCatchPattern":"match flush_inner(...).await {\n    Err(SinkError::Iceberg(e)) if format!(\"{:e}\", e).contains(\"flush before start_seed\") => {\n        warn!(\"handler not seeded yet; skipping flush\");\n    }\n    other => other?,\n}","preventionTips":["Always run start_seed before enabling the flush path.","Log handler state transitions (Unseeded → Seeding → Ready) for diagnosis.","Treat this error as an ordering bug: audit any code path that can flush during bootstrap."],"tags":["iceberg","state-machine","lifecycle","internal-invariant","streaming-sink"],"backgroundTag":"invalid-state-transition","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}