From 25c8a04ed803ac5223f730b826d1a45600c21764 Mon Sep 17 00:00:00 2001 From: Shanicky Chen Date: Wed, 18 Sep 2024 17:53:22 +0800 Subject: [PATCH] Add error handling to ScaleController when reducing actors, replacing `assert!` with `bail!` for robustness. Signed-off-by: Shanicky Chen --- src/meta/src/stream/scale.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/meta/src/stream/scale.rs b/src/meta/src/stream/scale.rs index 08a36ce3f7275..2ec25b75c72af 100644 --- a/src/meta/src/stream/scale.rs +++ b/src/meta/src/stream/scale.rs @@ -1587,7 +1587,9 @@ impl ScaleController { for (worker_id, n) in decreased_actor_count { if let Some(actor_ids) = worker_to_actors.get(worker_id) { - assert!(actor_ids.len() >= n); + if actor_ids.len() < n { + bail!("plan illegal, for fragment {}, worker {} only has {} actors, but needs to reduce {}",fragment_id, worker_id, actor_ids.len(), n); + } let removed_actors: Vec<_> = actor_ids .iter()