risingwavelabs/risingwave · critical · StreamExecutorError

iceberg pk-index writer {} replacement input closed before i

Error message

iceberg pk-index writer {} replacement input closed before its initial barrier for task {}

What it means

The Iceberg sink writer executor (pk-index variant) requires its replacement input to deliver an initial barrier before producing any other message, so it can align with the barrier expected from the previous phase of a compaction/resolution task. When the replacement input stream terminates (returns None) without ever yielding that barrier, the writer cannot complete the mode transition back to WriterInputMode::Normal and bails. This is a stream protocol/invariant violation: inputs are expected to stay alive until at least one barrier is delivered.

Source

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

                ),
                Message::Watermark(_) => bail!(
                    "iceberg pk-index writer {} received watermark from replacement input before its initial barrier for task {}",
                    self.sink_id,
                    task_id
                ),
                Message::Barrier(barrier) => {
                    self.validate_aligned_barriers(&barrier, &expected)?;
                    #[for_await]
                    for msg in self.checkpoint_barrier(barrier) {
                        yield msg?;
                    }
                    self.mode = WriterInputMode::Normal;
                    return Ok(());
                }
            }
        }

        bail!(
            "iceberg pk-index writer {} replacement input closed before its initial barrier for task {}",
            self.sink_id,
            task_id
        );
    }
}

impl<S, W> Execute for WriterExecutor<S, W>
where
    S: StateStore,
    W: IcebergWriter,
{
    fn execute(self: Box<Self>) -> BoxedMessageStream {
        self.execute_inner().boxed()
    }
}

async fn next_msg(input: &mut BoxedMessageStream) -> StreamExecutorResult<Message> {

View on GitHub (pinned to 6469eb736d)

Solutions

Inspect the failing upstream actor's logs (same fragment/task) for an earlier error or panic that ended its stream and fix that root cause first.
Retry the stream job/recovery — barrier-based recovery restarts the fragment and rebuilds the compaction state.
Check whether a rescale or schema-change was applied concurrently with the compaction task; re-run the operation without overlapping them.
If reproducible with no upstream error, file a bug with the fragment graph and task_id — this indicates a scheduler/writer protocol bug.

Defensive patterns

Strategy: try-catch

Try / catch

match writer.run().await {
    Err(e) if e.to_string().contains("replacement input closed before its initial barrier") => {
        // log task_id/sink_id, trigger fragment recovery/restart of the compaction task
        tracing::error!(%e, "iceberg writer replacement input died pre-barrier; recovering");
        recover_fragment().await?;
    }
    other => other?,
}

Prevention

Avoid rescaling or schema changes while an Iceberg compaction task is in the AligningReplacementInput phase.
Monitor upstream actor liveness/health so failures are caught before the writer observes a dead input.
Enable debug logging around sink/compaction barrier alignment to catch protocol anomalies early.
Ensure barrier delivery is not disabled or throttled in test/custom deployments.

When it happens

Trigger: execute_aligning_replacement_input is entered with an expected barrier from the resolving phase; the replacement (switched-to) input then ends — upstream actor terminated, failed, or the fragment scheduler stopped the upstream — before emitting its first Message::Barrier. Any Chunk or Watermark before the barrier also bails (different message), but stream end without a barrier triggers exactly this error.

Common situations: Upstream actor crash or failure during an Iceberg compaction task; cluster scaling/rescaling that removes the replacement upstream mid-task; stream graph rewrite (e.g. MV schema change) replacing inputs while the writer is mid-compaction; internal scheduler bugs that stop an upstream before barrier alignment completes.

Understand the failure class

Background: "This is a bug, please report it": internal invariant violations, unreachable panics, and SNH errors explained — this error's family across 47 libraries.

Related errors

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