fix: fix cancel of streaming job when create command is stashed (#10049)

Author: yezizp2012

src/meta/src/barrier/command.rs

@@ -477,6 +477,24 @@ where
         }
     }
 
+    /// For `CancelStreamingJob`, returns the table id of the target table.
+    pub fn table_to_cancel(&self) -> Option<TableId> {
+        match &self.command {
+            Command::CancelStreamingJob(table_fragments) => Some(table_fragments.table_id()),
+            _ => None,
+        }
+    }
+
+    /// For `CreateStreamingJob`, returns the table id of the target table.
+    pub fn table_to_create(&self) -> Option<TableId> {
+        match &self.command {
+            Command::CreateStreamingJob {
+                table_fragments, ..
+            } => Some(table_fragments.table_id()),
+            _ => None,
+        }
+    }
+
     /// Clean up actors in CNs if needed, used by drop, cancel and reschedule commands.
     async fn clean_up(
         &self,

src/meta/src/barrier/mod.rs

@@ -221,6 +221,11 @@ where
         }
     }
 
+    fn cancel_stashed_command(&mut self, id: TableId) {
+        self.finished_commands
+            .retain(|x| x.context.table_to_create() != Some(id));
+    }
+
     /// Before resolving the actors to be sent or collected, we should first record the newly
     /// created table and added actors into checkpoint control, so that `can_actor_send_or_collect`
     /// will return `true`.
@@ -964,6 +969,14 @@ where
                     checkpoint_control.stash_command_to_finish(command);
                 }
 
+                if let Some(command) = cancelled_command {
+                    checkpoint_control.cancel_command(command);
+                } else if let Some(table_id) = node.command_ctx.table_to_cancel() {
+                    // the cancelled command is possibly stashed in `finished_commands` and waiting
+                    // for checkpoint, we should also clear it.
+                    checkpoint_control.cancel_stashed_command(table_id);
+                }
+
                 let remaining = checkpoint_control.finish_commands(checkpoint).await?;
                 // If there are remaining commands (that requires checkpoint to finish), we force
                 // the next barrier to be a checkpoint.
@@ -972,10 +985,6 @@ where
                     self.scheduled_barriers.force_checkpoint_in_next_barrier();
                 }
 
-                if let Some(command) = cancelled_command {
-                    checkpoint_control.cancel_command(command);
-                }
-
                 node.timer.take().unwrap().observe_duration();
                 node.wait_commit_timer.take().unwrap().observe_duration();