github-actions[bot] commented on code in PR #65683:
URL: https://github.com/apache/doris/pull/65683#discussion_r3594467411
##########
be/src/cloud/cloud_warm_up_manager.cpp:
##########
@@ -815,7 +815,7 @@ Status CloudWarmUpManager::_do_warm_up_rowset(RowsetMeta&
rs_meta, int64_t table
Status st = Status::OK();
std::shared_ptr<PBackendService_Stub> brpc_stub =
ExecEnv::GetInstance()->brpc_internal_client_cache()->get_new_client_no_cache(
- brpc_addr);
+ brpc_addr, "", "", "", replica.host);
Review Comment:
[P0] Use the nested replica hostname at both call sites
This scope has no `replica` variable; the address is `info.replica.host`.
The analogous new call in `_recycle_cache()` at line 978 also uses
`replica.host` even though that loop element is `JobReplicaInfo` and the
surrounding code uses `replica.replica.host`. Both expressions fail the Cloud
BE compilation.
##########
be/src/exec/operator/exchange_sink_buffer.h:
##########
@@ -310,13 +310,13 @@ class ExchangeSinkBuffer : public HasTaskExecutionCtx {
#ifndef BE_TEST
inline void _ended(RpcInstance& ins);
- inline void _failed(InstanceLoId id, const std::string& err);
+ inline void _failed(InstanceLoId id, Status err);
inline void _set_receiver_eof(RpcInstance& ins);
inline void _turn_off_channel(RpcInstance& ins,
std::unique_lock<std::mutex>& with_lock);
#else
virtual void _ended(RpcInstance& ins);
- virtual void _failed(InstanceLoId id, const std::string& err);
+ virtual void _failed(InstanceLoId id, Status err);
Review Comment:
[P1] Migrate the BE test override with this signature
`MockSinkBuffer` in `be/test/exec/exchange/exchange_sink_test.h` still
declares `_failed(InstanceLoId, const std::string&) override`. After this
change it no longer overrides the base method, so the exchange-sink test target
fails to compile. Please update the mock to accept `Status` and assert the
newly preserved error code.
##########
be/src/util/thrift_rpc_helper.cpp:
##########
@@ -127,6 +129,79 @@ Status
ThriftRpcHelper::rpc(std::function<TNetworkAddress()> address_provider,
return Status::OK();
}
+template <typename T>
Review Comment:
[P0] Remove the duplicate template definition
This is a second definition of the exact
`rpc(std::function<TNetworkAddress()>, ..., int)` template already defined at
lines 68-130. C++ diagnoses the translation unit as a redefinition before the
explicit instantiations are reached, so the BE target cannot compile. Please
merge the intended changes into the existing body and keep only one definition.
##########
be/src/runtime/exec_env.h:
##########
@@ -140,6 +140,16 @@ class DeleteBitmapAggCache;
// set to true when BE is shutting down
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;
Review Comment:
[P1] Make the shutdown flags atomic
These flags are written by heartbeat Thrift workers but polled by the main
shutdown thread and read by FragmentMgr/RPC worker threads without a shared
lock. Concurrent access to these plain C++ `bool`s is a data race, so
acknowledgement and graceful-mode propagation have undefined behavior. Use
atomics with an explicit visibility contract (or one common synchronization
primitive) for both flags.
##########
be/src/service/doris_main.cpp:
##########
@@ -744,6 +751,10 @@ int main(int argc, char** argv) {
service.reset();
LOG(INFO) << "Backend Service stopped";
exec_env->destroy();
+ doris::ThreadLocalHandle::del_thread_local_if_count_is_zero();
Review Comment:
[P1] Let the scoped thread context release itself
`main()` already owns this handle through `SCOPED_INIT_THREAD_CONTEXT()`.
This explicit call deletes the only TLS context, and then the scope destructor
calls the same release routine again on `return 0`, reaching the fatal
no-pthread-context path. This breaks every clean `enable_graceful_exit_check`
shutdown; remove the manual deletion or end the RAII scope exactly once.
##########
be/src/agent/heartbeat_server.cpp:
##########
@@ -90,6 +90,12 @@ void HeartbeatServer::heartbeat(THeartbeatResult&
heartbeat_result,
heartbeat_result.backend_info.__set_be_node_role(config::be_node_role);
// If be is gracefully stop, then k_doris_exist is set to true
heartbeat_result.backend_info.__set_is_shutdown(doris::k_doris_exit);
+ if (doris::k_doris_exit) {
+ // FE Master has pulled at least one heartbeat carrying
+ // is_shutdown=true. Unblock Phase A in graceful_shutdown.
+ doris::k_shutdown_fe_known = true;
Review Comment:
[P1] Wait for a real FE acknowledgement
This flag is set while the BE is still constructing the heartbeat response.
FE cannot apply `is_shutdown` or write `OP_HEARTBEAT` until the RPC returns and
the heartbeat batch is collected, so a lost response makes BE proceed even
though no FE learned the state; a fixed sleep does not prove follower replay
either. Ack from a subsequent FE request only after the master has applied and
durably logged this generation.
##########
be/src/runtime/exec_env.cpp:
##########
@@ -170,20 +170,53 @@ std::map<TNetworkAddress, FrontendInfo>
ExecEnv::get_running_frontends() {
return res;
}
+void ExecEnv::wait_for_all_fe_known() {
+ // Phase A of graceful shutdown:
+ // Heartbeat callback (HeartbeatServer::heartbeat) sets
+ // k_shutdown_fe_known=true the next time Master pulls a heartbeat
+ // (which carries is_shutdown=true). Block here up to 30s for that to
+ // happen, then sleep an extra 10s so the resulting OP_HEARTBEAT EditLog
+ // can be replicated by BDBJE to all FE Followers before we start
+ // tearing down running tasks. The 30s deadline is a safety net for the
+ // case where Master is itself in drain / election and heartbeats are
+ // delayed.
+ constexpr int32_t kWaitLimitSeconds = 30;
+ constexpr int32_t kBufferSeconds = 10;
+ int32_t passed = 0;
+ while (!k_shutdown_fe_known && passed < kWaitLimitSeconds) {
+ sleep(1);
+ ++passed;
+ }
+ if (k_shutdown_fe_known) {
+ LOG(INFO) << "FE Master has acknowledged shutdown after " << passed
+ << "s, sleeping additional " << kBufferSeconds
+ << "s to let OP_HEARTBEAT EditLog replicate to all
Followers.";
+ } else {
+ LOG(WARNING) << "FE Master did not acknowledge shutdown within " <<
kWaitLimitSeconds
+ << "s, proceeding anyway.";
+ }
+ sleep(kBufferSeconds);
+}
+
void ExecEnv::wait_for_all_tasks_done() {
- // For graceful shutdown, need to wait for all running queries to stop
+ LOG(INFO) << "begin to wait for all tasks done before shutdown.
k_shutdown_fe_known: "
+ << k_shutdown_fe_known << " k_in_graceful_shutdown: " <<
k_in_graceful_shutdown;
+ // For graceful shutdown, need to wait for all running queries and load
channels to stop
int32_t wait_seconds_passed = 0;
while (true) {
int num_queries = _fragment_mgr->running_query_num();
- if (num_queries < 1) {
+ size_t num_load_channels =
_load_channel_mgr->get_active_load_channel_num();
Review Comment:
[P1] Include V2 receiver streams in the shutdown drain
This predicate counts only legacy `LoadChannelMgr` channels. The default
non-cloud `OlapTableSinkV2` receiver is registered in
`LoadStreamMgr::_load_streams_map` and can have active delta writers on a
receiver-only BE while both counters here are zero. The BE can therefore tear
down storage during an active write. Add synchronized drain accounting for
`LoadStreamMgr` (and audit `NewLoadStreamMgr`) before allowing shutdown.
##########
be/src/runtime/fragment_mgr.cpp:
##########
@@ -911,12 +911,27 @@ void FragmentMgr::_collect_invalid_queries(
itr->second.info.process_uuid == 0) {
continue;
} else {
+ if (doris::k_in_graceful_shutdown) {
Review Comment:
[P1] Still cancel after a confirmed coordinator epoch change
A nonzero FE process-UUID mismatch proves that the process owning this query
state is gone; its fragments cannot be adopted by the replacement FE. Skipping
this cleanup for the full rolling-restart flag strands BE work and resources
until query timeout. Limit any exemption to a bounded ambiguous liveness
window, and always cancel on a confirmed nonzero UUID change.
##########
fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java:
##########
@@ -1634,6 +1651,14 @@ public TLoadTxnCommitResult
loadTxnPreCommit(TLoadTxnCommitRequest request) thro
return result;
}
+ if (status.getStatusCode() != TStatusCode.OK) {
Review Comment:
[P1] Populate the precommit hint before the first return
The identical non-OK check immediately above already returns, so this branch
is unreachable and `loadTxnPreCommit` remains the one changed transaction RPC
that cannot redirect a NOT_MASTER retry. Set `masterAddress` in the first error
branch before returning and add a follower/precommit test.
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java:
##########
@@ -1733,7 +1734,17 @@ private void transferToMaster() {
// 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.
Review Comment:
[P1] Preserve the metadata-staleness bound during promotion
Once the replayer is joined, its `setCanRead()` loop is no longer available
to enforce `meta_delay_toleration_second`. If BDB open/fencing/replay stalls
beyond that bound, this change continues serving reads from the frozen catalog
indefinitely. Keep promotion reads only while the same staleness invariant
holds, with a timer/recheck that disables `canRead` when the configured bound
expires.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/VariableMgr.java:
##########
@@ -456,6 +461,29 @@ private static void setGlobalVarAndWriteEditLog(VarContext
ctx, String name, Str
}
}
+ 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;
Review Comment:
[P1] Restore the configured heartbeat interval, not a literal
Disabling this global mode always writes `10`, even when
`heartbeat_interval_second` was configured or dynamically set to another value.
For example, a deployment using 30 seconds becomes 3 while enabled and silently
becomes 10 afterward. Model the fast value as an override over the current base
configuration, or save/restore each FE's prior value.
##########
fe/fe-core/src/main/java/org/apache/doris/common/DNSCache.java:
##########
@@ -60,6 +60,27 @@ public String get(String hostname) {
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);
Review Comment:
[P1] Make DNS removal generation-safe
A refresh can resolve old IP A, this method can remove the entry during
restart cleanup, and that older refresh then observes `null` and publishes A
again. `ConcurrentHashMap` only makes individual operations safe; it does not
order invalidation against refresh publication. Track a per-host generation or
serialize refresh/remove per key, with a latch-based race test.
##########
fe/fe-core/src/main/java/org/apache/doris/system/Backend.java:
##########
@@ -968,9 +995,153 @@ public boolean handleHbResponse(BackendHbResponse
hbResponse, boolean isReplay)
lastMissingHeartbeatTime = System.currentTimeMillis();
}
+ // 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 {
+ 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("invalidateLocalConnections failed (heartbeat path),
backendId={}, "
+ + "host={}, err={}", id, host, t.getMessage(), t);
+ }
+ }
+
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
+ * 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.",
+ 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) {
+ 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);
+ }
+
+ 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);
Review Comment:
[P1] Prevent borrowed clients from re-entering after cleanup
`clearPool(key)` removes idle objects only. A client borrowed before this
restart cleanup can finish and return afterward, becoming an idle stale socket
that the next request borrows. Add a per-address generation (or close-on-return
marker) so clients from an invalidated generation are destroyed, and test
borrow/clear/return ordering.
##########
fe/fe-core/src/main/java/org/apache/doris/system/Backend.java:
##########
@@ -870,7 +878,25 @@ public String toString() {
*/
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
Review Comment:
[P1] Do not gate dead-to-alive cleanup on a nonzero old epoch
Older-version oplogs can legitimately leave an existing backend with epoch
0. In that state this outer gate also suppresses the explicit dead-to-alive
cleanup, so its first OK heartbeat can retain stale DNS/gRPC/Thrift clients.
Treat dead-to-alive as a cleanup trigger regardless of epoch; distinguish a
truly new backend by lifecycle state rather than a compatibility value.
##########
be/src/util/thrift_rpc_helper.cpp:
##########
@@ -143,4 +218,110 @@ template Status
ThriftRpcHelper::rpc<TPaloBrokerServiceClient>(
const std::string& ip, const int32_t port,
std::function<void(ClientConnection<TPaloBrokerServiceClient>&)>
callback, int timeout_ms);
+Status ThriftRpcHelper::rpc_fe_with_master_refresh(
+ std::function<TNetworkAddress()> address_provider,
+ std::function<void(ClientConnection<FrontendServiceClient>&)> callback,
+ std::function<TStatus()> status_extractor,
+ std::function<const TNetworkAddress*()> master_addr_extractor, int
timeout_ms) {
+ // First attempt: drives the existing one-shot transport-level retry inside
+ // 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;
+ bool have_new_master = false;
+
+ if (st.ok()) {
+ TStatus resp_status = status_extractor();
+ if (resp_status.status_code == TStatusCode::NOT_MASTER) {
+ not_master = true;
+ const TNetworkAddress* addr = master_addr_extractor();
+ if (addr != nullptr && !addr->hostname.empty() && addr->port != 0)
{
+ new_master = *addr;
+ have_new_master = true;
+ }
+ }
+ }
+
+ if (!transport_failed && !not_master) {
+ return st;
+ }
+
+ TNetworkAddress old_master = address_provider();
+ if (have_new_master) {
+ // Refresh cached master immediately from the response. No extra RPC.
+ // Note: unprotected write, same pattern as HeartbeatServer; the field
+ // is also written by heartbeat thread with epoch checks, which will
+ // self-correct any stale value we might race-write.
+ _s_exec_env->cluster_info()->master_fe_addr = new_master;
Review Comment:
[P1] Do not publish an unversioned hint into heartbeat-owned state
This unsynchronized assignment races on a string-bearing `TNetworkAddress`
with heartbeat writers/readers. It can also wedge routing: heartbeat may
install master C at epoch E, then a delayed response overwrites the address
with stale B without changing `_fe_epoch`; subsequent C/E heartbeats reject the
repair because the epoch is not greater. Keep the hint local to this retry or
route all master updates through a synchronized, epoch-aware owner.
##########
fe/fe-core/src/main/java/org/apache/doris/mysql/AcceptListener.java:
##########
@@ -58,9 +58,7 @@ public void handleEvent(AcceptingChannel<StreamConnection>
channel) {
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());
Review Comment:
[P2] Avoid per-connection INFO logging on the accept path
Graceful shutdown deliberately closes idle pooled connections and induces a
reconnect wave on surviving FEs. Promoting every accepted socket from debug to
INFO adds synchronous formatting/log I/O proportional to that storm. Keep this
at DEBUG, rate-limit it, or expose an aggregate reconnect metric.
##########
fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceProxy.java:
##########
@@ -113,7 +131,84 @@ public void removeProxy(TNetworkAddress address) {
}
if (serviceClientExtIp != null) {
- serviceClientExtIp.client.shutdown();
+ LOG.info("remove gRPC proxy on shard, address={}, cachedRealIp={},
channelState={}",
+ address, serviceClientExtIp.realIp,
serviceClientExtIp.client.getConnectivityState());
+ scheduleDeferredShutdown(address, serviceClientExtIp.client,
"removeProxy");
+ } 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;
Review Comment:
[P1] Order channel publication with restart invalidation
A concurrent `getProxy()` can resolve old IP A before cleanup, wait while
this loop removes the address from its shard, then acquire that shard lock and
insert a new A-backed client after eviction has passed. Cleanup returns with a
freshly stale channel. Capture an address/invalidation generation with DNS
resolution and revalidate it under the shard lock before publishing the client,
with a resolve-old / evict / insert race test.
##########
be/src/service/internal_service.cpp:
##########
@@ -355,6 +355,12 @@ void
PInternalService::exec_plan_fragment(google::protobuf::RpcController* contr
timeval tv {};
gettimeofday(&tv, nullptr);
response->set_received_time(tv.tv_sec * 1000LL + tv.tv_usec / 1000);
+ if (k_doris_exit) {
Review Comment:
[P1] Stop admitting fragments before taking the drain snapshot
This branch only logs and still queues the fragment. A request accepted
after `wait_for_all_tasks_done()` observes zero—or already waiting in
`_light_work_pool` before FragmentMgr registration—is absent from every drain
counter and can be terminated by the default `_exit(0)` path. Atomically reject
new prepare/execute admissions and account already-admitted work from queue
entry through registration/completion, while leaving required
completion/control RPCs available.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java:
##########
@@ -1369,6 +1389,12 @@ public Status shouldCancel(List<Backend>
currentBackends) {
// 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) {
Review Comment:
[P1] Do not suppress a confirmed BE epoch change
A nonzero process-epoch change means the old BE process and its submitted
fragments are gone; graceful draining cannot make that work resume. Returning
OK on every QueryCancelWorker pass leaves downstream fragments waiting and
retains resources until query timeout. Restrict the graceful exemption to a
transient `alive=false` state and keep cancellation for confirmed epoch changes.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java:
##########
@@ -2312,6 +2315,24 @@ public boolean isEnableHboNonStrictMatchingMode() {
"服务端 prepared statement 最大个数", "the maximum prepared
statements server holds."})
public int maxPreparedStmtCount = 100000;
+ @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 "
+ + "或 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;
Review Comment:
[P1] Publish the cluster flag to runtime readers
SET/replay writes this field while holding `VariableMgr.wlock`, but
heartbeat and cancellation workers read the mutable default `SessionVariable`
directly without acquiring the matching read lock. Because the field is not
`volatile`/atomic, enable or disable has no Java happens-before edge to those
workers. Provide a locked or volatile/atomic cluster-mode accessor and route
all runtime reads through it.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/QeService.java:
##########
@@ -66,7 +66,11 @@ public void start() throws Exception {
LOG.info("QE service start.");
}
- public void stop() {
+ /**
+ * Stop accepting new MySQL and Arrow Flight SQL connections while leaving
+ * existing connections to be drained by the graceful shutdown flow.
+ */
+ public void stopAcceptNewConnections() {
try {
if (flightStarter != null) {
flightStarter.stop();
Review Comment:
[P1] Do not close active Flight calls in the admission phase
`flightStarter.stop()` reaches `FlightServer.close()`. In [Arrow Java
19](https://github.com/apache/arrow-java/blob/v19.0.0/flight/flight-core/src/main/java/org/apache/arrow/flight/FlightServer.java#L145-L163),
`close()` waits only three seconds after shutdown and then calls
`shutdownNow()`, so a Flight query longer than three seconds is cancelled in
phase B despite the comment and the longer phase-C deadline. Separate new-call
rejection from server close, or defer `close()` until phase C completes.
##########
fe/fe-core/src/main/java/org/apache/doris/DorisFE.java:
##########
@@ -722,20 +735,95 @@ public static boolean isServerReady() {
return serverReady.get();
}
+ /**
+ * Returns true once graceful shutdown has entered phase B. New data-plane
+ * MySQL commands (COM_QUERY / COM_STMT_EXECUTE / ...) should be silently
+ * dropped so JDBC drivers fail over to another FE.
+ */
+ public static boolean shouldRejectNewQueries() {
+ return rejectNewQueries.get();
+ }
+
+ /**
+ * Graceful shutdown driven by SIGTERM (typically from K8s). Implements the
+ * A0 → A → B → C → D state machine described in the operator design doc.
+ *
+ * <pre>
+ * A0 serverReady=false → /api/health returns 503 (already done by
caller)
+ * A "drain": sleep graceful_shutdown_drain_seconds while still serving
+ * new queries, to let K8s LB/DNS converge.
+ * B atomic flip: rejectNewQueries=true, then close MySQL/ArrowFlight
+ * listeners, then close idle MySQL connections (FIN to client).
+ * C wait for in-flight Coordinators to finish (≤
grace_shutdown_wait_seconds),
+ * re-running closeIdleMysqlConnections() once per second.
+ * D return; JVM exits after shutdown hook completes.
+ * </pre>
+ */
private static void gracefulShutdown() {
- // wait for all queries to finish
try {
- long now = System.currentTimeMillis();
- List<Coordinator> allCoordinators =
QeProcessorImpl.INSTANCE.getAllCoordinators();
- while (!allCoordinators.isEmpty() && System.currentTimeMillis() -
now < 300 * 1000L) {
+ // === Phase A: drain — absorb K8s LB/DNS convergence gap ===
+ int drainSeconds = Math.max(0,
Config.graceful_shutdown_drain_seconds);
+ LOG.info("graceful shutdown phase A: drain for {}s (health=503 but
still serving)", drainSeconds);
+ for (int i = 0; i < drainSeconds; i++) {
+ Thread.sleep(1000);
+ }
+
+ // === Phase B: atomic flip + close listeners + close idle MySQL
connections ===
+ // Step 1: flip the reject flag FIRST so any thread that wins the
listener
+ // accept race still gets rejected at the dispatch layer.
+ rejectNewQueries.set(true);
+ LOG.info("graceful shutdown phase B step1: rejectNewQueries=true");
+
+ // Step 2: close MySQL / Arrow-Flight listeners. HTTP listener
stays open
+ // until after phase C so /api/health remains reachable for the
probe.
+ if (qeService != null) {
+ LOG.info("graceful shutdown phase B step2: close MySQL /
ArrowFlight listeners");
Review Comment:
[P1] Gate HTTP and Thrift admission in phase B
This closes MySQL/Flight only. The HTTP server and FE Thrift server remain
live throughout phase C, and the reject flag is not checked by HTTP
SQL/stream-load or `loadTxnBegin`/forward endpoints. New work can enter after
an empty coordinator snapshot and be killed when the hook exits. Add one
cross-protocol admission barrier for start/begin operations while still
allowing health and completion/commit traffic for already-admitted work.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/VariableMgr.java:
##########
@@ -540,6 +568,7 @@ public static void
replayGlobalVariableV2(GlobalVarPersistInfo info, boolean fro
try {
setValue(varContext.getObj(), new
SessionVariableField(varContext.getField()),
root.get(varName).toString());
+ applyGlobalVariableSideEffect((String) varName,
root.get(varName).toString());
Review Comment:
[P1] Restore the side effect when loading an image
The checkpoint stores this SessionVariable field through
`SessionVariable.write/readFields`, but the `GlobalVarPersistInfo` replayed
here contains only `GlobalVariable` names. After journals are pruned, an FE can
load `enable_graceful_shutdown=true` while leaving the heartbeat interval at
its configured/default value. Add a serving-image load hook that restores both
the flag and daemon override, and cover an image round trip.
##########
fe/fe-common/src/main/java/org/apache/doris/common/Config.java:
##########
@@ -2110,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)
Review Comment:
[P1] Validate the mutable heartbeat interval
Making this field mutable also allows live values of 0 or -1 because no
validator is attached. Zero makes `Daemon.run()` loop heartbeats continuously;
a negative interval makes `Thread.sleep()` throw `IllegalArgumentException`
outside its catch and terminates the master heartbeat thread. Reject
nonpositive and unreasonable values before mutating the config or daemon.
##########
fe/fe-core/src/main/java/org/apache/doris/system/SystemInfoService.java:
##########
@@ -872,6 +881,19 @@ public void updateBackendState(Backend be) {
memoryBe.setDisks(be.getDisks());
memoryBe.setCpuCores(be.getCputCores());
memoryBe.setPipelineExecutorSize(be.getPipelineExecutorSize());
+
+ boolean restartDetected =
Review Comment:
[P1] Invalidate the complete pre-update addresses
Only the old hostname is saved before mutating host, BE port, and BRPC port.
`invalidateLocalConnections` then builds gRPC/Thrift keys from the new fields,
so entries keyed by `(preHost, preBrpcPort)` and `(preHost, preBePort)` survive
indefinitely. Snapshot and evict both complete old addresses as well as the
current ones.
##########
be/src/util/dns_cache.cpp:
##########
@@ -97,6 +108,7 @@ Status DNSCache::_update(const std::string& hostname) {
cache[hostname] = real_ip;
LOG(INFO) << "update hostname " << hostname << "'s ip to " << real_ip;
}
+ dirty.erase(hostname); // here myabe need lock all update function.
Review Comment:
[P1] Do not clear a newer DNS invalidation
A lookup can resolve stale A, `invalidate(host)` can run, and this older
update then publishes A and unconditionally erases the dirty marker. The next
`get()` returns A without re-resolution, so the new invalidation contract is
lost. Capture a per-host generation before resolution and publish/clear only if
no newer invalidation occurred.
##########
fe/fe-core/src/main/java/org/apache/doris/rpc/BackendServiceProxy.java:
##########
@@ -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 =
Review Comment:
[P2] Bound the deferred channel-shutdown backlog
This single worker executes `BackendServiceClient.shutdown()`, which can
block for two five-second waits. One backend may enqueue clients from all 48
proxy shards, so the last channel can remain retained roughly eight minutes
beyond the advertised 120-second delay; rolling multiple backends extends the
queue without bound. Use a nonblocking/two-stage close or bounded concurrent
teardown with per-client deduplication.
##########
be/src/util/brpc_client_cache.h:
##########
@@ -166,66 +218,116 @@ class BrpcClientCache {
}
std::shared_ptr<T> get_client(const std::string& host, int port) {
- std::string realhost = host;
- auto dns_cache = ExecEnv::GetInstance()->dns_cache();
- if (dns_cache == nullptr) {
- LOG(WARNING) << "DNS cache is not initialized, skipping hostname
resolve";
- } 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();
- return nullptr;
- }
- }
-
- // Use original host:port as key (like Java's TNetworkAddress address)
- // This allows us to detect IP changes when DNS resolution changes
- std::string host_port = fmt::format("{}:{}", host, port);
+ // 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);
+ }
- std::shared_ptr<T> stub_ptr;
- bool need_remove = false;
-
- auto check_entry = [&](const auto& v) {
- const StubEntry<T>& entry = v.second;
- // Check if cached IP matches current resolved IP
- if (entry.real_ip != realhost) {
- // IP changed (DNS resolution changed)
- LOG(WARNING) << "Cached ip changed for " << host << ", before
ip: " << entry.real_ip
- << ", current ip: " << realhost;
- need_remove = true;
- } else if
(!static_cast<FailureDetectChannel*>(entry.stub->channel())
- ->channel_status()
- ->ok()) {
- // Client is not in normal state, need to recreate
- // 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.
- need_remove = true;
- } else {
- // Cache hit: IP matches and client is healthy
- stub_ptr = entry.stub;
+ // 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 5.
+ std::shared_ptr<T> get_client(const std::string& host, int port, bool
enable_handshake) {
+ const int max_attempts = enable_handshake ? 5 : 1;
+ std::string host_port;
+ for (int attempt = 1; attempt <= max_attempts; ++attempt) {
+ // Progressive backoff before retries so Kubernetes DNS/endpoints
+ // have more time to converge during graceful restart.
+ if (enable_handshake && attempt > 1) {
+ bthread_usleep(1000000 * (attempt - 1));
}
- };
-
- if (LIKELY(_stub_map.if_contains(host_port, check_entry))) {
- if (stub_ptr != nullptr) {
- return stub_ptr;
+ std::string realhost = host;
+ auto dns_cache = ExecEnv::GetInstance()->dns_cache();
+ if (dns_cache == nullptr) {
+ LOG(WARNING) << "DNS cache is not initialized, skipping
hostname resolve";
+ } 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()
+ << ", attempt=" << attempt << "/" <<
max_attempts;
+ if (enable_handshake && attempt < max_attempts) {
+ continue;
+ }
+ return nullptr;
+ }
}
- // IP changed or client unhealthy, need to remove old entry
- if (need_remove) {
+ // Keep the original hostname as the cache key so a DNS answer
+ // change can replace the cached channel for the same logical peer.
+ host_port = fmt::format("{}:{}", host, port);
+
+ std::shared_ptr<T> stub_ptr;
+ bool need_remove = false;
+ auto check_entry = [&](const auto& v) {
+ const StubEntry<T>& entry = v.second;
+ if (entry.real_ip != realhost) {
+ LOG(WARNING) << "Cached ip changed for " << host
+ << ", before ip: " << entry.real_ip
+ << ", current ip: " << realhost;
+ need_remove = true;
+ } else if
(!static_cast<FailureDetectChannel*>(entry.stub->channel())
+ ->channel_status()
+ ->ok()) {
+ need_remove = true;
+ } else {
+ stub_ptr = entry.stub;
+ }
+ };
+ 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)) {
Review Comment:
[P2] Keep healthy warm-cache hits free of RPCs
While the cluster-wide flag is enabled, every healthy cache hit performs a
synchronous `hand_shake` before returning, even though the overload contract
says this cost is paid only on miss/rebuild. Query and load setup fan out
through this cache, so a rolling restart adds a network round trip to every
acquisition. Probe only rebuilt/faulted generations or use a bounded
asynchronous health signal.
##########
fe/fe-core/src/main/java/org/apache/doris/DorisFE.java:
##########
@@ -722,20 +735,95 @@ public static boolean isServerReady() {
return serverReady.get();
}
+ /**
+ * Returns true once graceful shutdown has entered phase B. New data-plane
+ * MySQL commands (COM_QUERY / COM_STMT_EXECUTE / ...) should be silently
+ * dropped so JDBC drivers fail over to another FE.
+ */
+ public static boolean shouldRejectNewQueries() {
+ return rejectNewQueries.get();
+ }
+
+ /**
+ * Graceful shutdown driven by SIGTERM (typically from K8s). Implements the
+ * A0 → A → B → C → D state machine described in the operator design doc.
+ *
+ * <pre>
+ * A0 serverReady=false → /api/health returns 503 (already done by
caller)
+ * A "drain": sleep graceful_shutdown_drain_seconds while still serving
+ * new queries, to let K8s LB/DNS converge.
+ * B atomic flip: rejectNewQueries=true, then close MySQL/ArrowFlight
+ * listeners, then close idle MySQL connections (FIN to client).
+ * C wait for in-flight Coordinators to finish (≤
grace_shutdown_wait_seconds),
+ * re-running closeIdleMysqlConnections() once per second.
+ * D return; JVM exits after shutdown hook completes.
+ * </pre>
+ */
private static void gracefulShutdown() {
- // wait for all queries to finish
try {
- long now = System.currentTimeMillis();
- List<Coordinator> allCoordinators =
QeProcessorImpl.INSTANCE.getAllCoordinators();
- while (!allCoordinators.isEmpty() && System.currentTimeMillis() -
now < 300 * 1000L) {
+ // === Phase A: drain — absorb K8s LB/DNS convergence gap ===
+ int drainSeconds = Math.max(0,
Config.graceful_shutdown_drain_seconds);
+ LOG.info("graceful shutdown phase A: drain for {}s (health=503 but
still serving)", drainSeconds);
+ for (int i = 0; i < drainSeconds; i++) {
+ Thread.sleep(1000);
+ }
+
+ // === Phase B: atomic flip + close listeners + close idle MySQL
connections ===
+ // Step 1: flip the reject flag FIRST so any thread that wins the
listener
+ // accept race still gets rejected at the dispatch layer.
+ rejectNewQueries.set(true);
+ LOG.info("graceful shutdown phase B step1: rejectNewQueries=true");
+
+ // Step 2: close MySQL / Arrow-Flight listeners. HTTP listener
stays open
+ // until after phase C so /api/health remains reachable for the
probe.
+ if (qeService != null) {
+ LOG.info("graceful shutdown phase B step2: close MySQL /
ArrowFlight listeners");
+ qeService.stopAcceptNewConnections();
+ }
+
+ // Step 3: send FIN to currently-idle (COM_SLEEP) MySQL
connections so
+ // pooled JDBC clients notice and reconnect to another FE.
+ LOG.info("graceful shutdown phase B step3: close idle mysql
connections");
+ closeIdleMysqlConnections();
+
+ // === Phase C: wait for in-flight Coordinators to drain ===
+ long deadline = System.currentTimeMillis() + Math.max(0,
Config.grace_shutdown_wait_seconds) * 1000L;
+ LOG.info("graceful shutdown phase C: wait up to {}s for in-flight
queries to finish",
+ Config.grace_shutdown_wait_seconds);
+ while (System.currentTimeMillis() < deadline) {
+ List<Coordinator> allCoordinators =
QeProcessorImpl.INSTANCE.getAllCoordinators();
Review Comment:
[P1] Drain active requests outside the coordinator map
`getAllCoordinators()` is not a quiescence barrier. A request can pass the
phase-B gate, remain in parsing/planning before `StmtExecutor.registerQuery()`,
or execute wholly in FE without ever entering this map; this empty snapshot
then ends phase C and exits the JVM underneath it. After closing every
admission gate, wait on request-lifecycle accounting that spans admission
through pre-registration/FE-local work and coordinator completion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
