fix(cluster) try fix test failure

Author: zhang2014

src/query/service/src/api/rpc/exchange/statistics_receiver.rs

@@ -90,6 +90,7 @@ impl StatisticsReceiver {
                             notified = middle;
                             match Self::recv_data(&ctx, res) {
                                 Ok(true) => {
+                                    ctx.get_current_session().force_kill_query();
                                     return Ok(());
                                 }
                                 Ok(false) => {

src/query/service/src/api/rpc/exchange/statistics_sender.rs

@@ -90,13 +90,10 @@ impl StatisticsSender {
                         notified = right;
                         recv = Box::pin(flight_exchange.recv());
 
-                        if !shutdown_flag.load(Ordering::Relaxed) {
-                            if let Err(_cause) =
-                                Self::on_command(&ctx, command, &flight_exchange).await
-                            {
-                                ctx.get_exchange_manager().shutdown_query(&query_id);
-                                return;
-                            }
+                        if let Err(_cause) = Self::on_command(&ctx, command, &flight_exchange).await
+                        {
+                            ctx.get_exchange_manager().shutdown_query(&query_id);
+                            return;
                         }
                     }
                 }