{"record":{"id":"abf61ef580dd17ce","repo":"risingwavelabs/risingwave","slug":"end-of-stream-abf61e","errorCode":null,"errorMessage":"end of stream","messagePattern":"end of stream","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/stream/src/executor/backfill/snapshot_backfill/consume_upstream/executor.rs","lineNumber":154,"sourceCode":"            progress_state.latest_progress(),\n            &upstream_table,\n            self.snapshot_epoch,\n            self.chunk_size,\n            self.rate_limiter.rate_limit(),\n            snapshot_rebuild_interval,\n        );\n\n        'on_new_stream: loop {\n            loop {\n                let barrier = {\n                    loop {\n                        if self.rate_limiter.rate_limit().is_paused() {\n                            break receive_next_barrier(&mut self.barrier_rx).await?;\n                        }\n                        let future1 = receive_next_barrier(&mut self.barrier_rx);\n                        let future2 = stream.try_next().map(|result| {\n                            result\n                                .and_then(|opt| opt.ok_or_else(|| anyhow!(\"end of stream\").into()))\n                        });\n                        pin_mut!(future1);\n                        pin_mut!(future2);\n                        match drop_either_future(select(future1, future2).await) {\n                            Either::Left(Ok(barrier)) => {\n                                break barrier;\n                            }\n                            Either::Right(Ok(chunk)) => {\n                                assert!(!self.rate_limiter.rate_limit().is_paused());\n                                self.rate_limiter.wait(chunk.cardinality() as _).await;\n                                yield Message::Chunk(chunk);\n                            }\n                            Either::Left(Err(e)) | Either::Right(Err(e)) => {\n                                return Err(e);\n                            }\n                        }\n                    }\n                };","sourceCodeStart":136,"sourceCodeEnd":172,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/executor/backfill/snapshot_backfill/consume_upstream/executor.rs#L136-L172","documentation":"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.","triggerScenarios":"The upstream actor/fragment is cancelled or fails, closing the channel while this backfill actor is still consuming rows and waiting for barriers.","commonSituations":"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.","solutions":["Check the upstream actor/fragment logs for the failure or cancellation that closed the stream","Retry/recover the streaming job; RisingWave usually rebuilds the actor on another node","If the upstream source/table was dropped, recreate it or cancel this backfill job","Inspect meta service logs for fragment scheduling changes around the failure time"],"exampleFix":null,"handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":"// Recovery is handled by the stream framework; at the job level, verify upstream health then rebuild\nmatch backfill_result {\n    Err(e) if e.to_string().contains(\"end of stream\") => {\n        // check upstream MV/source exists and actor logs, then retry/recover the job\n    }\n    r => r?,\n}","preventionTips":["Do not drop the upstream table/source while backfills depend on it","Keep the cluster healthy during backfill (watch actor failures, resource limits)"],"tags":["rust","streaming","backfill","upstream","channel"],"backgroundTag":"upstream-api-error","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}