{"record":{"id":"d0e5f4e964ff9473","repo":"risingwavelabs/risingwave","slug":"end-of-stream","errorCode":null,"errorMessage":"end of stream","messagePattern":"end of stream","errorType":"exception","errorClass":"SinkError::BigQuery","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/big_query.rs","lineNumber":143,"sourceCode":"            offset_queue,\n            resp_stream: Box::pin(resp_stream),\n        }\n    }\n\n    pub fn add_offset(&mut self, offset: TruncateOffset, resp_num: usize) {\n        self.offset_queue.push_back((offset, resp_num));\n    }\n\n    pub async fn next_offset(&mut self) -> Result<TruncateOffset> {\n        if let Some((_offset, remaining_resp_num)) = self.offset_queue.front_mut() {\n            if *remaining_resp_num == 0 {\n                return Ok(self.offset_queue.pop_front().unwrap().0);\n            }\n            while *remaining_resp_num > 0 {\n                self.resp_stream\n                    .next()\n                    .await\n                    .ok_or_else(|| SinkError::BigQuery(anyhow::anyhow!(\"end of stream\")))??;\n                *remaining_resp_num -= 1;\n            }\n            Ok(self.offset_queue.pop_front().unwrap().0)\n        } else {\n            pending().await\n        }\n    }\n}\npub struct BigQueryLogSinker {\n    writer: BigQuerySinkWriter,\n    bigquery_future_manager: BigQueryFutureManager,\n    future_num: usize,\n}\nimpl BigQueryLogSinker {\n    pub fn new(\n        writer: BigQuerySinkWriter,\n        resp_stream: impl Stream<Item = Result<()>> + Send + 'static,\n        future_num: usize,","sourceCodeStart":125,"sourceCodeEnd":161,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/big_query.rs#L125-L161","documentation":"The BigQuery sink's BigQueryFutureManager tracks how many write responses must complete before a log-store offset can be truncated. `next_offset` polls the response stream and expects one item per pending write; if `resp_stream.next()` returns None, the stream has terminated prematurely (all responses consumed or the stream was dropped/aborted), so the sink cannot safely confirm the write and throws \"end of stream\".","triggerScenarios":"Calling `next_offset` while `remaining_resp_num > 0` and the pinned `resp_stream` yields None — e.g. the write-response stream ended after fewer items than `write_chunk` reported, or the stream was closed/dropped by the underlying gogo/gRPC writer client.","commonSituations":"Internal sink bookkeeping drift (resp_num count mismatch), the storage writer client failing and closing its stream silently, or task cancellation during barrier handling in the log-sinker loop.","solutions":["Verify each `write_chunk` call returns a resp_num exactly equal to the number of responses the stream will emit","Check BigQuery/storage service connectivity and auth; a failed writer often terminates the stream early","Restart or recreate the sink so a fresh StorageWriterClient and resp_stream are built","If reproducible, file an internal bug with the chunk size / future_num config, since it indicates resp_num/stream accounting drift"],"exampleFix":null,"handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":"// In the consume loop, treat a closed stream as fatal and rebuild the writer:\nmatch next_offset().await {\n    Ok(offset) => log_reader.truncate(offset)?,\n    Err(e) if e.to_string().contains(\"end of stream\") => {\n        rebuild_writer_client().await?; // fresh StorageWriterClient + stream\n    }\n    Err(e) => return Err(e),\n}","preventionTips":["Ensure write_chunk's returned resp_num exactly matches emitted stream items","Monitor stream closure events from StorageWriterClient","Add telemetry counting pending resp_num vs stream items seen"],"tags":["rust","stream","sink","bigquery"],"backgroundTag":"unexpected-response-shape","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"}