refactor(meta): readability improvements on source manager (#14170)

Co-authored-by: Richard Chien <stdrc@outlook.com>

src/meta/Cargo.toml

@@ -94,6 +94,7 @@ workspace-hack = { path = "../workspace-hack" }
 
 [dev-dependencies]
 assert_matches = "1"
+expect-test = "1.4"
 maplit = "1.0.2"
 rand = "0.8"
 risingwave_hummock_sdk = { workspace = true, features = ["enable_test_epoch"] }

src/meta/service/src/scale_service.rs

@@ -93,7 +93,7 @@ impl ScaleService for ScaleServiceImpl {
 
         let actor_splits = self
             .source_manager
-            .get_actor_splits()
+            .list_assignments()
             .await
             .into_iter()
             .map(|(actor_id, splits)| {

src/meta/src/barrier/command.rs

@@ -84,8 +84,8 @@ pub struct ReplaceTablePlan {
     pub init_split_assignment: SplitAssignment,
 }
 
-/// [`Command`] is the action of [`crate::barrier::GlobalBarrierManager`]. For different commands,
-/// we'll build different barriers to send, and may do different stuffs after the barrier is
+/// [`Command`] is the input of [`crate::barrier::GlobalBarrierManager`]. For different commands,
+/// it will build different barriers to send, and may do different stuffs after the barrier is
 /// collected.
 #[derive(Debug, Clone, strum::Display)]
 pub enum Command {
@@ -156,8 +156,8 @@ pub enum Command {
     /// of the Merge executors are changed additionally.
     ReplaceTable(ReplaceTablePlan),
 
-    /// `SourceSplitAssignment` generates Plain(Mutation::Splits) for pushing initialized splits or
-    /// newly added splits.
+    /// `SourceSplitAssignment` generates a `Splits` barrier for pushing initialized splits or
+    /// changed splits.
     SourceSplitAssignment(SplitAssignment),
 
     /// `Throttle` command generates a `Throttle` barrier with the given throttle config to change

src/meta/src/barrier/recovery.rs

@@ -104,7 +104,7 @@ impl GlobalBarrierManager {
 
         // clean up source connector dirty changes.
         self.source_manager
-            .drop_source_change(&to_drop_table_fragments)
+            .drop_source_fragments(&to_drop_table_fragments)
             .await;
 
         Ok(())

src/meta/src/stream/scale.rs

@@ -1091,7 +1091,7 @@ impl ScaleController {
 
                 let actor_splits = self
                     .source_manager
-                    .reallocate_splits(*fragment_id, &prev_actor_ids, &curr_actor_ids)
+                    .migrate_splits(*fragment_id, &prev_actor_ids, &curr_actor_ids)
                     .await?;
 
                 fragment_stream_source_actor_splits.insert(*fragment_id, actor_splits);