risingwavelabs/risingwave · error · StreamExecutorError

iceberg pk-index writer {} received unexpected End in Normal

Error message

iceberg pk-index writer {} received unexpected End in Normal mode for task {}

What it means

In Normal mode the writer handles a compaction Begin barrier to enter the compaction flow, but a Phase::End context must be consumed while already in compaction mode, not Normal mode. Receiving an End while Normal means the writer missed the corresponding Begin (or already returned to Normal prematurely), so the compaction lifecycle is inconsistent.

Source

Thrown at src/stream/src/executor/iceberg_with_pk_index/writer.rs:367

                "iceberg pk-index writer {} received mismatched left/right barriers: left={:?}, right={:?}",
                self.sink_id,
                left,
                right
            );
        }
        Ok(())
    }

    fn compaction_begin(
        &self,
        barrier: &Barrier,
    ) -> StreamExecutorResult<Option<IcebergCompactionTaskId>> {
        let context = match barrier.iceberg_pk_index_compaction() {
            Some(context) if context.sink_id == self.sink_id => context,
            _ => return Ok(None),
        };
        if context.phase == Phase::End as i32 {
            bail!(
                "iceberg pk-index writer {} received unexpected End in Normal mode for task {}",
                self.sink_id,
                context.task_id
            );
        }
        if context.phase != Phase::Begin as i32 {
            bail!(
                "iceberg pk-index writer {} expected Begin context for task {}, got {:?}",
                self.sink_id,
                context.task_id,
                context.phase
            );
        }
        Ok(Some(context.task_id))
    }

    #[try_stream(ok = Message, error = StreamExecutorError)]
    async fn execute_inner(mut self) {

View on GitHub (pinned to 6469eb736d)

Solutions

Restart the fragment from the last checkpoint to restore a consistent compaction lifecycle.
Check meta's compaction scheduler for an End issued without a matching Begin.
Inspect logs for the missing Begin barrier (task_id in the error identifies the task).
Verify meta/stream node version compatibility.
Report with the task id if the scheduler reproducibly skips Begin.

Defensive patterns

Strategy: validation

Validate before calling

// reject End contexts when the writer is in Normal mode, before dispatch
if mode == Mode::Normal && ctx.phase == Phase::End as i32 {
    // ignore or log; do not feed to compaction_begin
}

Type guard

fn begin_allowed(ctx: &IcebergPkIndexCompactionContext) -> bool {
    ctx.phase == Phase::Begin as i32
}

Try / catch

match res {
    Err(e) if e.to_string().contains("unexpected End in Normal mode") => {
        error!("compaction lifecycle desync (missed Begin); restart from checkpoint");
    }
    other => other?,
}

Prevention

Guarantee Begin/End phase pairs are delivered in order and never split across recoveries.
Log every compaction phase barrier per task_id for auditing.
Pin matching meta and stream versions.

When it happens

Trigger: Raised in compaction_begin (called from execute_normal) when barrier.iceberg_pk_index_compaction() yields a context for this sink whose phase is Phase::End — i.e. the compaction-terminating barrier arrives while the writer is still in Normal mode for that task.

Common situations: The Begin barrier was lost or consumed incorrectly during recovery/replay; meta sent End without a preceding Begin after a task restart; version skew where an older writer misses the Begin-phase barrier type.

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/fb649a4c1d665bb0. Report an issue: GitHub.