risingwavelabs/risingwave · error · SinkError

pk-index merger flush before start_seed

Error message

pk-index merger flush before start_seed

What it means

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.

Solutions

  1. Ensure start_seed is invoked before any flush_inner call in the handler's lifecycle.
  2. Check logs for a failed/aborted start_seed (e.g. the epoch-wait error) that left the handler Unseeded.
  3. If flushing must be optional pre-seed, guard the caller to skip flush when inner is Unseeded.
  4. Report as a bug with the executor's message sequence if the normal barrier flow triggers it.

Example fix

// before
if matches!(self.inner, HandlerInner::Unseeded) {
    self.await_ready().await?;
}
// after
if matches!(self.inner, HandlerInner::Unseeded) {
    return Ok(()); // nothing to flush before seeding
}
Defensive patterns

Strategy: validation

Validate before calling

// before flushing, confirm the handler is seeded
if matches!(handler.state(), HandlerState::Unseeded) {
    // skip flush or trigger start_seed first
}

Type guard

fn is_seeded(inner: &HandlerInner) -> bool {
    matches!(inner, HandlerInner::Ready(_) | HandlerInner::Seeding(_))
}

Try / catch

match flush_inner(...).await {
    Err(SinkError::Iceberg(e)) if format!("{:e}", e).contains("flush before start_seed") => {
        warn!("handler not seeded yet; skipping flush");
    }
    other => other?,
}

Prevention

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/aa653b11bc541eb6. Report an issue: GitHub.

Appendix: source

Thrown at src/stream/src/executor/iceberg_with_pk_index/position_delete_handler_impl.rs:139

    /// Await the background seed, transitioning `Seeding -> Ready`. Errors if called before
    /// `start_seed`.
    async fn await_ready(&mut self) -> SinkResult<&mut SeededState> {
        if matches!(self.inner, HandlerInner::Seeding(_)) {
            // Take the handle out before awaiting so an errored seed leaves `Unseeded` (a clean error
            // on any later flush) rather than a `Seeding` holding a consumed `JoinHandle` that would
            // panic if re-polled.
            let HandlerInner::Seeding(handle) =
                std::mem::replace(&mut self.inner, HandlerInner::Unseeded)
            else {
                unreachable!("guarded by matches! above");
            };
            let state = handle.await.map_err(|e| SinkError::Iceberg(anyhow!(e)))??;
            self.inner = HandlerInner::Ready(state);
        }
        match &mut self.inner {
            HandlerInner::Ready(state) => Ok(state),
            HandlerInner::Unseeded => Err(SinkError::Iceberg(anyhow!(
                "pk-index merger flush before start_seed"
            ))),
            HandlerInner::Seeding(_) => unreachable!("just transitioned out of Seeding"),
        }
    }

    async fn flush_inner(&mut self) -> SinkResult<Option<SinkMetadata>> {
        if self.pending.is_empty() {
            return Ok(None);
        }

        let config = self.config.clone();
        let pending = std::mem::take(&mut self.pending);
        let state = self.await_ready().await?;

        let mut delete_files = Vec::with_capacity(pending.len());
        let mut overwrite_files = Vec::new();

View on GitHub (pinned to 6469eb736d)