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 b34116063c84467875962038ce08cbd7407e9e44 Author: zhangstar333 <[email protected]> AuthorDate: Sun Jun 7 00:38:56 2026 +0800 enhance handshake in brpc --- be/src/util/brpc_client_cache.h | 31 +++++++++++- .../main/java/org/apache/doris/system/Backend.java | 58 +++++++++++++++++++++- 2 files changed, 87 insertions(+), 2 deletions(-) diff --git a/be/src/util/brpc_client_cache.h b/be/src/util/brpc_client_cache.h index 9daffd7d69e..7d767b4a077 100644 --- a/be/src/util/brpc_client_cache.h +++ b/be/src/util/brpc_client_cache.h @@ -17,6 +17,8 @@ #pragma once +#include <bthread/bthread.h> + #include <brpc/adaptive_connection_type.h> #include <brpc/adaptive_protocol_type.h> #include <brpc/channel.h> @@ -231,6 +233,11 @@ public: const int max_attempts = enable_handshake ? 3 : 1; std::string host_port; for (int attempt = 1; attempt <= max_attempts; ++attempt) { + // Brief backoff before retries so the new Pod has time to start + // listening. Only paid during graceful restart (enable_handshake). + if (enable_handshake && attempt > 1) { + bthread_usleep(500000); + } std::string realhost = host; auto dns_cache = ExecEnv::GetInstance()->dns_cache(); if (dns_cache == nullptr) { @@ -238,7 +245,11 @@ public: } else if (!is_valid_ip(host)) { Status status = dns_cache->get(host, &realhost); if (!status.ok()) { - LOG(WARNING) << "failed to get ip from host:" << status.to_string(); + LOG(WARNING) << "failed to get ip from host: " << status.to_string() + << ", attempt=" << attempt << "/" << max_attempts; + if (enable_handshake && attempt < max_attempts) { + continue; + } return nullptr; } } @@ -264,6 +275,21 @@ public: }; if (LIKELY(_stub_map.if_contains(host_port, check_entry))) { if (stub_ptr != nullptr) { + // When enable_handshake is on (during graceful shutdown), + // verify the cached channel is still reachable before + // returning it. Otherwise a cached stub pointing to an + // expired Pod IP will fail on the first real RPC. + if (enable_handshake && !available(stub_ptr, host_port)) { + LOG(WARNING) << "cached channel handshake failed to " + << host_port + << ", attempt=" << attempt << "/" + << max_attempts; + _stub_map.erase(host_port); + if (dns_cache != nullptr && !is_valid_ip(host)) { + dns_cache->invalidate(host); + } + continue; + } return stub_ptr; } DCHECK(need_remove); @@ -277,6 +303,9 @@ public: LOG(WARNING) << "failed to build brpc stub to " << real_host_port << ", attempt=" << attempt << "/" << max_attempts; if (enable_handshake) { + if (dns_cache != nullptr && !is_valid_ip(host)) { + dns_cache->invalidate(host); + } continue; } return nullptr; 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 c4ab8482409..ffcca0f6f53 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 @@ -33,6 +33,7 @@ import org.apache.doris.common.util.DebugPointUtil; import org.apache.doris.common.util.TimeUtils; import org.apache.doris.persist.gson.GsonUtils; import org.apache.doris.qe.SimpleScheduler; +import org.apache.doris.qe.VariableMgr; import org.apache.doris.resource.Tag; import org.apache.doris.rpc.BackendServiceProxy; import org.apache.doris.system.HeartbeatResponse.HbStatus; @@ -160,6 +161,10 @@ public class Backend implements Writable { // No need to persist, because only master FE handle heartbeat. private int heartbeatFailureCounter = 0; + private static final int GRACEFUL_ALIVE_HEARTBEAT_CONFIRM_COUNT = 3; + private int gracefulAliveHeartbeatOkCount = 0; + private long gracefulAliveHeartbeatStartTime = -1; + private long nextForceEditlogHeartbeatTime = System.currentTimeMillis() + (new SecureRandom()).nextInt(60 * 1000); public Backend() { @@ -888,6 +893,10 @@ public class Backend implements Writable { && preHbStartTime > 0 && ((newStartTime > 0 && preHbStartTime != newStartTime) || !preHbIsAlive); if (hbResponse.getStatus() == HbStatus.OK) { + if (shouldDelayAliveByGracefulHeartbeat(hbResponse, isReplay)) { + return false; + } + if (!this.version.equals(hbResponse.getVersion())) { isChanged = true; this.version = hbResponse.getVersion(); @@ -965,6 +974,7 @@ public class Backend implements Writable { this.nextForceEditlogHeartbeatTime = System.currentTimeMillis() + delaySecond * 1000L; } } else { + resetGracefulAliveHeartbeat(); // for a bad BackendHbResponse, its hbTime is last succ hbTime, not this hbTime if (hbResponse.getHbTime() > 0) { this.lastUpdateMs = hbResponse.getHbTime(); @@ -1014,6 +1024,53 @@ public class Backend implements Writable { return isChanged; } + private boolean shouldDelayAliveByGracefulHeartbeat(BackendHbResponse hbResponse, boolean isReplay) { + if (isReplay || hbResponse.isShutDown() || !isGracefulShutdownEnabled()) { + resetGracefulAliveHeartbeat(); + return false; + } + if (isAlive.get() && !isShutDown()) { + resetGracefulAliveHeartbeat(); + return false; + } + + long beStartTime = hbResponse.getBeStartTime(); + if (gracefulAliveHeartbeatStartTime != beStartTime) { + gracefulAliveHeartbeatStartTime = beStartTime; + gracefulAliveHeartbeatOkCount = 0; + } + gracefulAliveHeartbeatOkCount++; + + if (gracefulAliveHeartbeatOkCount < GRACEFUL_ALIVE_HEARTBEAT_CONFIRM_COUNT) { + LOG.info("{} delays marking backend alive during graceful shutdown, " + + "okHeartbeatCount={}/{}, beStartTime={} ({}), isReplay={}", + this.toString(), gracefulAliveHeartbeatOkCount, + GRACEFUL_ALIVE_HEARTBEAT_CONFIRM_COUNT, beStartTime, + TimeUtils.longToTimeString(beStartTime), isReplay); + return true; + } + + LOG.info("{} confirms backend alive during graceful shutdown after {} consecutive OK heartbeats, " + + "beStartTime={} ({})", + this.toString(), gracefulAliveHeartbeatOkCount, beStartTime, + TimeUtils.longToTimeString(beStartTime)); + resetGracefulAliveHeartbeat(); + return false; + } + + private boolean isGracefulShutdownEnabled() { + try { + return VariableMgr.getDefaultSessionVariable().enableGracefulShutdown; + } catch (Throwable t) { + return false; + } + } + + private void resetGracefulAliveHeartbeat() { + gracefulAliveHeartbeatOkCount = 0; + gracefulAliveHeartbeatStartTime = -1; + } + /** * Public entry to invalidate FE-side connection caches (DNSCache + gRPC channels + * Thrift pool) targeting this backend. Safe to call from any thread; all three @@ -1252,4 +1309,3 @@ public class Backend implements Writable { } } - --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
