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 c54bcbc9ec878f9b60843c1b3e73665776f1a7e1 Author: zhangstar333 <[email protected]> AuthorDate: Fri May 22 16:34:23 2026 +0800 fe graceful shutdown --- .../main/java/org/apache/doris/common/Config.java | 14 +++ .../src/main/java/org/apache/doris/DorisFE.java | 106 +++++++++++++++++++-- .../java/org/apache/doris/qe/ConnectScheduler.java | 31 ++++++ .../org/apache/doris/qe/MysqlConnectProcessor.java | 25 +++++ .../main/java/org/apache/doris/qe/QeService.java | 10 +- 5 files changed, 176 insertions(+), 10 deletions(-) diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java index 39f29b958d5..c7c4c65dae8 100644 --- a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java +++ b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java @@ -483,6 +483,20 @@ public class Config extends ConfigBase { @ConfField(description = {"The maximum number of task threads in the MySQL service"}) public static int max_mysql_service_task_threads_num = 4096; + @ConfField(mutable = true, description = { + "FE graceful shutdown A 阶段（假死 drain）的持续秒数。此阶段 /api/health 已返回 503，" + + "但 listener 不关、新 query 仍正常应答，用于吸收 K8s LB/DNS 收敛延迟。", + "Duration in seconds of FE graceful shutdown phase A (drain). During this phase /api/health " + + "returns 503, but listeners stay open and new queries are still served, to absorb " + + "K8s LB/DNS convergence delay."}) + public static int graceful_shutdown_drain_seconds = 15; + + @ConfField(mutable = true, description = { + "FE graceful shutdown C 阶段等待 in-flight Coordinator 完成的最大秒数。超过则强制退出 JVM。", + "Maximum seconds to wait for in-flight Coordinators to finish during FE graceful shutdown " + + "phase C. JVM will be forced to exit when exceeded."}) + public static int grace_shutdown_wait_seconds = 300; + @ConfField(description = {"BackendServiceProxy pool size for pooling GRPC channels."}) public static int backend_proxy_num = 48; diff --git a/fe/fe-core/src/main/java/org/apache/doris/DorisFE.java b/fe/fe-core/src/main/java/org/apache/doris/DorisFE.java index d313467b127..df251e8d02c 100755 --- a/fe/fe-core/src/main/java/org/apache/doris/DorisFE.java +++ b/fe/fe-core/src/main/java/org/apache/doris/DorisFE.java @@ -98,6 +98,10 @@ public class DorisFE { // set to true when all servers are ready. private static final AtomicBoolean serverReady = new AtomicBoolean(false); + // set to true once graceful shutdown enters phase B; new data-plane + // queries should then be silently rejected. + private static final AtomicBoolean rejectNewQueries = new AtomicBoolean(false); + // HTTP server instance, used for graceful shutdown private static HttpServer httpServer; private static ServerStarter thriftServerStarter; @@ -176,6 +180,9 @@ public class DorisFE { // Add shutdown hook for graceful exit Runtime.getRuntime().addShutdownHook(new Thread(() -> { LOG.info("Received shutdown signal, starting graceful shutdown..."); + // === A0: immediately mark not-ready so /api/health returns 503. + // K8s readinessProbe will fail and the pod will be removed from the + // EndpointSlice. Listeners stay open; new queries are still served. serverReady.set(false); gracefulShutdown(); @@ -191,9 +198,15 @@ public class DorisFE { } } - // Shutdown HTTP server after main process graceful shutdown is complete + // Shutdown HTTP server at the very end (after phase C). HTTP server is + // kept alive throughout drain so that K8s readiness probe can keep + // observing 503 until endpoint removal completes. if (httpServer != null) { - httpServer.shutdown(); + try { + httpServer.shutdown(); + } catch (Throwable t) { + LOG.warn("failed to shutdown http server", t); + } } LogManager.shutdown(); @@ -722,20 +735,95 @@ public class DorisFE { 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(); + if (allCoordinators.isEmpty()) { + LOG.info("all in-flight coordinators finished, exit phase C early"); + break; + } + LOG.info("phase C: waiting {} queries to finish before shutdown", allCoordinators.size()); Thread.sleep(1000); - allCoordinators = QeProcessorImpl.INSTANCE.getAllCoordinators(); - LOG.info("waiting {} queries to finish before shutdown", allCoordinators.size()); + // Re-scan for newly-idle connections produced by completed queries. + closeIdleMysqlConnections(); } + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + LOG.warn("graceful shutdown interrupted", ie); } catch (Throwable t) { - LOG.error("", t); + LOG.error("graceful shutdown encountered an error", t); } + // === Phase D: return, JVM exits after the shutdown hook completes. === LOG.info("graceful shutdown finished"); } + + private static void closeIdleMysqlConnections() { + try { + ExecuteEnv env = ExecuteEnv.getInstance(); + if (env == null) { + return; + } + org.apache.doris.qe.ConnectScheduler scheduler = env.getScheduler(); + if (scheduler != null) { + scheduler.closeIdleMysqlConnections(); + } + } catch (Throwable t) { + LOG.warn("failed to close idle mysql connections", t); + } + } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectScheduler.java b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectScheduler.java index b22e89cf209..ae8ffdfc9a7 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectScheduler.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectScheduler.java @@ -21,6 +21,7 @@ import org.apache.doris.analysis.UserIdentity; import org.apache.doris.common.Config; import org.apache.doris.common.Status; import org.apache.doris.common.ThreadPoolManager; +import org.apache.doris.mysql.MysqlCommand; import org.apache.doris.qe.ConnectContext.ThreadInfo; import org.apache.doris.service.arrowflight.sessions.FlightSqlConnectPoolMgr; @@ -169,4 +170,34 @@ public class ConnectScheduler { } } } + + /** + * Close MySQL connections that are currently idle (command == COM_SLEEP). + * Used during graceful shutdown phase B / C so that clients receive a FIN + * on idle sockets and can fail over to another FE. + * + * @return number of connections closed in this invocation + */ + public int closeIdleMysqlConnections() { + int closed = 0; + for (ConnectContext ctx : connectPoolMgr.getConnectionMap().values()) { + if (ctx == null) { + continue; + } + if (ctx.getCommand() == MysqlCommand.COM_SLEEP && !ctx.isKilled()) { + try { + // kill the connection (sends FIN to client) but do not cancel + // any running query (there is none in COM_SLEEP state). + ctx.kill(true); + closed++; + } catch (Throwable t) { + LOG.warn("failed to close idle mysql connection id={}", ctx.getConnectionId(), t); + } + } + } + if (closed > 0) { + LOG.info("closed {} idle mysql connections during graceful shutdown", closed); + } + return closed; + } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/MysqlConnectProcessor.java b/fe/fe-core/src/main/java/org/apache/doris/qe/MysqlConnectProcessor.java index f70a9d3f622..742327d3cc8 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/MysqlConnectProcessor.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/MysqlConnectProcessor.java @@ -17,6 +17,7 @@ package org.apache.doris.qe; +import org.apache.doris.DorisFE; import org.apache.doris.analysis.StatementBase; import org.apache.doris.analysis.UserIdentity; import org.apache.doris.catalog.Env; @@ -256,6 +257,17 @@ public class MysqlConnectProcessor extends ConnectProcessor { ctx.setStartTime(); resolveWorkloadGroupName(); + // Graceful shutdown: if FE is draining and rejecting new data-plane requests, + // silently close the connection (no MySQL error packet) so JDBC drivers fail + // over to another FE via multi-host config instead of surfacing an error. + if (DorisFE.shouldRejectNewQueries() && isDataPlaneCommand(command)) { + LOG.info("reject new query during graceful shutdown, connId={}, command={}, remote={}", + ctx.getConnectionId(), command, ctx.getMysqlChannel().getRemoteHostPortString()); + ctx.getState().setNoop(); + ctx.setKilled(); + return; + } + switch (command) { case COM_INIT_DB: handleInitDb(); @@ -309,6 +321,19 @@ public class MysqlConnectProcessor extends ConnectProcessor { } } + private static boolean isDataPlaneCommand(MysqlCommand command) { + switch (command) { + case COM_QUERY: + case COM_STMT_PREPARE: + case COM_STMT_EXECUTE: + case COM_FIELD_LIST: + case COM_INIT_DB: + return true; + default: + return false; + } + } + private void handleFieldList() throws ConnectionException { String tableName = new String(MysqlProto.readNulTerminateString(packetBuf), StandardCharsets.UTF_8); handleFieldList(tableName); diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/QeService.java b/fe/fe-core/src/main/java/org/apache/doris/qe/QeService.java index e070e602cd8..0b0e5f22024 100755 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/QeService.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/QeService.java @@ -66,7 +66,11 @@ public class QeService { 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(); @@ -82,4 +86,8 @@ public class QeService { LOG.warn("stop mysql server failed", e); } } + + public void stop() { + stopAcceptNewConnections(); + } } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
