risingwavelabs/risingwave · error · StreamExecutorError

iceberg pk-index writer {} expected Begin context for task {

Error message

iceberg pk-index writer {} expected Begin context for task {}, got {:?}

What it means

When compaction_begin receives a barrier carrying a compaction context for this sink in Normal mode, that context must be Phase::Begin (entering the compaction flow). Any other phase (Apply/End) indicates a lifecycle inconsistency: the writer is being asked to start a phase that can only occur mid-compaction.

Source

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

    }

    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) {
        let mut input = self.input.take().unwrap().execute();
        let mut resolver_input = self.resolver_input.take().unwrap().execute();

        // Consume the first barrier.
        let barrier = expect_first_barrier(&mut input).await?;
        let remap_first = expect_first_barrier(&mut resolver_input).await?;
        self.validate_aligned_barriers(&barrier, &remap_first)?;

View on GitHub (pinned to 6469eb736d)

Solutions

  1. Identify the observed phase (in the error) vs the expected Begin and trace which barrier was missed.
  2. Restart the fragment from the last checkpoint to reset the compaction state machine.
  3. Audit meta's compaction phase scheduling for this task_id to ensure Begin precedes Apply/End.
  4. Check for actor migration or scale changes that dropped the Begin barrier.
  5. File a bug with the task id and phase if reproducible.
Defensive patterns

Strategy: validation

Validate before calling

// only dispatch Begin-phase contexts into compaction_begin in Normal mode
if ctx.phase == Phase::Begin as i32 {
    writer.compaction_begin(barrier, task).await?;
}

Type guard

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

Try / catch

match res {
    Err(e) if e.to_string().contains("expected Begin context") => {
        error!("compaction phases out of order; restarting fragment from last checkpoint");
    }
    other => other?,
}

Prevention

When it happens

Trigger: Raised in compaction_begin (called from execute_normal) when context.phase != Phase::Begin as i32 — e.g. a Phase::Apply or Phase::End context barrier arrives while the writer is in Normal mode for the task.

Common situations: Missed or mis-ordered Begin barrier after recovery; meta issuing Apply/End phases to a writer that never entered the task; racing phase transitions when tasks are rescheduled across actors; node version skew.

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