fix

src/meta/src/barrier/mod.rs

@@ -273,15 +273,6 @@ impl CheckpointControl {
         {
             assert!(node.state.node_to_collect.remove(&worker_id));
             node.state.resps.push(resp);
-            if node.state.node_to_collect.is_empty() {
-                // change state to complete, and wait for nodes with the smaller epoch to commit
-                let wait_commit_timer = self
-                    .context
-                    .metrics
-                    .barrier_wait_commit_latency
-                    .start_timer();
-                node.wait_commit_timer = Some(wait_commit_timer);
-            }
         } else {
             panic!(
                 "collect barrier on non-existing barrier: {}, {}",
@@ -299,7 +290,20 @@ impl CheckpointControl {
             .iter()
             .position(|x| x.state.is_inflight())
             .unwrap_or(self.command_ctx_queue.len());
-        let complete_nodes = self.command_ctx_queue.drain(..index).collect_vec();
+        let complete_nodes = self
+            .command_ctx_queue
+            .drain(..index)
+            .map(|mut node| {
+                // change state to complete, and wait for nodes with the smaller epoch to commit
+                let wait_commit_timer = self
+                    .context
+                    .metrics
+                    .barrier_wait_commit_latency
+                    .start_timer();
+                node.wait_commit_timer = Some(wait_commit_timer);
+                node
+            })
+            .collect_vec();
         complete_nodes
     }