pre-apply and purge earlier

risingwavelabs · Jun 17, 2024 · 453d4a8 · 453d4a8
1 parent 0da9bf9
commit 453d4a8
Showing 1 changed file with 12 additions and 12 deletions.
diff --git a/src/meta/src/barrier/recovery.rs b/src/meta/src/barrier/recovery.rs
@@ -423,18 +423,6 @@ impl GlobalBarrierManager {
                             })?
                     };
 
-                    let mut control_stream_manager =
-                        ControlStreamManager::new(self.context.clone());
-
-                    control_stream_manager
-                        .reset(prev_epoch.value().0, active_streaming_nodes.current())
-                        .await
-                        .inspect_err(|err| {
-                            warn!(error = %err.as_report(), "reset compute nodes failed");
-                        })?;
-
-                    self.context.sink_manager.reset().await;
-
                     if self
                         .context
                         .pre_apply_drop_cancel(&self.scheduled_barriers)
@@ -454,6 +442,18 @@ impl GlobalBarrierManager {
                         .await
                         .context("purge state table from hummock")?;
 
+                    let mut control_stream_manager =
+                        ControlStreamManager::new(self.context.clone());
+
+                    control_stream_manager
+                        .reset(prev_epoch.value().0, active_streaming_nodes.current())
+                        .await
+                        .inspect_err(|err| {
+                            warn!(error = %err.as_report(), "reset compute nodes failed");
+                        })?;
+
+                    self.context.sink_manager.reset().await;
+
                     // update and build all actors.
                     self.context.update_actors(&info).await.inspect_err(|err| {
                         warn!(error = %err.as_report(), "update actors failed");