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 f13bd7ad7de779ffcc0dd08940fcc384a426082b Author: zhangstar333 <[email protected]> AuthorDate: Mon May 25 21:54:25 2026 +0800 follower fe invalidate dns and be rpc skip old master --- be/src/util/thrift_rpc_helper.cpp | 113 +++++++++++++++++++-- be/src/util/thrift_rpc_helper.h | 13 ++- .../apache/doris/service/FrontendServiceImpl.java | 4 + .../main/java/org/apache/doris/system/Backend.java | 90 +++++++++++----- .../org/apache/doris/system/SystemInfoService.java | 22 ++++ 5 files changed, 204 insertions(+), 38 deletions(-) diff --git a/be/src/util/thrift_rpc_helper.cpp b/be/src/util/thrift_rpc_helper.cpp index 39e614d324b..f7881ca755c 100644 --- a/be/src/util/thrift_rpc_helper.cpp +++ b/be/src/util/thrift_rpc_helper.cpp @@ -257,20 +257,115 @@ Status ThriftRpcHelper::rpc_fe_with_master_refresh( << ", refreshed master_fe_addr to " << new_master; } else { // Scenario A: transport failure; or NOT_MASTER without master_address - // (old FE without thrift change). Brief backoff so heartbeat thread - // may have time to push a new master. - std::this_thread::sleep_for( - std::chrono::milliseconds(config::thrift_client_retry_interval_ms * 2)); - TNetworkAddress refreshed = address_provider(); - if (refreshed.hostname == old_master.hostname && refreshed.port == old_master.port) { - // Heartbeat has not learned about a new master yet. Return the - // original error and let upper layer (stream load executor) decide. + // (old FE without thrift change). + // + // Replace the previous "sleep & re-read heartbeat" fallback with a + // bounded follower-probe loop: enumerate live FEs from + // get_running_frontends() (heartbeat-fresh, not just configured), skip + // the failed master, and try up to kMaxProbes candidates within a + // small time budget. + // + // * 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. + // + // We do NOT sleep, do NOT scan the whole FE list (max 2 probes), and + // give up to the caller if no candidate yields recovery within the + // budget. + constexpr int kMaxProbes = 2; + const auto deadline = std::chrono::steady_clock::now() + + std::chrono::milliseconds(config::thrift_client_retry_interval_ms * 4); + + auto running = _s_exec_env->get_running_frontends(); + int probed = 0; + bool master_refreshed = false; + + for (const auto& kv : running) { + if (probed >= kMaxProbes) { + break; + } + if (std::chrono::steady_clock::now() >= deadline) { + 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; + + // Each probe goes through the existing rpc<T>() helper. Its + // built-in one-shot transport retry is acceptable because it + // only re-reads address_provider() which here returns the + // constant candidate address; the worst case is one extra + // sleep of thrift_client_retry_interval_ms, still bounded by + // our deadline. + 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; + 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"; + continue; + } + _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; + } + // 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"; + continue; + } + + // 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(); + } + + if (!master_refreshed) { + LOG(WARNING) << "fe RPC failed against " << old_master + << ", probed " << probed + << " follower(s) without recovery; 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 - << ", heartbeat now reports master=" << refreshed << ", retrying"; + << ", master refreshed via probe, retrying original request."; } // Second (and only) attempt against the new master address. diff --git a/be/src/util/thrift_rpc_helper.h b/be/src/util/thrift_rpc_helper.h index c5f786599ce..749db480a45 100644 --- a/be/src/util/thrift_rpc_helper.h +++ b/be/src/util/thrift_rpc_helper.h @@ -59,18 +59,21 @@ public: static Status rpc(std::function<TNetworkAddress()> address_provider, std::function<void(ClientConnection<T>&)> callback, int timeout_ms); - // Wrapper around `rpc<FrontendServiceClient>` for the 6 stream-load FE RPCs + // Wrapper around `rpc<FrontendServiceClient>` for stream-load FE RPCs // (loadTxnBegin / loadTxnPreCommit / loadTxnCommit / loadTxn2PC / - // loadTxnRollback / streamLoadPut). It transparently retries **once** in - // two narrow situations: + // loadTxnRollback / streamLoadPut). It transparently recovers in two + // narrow master-transition situations: // 1) Transport-layer failure (master FE not reachable, e.g. restarting): // the first RPC never reached FE, so retrying with the same txn_id is - // side-effect-free. + // side-effect-free. If the cached master is still stale after the + // built-in rpc retry, this helper probes a bounded number of running + // non-master FEs to discover or hit the new master. // 2) The FE returned TStatusCode::NOT_MASTER (BE was still pointing at // the previous master after a leadership transfer). FE early-returns // before touching the txn, and now also sets `result.master_address` // so this helper can refresh `cluster_info()->master_fe_addr` without - // any extra round trip. + // any extra round trip, then retry the original request once against + // the refreshed master. // Other failures (ANALYSIS_ERROR, LABEL_ALREADY_EXISTS, ...) are returned // verbatim — they are not master-related and retrying is unsafe. // diff --git a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java index 6d7fd0d487e..2af1d6a9cac 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java +++ b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java @@ -2140,6 +2140,10 @@ public class FrontendServiceImpl implements FrontendService.Iface { TStatus status = checkMaster(); result.setStatus(status); if (status.getStatusCode() != TStatusCode.OK) { + TNetworkAddress masterAddr = getMasterAddressOrNull(); + if (masterAddr != null) { + result.setMasterAddress(masterAddr); + } return result; } try { diff --git a/fe/fe-core/src/main/java/org/apache/doris/system/Backend.java b/fe/fe-core/src/main/java/org/apache/doris/system/Backend.java index cdc250133c0..c4ab8482409 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/system/Backend.java +++ b/fe/fe-core/src/main/java/org/apache/doris/system/Backend.java @@ -985,43 +985,84 @@ public class Backend implements Writable { lastMissingHeartbeatTime = System.currentTimeMillis(); } - // If we detected a BE restart (new epoch, or dead->alive transition) and this is NOT - // a replay path, proactively invalidate FE-side caches (DNSCache + gRPC channels + - // Thrift pool) so the first query after restart does not hit a stale channel. - if (restartDetected && !isReplay) { + // If we detected a BE restart (new epoch, or dead->alive transition), proactively + // invalidate FE-side caches (DNSCache + gRPC channels + Thrift pool) so the first + // query after restart does not hit a stale channel. + // + // NOTE: This must run on BOTH master (isReplay=false) and follower FEs + // (isReplay=true, replaying OP_HEARTBEAT editlog). Each FE process holds its own + // in-memory caches, and follower FEs also serve client queries on port 9030. If we + // skipped follower replay, follower FEs would keep stale gRPC channels / Thrift + // sockets pointing to the previous BE IP and fail the first query after BE restart + if (restartDetected) { try { - onBackendRestartDetected(preHbStartTime, newStartTime, preHbIsAlive); + String reason = String.format( + "heartbeat: restart detected (preHbStartTime=%d (%s), " + + "postHbStartTime=%d (%s), preHbIsAlive=%s, postHbIsAlive=%s, " + + "isAvailable=%s, isShutDown=%s, isQueryDisabled=%s)", + preHbStartTime, TimeUtils.longToTimeString(preHbStartTime), + newStartTime, TimeUtils.longToTimeString(newStartTime), + preHbIsAlive, isAlive.get(), SimpleScheduler.isAvailable(this), + isShutDown(), isQueryDisabled()); + invalidateLocalConnections(host, reason); } catch (Throwable t) { - LOG.warn("onBackendRestartDetected failed, backendId={}, host={}, err={}", - id, host, t.getMessage(), t); + LOG.warn("invalidateLocalConnections failed (heartbeat path), backendId={}, " + + "host={}, err={}", id, host, t.getMessage(), t); } } return isChanged; } - // Invalidate FE-side caches that may still point to the previous BE process / IP. - // Called when the heartbeat response indicates the BE has restarted. - private void onBackendRestartDetected(long preStartTime, long postStartTime, boolean preIsAlive) { - LOG.info("BE restart detected on FE. backendId={}, host={}, bePort={}, brpcPort={}, " - + "preHbStartTime={} ({}), postHbStartTime={} ({}), preHbIsAlive={}, " - + "postHbIsAlive={}, isAvailable={}, isShutDown={}, isQueryDisabled={}. " + /** + * Public entry to invalidate FE-side connection caches (DNSCache + gRPC channels + + * Thrift pool) targeting this backend. Safe to call from any thread; all three + * underlying operations are idempotent. + * + * <p>Called from two replay paths: + * <ul> + * <li>{@link #handleHbResponse} when a heartbeat indicates a BE epoch change + * (covers {@code OP_HEARTBEAT} editlog replay on follower FEs);</li> + * <li>{@code SystemInfoService.updateBackendState()} which replays + * {@code OP_BACKEND_STATE_CHANGE} editlog entries and bypasses the heartbeat + * handler entirely.</li> + * </ul> + * Without this unified entry, follower FEs that learn about a BE restart via + * state-change editlog instead of heartbeat would never clear their in-memory + * connection caches and the next query would hit a stale gRPC channel / Thrift + * socket pointing to the previous BE IP. + * + * @param prevHost the host string seen BEFORE the update, used to invalidate the + * DNSCache entry for the old hostname in case the host field has + * been mutated by setters. Pass current host if it did not change. + * @param reason short tag included in the log line for diagnostics + * (e.g. {@code "OP_BACKEND_STATE_CHANGE replay"}). + */ + public void invalidateLocalConnections(String prevHost, String reason) { + LOG.info("BE restart detected on FE. reason={}, backendId={}, currentHost={}, " + + "prevHost={}, bePort={}, brpcPort={}, isAlive={}, lastStartTime={} ({}). " + "Will proactively invalidate DNSCache, gRPC channels and Thrift pool " + "to avoid stale connections on next query.", - id, host, bePort, brpcPort, - preStartTime, TimeUtils.longToTimeString(preStartTime), - postStartTime, TimeUtils.longToTimeString(postStartTime), - preIsAlive, isAlive.get(), SimpleScheduler.isAvailable(this), - isShutDown(), isQueryDisabled()); + reason, id, host, prevHost, bePort, brpcPort, isAlive.get(), + lastStartTime, TimeUtils.longToTimeString(lastStartTime)); Env env = Env.getCurrentEnv(); DNSCache dnsCache = env == null ? null : env.getDnsCache(); if (dnsCache != null) { - String prevIp = dnsCache.get(host); - dnsCache.remove(host); - LOG.info("BE restart cleanup: invalidated FE DNSCache. backendId={}, host={}, " - + "previousCachedIp={}. Next getProxy() will force re-resolution.", - id, host, prevIp); + if (prevHost != null && !prevHost.isEmpty()) { + String prevIp = dnsCache.get(prevHost); + dnsCache.remove(prevHost); + LOG.info("BE restart cleanup: invalidated FE DNSCache. backendId={}, host={}, " + + "previousCachedIp={}. Next getProxy() will force re-resolution.", + id, prevHost, prevIp); + } + if (host != null && !host.isEmpty() && !host.equals(prevHost)) { + String prevIp = dnsCache.get(host); + dnsCache.remove(host); + LOG.info("BE restart cleanup: invalidated FE DNSCache for current host. " + + "backendId={}, host={}, previousCachedIp={}.", + id, host, prevIp); + } } else { LOG.info("BE restart cleanup: DNSCache not available, skipped. backendId={}", id); } @@ -1040,7 +1081,8 @@ public class Backend implements Writable { id, bePoolAddr); } - LOG.info("BE restart cleanup done. backendId={}, host={}", id, host); + LOG.info("BE restart cleanup done. backendId={}, host={}, prevHost={}", + id, host, prevHost); } public void setTabletMaxCompactionScore(long compactionScore) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/system/SystemInfoService.java b/fe/fe-core/src/main/java/org/apache/doris/system/SystemInfoService.java index 8b5a80a978e..3152cf3f134 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/system/SystemInfoService.java +++ b/fe/fe-core/src/main/java/org/apache/doris/system/SystemInfoService.java @@ -859,6 +859,15 @@ public class SystemInfoService { // backend may already be dropped. this may happen when // drop and modify operations do not guarantee the order. if (memoryBe != null) { + // Snapshot pre-update values to detect a BE restart across this state change. + // This editlog path (OP_BACKEND_STATE_CHANGE) bypasses Backend.handleHbResponse() + // on follower FEs, so without this check follower FEs would never invalidate + // their stale gRPC channels / Thrift sockets when a BE restart is broadcast via + // state-change instead of heartbeat. + final long preStartTime = memoryBe.getLastStartTime(); + final boolean preAlive = memoryBe.isAlive(); + final String preHost = memoryBe.getHost(); + memoryBe.setHost(be.getHost()); memoryBe.setBePort(be.getBePort()); memoryBe.setAlive(be.isAlive()); @@ -872,6 +881,19 @@ public class SystemInfoService { memoryBe.setDisks(be.getDisks()); memoryBe.setCpuCores(be.getCputCores()); memoryBe.setPipelineExecutorSize(be.getPipelineExecutorSize()); + + boolean restartDetected = + (be.getLastStartTime() > 0 && be.getLastStartTime() != preStartTime) + || (be.isAlive() && !preAlive); + if (restartDetected) { + try { + memoryBe.invalidateLocalConnections(preHost, "OP_BACKEND_STATE_CHANGE replay"); + } catch (Throwable t) { + LOG.warn("invalidateLocalConnections failed in updateBackendState, " + + "backendId={}, host={}, err={}", + id, memoryBe.getHost(), t.getMessage(), t); + } + } } } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
