From fb77c1e25e38a2ad428779cd896c6d9b44e88cfd Mon Sep 17 00:00:00 2001
From: Winter Zhang <coswde@gmail.com>
Date: Wed, 11 Dec 2024 19:53:52 +0800
Subject: [PATCH] chore(cluster): disable parallel commit for cluster task
 (#17034)

* chore(cluster): disable parallel commit for cluster task

* chore(cluster): update default setting
---
 src/query/service/src/clusters/cluster.rs  | 14 ++++++++------
 src/query/settings/src/settings_default.rs |  6 +++---
 2 files changed, 11 insertions(+), 9 deletions(-)

diff --git a/src/query/service/src/clusters/cluster.rs b/src/query/service/src/clusters/cluster.rs
index 40cb98bcd343..8ab381200664 100644
--- a/src/query/service/src/clusters/cluster.rs
+++ b/src/query/service/src/clusters/cluster.rs
@@ -138,11 +138,11 @@ impl ClusterHelper for Cluster {
             )))
         }
 
-        let mut futures = Vec::with_capacity(message.len());
+        let mut response = HashMap::with_capacity(message.len());
         for (id, message) in message {
             let node = get_node(&self.nodes, &id)?;
 
-            futures.push({
+            let do_action_with_retry = {
                 let config = GlobalConfig::instance();
                 let flight_address = node.flight_address.clone();
                 let node_secret = node.secret.clone();
@@ -161,7 +161,7 @@ impl ClusterHelper for Cluster {
                             )
                             .await
                         {
-                            Ok(result) => return Ok((id, result)),
+                            Ok(result) => return Ok(result),
                             Err(e)
                                 if e.code() == ErrorCode::CANNOT_CONNECT_NODE
                                     && attempt < flight_params.retry_times =>
@@ -175,10 +175,12 @@ impl ClusterHelper for Cluster {
                         }
                     }
                 }
-            });
+            };
+
+            response.insert(id, do_action_with_retry.await?);
         }
-        let responses: Vec<(String, Res)> = futures::future::try_join_all(futures).await?;
-        Ok(responses.into_iter().collect::<HashMap<String, Res>>())
+
+        Ok(response)
     }
 }
 
diff --git a/src/query/settings/src/settings_default.rs b/src/query/settings/src/settings_default.rs
index 542b463be0e2..c614e1361058 100644
--- a/src/query/settings/src/settings_default.rs
+++ b/src/query/settings/src/settings_default.rs
@@ -965,16 +965,16 @@ impl DefaultSettings {
                     range: Some(SettingRange::Numeric(0..=1)),
                 }),
                 ("flight_connection_max_retry_times", DefaultSettingValue {
-                    value: UserSettingValue::UInt64(3),
+                    value: UserSettingValue::UInt64(0),
                     desc: "The maximum retry count for cluster flight. Disable if 0.",
                     mode: SettingMode::Both,
                     range: Some(SettingRange::Numeric(0..=10)),
                 }),
                 ("flight_connection_retry_interval", DefaultSettingValue {
-                    value: UserSettingValue::UInt64(3),
+                    value: UserSettingValue::UInt64(1),
                     desc: "The retry interval of cluster flight is in seconds.",
                     mode: SettingMode::Both,
-                    range: Some(SettingRange::Numeric(0..=30)),
+                    range: Some(SettingRange::Numeric(0..=10)),
                 }),
             ]);