From da017042bb01a7420db4b4a52a2e4c5c90ba03eb Mon Sep 17 00:00:00 2001 From: StrikeW Date: Fri, 9 Aug 2024 23:05:50 +0800 Subject: [PATCH] fix --- .../src/executor/backfill/cdc/cdc_backfill.rs | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/src/stream/src/executor/backfill/cdc/cdc_backfill.rs b/src/stream/src/executor/backfill/cdc/cdc_backfill.rs index 943059355f054..2b24e57af1a0b 100644 --- a/src/stream/src/executor/backfill/cdc/cdc_backfill.rs +++ b/src/stream/src/executor/backfill/cdc/cdc_backfill.rs @@ -343,8 +343,17 @@ impl CdcBackfillExecutor { && *new_rate_limit != self.rate_limit_rps { self.rate_limit_rps = *new_rate_limit; - // rebuild the new reader stream with new rate limit - continue 'backfill_loop; + // staging the barrier for post processing + pending_barrier = Some(barrier); + tracing::info!( + table_id, + ?current_pk_pos, + ?snapshot_read_row_cnt, + "Break snapshot loop to apply rate limit" + ); + // break the snapshopt read loop, + // so that we can rebuild the new snapshot stream with the new rate limit + break; } } Mutation::Update(UpdateMutation {