You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by mo...@apache.org on 2025/11/07 06:26:29 UTC
(doris) branch master updated: [opt](scheduler) Improve Graceful Shutdown Behavior for BE and FE, and Optimize Query Retry During BE Shutdown (#56601)
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new bbded3ab346 [opt](scheduler) Improve Graceful Shutdown Behavior for BE and FE, and Optimize Query Retry During BE Shutdown (#56601)
bbded3ab346 is described below
commit bbded3ab3461dd0e9ba489fbb8e7f0ad5cfa75e6
Author: Mingyu Chen (Rayner) <mo...@163.com>
AuthorDate: Fri Nov 7 14:26:22 2025 +0800
[opt](scheduler) Improve Graceful Shutdown Behavior for BE and FE, and Optimize Query Retry During BE Shutdown (#56601)
### What problem does this PR solve?
Related PR: #23865
This PR includes the following main changes:
#### BE Graceful Shutdown Improvements
1. New BE Parameter: `grace_shutdown_post_delay_seconds`
When using the BE graceful stop feature, after the main process waits
for all currently running tasks to complete, it will continue to wait
for an additional period to ensure that queries still running on other
nodes have also finished.
Since a BE node cannot detect the execution status of tasks on other BE
nodes, this threshold may need to be increased to allow a longer waiting
time.
2. Enhanced BE `api/health` Endpoint
* When the BE has not yet fully started or is in the process of shutting
down, the endpoint will return:
* Message: `"Server is not available"`
* HTTP Code: `200`
* Under normal circumstances:
* Message: `"OK"`
* HTTP Code: `200`
#### Added FE Graceful Shutdown Support
When using `stop_fe.sh --grace`, the FE will wait for currently running
queries to finish before exiting.
Note, Currently, only query tasks are waited for; import and other types
of tasks are not yet included.
#### Query Retry Optimization During BE Shutdown
In cloud mode, when encountering the error `"No backend available as
scan node"`,
the FE will now internally retry the query to reassign it to other
available BE nodes.
---
be/src/common/config.cpp | 1 +
be/src/common/config.h | 6 ++
be/src/http/action/health_action.cpp | 27 +++++++-
be/src/runtime/exec_env.cpp | 7 ++
be/src/runtime/exec_env.h | 3 +
be/src/service/doris_main.cpp | 3 +
be/test/http/http_client_test.cpp | 4 +-
.../src/main/java/org/apache/doris/DorisFE.java | 52 +++++++++++++-
.../apache/doris/cloud/catalog/CloudReplica.java | 4 +-
.../java/org/apache/doris/common/FeConstants.java | 26 -------
.../java/org/apache/doris/httpv2/HttpServer.java | 24 ++++++-
.../doris/httpv2/entity/ResponseEntityBuilder.java | 5 ++
.../org/apache/doris/httpv2/rest/HealthAction.java | 5 ++
.../doris/httpv2/rest/RestApiStatusCode.java | 3 +-
.../apache/doris/job/extensions/mtmv/MTMVTask.java | 3 +-
.../org/apache/doris/load/StreamLoadHandler.java | 2 +-
.../commands/insert/AbstractInsertExecutor.java | 6 +-
.../plans/commands/insert/OlapInsertExecutor.java | 6 +-
.../java/org/apache/doris/qe/StmtExecutor.java | 21 ++++--
.../main/java/org/apache/doris/system/Backend.java | 52 ++++++++++----
.../org/apache/doris/system/SystemInfoService.java | 40 ++++++-----
.../doris/utframe/DemoMultiBackendsTest.java | 3 +-
regression-test/framework/pom.xml | 4 +-
.../query_retry/test_retry_no_scan_node.groovy | 80 ++++++++++++++++++++++
24 files changed, 303 insertions(+), 84 deletions(-)
diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp
index 606fdd406cf..7c74629cab9 100644
--- a/be/src/common/config.cpp
+++ b/be/src/common/config.cpp
@@ -1317,6 +1317,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");
diff --git a/be/src/common/config.h b/be/src/common/config.h
index 2e105b36182..313145330d8 100644
--- a/be/src/common/config.h
+++ b/be/src/common/config.h
@@ -1368,6 +1368,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);
diff --git a/be/src/http/action/health_action.cpp b/be/src/http/action/health_action.cpp
index 91be5f78f30..8c1976dea2e 100644
--- a/be/src/http/action/health_action.cpp
+++ b/be/src/http/action/health_action.cpp
@@ -24,21 +24,42 @@
#include "http/http_headers.h"
#include "http/http_request.h"
#include "http/http_status.h"
+#include "runtime/exec_env.h"
namespace doris {
const static std::string HEADER_JSON = "application/json";
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::OK;
+ } else if (doris::k_doris_exit) {
+ status = "Server is not available";
+ msg = "Server is shutting down";
+ st = HttpStatus::OK;
+ } else {
+ status = "OK";
+ msg = "OK";
+ st = HttpStatus::OK;
+ }
+
std::stringstream ss;
ss << "{";
- ss << "\"status\": \"OK\",";
- ss << "\"msg\": \"To Be Added\"";
+ ss << "\"status\": \"" << status << "\",";
+ ss << "\"msg\": \"" << msg << "\"";
ss << "}";
std::string result = ss.str();
req->add_output_header(HttpHeaders::CONTENT_TYPE, HEADER_JSON.c_str());
- HttpChannel::send_reply(req, HttpStatus::OK, result);
+ HttpChannel::send_reply(req, st, result);
}
} // end namespace doris
diff --git a/be/src/runtime/exec_env.cpp b/be/src/runtime/exec_env.cpp
index a31a5b00d0b..a8fefb308b6 100644
--- a/be/src/runtime/exec_env.cpp
+++ b/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) {
diff --git a/be/src/runtime/exec_env.h b/be/src/runtime/exec_env.h
index 1f68b435a5b..d6112373efa 100644
--- a/be/src/runtime/exec_env.h
+++ b/be/src/runtime/exec_env.h
@@ -135,7 +135,10 @@ class IndexPolicyMgr;
struct SyncRowsetStats;
class DeleteBitmapAggCache;
+// set to true when BE is shutting down
inline bool k_doris_exit = false;
+// set to true after BE start ready
+inline bool k_is_server_ready = false;
// Execution environment for queries/plan fragments.
// Contains all required global structures, and handles to
diff --git a/be/src/service/doris_main.cpp b/be/src/service/doris_main.cpp
index 401456c6f3a..ead7426a564 100644
--- a/be/src/service/doris_main.cpp
+++ b/be/src/service/doris_main.cpp
@@ -602,12 +602,15 @@ int main(int argc, char** argv) {
exec_env->storage_engine().notify_listeners();
+ doris::k_is_server_ready = true;
+
while (!doris::k_doris_exit) {
#if defined(LEAK_SANITIZER)
__lsan_do_leak_check();
#endif
sleep(3);
}
+ doris::k_is_server_ready = false;
LOG(INFO) << "Doris main exiting.";
#if defined(LLVM_PROFILE)
__llvm_profile_write_file();
diff --git a/be/test/http/http_client_test.cpp b/be/test/http/http_client_test.cpp
index 02699f0baef..d874df5bcd6 100644
--- a/be/test/http/http_client_test.cpp
+++ b/be/test/http/http_client_test.cpp
@@ -392,7 +392,7 @@ TEST_F(HttpClientTest, enable_http_auth) {
st = client.execute(&response);
EXPECT_TRUE(st.ok());
std::cout << "response = " << response << "\n";
- EXPECT_TRUE(response.find("To Be Added") != std::string::npos);
+ EXPECT_TRUE(response.find("Server is not ready") != std::string::npos);
}
{
@@ -423,7 +423,7 @@ TEST_F(HttpClientTest, enable_http_auth) {
st = client.execute(&response);
EXPECT_TRUE(st.ok());
std::cout << "response = " << response << "\n";
- EXPECT_TRUE(response.find("To Be Added") != std::string::npos);
+ EXPECT_TRUE(response.find("Server is not ready") != std::string::npos);
}
{
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 d0e8b7fe5bc..0fa4fdb7822 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
@@ -36,6 +36,8 @@ import org.apache.doris.journal.bdbje.BDBDebugger;
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,7 +65,9 @@ import java.nio.channels.FileLock;
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;
public class DorisFE {
private static final Logger LOG = LogManager.getLogger(DorisFE.class);
@@ -82,6 +86,12 @@ public class DorisFE {
private static FileChannel processLockFileChannel;
private static FileLock processFileLock;
+ // set to true when all servers are ready.
+ 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
// between small releases. Add a check here to avoid mistake.
@@ -144,7 +154,19 @@ public class DorisFE {
}
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");
@@ -207,7 +229,7 @@ public class DorisFE {
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);
@@ -236,11 +258,14 @@ public class DorisFE {
ThreadPoolManager.registerAllThreadPoolMetric();
startMonitor();
+
+ serverReady.set(true);
+ // JVM will exit when shutdown hook is completed
while (true) {
Thread.sleep(2000);
}
} 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);
@@ -589,4 +614,25 @@ public class DorisFE {
public boolean enableHttpServer = true;
public boolean enableQeService = true;
}
+
+ 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");
+ }
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java
index 3fc7c652f58..df59b3d67cc 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudReplica.java
@@ -101,7 +101,7 @@ public class CloudReplica extends Replica {
public long getColocatedBeId(String clusterId) throws ComputeGroupException {
CloudSystemInfoService infoService = ((CloudSystemInfoService) Env.getCurrentSystemInfo());
List<Backend> bes = infoService.getBackendsByClusterId(clusterId).stream()
- .filter(be -> !be.isQueryDisabled()).collect(Collectors.toList());
+ .filter(be -> be.isQueryAvailable()).collect(Collectors.toList());
String clusterName = infoService.getClusterNameByClusterId(clusterId);
if (bes.isEmpty()) {
LOG.warn("failed to get available be, cluster: {}-{}", clusterName, clusterId);
@@ -422,7 +422,7 @@ public class CloudReplica extends Replica {
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);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/FeConstants.java b/fe/fe-core/src/main/java/org/apache/doris/common/FeConstants.java
index 5303d11b0a6..491a8e036af 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/common/FeConstants.java
+++ b/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";
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/httpv2/HttpServer.java b/fe/fe-core/src/main/java/org/apache/doris/httpv2/HttpServer.java
index 1902f471f1b..3c9a96140f5 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/httpv2/HttpServer.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/httpv2/HttpServer.java
@@ -23,11 +23,14 @@ import org.apache.doris.common.FeConstants;
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 @@ import java.util.Map;
@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 class HttpServer extends SpringBootServletInitializer {
} 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();
}
+
+ /**
+ * 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");
+ }
+ }
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/httpv2/entity/ResponseEntityBuilder.java b/fe/fe-core/src/main/java/org/apache/doris/httpv2/entity/ResponseEntityBuilder.java
index 30564945505..7cc4b6774f7 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/httpv2/entity/ResponseEntityBuilder.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/httpv2/entity/ResponseEntityBuilder.java
@@ -67,4 +67,9 @@ public class ResponseEntityBuilder {
ResponseBody body = new ResponseBody().code(RestApiStatusCode.NOT_FOUND).msg("Not Found").data(data);
return ResponseEntity.status(HttpStatus.OK).body(body);
}
+
+ public static ResponseEntity serviceUnavailable(String msg) {
+ ResponseBody body = new ResponseBody().code(RestApiStatusCode.SERVICE_UNAVAILABLE).msg(msg);
+ return ResponseEntity.status(HttpStatus.SERVICE_UNAVAILABLE).body(body);
+ }
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/HealthAction.java b/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/HealthAction.java
index fe27402f9a9..36e9e13751f 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/HealthAction.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/HealthAction.java
@@ -17,6 +17,7 @@
package org.apache.doris.httpv2.rest;
+import org.apache.doris.DorisFE;
import org.apache.doris.catalog.Env;
import org.apache.doris.common.Config;
import org.apache.doris.httpv2.entity.ResponseEntityBuilder;
@@ -39,6 +40,10 @@ public class HealthAction extends RestBaseController {
executeCheckPassword(request, response);
}
+ if (!DorisFE.isServerReady()) {
+ return ResponseEntityBuilder.serviceUnavailable("Server is not ready");
+ }
+
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());
diff --git a/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/RestApiStatusCode.java b/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/RestApiStatusCode.java
index 6952f2abb7c..fc803b2d955 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/RestApiStatusCode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/RestApiStatusCode.java
@@ -23,7 +23,8 @@ public enum RestApiStatusCode {
UNAUTHORIZED(401),
BAD_REQUEST(403),
NOT_FOUND(404),
- INTERNAL_SERVER_ERROR(500);
+ INTERNAL_SERVER_ERROR(500),
+ SERVICE_UNAVAILABLE(503);
public int code;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java
index c8b52ab1f58..6123f1ae85a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java
@@ -64,6 +64,7 @@ import org.apache.doris.qe.AuditLogHelper;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.QueryState.MysqlStateType;
import org.apache.doris.qe.StmtExecutor;
+import org.apache.doris.system.SystemInfoService;
import org.apache.doris.thrift.TCell;
import org.apache.doris.thrift.TRow;
import org.apache.doris.thrift.TStatusCode;
@@ -291,7 +292,7 @@ public class MTMVTask extends AbstractTask {
exec(execPartitionNames, tableWithPartKey);
break; // Exit loop if execution is successful
} catch (Exception e) {
- if (!(Config.isCloudMode() && e.getMessage().contains(FeConstants.CLOUD_RETRY_E230))) {
+ if (!(Config.isCloudMode() && SystemInfoService.needRetryWithReplan(e.getMessage()))) {
throw e; // Re-throw if it's not a retryable exception
}
lastException = e;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/StreamLoadHandler.java b/fe/fe-core/src/main/java/org/apache/doris/load/StreamLoadHandler.java
index 3bed51c1e47..6c0648c6022 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/StreamLoadHandler.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/StreamLoadHandler.java
@@ -92,7 +92,7 @@ public class StreamLoadHandler {
public static Backend selectBackend(String clusterName) throws LoadException {
List<Backend> backends = ((CloudSystemInfoService) Env.getCurrentSystemInfo())
.getBackendsByClusterName(clusterName)
- .stream().filter(Backend::isAlive)
+ .stream().filter(Backend::isLoadAvailable)
.collect(Collectors.toList());
if (backends.isEmpty()) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/AbstractInsertExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/AbstractInsertExecutor.java
index f1afbd135ec..9f490c090e8 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/AbstractInsertExecutor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/AbstractInsertExecutor.java
@@ -23,7 +23,6 @@ import org.apache.doris.catalog.TableIf;
import org.apache.doris.common.Config;
import org.apache.doris.common.ErrorCode;
import org.apache.doris.common.ErrorReport;
-import org.apache.doris.common.FeConstants;
import org.apache.doris.common.Status;
import org.apache.doris.common.UserException;
import org.apache.doris.common.util.DebugUtil;
@@ -37,6 +36,7 @@ import org.apache.doris.qe.Coordinator;
import org.apache.doris.qe.QeProcessorImpl;
import org.apache.doris.qe.QeProcessorImpl.QueryInfo;
import org.apache.doris.qe.StmtExecutor;
+import org.apache.doris.system.SystemInfoService;
import org.apache.doris.task.LoadEtlTask;
import org.apache.doris.thrift.TQueryType;
import org.apache.doris.thrift.TStatusCode;
@@ -246,8 +246,8 @@ public abstract class AbstractInsertExecutor {
}
} catch (Throwable t) {
onFail(t);
- // retry insert into from select when meet E-230 in cloud
- if (Config.isCloudMode() && t.getMessage().contains(FeConstants.CLOUD_RETRY_E230)) {
+ // retry insert into from select when meet "need re-plan error" or no scan node in cloud
+ if (Config.isCloudMode() && SystemInfoService.needRetryWithReplan(t.getMessage())) {
throw t;
}
return;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapInsertExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapInsertExecutor.java
index 7fceb289ef1..463da489f3d 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapInsertExecutor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/OlapInsertExecutor.java
@@ -26,7 +26,6 @@ import org.apache.doris.catalog.TableIf.TableType;
import org.apache.doris.cloud.system.CloudSystemInfoService;
import org.apache.doris.common.Config;
import org.apache.doris.common.ErrorCode;
-import org.apache.doris.common.FeConstants;
import org.apache.doris.common.MetaNotFoundException;
import org.apache.doris.common.UserException;
import org.apache.doris.common.util.DebugPointUtil;
@@ -51,6 +50,7 @@ import org.apache.doris.qe.StmtExecutor;
import org.apache.doris.service.ExecuteEnv;
import org.apache.doris.service.FrontendOptions;
import org.apache.doris.system.Backend;
+import org.apache.doris.system.SystemInfoService;
import org.apache.doris.thrift.TOlapTableLocationParam;
import org.apache.doris.thrift.TPartitionType;
import org.apache.doris.transaction.BeginTransactionException;
@@ -286,8 +286,8 @@ public class OlapInsertExecutor extends AbstractInsertExecutor {
labelName, queryId, txnId, abortTxnException);
}
}
- // retry insert into from select when meet E-230 in cloud
- if (Config.isCloudMode() && t.getMessage().contains(FeConstants.CLOUD_RETRY_E230)) {
+ // retry insert into from select when meet "need re-plan error" in cloud
+ if (Config.isCloudMode() && SystemInfoService.needRetryWithReplan(t.getMessage())) {
return;
}
StringBuilder sb = new StringBuilder(t.getMessage());
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
index 2bf55f150b6..0c2478a5ae0 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
@@ -500,7 +500,10 @@ public class StmtExecutor {
execute(queryId);
return;
} catch (UserException e) {
- if (!e.getMessage().contains(FeConstants.CLOUD_RETRY_E230) || i == retryTime) {
+ if (!SystemInfoService.needRetryWithReplan(e.getMessage()) || i == retryTime) {
+ // We have retried internally(in handleQueryWithRetry()) for other kinds of exceptions.
+ // And for error in SystemInfoService.NEED_REPLAN_ERRORS, they are not handled internally but here
+ // so we just handle these errors, and throw exception for other errors.
throw e;
}
if (this.coord != null && this.coord.isQueryCancelled()) {
@@ -516,8 +519,8 @@ public class StmtExecutor {
if (DebugPointUtil.isEnable("StmtExecutor.retry.longtime")) {
randomMillis = 1000;
}
- LOG.warn("receive E-230 tried={} first queryId={} last queryId={} new queryId={} sleep={}ms",
- i, DebugUtil.printId(firstQueryId), DebugUtil.printId(lastQueryId),
+ LOG.warn("receive '{}' tried={} first queryId={} last queryId={} new queryId={} sleep={}ms",
+ e.getMessage(), i, DebugUtil.printId(firstQueryId), DebugUtil.printId(lastQueryId),
DebugUtil.printId(queryId), randomMillis);
Thread.sleep(randomMillis);
context.getState().reset();
@@ -685,7 +688,9 @@ public class StmtExecutor {
if (LOG.isDebugEnabled()) {
LOG.debug("Command({}) process failed.", originStmt.originStmt, e);
}
- if (Config.isCloudMode() && e.getDetailMessage().contains(FeConstants.CLOUD_RETRY_E230)) {
+ if (Config.isCloudMode() && SystemInfoService.needRetryWithReplan(e.getDetailMessage())) {
+ // For errors in SystemInfoService.NEED_REPLAN_ERRORS,
+ // throw exception directly to trigger a replan retry outside(in StmtExecutor.queryRetry())
throw e;
}
context.getState().setError(e.getMysqlErrorCode(), e.getMessage());
@@ -907,14 +912,16 @@ public class StmtExecutor {
LOG.info("Query {} finished", DebugUtil.printId(context.queryId));
break;
} catch (RpcException | UserException e) {
- if (Config.isCloudMode() && e.getMessage().contains(FeConstants.CLOUD_RETRY_E230)) {
+ if (Config.isCloudMode() && SystemInfoService.needRetryWithReplan(e.getMessage())) {
+ // For errors in SystemInfoService.NEED_REPLAN_ERRORS,
+ // throw exception directly to trigger a replan retry outside(in StmtExecutor.queryRetry())
throw e;
}
// If the previous try is timeout or cancelled, then do not need try again.
if (this.coord != null && (this.coord.isQueryCancelled() || this.coord.isTimeout())) {
throw e;
}
- LOG.warn("due to exception {} retry {} rpc {} user {}",
+ LOG.warn("retry due to exception {}. retried {} times. is rpc error: {}, is user error: {}.",
e.getMessage(), i, e instanceof RpcException, e instanceof UserException);
boolean isNeedRetry = false;
@@ -1406,6 +1413,8 @@ public class StmtExecutor {
DebugUtil.printId(context.queryId()), e.getMessage());
LOG.warn(internalErrorSt.getErrorMsg());
coordBase.cancel(internalErrorSt);
+ // set to null so that the retry logic will generate a new coordinator
+ this.coord = null;
throw e;
} finally {
coordBase.close();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/system/Backend.java b/fe/fe-core/src/main/java/org/apache/doris/system/Backend.java
index 0f3ac2e2226..3f54db60a86 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/system/Backend.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/system/Backend.java
@@ -26,6 +26,7 @@ import org.apache.doris.common.Config;
import org.apache.doris.common.FeConstants;
import org.apache.doris.common.io.Text;
import org.apache.doris.common.io.Writable;
+import org.apache.doris.common.util.DebugPointUtil;
import org.apache.doris.common.util.PrintableMap;
import org.apache.doris.common.util.TimeUtils;
import org.apache.doris.persist.gson.GsonUtils;
@@ -37,6 +38,7 @@ import org.apache.doris.thrift.TNetworkAddress;
import org.apache.doris.thrift.TStorageMedium;
import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
@@ -49,6 +51,7 @@ import java.io.DataOutput;
import java.io.IOException;
import java.security.SecureRandom;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -153,10 +156,6 @@ public class Backend implements Writable {
// No need to persist, because only master FE handle heartbeat.
private int heartbeatFailureCounter = 0;
- // Not need serialize this field. If fe restart the state is reset to false. Maybe fe will
- // send some queries to this BE, it is not an important problem.
- private AtomicBoolean isShutDown = new AtomicBoolean(false);
-
private long nextForceEditlogHeartbeatTime = System.currentTimeMillis() + (new SecureRandom()).nextInt(60 * 1000);
public Backend() {
@@ -293,20 +292,38 @@ public class Backend implements Writable {
this.backendStatus.lastStreamLoadTime = lastStreamLoadTime;
}
+ // ATTN: This method only return the value of "isQueryDisabled",
+ // it does not determine the backend IS queryable or not, use isQueryAvailable instead.
public boolean isQueryDisabled() {
return backendStatus.isQueryDisabled;
}
- public void setQueryDisabled(boolean isQueryDisabled) {
- this.backendStatus.isQueryDisabled = isQueryDisabled;
+ // return true if be status is changed
+ public boolean setQueryDisabled(boolean isQueryDisabled) {
+ if (this.backendStatus.isQueryDisabled != isQueryDisabled) {
+ this.backendStatus.isQueryDisabled = isQueryDisabled;
+ return true;
+ }
+ return false;
}
+ // ATTN: This method only return the value of "isLoadDisabled",
+ // it does not determine the backend IS loadable or not, use isLoadAvailable instead.
public boolean isLoadDisabled() {
return backendStatus.isLoadDisabled;
}
- public void setLoadDisabled(boolean isLoadDisabled) {
- this.backendStatus.isLoadDisabled = isLoadDisabled;
+ // return true if be status is changed
+ public boolean setLoadDisabled(boolean isLoadDisabled) {
+ if (this.backendStatus.isLoadDisabled != isLoadDisabled) {
+ this.backendStatus.isLoadDisabled = isLoadDisabled;
+ return true;
+ }
+ return false;
+ }
+
+ private boolean isShutDown() {
+ return backendStatus.isShutdown;
}
public void setActive(boolean isActive) {
@@ -524,15 +541,21 @@ public class Backend implements Writable {
}
public boolean isQueryAvailable() {
- return isAlive() && !isQueryDisabled() && !isShutDown.get();
+ String debugDeadBeIds = DebugPointUtil.getDebugParamOrDefault(
+ "Backend.isQueryAvailable", "unavailableBeIds", "");
+ if (!Strings.isNullOrEmpty(debugDeadBeIds)
+ && Arrays.stream(debugDeadBeIds.split(",")).anyMatch(id -> Long.parseLong(id) == this.id)) {
+ return false;
+ }
+ return isAlive() && !isQueryDisabled() && !isShutDown();
}
public boolean isScheduleAvailable() {
- return isAlive() && !isDecommissioned();
+ return isAlive() && !isDecommissioned() && !isShutDown();
}
public boolean isLoadAvailable() {
- return isAlive() && !isLoadDisabled();
+ return isAlive() && !isLoadDisabled() && !isShutDown();
}
public void setDisks(ImmutableMap<String, DiskInfo> disks) {
@@ -872,10 +895,10 @@ public class Backend implements Writable {
this.arrowFlightSqlPort = hbResponse.getArrowFlightSqlPort();
}
- if (this.isShutDown.get() != hbResponse.isShutDown()) {
+ if (this.backendStatus.isShutdown != hbResponse.isShutDown()) {
isChanged = true;
LOG.info("{} shutdown state is changed", this.toString());
- this.isShutDown.set(hbResponse.isShutDown());
+ this.backendStatus.isShutdown = hbResponse.isShutDown();
}
if (!this.getNodeRoleTag().value.equals(hbResponse.getNodeRole()) && Tag.validNodeRoleTag(
@@ -990,6 +1013,8 @@ public class Backend implements Writable {
public volatile boolean isLoadDisabled = false;
@SerializedName("isActive")
public volatile boolean isActive = true;
+ @SerializedName("isShutdown")
+ public volatile boolean isShutdown = false;
// cloud mode, cloud control just query master, so not need SerializedName
public volatile long currentFragmentNum = 0;
@@ -1101,3 +1126,4 @@ public class Backend implements Writable {
}
}
+
diff --git a/fe/fe-core/src/main/java/org/apache/doris/system/SystemInfoService.java b/fe/fe-core/src/main/java/org/apache/doris/system/SystemInfoService.java
index 1e41eb8d501..b9c63f813b1 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/system/SystemInfoService.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/system/SystemInfoService.java
@@ -43,6 +43,7 @@ import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Multimap;
@@ -79,6 +80,13 @@ public class SystemInfoService {
public static final String NOT_USING_VALID_CLUSTER_MSG =
"Not using valid cloud clusters, please use a cluster before issuing any queries";
+ public static final String ERROR_E230 = "E-230";
+
+ public static final ImmutableSet<String> NEED_REPLAN_ERRORS = ImmutableSet.of(
+ NO_SCAN_NODE_BACKEND_AVAILABLE_MSG,
+ ERROR_E230
+ );
+
protected volatile ImmutableMap<Long, Backend> idToBackendRef = ImmutableMap.of();
protected volatile ImmutableMap<Long, AtomicLong> idToReportVersionRef = ImmutableMap.of();
@@ -998,17 +1006,11 @@ public class SystemInfoService {
}
if (alterClause.isQueryDisabled() != null) {
- if (!alterClause.isQueryDisabled().equals(be.isQueryDisabled())) {
- be.setQueryDisabled(alterClause.isQueryDisabled());
- shouldModify = true;
- }
+ shouldModify = be.setQueryDisabled(alterClause.isQueryDisabled());
}
if (alterClause.isLoadDisabled() != null) {
- if (!alterClause.isLoadDisabled().equals(be.isLoadDisabled())) {
- be.setLoadDisabled(alterClause.isLoadDisabled());
- shouldModify = true;
- }
+ shouldModify = be.setLoadDisabled(alterClause.isLoadDisabled());
}
if (shouldModify) {
@@ -1052,17 +1054,11 @@ public class SystemInfoService {
}
if (op.isQueryDisabled() != null) {
- if (!op.isQueryDisabled().equals(be.isQueryDisabled())) {
- be.setQueryDisabled(op.isQueryDisabled());
- shouldModify = true;
- }
+ shouldModify = be.setQueryDisabled(op.isQueryDisabled());
}
if (op.isLoadDisabled() != null) {
- if (!op.isLoadDisabled().equals(be.isLoadDisabled())) {
- be.setLoadDisabled(op.isLoadDisabled());
- shouldModify = true;
- }
+ shouldModify = be.setLoadDisabled(op.isLoadDisabled());
}
if (shouldModify) {
@@ -1156,4 +1152,16 @@ public class SystemInfoService {
return Env.getCurrentInvertedIndex().getTabletNumByBackendId(beId);
}
+ // If the error msg contains certain keywords, we need to retry the query with re-plan.
+ public static boolean needRetryWithReplan(String errorMsg) {
+ if (Strings.isNullOrEmpty(errorMsg)) {
+ return false;
+ }
+ for (String keyword : NEED_REPLAN_ERRORS) {
+ if (errorMsg.contains(keyword)) {
+ return true;
+ }
+ }
+ return false;
+ }
}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/utframe/DemoMultiBackendsTest.java b/fe/fe-core/src/test/java/org/apache/doris/utframe/DemoMultiBackendsTest.java
index 62c7d140bf1..800af43aaa1 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/utframe/DemoMultiBackendsTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/utframe/DemoMultiBackendsTest.java
@@ -212,7 +212,8 @@ public class DemoMultiBackendsTest {
result.getRows().get(0).get(BackendsProcDir.TITLE_NAMES.size() - 10));
Assert.assertEquals(
"{\"lastSuccessReportTabletsTime\":\"N/A\",\"lastStreamLoadTime\":-1,\"isQueryDisabled\":false,"
- + "\"isLoadDisabled\":false,\"isActive\":true,\"currentFragmentNum\":0,\"lastFragmentUpdateTime\":0}",
+ + "\"isLoadDisabled\":false,\"isActive\":true,\"isShutdown\":false,\"currentFragmentNum\":0,"
+ + "\"lastFragmentUpdateTime\":0}",
result.getRows().get(0).get(BackendsProcDir.TITLE_NAMES.size() - 7));
Assert.assertEquals("0", result.getRows().get(0).get(BackendsProcDir.TITLE_NAMES.size() - 6));
Assert.assertEquals(Tag.VALUE_MIX, result.getRows().get(0).get(BackendsProcDir.TITLE_NAMES.size() - 5));
diff --git a/regression-test/framework/pom.xml b/regression-test/framework/pom.xml
index 93fabc5f739..c946bb9bcf9 100644
--- a/regression-test/framework/pom.xml
+++ b/regression-test/framework/pom.xml
@@ -321,11 +321,11 @@ under the License.
<artifactId>aliyun-java-sdk-ram</artifactId>
<version>3.3.1</version>
</dependency>
- <dependency>
+ <!--dependency>
<groupId>com.google.code.gson</groupId>
<artifactId>gson</artifactId>
<version>2.2.4</version>
- </dependency>
+ </dependency-->
<!---->
<!-- txcloud ram -->
<dependency>
diff --git a/regression-test/suites/cloud_p0/query_retry/test_retry_no_scan_node.groovy b/regression-test/suites/cloud_p0/query_retry/test_retry_no_scan_node.groovy
new file mode 100644
index 00000000000..ce5417924e5
--- /dev/null
+++ b/regression-test/suites/cloud_p0/query_retry/test_retry_no_scan_node.groovy
@@ -0,0 +1,80 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+import org.apache.doris.regression.suite.ClusterOptions
+import org.apache.doris.regression.util.NodeType
+import org.apache.doris.regression.suite.SuiteCluster
+
+suite("test_retry_no_scan_node", "p0, docker") {
+ if (!isCloudMode()) {
+ return
+ }
+ def options = new ClusterOptions()
+ options.enableDebugPoints()
+ options.setFeNum(1)
+ options.feConfigs.add('max_query_retry_time=100')
+ options.feConfigs.add('sys_log_verbose_modules=org')
+ options.setBeNum(2)
+ options.cloudMode = true
+ // 1. connect to master
+ options.connectToFollower = false
+
+ def queryTask = {
+ for (int i = 0; i < 100; i++) {
+ try {
+ log.info("query count: {}", i)
+ sql """select * from test_no_scan_node_table"""
+ Thread.sleep(100)
+ } catch (Exception e) {
+ logger.warn("select failed: ${e.message}")
+ assertFalse(true);
+ }
+ }
+ }
+
+ docker(options) {
+ def be1 = cluster.getBeByIndex(1)
+ def beId = be1.backendId;
+
+ try {
+ sql """
+ CREATE TABLE test_no_scan_node_table
+ ( k1 TINYINT, k2 INT not null )
+ DISTRIBUTED BY HASH(k2) BUCKETS 2 PROPERTIES ( "replication_num" = "1" );
+ """
+ sql """
+ INSERT INTO test_no_scan_node_table VALUES (1, 1), (2, 2), (3, 3);
+ """
+
+ def result = sql """select * from test_no_scan_node_table order by k2;"""
+ log.info("insert result : {}", result)
+ assertEquals([[1, 1], [2, 2], [3, 3]], result)
+
+ // this should be run at least 10 seconds
+ def queryThread = Thread.start(queryTask)
+
+ // inject query not available error
+ cluster.injectDebugPoints(NodeType.FE, ['Backend.isQueryAvailable' : [unavailableBeIds:beId]])
+ // query should have no failure
+ // wait query thread finish
+ queryThread.join(15000);
+ } finally {
+ cluster.clearFrontendDebugPoints()
+ cluster.clearBackendDebugPoints()
+ }
+ }
+}
+
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org