risingwavelabs/risingwave · error · SinkError

wait for pk-index sink commit epoch before seeding merger st

Error message

wait for pk-index sink commit epoch before seeding merger staging

What it means

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.

Source

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

            std::mem::replace(&mut self.inner, HandlerInner::Unseeded)
        {
            handle.abort();
        }

        let config = self.config.clone();
        let actor_id = self.actor_id;
        let vnode_bitmap = self.vnode_bitmap.clone();
        let sink_id = self.sink_id;
        let meta_client = self.meta_client.clone();
        self.inner = HandlerInner::Seeding(tokio::spawn(async move {
            let result: SinkResult<SeededState> = async move {
                // 1. Block until meta has committed through `wait_epoch`; get the committed snapshot
                //    lower bound.
                let expected_snapshot = meta_client
                    .wait_iceberg_pk_index_sink_epoch(sink_id, wait_epoch)
                    .await
                    .map_err(|e| {
                        SinkError::Iceberg(anyhow!(e).context(
                            "wait for pk-index sink commit epoch before seeding merger staging",
                        ))
                    })?;

                // 2. Load the table, retrying until the catalog reflects at least `expected_snapshot`.
                let table = load_table_at_least(&config, expected_snapshot).await?;

                // 3. Derive per-table state + seed staging (shard-filtered).
                let location_generator = DefaultLocationGenerator::new(table.metadata())?;
                let uuid_suffix = Uuid::now_v7();
                let puffin_file_name_generator = DefaultFileNameGenerator::new(
                    actor_id.to_string(),
                    Some(format!("delvec-{}", uuid_suffix)),
                    DataFileFormat::Puffin,
                );
                let parquet_file_name_generator = DefaultFileNameGenerator::new(
                    actor_id.to_string(),
                    Some(format!("pos-del-{}", uuid_suffix)),

View on GitHub (pinned to 6469eb736d)

Solutions

1. Check the pk-index sink's health and confirm it is committing epochs (sink logs/metrics).
2. Verify the sink_id recorded by the merger still exists in the meta service.
3. Check meta service connectivity and RPC error details in the wrapped cause.
4. Restart/recover the merger after the sink has advanced past wait_epoch.
5. If the sink is permanently gone, resync/rebuild the merger state or recreate the pipeline.

// operator pre-check: confirm the sink has committed up to the epoch
let sink_epoch = meta_client.get_iceberg_pk_index_sink_epoch(sink_id).await?;
if sink_epoch < wait_epoch {
    eprintln!("sink at epoch {sink_epoch} has not reached {wait_epoch}; defer seeding");
}

match start_seed(...).await {
    Err(SinkError::Iceberg(e)) if e.root_cause().to_string().contains("wait for pk-index sink commit epoch") => {
        warn!("sink epoch not committed yet; will retry after sink advances");
        // schedule retry with backoff
    }
    other => other?,
}

When it happens

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

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

Understand the failure class

Background: Database query failed: Internal Server Error 500s wrapping SQL, Prisma, and connection failures — what to check first — this error's family across 16 libraries.

Related errors

• Iceberg branch {} disappeared after manifest rewrite for sin
• Manual iceberg compaction task {} for sink {} failed: {}
• No iceberg compactor available
• Manual iceberg compaction waiter dropped unexpectedly for si
• Sink not found: {}

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