fix: join actor when dropping (#7263)

Ensure actor task is completely dropped after calling `drop_all_actors`, see #7208 (comment).
Also remove unused method `wait_all`.

Approved-By: yezizp2012

Co-Authored-By: zwang28 <84491488@qq.com>
Co-Authored-By: Runji Wang <wangrunji0408@163.com>

src/stream/src/task/stream_manager.rs

@@ -311,7 +311,7 @@ impl LocalStreamManager {
 
     /// Force stop all actors on this worker.
     pub async fn stop_all_actors(&self) -> StreamResult<()> {
-        self.core.lock().await.drop_all_actors();
+        self.core.lock().await.drop_all_actors().await;
         // Clear shared buffer in storage to release memory
         self.clear_storage_buffer().await;
         self.clear_all_senders_and_collect_rx();
@@ -333,15 +333,6 @@ impl LocalStreamManager {
         core.update_actors(actors, hanging_channels)
     }
 
-    /// This function was called while [`LocalStreamManager`] exited.
-    pub async fn wait_all(self) -> StreamResult<()> {
-        let handles = self.core.lock().await.take_all_handles()?;
-        for (_id, handle) in handles {
-            handle.await.unwrap();
-        }
-        Ok(())
-    }
-
     /// This function could only be called once during the lifecycle of `LocalStreamManager` for
     /// now.
     pub async fn update_actor_info(&self, actor_infos: &[ActorInfo]) -> StreamResult<()> {
@@ -783,11 +774,16 @@ impl LocalStreamManagerCore {
     }
 
     /// `drop_all_actors` is invoked by meta node via RPC for recovery purpose.
-    fn drop_all_actors(&mut self) {
-        for (actor_id, handle) in self.handles.drain() {
+    async fn drop_all_actors(&mut self) {
+        for (actor_id, handle) in &self.handles {
             tracing::debug!("force stopping actor {}", actor_id);
             handle.abort();
         }
+        for (actor_id, handle) in self.handles.drain() {
+            tracing::debug!("join actor {}", actor_id);
+            let result = handle.await;
+            assert!(result.is_ok() || result.unwrap_err().is_cancelled());
+        }
         self.actors.clear();
         self.context.clear_channels();
         if let Some(m) = self.stack_trace_manager.as_mut() {

src/tests/simulation/Cargo.toml

@@ -13,7 +13,7 @@ etcd-client = { version = "0.2.11", package = "madsim-etcd-client" }
 futures = { version = "0.3", default-features = false, features = ["alloc"] }
 glob = "0.3"
 itertools = "0.10"
-madsim = "0.2.12"
+madsim = "0.2.13"
 paste = "1"
 rand = "0.8"
 rdkafka = { package = "madsim-rdkafka", version = "=0.2.8-alpha", features = ["cmake-build"] }