fix test

src/meta/src/barrier/mod.rs

@@ -673,7 +673,12 @@ impl GlobalBarrierManager {
 
         send_latency_timer.observe_duration();
 
-        self.rpc_manager.inject_barrier(command_ctx.clone()).await?;
+        self.rpc_manager
+            .inject_barrier(command_ctx.clone())
+            .await
+            .inspect_err(|_| {
+                fail_point!("inject_barrier_err_success");
+            })?;
 
         // Notify about the injection.
         let prev_paused_reason = self.state.paused_reason();
@@ -709,7 +714,6 @@ impl GlobalBarrierManager {
         if let Err(err) = result {
             // FIXME: If it is a connector source error occurred in the init barrier, we should pass
             // back to frontend
-            fail_point!("inject_barrier_err_success");
             let fail_node = self.checkpoint_control.barrier_failed();
             warn!(%prev_epoch, error = %err.as_report(), "Failed to complete epoch");
             self.failure_recovery(err, fail_node).await;

src/stream/src/task/barrier_manager.rs

@@ -302,13 +302,13 @@ impl LocalBarrierWorker {
     }
 
     fn send_response(&mut self, response: StreamingControlStreamResponse) {
-        let (sender, _) = self
-            .control_stream_handle
-            .as_ref()
-            .expect("should not be None");
-        if sender.send(Ok(response)).is_err() {
-            self.control_stream_handle = None;
-            warn!("fail to send response. control stream reset");
+        if let Some((sender, _)) = self.control_stream_handle.as_ref() {
+            if sender.send(Ok(response)).is_err() {
+                self.control_stream_handle = None;
+                warn!("fail to send response. control stream reset");
+            }
+        } else {
+            debug!(?response, "control stream has been reset. ignore response");
         }
     }