1

Author: morningman

be/src/common/config.cpp

@@ -1315,6 +1315,7 @@ DEFINE_String(user_files_secure_path, "${DORIS_HOME}");
 DEFINE_Int32(fe_expire_duration_seconds, "60");
 
 DEFINE_Int32(grace_shutdown_wait_seconds, "120");
+DEFINE_Int32(grace_shutdown_post_delay_seconds, "30");
 
 DEFINE_Int16(bitmap_serialize_version, "1");
 

be/src/common/config.h

@@ -1367,6 +1367,12 @@ DECLARE_Int32(fe_expire_duration_seconds);
 // , but if the waiting time exceed the limit, then be will exit directly.
 // During this period, FE will not send any queries to BE and waiting for all running queries to stop.
 DECLARE_Int32(grace_shutdown_wait_seconds);
+// When using the graceful stop feature, after the main process waits for
+// all currently running tasks to finish, it will continue to wait for
+// an additional period to ensure that queries still running on other nodes have also completed.
+// Since a BE node cannot detect the task execution status on other BE nodes,
+// you may need to increase this threshold to allow for a longer waiting time.
+DECLARE_Int32(grace_shutdown_post_delay_seconds);
 
 // BitmapValue serialize version.
 DECLARE_Int16(bitmap_serialize_version);

be/src/http/action/health_action.cpp

@@ -34,14 +34,17 @@ void HealthAction::handle(HttpRequest* req) {
     std::string status;
     std::string msg;
     HttpStatus st;
+    // always return HttpStatus::OK
+    // because in k8s, we don't want the pod to be removed
+    // from service during shutdown
     if (!doris::k_is_server_ready) {
         status = "Server is not available";
         msg = "Server is not ready";
-        st = HttpStatus::SERVICE_UNAVAILABLE;
+        st = HttpStatus::OK;
     } else if (doris::k_doris_exit) {
         status = "Server is not available";
         msg = "Server is shutting down";
-        st = HttpStatus::SERVICE_UNAVAILABLE;
+        st = HttpStatus::OK;
     } else {
         status = "OK";
         msg = "OK";

be/src/runtime/exec_env.cpp

@@ -177,6 +177,13 @@ void ExecEnv::wait_for_all_tasks_done() {
         sleep(1);
         ++wait_seconds_passed;
     }
+    // This is a conservative strategy.
+    // Because a query might still have fragments running on other BE nodes.
+    // In other words, the query hasn't truly terminated.
+    // If the current BE is shut down at this point,
+    // the FE will detect the downtime of a related BE and cancel the entire query,
+    // defeating the purpose of a graceful stop.
+    sleep(config::grace_shutdown_post_delay_seconds);
 }
 
 bool ExecEnv::check_auth_token(const std::string& auth_token) {

fe/fe-common/src/main/java/org/apache/doris/common/Config.java

@@ -3654,12 +3654,6 @@ public static int metaServiceRpcRetryTimes() {
     })
     public static int first_error_msg_max_length = 256;
 
-    @ConfField(mutable = true, masterOnly = false, description = {
-            "如果设置为 false，则 health 接口将返回 503，表示此 FE 不应再接收服务请求",
-            "If set to false, the health endpoint will return 503, "
-                    + "indicating that this FE should no longer receive service requests"})
-    public static boolean frontend_service_available = true;
-
     @ConfField
     public static String cloud_snapshot_handler_class = "org.apache.doris.cloud.snapshot.CloudSnapshotHandler";
     @ConfField

fe/fe-core/src/main/java/org/apache/doris/DorisFE.java

@@ -36,6 +36,8 @@
 import org.apache.doris.journal.bdbje.BDBTool;
 import org.apache.doris.journal.bdbje.BDBToolOptions;
 import org.apache.doris.persist.meta.MetaReader;
+import org.apache.doris.qe.Coordinator;
+import org.apache.doris.qe.QeProcessorImpl;
 import org.apache.doris.qe.QeService;
 import org.apache.doris.qe.SimpleScheduler;
 import org.apache.doris.service.ExecuteEnv;
@@ -63,6 +65,7 @@
 import java.nio.channels.OverlappingFileLockException;
 import java.nio.file.StandardOpenOption;
 import java.time.LocalDate;
+import java.util.List;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
@@ -84,7 +87,10 @@ public class DorisFE {
     private static FileLock processFileLock;
 
     // set to true when all servers are ready.
-    private static AtomicBoolean serverReady = new AtomicBoolean(false);
+    private static final AtomicBoolean serverReady = new AtomicBoolean(false);
+
+    // HTTP server instance, used for graceful shutdown
+    private static HttpServer httpServer;
 
     public static void main(String[] args) {
         // Every doris version should have a final meta version, it should not change
@@ -148,7 +154,19 @@ public static void start(String dorisHomeDir, String pidDir, String[] args, Star
             }
 
             Log4jConfig.initLogging(dorisHomeDir + "/conf/");
-            Runtime.getRuntime().addShutdownHook(new Thread(LogManager::shutdown));
+            // Add shutdown hook for graceful exit
+            Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+                LOG.info("Received shutdown signal, starting graceful shutdown...");
+                serverReady.set(false);
+                gracefulShutdown();
+
+                // Shutdown HTTP server after main process graceful shutdown is complete
+                if (httpServer != null) {
+                    httpServer.shutdown();
+                }
+
+                LogManager.shutdown();
+            }));
 
             // set dns cache ttl
             java.security.Security.setProperty("networkaddress.cache.ttl", "60");
