This is an automated email from the ASF dual-hosted git repository. zhangstar333 pushed a commit to branch doris_master_new in repository https://gitbox.apache.org/repos/asf/doris.git
commit a824e2e43a2f7d121e713868b1424f10e9c9a00d Author: zhangstar333 <[email protected]> AuthorDate: Fri Jun 5 15:51:59 2026 +0800 kMaxRetries to 5 for rpc retry add load nums --- be/src/load/channel/load_channel_mgr.cpp | 5 +++++ be/src/load/channel/load_channel_mgr.h | 3 +++ be/src/runtime/exec_env.cpp | 13 ++++++++----- be/src/util/thrift_rpc_helper.cpp | 16 +++++++--------- 4 files changed, 23 insertions(+), 14 deletions(-) diff --git a/be/src/load/channel/load_channel_mgr.cpp b/be/src/load/channel/load_channel_mgr.cpp index a5f655b5fb6..18079d1c78d 100644 --- a/be/src/load/channel/load_channel_mgr.cpp +++ b/be/src/load/channel/load_channel_mgr.cpp @@ -286,4 +286,9 @@ Status LoadChannelMgr::_start_load_channels_clean() { return Status::OK(); } + +size_t LoadChannelMgr::get_active_load_channel_num() { + std::lock_guard<std::mutex> l(_lock); + return _load_channels.size(); +} } // namespace doris diff --git a/be/src/load/channel/load_channel_mgr.h b/be/src/load/channel/load_channel_mgr.h index fdddbee0400..eca9b372b06 100644 --- a/be/src/load/channel/load_channel_mgr.h +++ b/be/src/load/channel/load_channel_mgr.h @@ -62,6 +62,9 @@ public: // cancel all tablet stream for 'load_id' load Status cancel(const PTabletWriterCancelRequest& request); + // get the number of active load channels + size_t get_active_load_channel_num(); + void stop(); std::vector<std::string> get_all_load_channel_ids() { diff --git a/be/src/runtime/exec_env.cpp b/be/src/runtime/exec_env.cpp index e1ef845ccc0..0724c5c6515 100644 --- a/be/src/runtime/exec_env.cpp +++ b/be/src/runtime/exec_env.cpp @@ -201,19 +201,22 @@ void ExecEnv::wait_for_all_fe_known() { void ExecEnv::wait_for_all_tasks_done() { LOG(INFO) << "begin to wait for all tasks done before shutdown. k_shutdown_fe_known: " << k_shutdown_fe_known << " k_in_graceful_shutdown: " << k_in_graceful_shutdown; - // For graceful shutdown, need to wait for all running queries to stop + // For graceful shutdown, need to wait for all running queries and load channels to stop int32_t wait_seconds_passed = 0; while (true) { int num_queries = _fragment_mgr->running_query_num(); - if (num_queries < 1) { + size_t num_load_channels = _load_channel_mgr->get_active_load_channel_num(); + if (num_queries < 1 && num_load_channels < 1) { break; } if (wait_seconds_passed > doris::config::grace_shutdown_wait_seconds) { - LOG(INFO) << "There are still " << num_queries << " queries running, but " - << wait_seconds_passed << " seconds passed, has to exist now"; + LOG(INFO) << "There are still " << num_queries << " queries running and " + << num_load_channels << " load channels active, but " + << wait_seconds_passed << " seconds passed, has to exit now"; break; } - LOG(INFO) << "There are still " << num_queries << " queries running, waiting..."; + LOG(INFO) << "There are still " << num_queries << " queries running, " + << num_load_channels << " load channels active, waiting..."; sleep(1); ++wait_seconds_passed; } diff --git a/be/src/util/thrift_rpc_helper.cpp b/be/src/util/thrift_rpc_helper.cpp index 9f3831f94cf..49b5e3dae4a 100644 --- a/be/src/util/thrift_rpc_helper.cpp +++ b/be/src/util/thrift_rpc_helper.cpp @@ -266,17 +266,17 @@ Status ThriftRpcHelper::rpc_fe_with_master_refresh( // is also written by heartbeat thread with epoch checks, which will // self-correct any stale value we might race-write. _s_exec_env->cluster_info()->master_fe_addr = new_master; - LOG(INFO) << "fe RPC got NOT_MASTER from " << old_master - << ", refreshed master_fe_addr to " << new_master; + LOG(INFO) << "fe RPC got NOT_MASTER from " << old_master << ", refreshed master_fe_addr to " + << new_master; } else { // Transport failure or NOT_MASTER without master_address hint. - // During graceful shutdown, the heartbeat interval is reduced to ~1s, + // During graceful shutdown, the heartbeat interval is reduced to ~3s, // so BE will quickly pick up the new master FE address. Retry with // increasing backoff, relying on heartbeat to update master_fe_addr // which is read fresh by address_provider on each attempt. - constexpr int kMaxRetries = 3; + constexpr int kMaxRetries = 5; for (int i = 0; i < kMaxRetries; i++) { - int backoff_ms = 1000 * (1 << i); + int backoff_ms = 1000 * (2 * i + 1); // 1s, 3s, 5s, 7s, 9s std::this_thread::sleep_for(std::chrono::milliseconds(backoff_ms)); Status retry_st = rpc<FrontendServiceClient>(address_provider, callback, timeout_ms); @@ -284,8 +284,7 @@ Status ThriftRpcHelper::rpc_fe_with_master_refresh( TStatus retry_resp = status_extractor(); if (retry_resp.status_code == TStatusCode::NOT_MASTER) { const TNetworkAddress* hinted = master_addr_extractor(); - if (hinted != nullptr && !hinted->hostname.empty() - && hinted->port != 0) { + if (hinted != nullptr && !hinted->hostname.empty() && hinted->port != 0) { _s_exec_env->cluster_info()->master_fe_addr = *hinted; new_master = *hinted; have_new_master = true; @@ -307,8 +306,7 @@ Status ThriftRpcHelper::rpc_fe_with_master_refresh( } if (!have_new_master) { - LOG(WARNING) << "fe RPC failed against " << old_master - << " after " << kMaxRetries + LOG(WARNING) << "fe RPC failed against " << old_master << " after " << kMaxRetries << " backoff retries, returning original error."; return st.ok() ? Status::Error<ErrorCode::NOT_MASTER>( "FE returned NOT_MASTER but no new master address available") --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
