risingwavelabs/risingwave · error · SinkError::BigQuery

end of stream

Error message

end of stream

What it means

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".

Source

Thrown at src/connector/src/sink/big_query.rs:143

            offset_queue,
            resp_stream: Box::pin(resp_stream),
        }
    }

    pub fn add_offset(&mut self, offset: TruncateOffset, resp_num: usize) {
        self.offset_queue.push_back((offset, resp_num));
    }

    pub async fn next_offset(&mut self) -> Result<TruncateOffset> {
        if let Some((_offset, remaining_resp_num)) = self.offset_queue.front_mut() {
            if *remaining_resp_num == 0 {
                return Ok(self.offset_queue.pop_front().unwrap().0);
            }
            while *remaining_resp_num > 0 {
                self.resp_stream
                    .next()
                    .await
                    .ok_or_else(|| SinkError::BigQuery(anyhow::anyhow!("end of stream")))??;
                *remaining_resp_num -= 1;
            }
            Ok(self.offset_queue.pop_front().unwrap().0)
        } else {
            pending().await
        }
    }
}
pub struct BigQueryLogSinker {
    writer: BigQuerySinkWriter,
    bigquery_future_manager: BigQueryFutureManager,
    future_num: usize,
}
impl BigQueryLogSinker {
    pub fn new(
        writer: BigQuerySinkWriter,
        resp_stream: impl Stream<Item = Result<()>> + Send + 'static,
        future_num: usize,

View on GitHub (pinned to 6469eb736d)

Solutions

1. Verify each `write_chunk` call returns a resp_num exactly equal to the number of responses the stream will emit
2. Check BigQuery/storage service connectivity and auth; a failed writer often terminates the stream early
3. Restart or recreate the sink so a fresh StorageWriterClient and resp_stream are built
4. If reproducible, file an internal bug with the chunk size / future_num config, since it indicates resp_num/stream accounting drift

// In the consume loop, treat a closed stream as fatal and rebuild the writer:
match next_offset().await {
    Ok(offset) => log_reader.truncate(offset)?,
    Err(e) if e.to_string().contains("end of stream") => {
        rebuild_writer_client().await?; // fresh StorageWriterClient + stream
    }
    Err(e) => return Err(e),
}

When it happens

Trigger: 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.

Common situations: 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.

Related errors

• MAP is not supported for BigQuery sink.
• VECTOR is not supported for BigQuery sink.
• INTERVAL is not supported for BigQuery sink. Please convert
• license feature check failed (BigQuerySink not available): w
• Primary key not defined for upsert bigquery sink (please def

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