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 05f2248c44621e04d179f4ddb3c8a1f94de6ada4 Author: zhangstar333 <[email protected]> AuthorDate: Wed May 27 23:57:14 2026 +0800 add session enableGracefulShutdown avoid cancel_worker --- be/src/agent/heartbeat_server.cpp | 17 +++++ be/src/common/config.cpp | 3 - be/src/common/config.h | 9 --- be/src/runtime/exec_env.h | 7 +++ be/src/runtime/fragment_mgr.cpp | 17 ++++- be/src/util/brpc_client_cache.h | 17 +++-- be/src/util/thrift_rpc_helper.cpp | 45 ++++++------- .../main/java/org/apache/doris/qe/Coordinator.java | 26 ++++++++ .../java/org/apache/doris/qe/SessionVariable.java | 18 ++++++ .../org/apache/doris/rpc/BackendServiceProxy.java | 73 +++++++++++++++++++++- .../java/org/apache/doris/system/HeartbeatMgr.java | 10 +++ gensrc/thrift/HeartbeatService.thrift | 7 +++ 12 files changed, 201 insertions(+), 48 deletions(-) diff --git a/be/src/agent/heartbeat_server.cpp b/be/src/agent/heartbeat_server.cpp index 2253e096b17..390f46f0f8d 100644 --- a/be/src/agent/heartbeat_server.cpp +++ b/be/src/agent/heartbeat_server.cpp @@ -251,6 +251,23 @@ Status HeartbeatServer::_heartbeat(const TMasterInfo& master_info) { master_info.network_address.hostname, master_info.network_address.port); } + // Propagate the cluster-level "graceful shutdown" flag pushed from FE master + // (set via SET GLOBAL enable_graceful_shutdown = true). When true, BE + // FragmentMgr::cancel_worker will skip cancelling queries for reasons related + // to coord FE process_uuid change / missing, so rolling restart does not kill + // in-flight stream load / select. + { + const bool new_flag = + master_info.__isset.in_graceful_shutdown ? master_info.in_graceful_shutdown : false; + if (new_flag != doris::k_in_graceful_shutdown) { + LOG(INFO) << "cluster in_graceful_shutdown flag changed: " + << doris::k_in_graceful_shutdown << " -> " << new_flag + << " (master=" << master_info.network_address.hostname << ":" + << master_info.network_address.port << ")"; + doris::k_in_graceful_shutdown = new_flag; + } + } + if (master_info.__isset.meta_service_endpoint != config::is_cloud_mode()) { LOG(WARNING) << "Detected mismatch in cloud mode configuration between FE and BE. " << "FE cloud mode: " diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp index 5c193ec603c..b2d51ecbb6d 100644 --- a/be/src/common/config.cpp +++ b/be/src/common/config.cpp @@ -885,9 +885,6 @@ DEFINE_Int64(brpc_max_body_size, "3147483648"); DEFINE_Int64(brpc_socket_max_unwritten_bytes, "-1"); DEFINE_mBool(brpc_usercode_in_pthread, "false"); -DEFINE_mBool(enable_brpc_get_client_handshake, "true"); // test it -DEFINE_mInt32(brpc_get_client_handshake_max_retries, "3"); - // TODO(zxy): expect to be true in v1.3 // Whether to embed the ProtoBuf Request serialized string together with Tuple/Block data into // Controller Attachment and send it through http brpc when the length of the Tuple/Block data diff --git a/be/src/common/config.h b/be/src/common/config.h index 670f0284a83..ad8b0e4f1fb 100644 --- a/be/src/common/config.h +++ b/be/src/common/config.h @@ -944,15 +944,6 @@ DECLARE_Int64(brpc_socket_max_unwritten_bytes); // Whether to set FLAGS_usercode_in_pthread to true in brpc DECLARE_mBool(brpc_usercode_in_pthread); -// Whether BrpcClientCache::get_client() should handshake-validate a freshly -// rebuilt stub before caching/returning it, and retry on failure. Costs 1 RTT -// per cache-miss / rebuild (the warm-cache fast path is untouched). Default -// false to preserve current behavior; flip on (or wire a session variable into -// get_client) to harden rolling-restart scenarios. -DECLARE_mBool(enable_brpc_get_client_handshake); -// Max attempts in get_client when handshake validation is enabled. Each -// failed attempt evicts the stub and marks DNSCache dirty for re-resolution. -DECLARE_mInt32(brpc_get_client_handshake_max_retries); // TODO(zxy): expect to be true in v1.3 // Whether to embed the ProtoBuf Request serialized string together with Tuple/Block data into // Controller Attachment and send it through http brpc when the length of the Tuple/Block data diff --git a/be/src/runtime/exec_env.h b/be/src/runtime/exec_env.h index dfbd1eee0dd..8a67fccd42d 100644 --- a/be/src/runtime/exec_env.h +++ b/be/src/runtime/exec_env.h @@ -143,6 +143,13 @@ inline bool k_doris_exit = false; // set to true once FE Master has pulled a heartbeat carrying is_shutdown=true. // Used by Phase A of graceful shutdown to know FE has been informed. inline bool k_shutdown_fe_known = false; +// Cluster-level "graceful shutdown" flag pushed by FE master via +// TMasterInfo.in_graceful_shutdown (operator does `SET GLOBAL +// enable_graceful_shutdown=true` before rolling restart). When true, +// FragmentMgr::cancel_worker will skip cancelling queries for reasons related to +// coord FE process_uuid change / coord-missing. Timeout / pipeline_task_leak +// cancels are NOT affected. Updated in HeartbeatServer::_heartbeat. +inline bool k_in_graceful_shutdown = false; // set to true after BE start ready inline bool k_is_server_ready = false; diff --git a/be/src/runtime/fragment_mgr.cpp b/be/src/runtime/fragment_mgr.cpp index e56fe738fa0..8b915da8cc7 100644 --- a/be/src/runtime/fragment_mgr.cpp +++ b/be/src/runtime/fragment_mgr.cpp @@ -907,16 +907,31 @@ void FragmentMgr::_collect_invalid_queries( auto itr = running_fes.find(q_ctx->coord_addr); if (itr != running_fes.end()) { - if (fe_process_uuid >= itr->second.info.process_uuid || + if (fe_process_uuid == itr->second.info.process_uuid || itr->second.info.process_uuid == 0) { continue; } else { + if (doris::k_in_graceful_shutdown) { + LOG_EVERY_N(INFO, 50) + << "Skip cancel coord-restarted query " + << print_id(q_ctx->query_id()) + << " because cluster is in graceful shutdown. coord=" + << q_ctx->coord_addr.hostname << ":" << q_ctx->coord_addr.port; + continue; + } LOG_WARNING( "Coordinator of query {} restarted, going to cancel " "it.", print_id(q_ctx->query_id())); } } else { + if (doris::k_in_graceful_shutdown) { + LOG_EVERY_N(INFO, 50) + << "Skip cancel coord-missing query " << print_id(q_ctx->query_id()) + << " because cluster is in graceful shutdown. coord=" + << q_ctx->coord_addr.hostname << ":" << q_ctx->coord_addr.port; + continue; + } // In some rear cases, the rpc port of follower is not updated in time, // then the port of this follower will be zero, but acutally it is still running, // and be has already received the query from follower. diff --git a/be/src/util/brpc_client_cache.h b/be/src/util/brpc_client_cache.h index b4ebb10acea..9daffd7d69e 100644 --- a/be/src/util/brpc_client_cache.h +++ b/be/src/util/brpc_client_cache.h @@ -217,21 +217,18 @@ public: } std::shared_ptr<T> get_client(const std::string& host, int port) { - // Default: read the global config switch. Callers with session context - // (e.g. a RuntimeState query option) should call the 3-arg overload - // below and pass an explicit bool — that's the integration point for a - // future session-level "enable handshake on get_client" variable. - return get_client(host, port, config::enable_brpc_get_client_handshake); + // The handshake-and-retry path is only enabled while the cluster-level + // graceful shutdown flag is on (operator did + // `SET GLOBAL enable_graceful_shutdown=true` before rolling restart). + // Outside of rolling restart, get_client stays one-shot. + return get_client(host, port, doris::k_in_graceful_shutdown); } // Explicit-override entry point. Pass `enable_handshake=true` to force the // handshake-and-retry path (paid only on cache miss / rebuild — warm cache - // hits stay free). Pass `false` to disable. Max attempts is governed by - // config::brpc_get_client_handshake_max_retries (default 3). + // hits stay free). Pass `false` to disable. Max attempts is 3. std::shared_ptr<T> get_client(const std::string& host, int port, bool enable_handshake) { - const int max_attempts = enable_handshake - ? std::max(1, config::brpc_get_client_handshake_max_retries) - : 1; + const int max_attempts = enable_handshake ? 3 : 1; std::string host_port; for (int attempt = 1; attempt <= max_attempts; ++attempt) { std::string realhost = host; diff --git a/be/src/util/thrift_rpc_helper.cpp b/be/src/util/thrift_rpc_helper.cpp index f7881ca755c..63b3cd54125 100644 --- a/be/src/util/thrift_rpc_helper.cpp +++ b/be/src/util/thrift_rpc_helper.cpp @@ -48,6 +48,8 @@ class TTransport; namespace doris { +using namespace std::chrono_literals; + using apache::thrift::protocol::TProtocol; using apache::thrift::protocol::TBinaryProtocol; using apache::thrift::transport::TSocket; @@ -225,6 +227,17 @@ Status ThriftRpcHelper::rpc_fe_with_master_refresh( // rpc<FrontendServiceClient>. Status st = rpc<FrontendServiceClient>(address_provider, callback, timeout_ms); + // Outside of a graceful rolling restart (operator-controlled + // `SET GLOBAL enable_graceful_shutdown=true`), keep strictly one-shot + // behavior: return the first-attempt result as-is, without probing + // followers or refreshing the cached master. The caller will see the + // raw TStatus (including NOT_MASTER) and decide what to do. This is + // equivalent to the legacy behavior before rpc_fe_with_master_refresh + // was introduced. + if (!doris::k_in_graceful_shutdown) { + return st; + } + bool transport_failed = !st.ok(); bool not_master = false; TNetworkAddress new_master; @@ -259,12 +272,6 @@ Status ThriftRpcHelper::rpc_fe_with_master_refresh( // Scenario A: transport failure; or NOT_MASTER without master_address // (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 @@ -274,13 +281,7 @@ Status ThriftRpcHelper::rpc_fe_with_master_refresh( // 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); - + constexpr int kMaxProbes = 3; auto running = _s_exec_env->get_running_frontends(); int probed = 0; bool master_refreshed = false; @@ -289,9 +290,6 @@ Status ThriftRpcHelper::rpc_fe_with_master_refresh( 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) { @@ -302,12 +300,6 @@ Status ThriftRpcHelper::rpc_fe_with_master_refresh( } ++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); @@ -315,6 +307,9 @@ Status ThriftRpcHelper::rpc_fe_with_master_refresh( 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; } @@ -329,6 +324,9 @@ Status ThriftRpcHelper::rpc_fe_with_master_refresh( << " 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; } _s_exec_env->cluster_info()->master_fe_addr = *hinted; @@ -341,6 +339,9 @@ Status ThriftRpcHelper::rpc_fe_with_master_refresh( // 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; } diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java index 74e0128e8da..ab8a318094b 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java @@ -1347,17 +1347,37 @@ public class Coordinator implements CoordInterface { // We use a very conservative cancel strategy. // 0. If backends has zero process epoch, do not cancel. Zero process epoch usually arises in cluster upgrading. // 1. If process epoch is same, do not cancel. Means backends does not restart or die. + // 2. If process epoch is different, but cluster-level "graceful shutdown" flag is set, do not cancel. + // Means backends may restart, but we want to defer cancel until + // the backend actually restarts and becomes unavailable. public Status shouldCancel(List<Backend> currentBackends) { Map<Long, Backend> curBeMap = Maps.newHashMap(); for (Backend be : currentBackends) { curBeMap.put(be.getId(), be); } + // Cluster-level "graceful shutdown" flag set via `SET GLOBAL enable_graceful_shutdown`. + // When true, fully skip cancel for BE-death / epoch-change reasons cluster-wide + // so a rolling restart does not kill in-flight queries / stream loads. + // Timeout / explicit KILL paths are NOT affected by this flag. + boolean clusterGraceful = false; + try { + clusterGraceful = VariableMgr.getDefaultSessionVariable().enableGracefulShutdown; + } catch (Throwable t) { + // Defensive: ignore. + } + try { lock(); for (PipelineExecContext pipelineExecContext : pipelineExecContexts.values()) { Backend be = curBeMap.get(pipelineExecContext.backend.getId()); if (be == null || !be.isAlive()) { + if (clusterGraceful) { + LOG.info("enable_graceful_shutdown=true, defer cancel for query {} " + + "(BE {} not alive)", DebugUtil.printId(queryId), + pipelineExecContext.backend.toString()); + continue; + } Status errorStatus = new Status(TStatusCode.CANCELLED, "Backend {} not exists or dead, query {} should be cancelled", pipelineExecContext.backend.toString(), DebugUtil.printId(queryId)); @@ -1369,6 +1389,12 @@ public class Coordinator implements CoordInterface { // Check zero since during upgrading, older version oplog will not persistent be start time // so newer version follower will get zero epoch when replaying oplog or snapshot if (pipelineExecContext.beProcessEpoch != be.getProcessEpoch() && be.getProcessEpoch() != 0) { + if (clusterGraceful) { + LOG.info("enable_graceful_shutdown=true, defer cancel for query {} " + + "(BE {} epoch {} -> {})", DebugUtil.printId(queryId), + be.toString(), pipelineExecContext.beProcessEpoch, be.getProcessEpoch()); + continue; + } Status errorStatus = new Status(TStatusCode.CANCELLED, "Backend process epoch changed, previous {} now {}, " + "means this be has already restarted, should cancel this coordinator," 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 aa70a4f6ea6..1dcb8562dc3 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 @@ -188,6 +188,7 @@ public class SessionVariable implements Serializable, Writable { public static final String ENABLE_SERVER_SIDE_PREPARED_STATEMENT = "enable_server_side_prepared_statement"; public static final String MAX_PREPARED_STMT_COUNT = "max_prepared_stmt_count"; + public static final String ENABLE_GRACEFUL_SHUTDOWN = "enable_graceful_shutdown"; public static final String ENABLE_GROUP_COMMIT_FULL_PREPARE = "enable_group_commit_full_prepare"; public static final String PREFER_JOIN_METHOD = "prefer_join_method"; @@ -2312,6 +2313,23 @@ 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, + description = { + "集群级别开关：当 FE 或 BE 准备做滚动重启 / 优雅退出时，由运维 SET GLOBAL 打开。" + + "打开后：(1) FE QueryCancelWorker / Coordinator.shouldCancel 不再因 BE alive=false " + + "或 process epoch 变化 cancel in-flight query；(2) FE master 通过心跳把该标记下发" + + "给所有 BE，BE 端 FragmentMgr::cancel_worker 也跳过 'Coordinator restarted' / " + + "'Source frontend is not running' cancel。timeout / 用户主动 KILL / pipeline task " + + "leak 等其他 cancel 不受影响。滚动结束后必须 SET GLOBAL ... = false 复位。", + "Cluster-level switch operators flip on via SET GLOBAL before performing a rolling restart " + + "or graceful shutdown. When true: (1) FE QueryCancelWorker / Coordinator.shouldCancel " + + "skip cancelling in-flight queries on BEs whose alive=false or process_epoch changed; " + + "(2) FE master propagates the flag to all BEs via heartbeat so that BE " + + "FragmentMgr::cancel_worker also skips 'Coordinator restarted' / 'Source frontend is " + + "not running' cancel. Timeout / explicit KILL / pipeline-task-leak cancels are not " + + "affected. Operators MUST flip it back to false after rolling restart finishes."}) + public boolean enableGracefulShutdown = false; + @VarAttrDef.VarAttr(name = ENABLE_GROUP_COMMIT_FULL_PREPARE) public boolean enableGroupCommitFullPrepare = true; 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 4fa6f346f10..48fae0674b6 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 @@ -51,7 +51,10 @@ import java.net.UnknownHostException; import java.util.List; import java.util.Map; import java.util.concurrent.Executor; +import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.locks.ReentrantLock; @@ -65,6 +68,21 @@ public class BackendServiceProxy { Config.grpc_threadmgr_threads_nums, "grpc_thread_pool", true); + // Single-threaded scheduler used to *deferred-shutdown* gRPC channels that were + // invalidated by removeProxy / removeProxyForAll on BE restart. We MUST NOT call + // BackendServiceClient.shutdown() inline because it triggers channel.shutdownNow() + // (after a 5s grace) and instantly kills any in-flight RPC on the old channel with + // "UNAVAILABLE: Channel shutdownNow invoked". Instead we drop the client from the + // serviceMap (so new RPCs go to a freshly-created channel pointing at the new BE IP) + // and asynchronously close the old channel after a grace period long enough for + // in-flight RPCs to finish or fail naturally. + private static final ScheduledExecutorService deferredShutdownExecutor = + Executors.newSingleThreadScheduledExecutor(r -> { + Thread t = new Thread(r, "grpc-deferred-shutdown"); + t.setDaemon(true); + return t; + }); + private final Map<TNetworkAddress, BackendServiceClientExtIp> serviceMap; public BackendServiceProxy() { @@ -115,7 +133,7 @@ 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(); + scheduleDeferredShutdown(address, serviceClientExtIp.client, "removeProxy"); } else { LOG.info("remove gRPC proxy on shard, address={}, no cached channel", address); } @@ -140,7 +158,7 @@ public class BackendServiceProxy { LOG.info("removeProxyForAll shard={}, address={}, cachedRealIp={}, channelState={}", i, address, serviceClientExtIp.realIp, serviceClientExtIp.client.getConnectivityState()); - serviceClientExtIp.client.shutdown(); + scheduleDeferredShutdown(address, serviceClientExtIp.client, "removeProxyForAll"); removed++; } } @@ -148,6 +166,52 @@ public class BackendServiceProxy { address, removed, shardCount); } + // Schedule a deferred shutdown of the given gRPC client. The client has already been + // removed from serviceMap, so new RPCs will create a fresh client (pointing at the + // possibly-new BE IP). The old channel is kept alive for a grace window so that + // in-flight RPCs can either complete naturally or fail with the real underlying + // brpc error (e.g. "Network closed" when the BE socket RSTs) instead of being killed + // by a synchronous channel.shutdownNow(). + private static void scheduleDeferredShutdown(TNetworkAddress address, + BackendServiceClient client, String cause) { + // 120 seconds, should be long enough for in-flight RPCs to finish or fail naturally. + long delayMs = 120 * 1000L; + if (delayMs <= 0) { + // Disabled: fall back to legacy synchronous shutdown. + LOG.info("deferred shutdown disabled (delayMs={}), shutdown inline. address={}, cause={}", + delayMs, address, cause); + try { + client.shutdown(); + } catch (Throwable t) { + LOG.warn("inline client shutdown failed, address={}, err={}", address, t.getMessage(), t); + } + return; + } + LOG.info("schedule deferred gRPC channel shutdown, address={}, delayMs={}, cause={}, " + + "channelState={}", address, delayMs, cause, client.getConnectivityState()); + try { + deferredShutdownExecutor.schedule(() -> { + try { + LOG.info("deferred gRPC channel shutdown firing, address={}, cause={}, " + + "channelState={}", address, cause, client.getConnectivityState()); + client.shutdown(); + } catch (Throwable t) { + LOG.warn("deferred client shutdown failed, address={}, err={}", + address, t.getMessage(), t); + } + }, delayMs, TimeUnit.MILLISECONDS); + } catch (Throwable t) { + LOG.warn("failed to schedule deferred shutdown, fallback to inline. address={}, err={}", + address, t.getMessage(), t); + try { + client.shutdown(); + } catch (Throwable inner) { + LOG.warn("inline fallback shutdown failed, address={}, err={}", + address, inner.getMessage(), inner); + } + } + } + private BackendServiceClient getProxy(TNetworkAddress address) throws UnknownHostException { String realIp = Env.getCurrentEnv().getDnsCache().get(address.hostname); @@ -215,7 +279,10 @@ public class BackendServiceProxy { } finally { lock.unlock(); if (removedClient != null) { - removedClient.shutdown(); + // Use deferred shutdown to avoid killing in-flight RPCs on the old channel + // (same rationale as removeProxy/removeProxyForAll). + scheduleDeferredShutdown(address, removedClient, + removedCause == null ? "getProxy-replace" : "getProxy-" + removedCause); } } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/system/HeartbeatMgr.java b/fe/fe-core/src/main/java/org/apache/doris/system/HeartbeatMgr.java index dbd36715d53..b168c111c0c 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/system/HeartbeatMgr.java +++ b/fe/fe-core/src/main/java/org/apache/doris/system/HeartbeatMgr.java @@ -302,6 +302,16 @@ public class HeartbeatMgr extends MasterDaemon { copiedMasterInfo.setBackendId(backendId); copiedMasterInfo.setFrontendInfos(feInfos); copiedMasterInfo.setAuthToken(Env.getCurrentEnv().getTokenManager().acquireToken()); + // Propagate cluster-level "graceful shutdown" flag (SessionVariable set GLOBAL) + // to BE so that BE cancel_worker knows to skip coord-restart cancellation. + try { + copiedMasterInfo.setInGracefulShutdown( + org.apache.doris.qe.VariableMgr.getDefaultSessionVariable() + .enableGracefulShutdown); + } catch (Throwable t) { + // Defensive: VariableMgr might not be initialized in unit tests. + copiedMasterInfo.setInGracefulShutdown(false); + } if (Config.isCloudMode()) { String cloudUniqueId = backend.getTagMap().get(Tag.CLOUD_UNIQUE_ID); copiedMasterInfo.setCloudUniqueId(cloudUniqueId); diff --git a/gensrc/thrift/HeartbeatService.thrift b/gensrc/thrift/HeartbeatService.thrift index 3fde7fdf2df..4ed8ab280da 100644 --- a/gensrc/thrift/HeartbeatService.thrift +++ b/gensrc/thrift/HeartbeatService.thrift @@ -50,6 +50,13 @@ struct TMasterInfo { 12: optional i64 tablet_report_inactive_duration_ms; 13: optional string auth_token; 14: optional TCloudClusterInfo cloud_cluster_info; + // True when the FE master cluster is in "graceful shutdown" mode + // (SessionVariable.enable_graceful_shutdown set to true via SET GLOBAL). + // When true, BE.fragment_mgr cancel_worker MUST skip cancelling queries + // due to "Coordinator restarted" / "Source frontend is not running", + // so that in-flight stream load / select can finish or be retried via + // master-failover instead of being killed within 1s of FE restart. + 15: optional bool in_graceful_shutdown } struct TBackendInfo { --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
