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 7ab0e9b8447a024d22bcfe1d4efc8921d143bd42 Author: zhangstar333 <[email protected]> AuthorDate: Wed Jun 3 23:16:35 2026 +0800 change heart beat time and rpc retry with master --- be/src/util/thrift_rpc_helper.cpp | 121 ++++++--------------- .../main/java/org/apache/doris/common/Config.java | 2 +- .../main/java/org/apache/doris/catalog/Env.java | 1 + .../java/org/apache/doris/qe/SessionVariable.java | 14 ++- .../main/java/org/apache/doris/qe/VariableMgr.java | 29 +++++ 5 files changed, 79 insertions(+), 88 deletions(-) diff --git a/be/src/util/thrift_rpc_helper.cpp b/be/src/util/thrift_rpc_helper.cpp index 63b3cd54125..9f3831f94cf 100644 --- a/be/src/util/thrift_rpc_helper.cpp +++ b/be/src/util/thrift_rpc_helper.cpp @@ -269,104 +269,53 @@ Status ThriftRpcHelper::rpc_fe_with_master_refresh( LOG(INFO) << "fe RPC got NOT_MASTER from " << old_master << ", refreshed master_fe_addr to " << new_master; } else { - // Scenario A: transport failure; or NOT_MASTER without master_address - // (old FE without thrift change). - // - // * candidate RPC returns ok -> candidate IS current master, - // cache it and return OK - // (the response was filled by - // callback during the probe). - // * candidate returns NOT_MASTER hint -> refresh master_fe_addr and - // fall through to the single - // retry against new master. - // * candidate transport-failed -> try the next one. - // - constexpr int kMaxProbes = 3; - auto running = _s_exec_env->get_running_frontends(); - int probed = 0; - bool master_refreshed = false; + // Transport failure or NOT_MASTER without master_address hint. + // During graceful shutdown, the heartbeat interval is reduced to ~1s, + // 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; + for (int i = 0; i < kMaxRetries; i++) { + int backoff_ms = 1000 * (1 << i); + std::this_thread::sleep_for(std::chrono::milliseconds(backoff_ms)); - for (const auto& kv : running) { - if (probed >= kMaxProbes) { - break; - } - const TNetworkAddress& candidate_addr = kv.first; - if (candidate_addr.hostname == old_master.hostname && - candidate_addr.port == old_master.port) { - continue; // Skip the already-failed master. - } - if (candidate_addr.hostname.empty() || candidate_addr.port == 0) { - continue; - } - ++probed; - - auto candidate_provider = [candidate_addr]() { return candidate_addr; }; - Status probe_st = rpc<FrontendServiceClient>( - candidate_provider, callback, timeout_ms); - if (!probe_st.ok()) { - LOG(INFO) << "fe RPC probe to " << candidate_addr - << " transport-failed after master " << old_master - << " was unreachable: " << probe_st; - if (probed < kMaxProbes) { - std::this_thread::sleep_for(1000ms); // Back off a bit before the next probe. - } - continue; - } - - TStatus probe_resp = status_extractor(); - if (probe_resp.status_code == TStatusCode::NOT_MASTER) { - const TNetworkAddress* hinted = master_addr_extractor(); - if (hinted != nullptr && !hinted->hostname.empty() - && hinted->port != 0) { - if (hinted->hostname == old_master.hostname && - hinted->port == old_master.port) { - LOG(INFO) << "fe RPC probe to " << candidate_addr - << " returned NOT_MASTER with stale master_address " - << *hinted << ", same as failed master " << old_master - << "; trying next"; - if (probed < kMaxProbes) { - std::this_thread::sleep_for(1000ms); // Back off a bit before the next probe. - } - continue; + Status retry_st = rpc<FrontendServiceClient>(address_provider, callback, timeout_ms); + if (retry_st.ok()) { + 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) { + _s_exec_env->cluster_info()->master_fe_addr = *hinted; + new_master = *hinted; + have_new_master = true; + LOG(INFO) << "fe RPC retry " << (i + 1) << "/" << kMaxRetries + << " returned NOT_MASTER, refreshed master_fe_addr to " + << *hinted; + break; // fall through to the single retry below } - _s_exec_env->cluster_info()->master_fe_addr = *hinted; - LOG(INFO) << "fe RPC probe to " << candidate_addr - << " returned NOT_MASTER, refreshed master_fe_addr to " - << *hinted << " (after transport failure on " << old_master << ")"; - master_refreshed = true; - break; + LOG(INFO) << "fe RPC retry " << (i + 1) << "/" << kMaxRetries + << " returned NOT_MASTER without master_address"; + continue; } - // NOT_MASTER without master_address (old FE). Try next candidate. - LOG(INFO) << "fe RPC probe to " << candidate_addr - << " returned NOT_MASTER without master_address, trying next"; - if (probed < kMaxProbes) { - std::this_thread::sleep_for(2000ms); // Back off a bit before the next probe. - } - continue; + // RPC succeeded: address_provider already yielded the correct master. + LOG(INFO) << "fe RPC succeeded on retry " << (i + 1) << "/" << kMaxRetries; + return Status::OK(); } - - // Probe succeeded on the candidate: it IS the current master. - // The callback has already populated the response on this probe - // (status_extractor returned non-NOT_MASTER OK), so the original - // operation has completed against the new master. Refresh the - // cache so subsequent RPCs go directly there. - _s_exec_env->cluster_info()->master_fe_addr = candidate_addr; - LOG(INFO) << "fe RPC succeeded on probe candidate " << candidate_addr - << " after transport failure on " << old_master - << ", refreshed master_fe_addr."; - return Status::OK(); + LOG(INFO) << "fe RPC retry " << (i + 1) << "/" << kMaxRetries + << " transport-failed: " << retry_st; } - if (!master_refreshed) { + if (!have_new_master) { LOG(WARNING) << "fe RPC failed against " << old_master - << ", probed " << probed - << " follower(s) without recovery; returning original error."; + << " 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") : st; } LOG(INFO) << "fe RPC failed against " << old_master - << ", master refreshed via probe, retrying original request."; + << ", master refreshed via backoff retry, retrying original request."; } // Second (and only) attempt against the new master address. diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java index c7c4c65dae8..3fae58d6dd9 100644 --- a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java +++ b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java @@ -2124,7 +2124,7 @@ public class Config extends ConfigBase { * Heartbeat interval in seconds. * Default is 10, which means every 10 seconds, the master will send a heartbeat to all backends. */ - @ConfField(mutable = false, masterOnly = false) + @ConfField(mutable = true, masterOnly = false) public static int heartbeat_interval_second = 10; /** diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java index 7605a93639c..41c86f0d414 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java +++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java @@ -616,6 +616,7 @@ public class Env { .put("dynamic_partition_check_interval_seconds", this::getDynamicPartitionScheduler) .put("table_stream_partition_offset_cleanup_interval_second", this::getTableStreamManager) + .put("heartbeat_interval_second", this::getHeartbeatMgr) .build(); private TSOService tsoService; diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java index 1dcb8562dc3..8638a95840b 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java @@ -24,6 +24,8 @@ import org.apache.doris.catalog.Env; import org.apache.doris.cloud.qe.ComputeGroupException; import org.apache.doris.common.Config; import org.apache.doris.common.DdlException; +import org.apache.doris.common.ErrorCode; +import org.apache.doris.common.ErrorReport; import org.apache.doris.common.VariableAnnotation; import org.apache.doris.common.io.Text; import org.apache.doris.common.io.Writable; @@ -2313,7 +2315,8 @@ public class SessionVariable implements Serializable, Writable { "服务端 prepared statement 最大个数", "the maximum prepared statements server holds."}) public int maxPreparedStmtCount = 100000; - @VarAttrDef.VarAttr(name = ENABLE_GRACEFUL_SHUTDOWN, flag = VarAttrDef.GLOBAL, needForward = true, + @VarAttrDef.VarAttr(name = ENABLE_GRACEFUL_SHUTDOWN, flag = VarAttrDef.GLOBAL, + setter = "setEnableGracefulShutdown", needForward = true, description = { "集群级别开关：当 FE 或 BE 准备做滚动重启 / 优雅退出时，由运维 SET GLOBAL 打开。" + "打开后：(1) FE QueryCancelWorker / Coordinator.shouldCancel 不再因 BE alive=false " @@ -6864,3 +6867,12 @@ public class SessionVariable implements Serializable, Writable { } } } + public void setEnableGracefulShutdown(String value) throws DdlException { + if (value.equalsIgnoreCase("TRUE")) { + enableGracefulShutdown = true; + } else if (value.equalsIgnoreCase("FALSE")) { + enableGracefulShutdown = false; + } else { + ErrorReport.reportDdlException(ErrorCode.ERR_WRONG_VALUE_FOR_VAR, ENABLE_GRACEFUL_SHUTDOWN, value); + } + } diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/VariableMgr.java b/fe/fe-core/src/main/java/org/apache/doris/qe/VariableMgr.java index 5f6a60a2d16..768de26cb9d 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/VariableMgr.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/VariableMgr.java @@ -26,6 +26,7 @@ import org.apache.doris.catalog.Env; import org.apache.doris.catalog.Type; import org.apache.doris.common.AnalysisException; import org.apache.doris.common.Config; +import org.apache.doris.common.ConfigException; import org.apache.doris.common.DdlException; import org.apache.doris.common.ErrorCode; import org.apache.doris.common.ErrorReport; @@ -103,6 +104,9 @@ import javax.annotation.Nullable; */ public class VariableMgr { private static final Logger LOG = LogManager.getLogger(VariableMgr.class); + private static final String HEARTBEAT_INTERVAL_SECOND = "heartbeat_interval_second"; + private static final String FAST_HEARTBEAT_INTERVAL_SECONDS = "1"; + private static final String DEFAULT_HEARTBEAT_INTERVAL_SECONDS = "10"; // Map variable name to variable context which have enough information to change variable value. // This map contains info of all session and global variables. @@ -448,6 +452,7 @@ public class VariableMgr { try { setValue(ctx.getObj(), new SessionVariableField(ctx.getField()), value); + applyGlobalVariableSideEffect(name, value); // write edit log GlobalVarPersistInfo info = new GlobalVarPersistInfo(defaultSessionVariable, Lists.newArrayList(name)); Env.getCurrentEnv().getEditLog().logGlobalVariableV2(info); @@ -456,6 +461,29 @@ public class VariableMgr { } } + private static void applyGlobalVariableSideEffect(String name, String value) throws DdlException { + if (!SessionVariable.ENABLE_GRACEFUL_SHUTDOWN.equals(name) || Env.isCheckpointThread()) { + return; + } + String heartbeatInterval = parseBooleanValue(name, value) + ? FAST_HEARTBEAT_INTERVAL_SECONDS : DEFAULT_HEARTBEAT_INTERVAL_SECONDS; + try { + Env.getCurrentEnv().setMutableConfigWithCallback(HEARTBEAT_INTERVAL_SECOND, heartbeatInterval); + } catch (ConfigException e) { + throw new DdlException(e.getMessage()); + } + } + + private static boolean parseBooleanValue(String name, String value) throws DdlException { + if (value.equalsIgnoreCase("TRUE")) { + return true; + } else if (value.equalsIgnoreCase("FALSE")) { + return false; + } + ErrorReport.reportDdlException(ErrorCode.ERR_WRONG_VALUE_FOR_VAR, name, value); + return false; + } + public static void refreshDefaultSessionVariables(String versionMsg, String sessionVar, String value) { wlock.lock(); try { @@ -540,6 +568,7 @@ public class VariableMgr { try { setValue(varContext.getObj(), new SessionVariableField(varContext.getField()), root.get(varName).toString()); + applyGlobalVariableSideEffect((String) varName, root.get(varName).toString()); } catch (Exception exception) { LOG.warn("Exception during replay global variabl {} oplog, {}, THIS EXCEPTION WILL BE IGNORED.", (String) varName, exception.getMessage()); --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
