chore: Use force mode with risectl unregister-workers. (#14575)

Co-authored-by: Shanicky Chen <[email protected]>

src/ctl/src/cmd_impl/meta/reschedule.rs

@@ -241,6 +241,7 @@ pub async fn unregister_workers(
     workers: Vec<String>,
     yes: bool,
     ignore_not_found: bool,
+    check_fragment_occupied: bool,
 ) -> Result<()> {
     let meta_client = context.meta_client().await?;
 
@@ -320,7 +321,7 @@ pub async fn unregister_workers(
                 .intersection(&target_worker_ids)
                 .collect();
 
-            if !intersection_worker_ids.is_empty() {
+            if check_fragment_occupied && !intersection_worker_ids.is_empty() {
                 println!(
                     "worker ids {:?} are still occupied by fragment #{}",
                     intersection_worker_ids, fragment_id

src/ctl/src/lib.rs

@@ -486,6 +486,10 @@ enum MetaCommands {
         /// The worker not found will be ignored
         #[clap(long, default_value_t = false)]
         ignore_not_found: bool,
+
+        /// Checking whether the fragment is occupied by workers
+        #[clap(long, default_value_t = false)]
+        check_fragment_occupied: bool,
     },
 
     /// Validate source interface for the cloud team
@@ -706,7 +710,17 @@ pub async fn start_impl(opts: CliOpts, context: &CtlContext) -> Result<()> {
             workers,
             yes,
             ignore_not_found,
-        }) => cmd_impl::meta::unregister_workers(context, workers, yes, ignore_not_found).await?,
+            check_fragment_occupied,
+        }) => {
+            cmd_impl::meta::unregister_workers(
+                context,
+                workers,
+                yes,
+                ignore_not_found,
+                check_fragment_occupied,
+            )
+            .await?
+        }
         Commands::Meta(MetaCommands::ValidateSource { props }) => {
             cmd_impl::meta::validate_source(context, props).await?
         }