risingwave_stream/executor/backfill/cdc/upstream_table/
reader.rs

1// Copyright 2026 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::future::Future;
16use std::pin::Pin;
17
18use either::Either;
19use futures::{Stream, StreamExt};
20use risingwave_connector::source::cdc::external::ExternalTableReaderImpl;
21use thiserror_ext::AsReport;
22use tracing::Instrument;
23
24use super::external::ExternalStorageTable;
25use crate::executor::prelude::{Message, StreamExecutorResult};
26use crate::executor::source::get_infinite_backoff_strategy;
27use crate::task::{ActorId, FragmentId};
28
29pub(crate) async fn build_reader_and_poll_upstream(
30    upstream: &mut (impl Stream<Item = StreamExecutorResult<Message>> + Unpin),
31    future: &mut Pin<Box<impl Future<Output = ExternalTableReaderImpl>>>,
32) -> StreamExecutorResult<Either<Message, ExternalTableReaderImpl>> {
33    tokio::select! {
34        biased;
35        reader = &mut *future => Ok(Either::Right(reader)),
36        msg = upstream.next() => {
37            msg.transpose()?
38                .map(Either::Left)
39                .ok_or_else(|| anyhow::anyhow!(
40                    "upstream closed while creating CDC table reader"
41                ).into())
42        }
43    }
44}
45
46pub(crate) async fn create_table_reader_with_retry(
47    external_table: ExternalStorageTable,
48    actor_id: ActorId,
49    fragment_id: FragmentId,
50) -> ExternalTableReaderImpl {
51    let backoff = get_infinite_backoff_strategy();
52
53    tokio_retry::Retry::spawn(backoff, || async {
54        match external_table.create_table_reader().await {
55            Ok(reader) => Ok(reader),
56            Err(error) => {
57                tracing::warn!(
58                    error = %error.as_report(),
59                    actor_id = %actor_id,
60                    fragment_id = %fragment_id,
61                    "failed to create CDC table reader; retrying"
62                );
63                Err(error)
64            }
65        }
66    })
67    .instrument(tracing::info_span!("create_cdc_table_reader_with_retry"))
68    .await
69    .expect("retry creating CDC table reader until success")
70}
risingwave_stream/executor/backfill/cdc/upstream_table/reader.rs

risingwave_stream/executor/backfill/cdc/upstream_table/
reader.rs