@@ -211,7 +229,7 @@ public static void start(String dorisHomeDir, String pidDir, String[] args, Star
             feServer.start();
 
             if (options.enableHttpServer) {
-                HttpServer httpServer = new HttpServer();
+                httpServer = new HttpServer();
                 httpServer.setPort(Config.http_port);
                 httpServer.setHttpsPort(Config.https_port);
                 httpServer.setMaxHttpPostSize(Config.jetty_server_max_http_post_size);
@@ -242,11 +260,13 @@ public static void start(String dorisHomeDir, String pidDir, String[] args, Star
             startMonitor();
 
             serverReady.set(true);
+            // JVM will exit when shutdown hook is completed
             while (true) {
                 Thread.sleep(2000);
             }
+            LOG.info("Doris FE main loop exited, shutting down gracefully...");
         } catch (Throwable e) {
-            // Some exception may thrown before LOG is inited.
+            // Some exception may throw before LOG is inited.
             // So need to print to stdout
             e.printStackTrace();
             LOG.error("", e);
@@ -599,4 +619,21 @@ public static class StartupOptions {
     public static boolean isServerReady() {
         return serverReady.get();
     }
+
+    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) {
+                Thread.sleep(1000);
+                allCoordinators = QeProcessorImpl.INSTANCE.getAllCoordinators();
+                LOG.info("waiting {} queries to finish before shutdown", allCoordinators.size());
+            }
+        } catch (Throwable t) {
+            LOG.error("", t);
+        }
+
+        LOG.info("graceful shutdown finished");
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java

@@ -422,7 +422,7 @@ public long hashReplicaToBe(String clusterId, boolean isBackGround) throws Compu
             long lastUpdateMs = be.getLastUpdateMs();
             long missTimeMs = Math.abs(lastUpdateMs - System.currentTimeMillis());
             // be core or restart must in heartbeat_interval_second
-            if ((be.isAlive() || missTimeMs <= Config.heartbeat_interval_second * 1000L)
+            if ((be.isQueryAvailable() || missTimeMs <= Config.heartbeat_interval_second * 1000L)
                     && !be.isSmoothUpgradeSrc()) {
                 if (be.isDecommissioned()) {
                     decommissionAvailBes.add(be);

fe/fe-core/src/main/java/org/apache/doris/common/FeConstants.java

@@ -36,16 +36,11 @@ public class FeConstants {
 
     public static int checkpoint_interval_second = 60; // 1 minutes
 
-    // dpp version
-    public static String dpp_version = "3_2_0";
-
     // bloom filter false positive probability
     public static double default_bloom_filter_fpp = 0.05;
 
     // set to true to skip some step when running FE unit test
     public static boolean runningUnitTest = false;
-    // use to set some mocked values for FE unit test
-    public static Object unitTestConstant = null;
 
     // set to false to disable internal schema db
     public static boolean enableInternalSchemaDb = true;
@@ -68,29 +63,8 @@ public class FeConstants {
     // use for copy into test
     public static boolean disablePreHeat = false;
 
-    public static final String FS_PREFIX_S3 = "s3";
-    public static final String FS_PREFIX_S3A = "s3a";
-    public static final String FS_PREFIX_S3N = "s3n";
-    public static final String FS_PREFIX_OSS = "oss";
-    public static final String FS_PREFIX_GCS = "gs";
-    public static final String FS_PREFIX_BOS = "bos";
-    public static final String FS_PREFIX_COS = "cos";
-    public static final String FS_PREFIX_COSN = "cosn";
-    public static final String FS_PREFIX_LAKEFS = "lakefs";
-    public static final String FS_PREFIX_OBS = "obs";
-    public static final String FS_PREFIX_OFS = "ofs";
-    public static final String FS_PREFIX_GFS = "gfs";
-    public static final String FS_PREFIX_JFS = "jfs";
-    public static final String FS_PREFIX_HDFS = "hdfs";
-    public static final String FS_PREFIX_VIEWFS = "viewfs";
-    public static final String FS_PREFIX_FILE = "file";
-
     public static final String INTERNAL_DB_NAME = "__internal_schema";
     public static final String INTERNAL_FILE_CACHE_HOTSPOT_TABLE_NAME = "cloud_cache_hotspot";
 
     public static String METADATA_FAILURE_RECOVERY_KEY = "metadata_failure_recovery";
-
-    public static String CLOUD_RETRY_E230 = "E-230";
-
-    public static String BUILT_IN_STORAGE_VAULT_NAME = "built_in_storage_vault";
 }

fe/fe-core/src/main/java/org/apache/doris/httpv2/HttpServer.java

@@ -23,11 +23,14 @@
 import org.apache.doris.httpv2.config.SpringLog4j2Config;
 import org.apache.doris.service.FrontendOptions;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.builder.SpringApplicationBuilder;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.boot.web.servlet.ServletComponentScan;
 import org.springframework.boot.web.servlet.support.SpringBootServletInitializer;
+import org.springframework.context.ConfigurableApplicationContext;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -36,6 +39,8 @@
 @EnableConfigurationProperties
 @ServletComponentScan
 public class HttpServer extends SpringBootServletInitializer {
+    private static final Logger LOG = LogManager.getLogger(HttpServer.class);
+    private ConfigurableApplicationContext applicationContext;
     private int port;
     private int httpsPort;
     private int acceptors;
@@ -176,9 +181,26 @@ public void start() {
         } else {
             properties.put("logging.config", Config.custom_config_dir + "/" + SpringLog4j2Config.SPRING_LOG_XML_FILE);
         }
-        new SpringApplicationBuilder()
+        // Disable automatic shutdown hook registration
+        // This prevents Spring Boot from responding to SIGTERM automatically
+        // allowing the main process (DorisFE) to control when the HTTP server shuts down
+        this.applicationContext = new SpringApplicationBuilder()
                 .sources(HttpServer.class)
                 .properties(properties)
+                // Disable the automatic shutdown hook registration, there is a shutdown hook in DorisFE.
+                .registerShutdownHook(false)
                 .run(new String[]{});
     }
+
+    /**
+     * Explicitly shutdown the HTTP server.
+     * This method should be called by the main process (DorisFE) after its graceful shutdown is complete.
+     */
+    public void shutdown() {
+        if (applicationContext != null) {
+            LOG.info("Shutting down HTTP server gracefully...");
+            applicationContext.close();
+            LOG.info("HTTP server shutdown complete");
+        }
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/HealthAction.java

@@ -44,10 +44,6 @@ public Object execute(HttpServletRequest request, HttpServletResponse response)
             return ResponseEntityBuilder.serviceUnavailable("Server is not ready");
         }
 
-        if (!Config.frontend_service_available) {
-            return ResponseEntityBuilder.serviceUnavailable("Server is not available");
-        }
-
         Map<String, Object> result = new HashMap<>();
         result.put("total_backend_num", Env.getCurrentSystemInfo().getAllBackendIds(false).size());
         result.put("online_backend_num", Env.getCurrentSystemInfo().getAllBackendIds(true).size());