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 461842ebf9c9e465be9d55a993b3b47b47518cf6 Author: Mingyu Chen (Rayner) <[email protected]> AuthorDate: Fri Apr 24 08:28:53 2026 -0700 [fix](rpc) proactively invalidate FE-side caches on BE restart to avoid first-query RPC failure (#8576) After a BE pod restart in K8s, the first fragment RPC often fails with "UNAVAILABLE: io exception" even though heartbeat has already succeeded. Root cause: FE keeps stale state that heartbeat does not touch — DNSCache, per-shard gRPC channels (BackendServiceProxy), and Thrift pool. The first query racing with these stale entries hits a channel pointing at the old pod IP. Fix: when handleHbResponse sees a new beStartTime (or dead->alive with a prior valid epoch), run onBackendRestartDetected to drop DNSCache, shut down cached gRPC channels on every proxy shard, and clear the Thrift backendPool. Also keep the existing blacklist-stability guard (SimpleScheduler.isRecentlyRecorded) as a backstop for any residual first-failure before cleanup lands. Add INFO-level diagnostics so future incidents have full traceability: per-shard channel ConnectivityState at removal time, DNSCache entry transitions, getProxy slow-path branch reasons, and a pre/post state snapshot (epoch, isAlive, isAvailable, isShutDown, isQueryDisabled) when restart is detected. --------- Co-authored-by: Claude Opus 4.7 <[email protected]> Co-authored-by: Copilot <[email protected]> --- .../main/java/org/apache/doris/catalog/Env.java | 12 +++- .../java/org/apache/doris/common/DNSCache.java | 22 +++++++ .../org/apache/doris/mysql/AcceptListener.java | 4 +- .../java/org/apache/doris/qe/SimpleScheduler.java | 33 +++++++++- .../java/org/apache/doris/qe/StmtExecutor.java | 16 +++++ .../org/apache/doris/rpc/BackendServiceClient.java | 5 ++ .../org/apache/doris/rpc/BackendServiceProxy.java | 50 ++++++++++++++- .../main/java/org/apache/doris/system/Backend.java | 72 ++++++++++++++++++++++ .../org/apache/doris/qe/SimpleSchedulerTest.java | 3 + 9 files changed, 208 insertions(+), 9 deletions(-) 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 dc8cac76500..7605a93639c 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 @@ -1733,7 +1733,17 @@ public class Env { // set this after replay thread stopped. to avoid replay thread modify them. isReady.set(false); - canRead.set(false); + // Do NOT set canRead to false here. + // At this point, the replayer thread has been stopped (see above), so the metadata + // is frozen at the last successfully replayed state. This metadata is consistent and + // safe for read queries — similar to the normal follower read delay. + // By keeping canRead=true, read queries can still be served locally during the + // transferToMaster() window (which can take several seconds due to editLog.open(), + // fencing, etc.), instead of failing with "Node catalog is not ready". + // This is consistent with the design in transferToNonMaster() which also preserves + // canRead when transitioning from FOLLOWER to UNKNOWN. + // canRead will be explicitly set to true at the end of this method (along with isReady) + // once the full master transition is complete. toMasterProgress = "open editlog"; editLog.open(); diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/DNSCache.java b/fe/fe-core/src/main/java/org/apache/doris/common/DNSCache.java index 395130e2659..1dddd4f7e8e 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/common/DNSCache.java +++ b/fe/fe-core/src/main/java/org/apache/doris/common/DNSCache.java @@ -60,6 +60,27 @@ public class DNSCache { return cache.computeIfAbsent(hostname, this::resolveHostname); } + /** + * Remove the cached entry for a hostname so the next get() will trigger a fresh DNS resolution. + * Used to proactively invalidate stale entries when a backend is known to have restarted. + * + * @param hostname The hostname whose cached entry should be evicted. + * @return The previously cached IP (may be null if not cached). + */ + public String remove(String hostname) { + String old = cache.remove(hostname); + LOG.info("DNSCache remove entry, hostname={}, previousCachedIp={}, remainingSize={}", + hostname, old, cache.size()); + return old; + } + + /** + * Returns a snapshot of current cache state. For diagnostic logging only. + */ + public String describeState() { + return "DNSCache{size=" + cache.size() + ", entries=" + cache + "}"; + } + /** * The resolveHostname method resolves a hostname to an IP address. * If the hostname cannot be resolved, it returns an empty string. @@ -80,6 +101,7 @@ public class DNSCache { return ""; } } + LOG.info("DNSCache resolved hostname={}, ip={}", hostname, ip); return ip; } catch (UnknownHostException e) { if (cachedIp != null && !cachedIp.isEmpty()) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/mysql/AcceptListener.java b/fe/fe-core/src/main/java/org/apache/doris/mysql/AcceptListener.java index edc35d99d57..5597330b072 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/mysql/AcceptListener.java +++ b/fe/fe-core/src/main/java/org/apache/doris/mysql/AcceptListener.java @@ -58,9 +58,7 @@ public class AcceptListener implements ChannelListener<AcceptingChannel<StreamCo return; } connection.setOption(Options.KEEP_ALIVE, true); - if (LOG.isDebugEnabled()) { - LOG.debug("Connection established. remote={}", connection.getPeerAddress()); - } + LOG.info("MySQL connection established. remote={}", connection.getPeerAddress()); // connection has been established, so need to call context.cleanup() // if exception happens. ConnectContext context = new ConnectContext(connection); diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SimpleScheduler.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SimpleScheduler.java index a9a593f41a6..a3b3ca22e37 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/SimpleScheduler.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SimpleScheduler.java @@ -154,6 +154,24 @@ public class SimpleScheduler { } } + // Check if there are recent error records being accumulated. + // This prevents clearing a backend entry that is actively receiving errors + // but hasn't yet reached the blacklist threshold count. + // Without this, the UpdateBlacklistThread would clear the entry every second + // (because isAlive=true), resetting the counter, so the threshold would never be reached. + public boolean isRecentlyRecorded() { + lock.lock(); + try { + if (lastRecordBlackTimestampMs <= 0) { + return false; + } + return System.currentTimeMillis() - lastRecordBlackTimestampMs + < Config.do_add_backend_black_list_threshold_seconds * 1000; + } finally { + lock.unlock(); + } + } + public String getReasonForBlackList() { lock.lock(); try { @@ -384,10 +402,19 @@ public class SimpleScheduler { LOG.info("remove backend {} from black list because it does not exist", backendId); } else { BlackListInfo blackListInfo = entry.getValue(); - if (backend.isAlive() || blackListInfo.shouldBeRemoved()) { + if (blackListInfo.shouldBeRemoved()) { + // Backend has been in blacklist long enough, remove it + iterator.remove(); + LOG.info("remove backend {} from black list. timeout reached," + + " backend is alive: {}", backendId, backend.isAlive()); + } else if (backend.isAlive() && !blackListInfo.isBlacked() + && !blackListInfo.isRecentlyRecorded()) { + // Backend is alive, was never fully blacklisted, + // and has no recent error records. Safe to remove. iterator.remove(); - LOG.info("remove backend {} from black list. backend is alive: {}", - backendId, backend.isAlive()); + LOG.info("remove backend {} from black list." + + " backend is alive, not fully blacked," + + " and no recent error records", backendId); } else { if (LOG.isDebugEnabled()) { LOG.debug("blacklistBackends {}", blackListInfo.toString()); diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java index dac8b50ac05..d65efb5e88d 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java @@ -1064,6 +1064,22 @@ public class StmtExecutor { } catch (InterruptedException e) { LOG.info("stmt executor sleep wait InterruptedException: ", e); } + } else { + // For non-cloud mode, also add a backoff delay before retry. + // During K8s rolling restarts, BE pods may be temporarily unreachable + // (NoRouteToHostException / gRPC UNAVAILABLE) even after heartbeat + // reports them alive. Without backoff, retries fire immediately and + // hit the same broken gRPC channel. The backoff gives K8s network + // (conntrack/iptables) time to settle and the gRPC channel time + // to reconnect. + int backoffMs = Math.min(1000 * i, 3000); + LOG.info("retry {} with backoff {}ms due to RPC error, query id: {}", + i, backoffMs, DebugUtil.printId(queryId)); + try { + Thread.sleep(backoffMs); + } catch (InterruptedException e) { + LOG.info("stmt executor retry sleep interrupted: ", e); + } } } if (context.getConnectType() == ConnectType.ARROW_FLIGHT_SQL) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceClient.java b/fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceClient.java index 13489e1894f..fb894f75e38 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceClient.java +++ b/fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceClient.java @@ -65,6 +65,11 @@ public class BackendServiceClient { return channelConfigVersion == CHANNEL_PROVIDER.currentConfigVersion(); } + // Return the current gRPC ConnectivityState for diagnostics/logging. + public ConnectivityState getConnectivityState() { + return channel.getState(false); + } + public Future<InternalService.PExecPlanFragmentResult> execPlanFragmentAsync( InternalService.PExecPlanFragmentRequest request) { return stub.withDeadlineAfter(execPlanTimeout, TimeUnit.MILLISECONDS) diff --git a/fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceProxy.java b/fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceProxy.java index 15f35d37117..4fa6f346f10 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceProxy.java +++ b/fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceProxy.java @@ -113,8 +113,39 @@ public class BackendServiceProxy { } if (serviceClientExtIp != null) { + LOG.info("remove gRPC proxy on shard, address={}, cachedRealIp={}, channelState={}", + address, serviceClientExtIp.realIp, serviceClientExtIp.client.getConnectivityState()); serviceClientExtIp.client.shutdown(); + } else { + LOG.info("remove gRPC proxy on shard, address={}, no cached channel", address); + } + } + + // Remove the cached gRPC channel for this address on ALL proxy shards. + // Used when a BE restart is detected, to make sure no stale channel survives on any shard. + public static void removeProxyForAll(TNetworkAddress address) { + int shardCount = Holder.proxies.length; + LOG.info("removeProxyForAll begin, address={}, shardCount={}", address, shardCount); + int removed = 0; + for (int i = 0; i < shardCount; i++) { + BackendServiceProxy proxy = Holder.proxies[i]; + BackendServiceClientExtIp serviceClientExtIp; + proxy.lock.lock(); + try { + serviceClientExtIp = proxy.serviceMap.remove(address); + } finally { + proxy.lock.unlock(); + } + if (serviceClientExtIp != null) { + LOG.info("removeProxyForAll shard={}, address={}, cachedRealIp={}, channelState={}", + i, address, serviceClientExtIp.realIp, + serviceClientExtIp.client.getConnectivityState()); + serviceClientExtIp.client.shutdown(); + removed++; + } } + LOG.info("removeProxyForAll done, address={}, removedShardCount={}, totalShardCount={}", + address, removed, shardCount); } private BackendServiceClient getProxy(TNetworkAddress address) throws UnknownHostException { @@ -137,17 +168,25 @@ public class BackendServiceProxy { // not exist, create one and return. BackendServiceClient removedClient = null; + String removedCause = null; + String removedPrevIp = null; lock.lock(); try { serviceClientExtIp = serviceMap.get(address); if (serviceClientExtIp != null && !serviceClientExtIp.realIp.equals(realIp)) { - LOG.warn("Cached ip changed, before ip: {}, curIp: {}", serviceClientExtIp.realIp, realIp); + LOG.info("getProxy: cached ip changed, address={}, beforeIp={}, curIp={}, channelState={}", + address, serviceClientExtIp.realIp, realIp, + serviceClientExtIp.client.getConnectivityState()); + removedPrevIp = serviceClientExtIp.realIp; + removedCause = "ip-changed"; serviceMap.remove(address); removedClient = serviceClientExtIp.client; serviceClientExtIp = null; } if (serviceClientExtIp != null && !serviceClientExtIp.client.isUsingLatestChannelConfig()) { LOG.info("BackendServiceClient channel config changed, recreate client for {}", address); + removedPrevIp = serviceClientExtIp.realIp; + removedCause = "config-changed"; serviceMap.remove(address); removedClient = serviceClientExtIp.client; serviceClientExtIp = null; @@ -156,12 +195,19 @@ public class BackendServiceProxy { // At this point we cannot judge the progress of reconnecting the underlying channel. // In the worst case, it may take two minutes. But we can't stand the connection refused // for two minutes, so rebuild the channel directly. + LOG.info("getProxy: cached channel not in normal state, address={}, cachedRealIp={}, " + + "channelState={}", address, serviceClientExtIp.realIp, + serviceClientExtIp.client.getConnectivityState()); + removedPrevIp = serviceClientExtIp.realIp; + removedCause = "abnormal-state"; serviceMap.remove(address); removedClient = serviceClientExtIp.client; serviceClientExtIp = null; } if (serviceClientExtIp == null) { - // Pass resolved IP to BackendServiceClient to avoid DNS resolution at gRPC layer + LOG.info("getProxy: creating new channel, address={}, realIpFromDnsCache={}, " + + "prevIp={}, cause={}", address, realIp, removedPrevIp, + removedCause == null ? "no-cached-client" : removedCause); BackendServiceClient client = new BackendServiceClient(address, realIp, grpcThreadPool); serviceMap.put(address, new BackendServiceClientExtIp(realIp, client)); } 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 d403c88732e..cdc250133c0 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 @@ -22,7 +22,9 @@ import org.apache.doris.catalog.DiskInfo.DiskState; import org.apache.doris.catalog.Env; import org.apache.doris.cloud.proto.Cloud; import org.apache.doris.cloud.system.CloudSystemInfoService; +import org.apache.doris.common.ClientPool; import org.apache.doris.common.Config; +import org.apache.doris.common.DNSCache; import org.apache.doris.common.FeConstants; import org.apache.doris.common.io.Text; import org.apache.doris.common.io.Writable; @@ -32,6 +34,7 @@ import org.apache.doris.common.util.TimeUtils; import org.apache.doris.persist.gson.GsonUtils; import org.apache.doris.qe.SimpleScheduler; import org.apache.doris.resource.Tag; +import org.apache.doris.rpc.BackendServiceProxy; import org.apache.doris.system.HeartbeatResponse.HbStatus; import org.apache.doris.thrift.TBackend; import org.apache.doris.thrift.TDisk; @@ -870,6 +873,20 @@ public class Backend implements Writable { */ public boolean handleHbResponse(BackendHbResponse hbResponse, boolean isReplay) { boolean isChanged = false; + // Snapshot state BEFORE any mutation, so we can detect a BE process restart + // (new beStartTime/epoch) and run proactive cleanup at the end of this method. + // Trigger cleanup when: + // 1. beStartTime changed (process restarted; covers the "alive whole time" case too), OR + // 2. BE was dead and is now coming back alive (may or may not have a new epoch, but + // stale FE-side channels are still expected). + final long preHbStartTime = this.lastStartTime; + final boolean preHbIsAlive = this.isAlive.get(); + final long newStartTime = hbResponse.getStatus() == HbStatus.OK ? hbResponse.getBeStartTime() : 0L; + // Require a prior valid epoch (preHbStartTime > 0) so the very first heartbeat of a + // freshly-added BE does not trigger a bogus "restart detected" cleanup/log. + final boolean restartDetected = hbResponse.getStatus() == HbStatus.OK + && preHbStartTime > 0 + && ((newStartTime > 0 && preHbStartTime != newStartTime) || !preHbIsAlive); if (hbResponse.getStatus() == HbStatus.OK) { if (!this.version.equals(hbResponse.getVersion())) { isChanged = true; @@ -968,9 +985,64 @@ 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) { + try { + onBackendRestartDetected(preHbStartTime, newStartTime, preHbIsAlive); + } catch (Throwable t) { + LOG.warn("onBackendRestartDetected failed, 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={}. " + + "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()); + + 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); + } else { + LOG.info("BE restart cleanup: DNSCache not available, skipped. backendId={}", id); + } + + if (brpcPort > 0) { + TNetworkAddress brpcAddr = new TNetworkAddress(host, brpcPort); + LOG.info("BE restart cleanup: dropping gRPC channels on all shards. backendId={}, " + + "brpcAddr={}", id, brpcAddr); + BackendServiceProxy.removeProxyForAll(brpcAddr); + } + + if (bePort > 0) { + TNetworkAddress bePoolAddr = new TNetworkAddress(host, bePort); + ClientPool.backendPool.clearPool(bePoolAddr); + LOG.info("BE restart cleanup: cleared Thrift backendPool. backendId={}, bePoolAddr={}", + id, bePoolAddr); + } + + LOG.info("BE restart cleanup done. backendId={}, host={}", id, host); + } + public void setTabletMaxCompactionScore(long compactionScore) { tabletMaxCompactionScore = compactionScore; } diff --git a/fe/fe-core/src/test/java/org/apache/doris/qe/SimpleSchedulerTest.java b/fe/fe-core/src/test/java/org/apache/doris/qe/SimpleSchedulerTest.java index f33508ca21d..ed877e24acf 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/qe/SimpleSchedulerTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/qe/SimpleSchedulerTest.java @@ -53,6 +53,9 @@ public class SimpleSchedulerTest { public static void setUp() { SimpleScheduler.init(); Config.heartbeat_interval_second = 2; + // Shrink blacklist removal threshold so the test does not need to wait + // the default 60s for an alive backend to be evicted from the blacklist. + Config.stay_in_backend_black_list_threshold_seconds = 3; be1 = new Backend(1000L, "192.168.100.0", 9050); be2 = new Backend(1001L, "192.168.100.1", 9050); be3 = new Backend(1002L, "192.168.100.2", 9050); --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
