risingwavelabs/risingwave · error

end of stream

Error message

end of stream

What it means

While the snapshot backfill consume-upstream executor races reading the upstream chunk stream against receiving the next barrier, a None from the upstream stream means the upstream channel closed without an end-of-stream signal. The executor converts it into an 'end of stream' error instead of treating it as a normal completion.

Source

Thrown at src/stream/src/executor/backfill/snapshot_backfill/consume_upstream/executor.rs:154

            progress_state.latest_progress(),
            &upstream_table,
            self.snapshot_epoch,
            self.chunk_size,
            self.rate_limiter.rate_limit(),
            snapshot_rebuild_interval,
        );

        'on_new_stream: loop {
            loop {
                let barrier = {
                    loop {
                        if self.rate_limiter.rate_limit().is_paused() {
                            break receive_next_barrier(&mut self.barrier_rx).await?;
                        }
                        let future1 = receive_next_barrier(&mut self.barrier_rx);
                        let future2 = stream.try_next().map(|result| {
                            result
                                .and_then(|opt| opt.ok_or_else(|| anyhow!("end of stream").into()))
                        });
                        pin_mut!(future1);
                        pin_mut!(future2);
                        match drop_either_future(select(future1, future2).await) {
                            Either::Left(Ok(barrier)) => {
                                break barrier;
                            }
                            Either::Right(Ok(chunk)) => {
                                assert!(!self.rate_limiter.rate_limit().is_paused());
                                self.rate_limiter.wait(chunk.cardinality() as _).await;
                                yield Message::Chunk(chunk);
                            }
                            Either::Left(Err(e)) | Either::Right(Err(e)) => {
                                return Err(e);
                            }
                        }
                    }
                };

View on GitHub (pinned to 6469eb736d)

Solutions

1. Check the upstream actor/fragment logs for the failure or cancellation that closed the stream
2. Retry/recover the streaming job; RisingWave usually rebuilds the actor on another node
3. If the upstream source/table was dropped, recreate it or cancel this backfill job
4. Inspect meta service logs for fragment scheduling changes around the failure time

// Recovery is handled by the stream framework; at the job level, verify upstream health then rebuild
match backfill_result {
    Err(e) if e.to_string().contains("end of stream") => {
        // check upstream MV/source exists and actor logs, then retry/recover the job
    }
    r => r?,
}

When it happens

Trigger: The upstream actor/fragment is cancelled or fails, closing the channel while this backfill actor is still consuming rows and waiting for barriers.

Common situations: Upstream MV/table dropped or its fragment failed during a backfill; cluster scale-in or actor rescheduling terminating the upstream; a barrier injection failure upstream.

Related errors

• no upstream while snapshot epoch not set
• locality provider upstream ended unexpectedly during backfil
• legacy no-shuffle backfill recovered unfinished progress; ca
• mismatch initial vnode bitmap: {:?}, expect: {:?}
• should not update vnode bitmap during consuming log store

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