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 64f16e69a586c49f264c5184dfb1ca94908ac157 Author: zhangstar333 <[email protected]> AuthorDate: Mon May 25 00:02:09 2026 +0800 retry stream load rpc --- be/src/cloud/cloud_warm_up_manager.cpp | 4 +- be/src/load/stream_load/stream_load_executor.cpp | 36 ++++-- be/src/service/http/action/http_stream.cpp | 11 +- be/src/service/http/action/stream_load.cpp | 13 +- be/src/util/thrift_rpc_helper.cpp | 138 +++++++++++++++++++++ be/src/util/thrift_rpc_helper.h | 28 +++++ .../apache/doris/service/FrontendServiceImpl.java | 33 +++++ gensrc/thrift/FrontendService.thrift | 13 ++ 8 files changed, 259 insertions(+), 17 deletions(-) diff --git a/be/src/cloud/cloud_warm_up_manager.cpp b/be/src/cloud/cloud_warm_up_manager.cpp index 5633794ddf9..c4d924fcfd9 100644 --- a/be/src/cloud/cloud_warm_up_manager.cpp +++ b/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, "", "", "", info.replica.host); + brpc_addr, "", "", "", replica.host); if (!brpc_stub) { st = Status::RpcError("Address {} is wrong", brpc_addr); add_failure(info, target, st); @@ -975,7 +975,7 @@ void CloudWarmUpManager::_recycle_cache(int64_t tablet_id, Status st = Status::OK(); std::shared_ptr<PBackendService_Stub> brpc_stub = ExecEnv::GetInstance()->brpc_internal_client_cache()->get_new_client_no_cache( - brpc_addr, "", "", "", replica.replica.host); + brpc_addr, "", "", "", replica.host); if (!brpc_stub) { st = Status::RpcError("Address {} is wrong", brpc_addr); continue; diff --git a/be/src/load/stream_load/stream_load_executor.cpp b/be/src/load/stream_load/stream_load_executor.cpp index 9b307c932d4..f5ce55b7f73 100644 --- a/be/src/load/stream_load/stream_load_executor.cpp +++ b/be/src/load/stream_load/stream_load_executor.cpp @@ -208,9 +208,14 @@ Status StreamLoadExecutor::begin_txn(StreamLoadContext* ctx) { } else { SCOPED_RAW_TIMER(&duration_ns); #ifndef BE_TEST - RETURN_IF_ERROR(ThriftRpcHelper::rpc<FrontendServiceClient>( - master_addr_provider, [&request, &result](FrontendServiceConnection& client) { + RETURN_IF_ERROR(ThriftRpcHelper::rpc_fe_with_master_refresh( + master_addr_provider, + [&request, &result](FrontendServiceConnection& client) { client->loadTxnBegin(result, request); + }, + [&result]() { return result.status; }, + [&result]() -> const TNetworkAddress* { + return result.__isset.master_address ? &result.master_address : nullptr; })); #else result = k_stream_load_begin_result; @@ -253,11 +258,15 @@ Status StreamLoadExecutor::pre_commit_txn(StreamLoadContext* ctx) { SCOPED_RAW_TIMER(&duration_ns); #ifndef BE_TEST auto master_addr_provider = [this]() { return _exec_env->cluster_info()->master_fe_addr; }; - RETURN_IF_ERROR(ThriftRpcHelper::rpc<FrontendServiceClient>( + RETURN_IF_ERROR(ThriftRpcHelper::rpc_fe_with_master_refresh( master_addr_provider, [&request, &result](FrontendServiceConnection& client) { client->loadTxnPreCommit(result, request); }, + [&result]() { return result.status; }, + [&result]() -> const TNetworkAddress* { + return result.__isset.master_address ? &result.master_address : nullptr; + }, config::txn_commit_rpc_timeout_ms)); #else result = k_stream_load_commit_result; @@ -297,11 +306,15 @@ Status StreamLoadExecutor::operate_txn_2pc(StreamLoadContext* ctx) { { SCOPED_RAW_TIMER(&duration_ns); auto master_addr_provider = [this]() { return _exec_env->cluster_info()->master_fe_addr; }; - RETURN_IF_ERROR(ThriftRpcHelper::rpc<FrontendServiceClient>( + RETURN_IF_ERROR(ThriftRpcHelper::rpc_fe_with_master_refresh( master_addr_provider, [&request, &result](FrontendServiceConnection& client) { client->loadTxn2PC(result, request); }, + [&result]() { return result.status; }, + [&result]() -> const TNetworkAddress* { + return result.__isset.master_address ? &result.master_address : nullptr; + }, config::txn_commit_rpc_timeout_ms)); } g_stream_load_commit_txn_latency << duration_ns / 1000; @@ -347,11 +360,15 @@ Status StreamLoadExecutor::commit_txn(StreamLoadContext* ctx) { TLoadTxnCommitResult result; #ifndef BE_TEST auto master_addr_provider = [this]() { return _exec_env->cluster_info()->master_fe_addr; }; - RETURN_IF_ERROR(ThriftRpcHelper::rpc<FrontendServiceClient>( + RETURN_IF_ERROR(ThriftRpcHelper::rpc_fe_with_master_refresh( master_addr_provider, [&request, &result](FrontendServiceConnection& client) { client->loadTxnCommit(result, request); }, + [&result]() { return result.status; }, + [&result]() -> const TNetworkAddress* { + return result.__isset.master_address ? &result.master_address : nullptr; + }, config::txn_commit_rpc_timeout_ms)); #else result = k_stream_load_commit_result; @@ -397,9 +414,14 @@ void StreamLoadExecutor::rollback_txn(StreamLoadContext* ctx) { TLoadTxnRollbackResult result; #ifndef BE_TEST auto master_addr_provider = [this]() { return _exec_env->cluster_info()->master_fe_addr; }; - auto rpc_st = ThriftRpcHelper::rpc<FrontendServiceClient>( - master_addr_provider, [&request, &result](FrontendServiceConnection& client) { + auto rpc_st = ThriftRpcHelper::rpc_fe_with_master_refresh( + master_addr_provider, + [&request, &result](FrontendServiceConnection& client) { client->loadTxnRollback(result, request); + }, + [&result]() { return result.status; }, + [&result]() -> const TNetworkAddress* { + return result.__isset.master_address ? &result.master_address : nullptr; }); if (!rpc_st.ok()) { LOG(WARNING) << "transaction rollback failed. errmsg=" << rpc_st << ctx->brief(); diff --git a/be/src/service/http/action/http_stream.cpp b/be/src/service/http/action/http_stream.cpp index e8b8d974211..a301d1c2358 100644 --- a/be/src/service/http/action/http_stream.cpp +++ b/be/src/service/http/action/http_stream.cpp @@ -376,13 +376,16 @@ Status HttpStreamAction::process_put(HttpRequest* http_req, } // plan this load - TNetworkAddress master_addr = _exec_env->cluster_info()->master_fe_addr; int64_t stream_load_put_start_time = MonotonicNanos(); - RETURN_IF_ERROR(ThriftRpcHelper::rpc<FrontendServiceClient>( - master_addr.hostname, master_addr.port, + auto master_addr_provider = [this]() { return _exec_env->cluster_info()->master_fe_addr; }; + RETURN_IF_ERROR(ThriftRpcHelper::rpc_fe_with_master_refresh( + master_addr_provider, [&request, ctx](FrontendServiceConnection& client) { client->streamLoadPut(ctx->put_result, request); - })); + }, + [ctx]() { return ctx->put_result.status; }, + // TStreamLoadPutResult does not carry master_address. + []() -> const TNetworkAddress* { return nullptr; })); ctx->put_result.pipeline_params.query_options.__set_enable_strict_cast(false); ctx->stream_load_put_cost_nanos = MonotonicNanos() - stream_load_put_start_time; Status plan_status(Status::create(ctx->put_result.status)); diff --git a/be/src/service/http/action/stream_load.cpp b/be/src/service/http/action/stream_load.cpp index 3b312a6d153..3c414ad858f 100644 --- a/be/src/service/http/action/stream_load.cpp +++ b/be/src/service/http/action/stream_load.cpp @@ -800,13 +800,18 @@ Status StreamLoadAction::_process_put(HttpRequest* http_req, #ifndef BE_TEST // plan this load - TNetworkAddress master_addr = _exec_env->cluster_info()->master_fe_addr; int64_t stream_load_put_start_time = MonotonicNanos(); - RETURN_IF_ERROR(ThriftRpcHelper::rpc<FrontendServiceClient>( - master_addr.hostname, master_addr.port, + auto master_addr_provider = [this]() { return _exec_env->cluster_info()->master_fe_addr; }; + RETURN_IF_ERROR(ThriftRpcHelper::rpc_fe_with_master_refresh( + master_addr_provider, [&request, ctx](FrontendServiceConnection& client) { client->streamLoadPut(ctx->put_result, request); - })); + }, + [ctx]() { return ctx->put_result.status; }, + // TStreamLoadPutResult does not carry master_address (FE + // streamLoadPut does not call checkMaster), so on transport-layer + // failure we fall back to heartbeat-driven refresh. + []() -> const TNetworkAddress* { return nullptr; })); ctx->stream_load_put_cost_nanos = MonotonicNanos() - stream_load_put_start_time; #else ctx->put_result = k_stream_load_put_result; diff --git a/be/src/util/thrift_rpc_helper.cpp b/be/src/util/thrift_rpc_helper.cpp index 05cfc2ffc00..39e614d324b 100644 --- a/be/src/util/thrift_rpc_helper.cpp +++ b/be/src/util/thrift_rpc_helper.cpp @@ -127,6 +127,79 @@ Status ThriftRpcHelper::rpc(std::function<TNetworkAddress()> address_provider, return Status::OK(); } +template <typename T> +Status ThriftRpcHelper::rpc(std::function<TNetworkAddress()> address_provider, + std::function<void(ClientConnection<T>&)> callback, int timeout_ms) { + TNetworkAddress address = address_provider(); + if (address.hostname.empty() || address.port == 0) { + return Status::Error<ErrorCode::SERVICE_UNAVAILABLE>("FE address is not available"); + } + Status status; + ClientConnection<T> client(_s_exec_env->get_client_cache<T>(), address, timeout_ms, &status); + if (!status.ok()) { +#ifndef ADDRESS_SANITIZER + LOG(WARNING) << "Connect frontend failed, address=" << address << ", status=" << status; +#endif + return status; + } + try { + try { + callback(client); + } catch (apache::thrift::transport::TTransportException& e) { + std::cerr << "thrift error, reason=" << e.what(); +#ifndef ADDRESS_SANITIZER + LOG(WARNING) << "retrying call frontend service after " + << config::thrift_client_retry_interval_ms << " ms, address=" << address + << ", reason=" << e.what(); +#endif + std::this_thread::sleep_for( + std::chrono::milliseconds(config::thrift_client_retry_interval_ms)); + TNetworkAddress retry_address = address_provider(); + if (retry_address.hostname.empty() || retry_address.port == 0) { + return Status::Error<ErrorCode::SERVICE_UNAVAILABLE>("FE address is not available"); + } + if (retry_address.hostname != address.hostname || retry_address.port != address.port) { +#ifndef ADDRESS_SANITIZER + LOG(INFO) << "retrying call frontend service with new address=" << retry_address; +#endif + Status retry_status; + ClientConnection<T> retry_client(_s_exec_env->get_client_cache<T>(), retry_address, + timeout_ms, &retry_status); + if (!retry_status.ok()) { +#ifndef ADDRESS_SANITIZER + LOG(WARNING) << "Connect frontend failed, address=" << retry_address + << ", status=" << retry_status; +#endif + return retry_status; + } + callback(retry_client); + } else { + status = client.reopen(timeout_ms); + if (!status.ok()) { +#ifndef ADDRESS_SANITIZER + LOG(WARNING) << "client reopen failed. address=" << address + << ", status=" << status; +#endif + return status; + } + callback(client); + } + } + } catch (apache::thrift::TException& e) { +#ifndef ADDRESS_SANITIZER + LOG(WARNING) << "call frontend service failed, address=" << address + << ", reason=" << e.what(); +#endif + std::this_thread::sleep_for( + std::chrono::milliseconds(config::thrift_client_retry_interval_ms * 2)); + // just reopen to disable this connection + static_cast<void>(client.reopen(timeout_ms)); + return Status::RpcError("failed to call frontend service, FE address={}:{}, reason: {}", + address.hostname, address.port, e.what()); + } + return Status::OK(); +} + template Status ThriftRpcHelper::rpc<FrontendServiceClient>( const std::string& ip, const int32_t port, std::function<void(ClientConnection<FrontendServiceClient>&)> callback, int timeout_ms); @@ -143,4 +216,69 @@ 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); + + 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; + LOG(INFO) << "fe RPC got NOT_MASTER from " << old_master + << ", refreshed master_fe_addr to " << new_master; + } else { + // Scenario A: transport failure; or NOT_MASTER without master_address + // (old FE without thrift change). Brief backoff so heartbeat thread + // may have time to push a new master. + std::this_thread::sleep_for( + std::chrono::milliseconds(config::thrift_client_retry_interval_ms * 2)); + TNetworkAddress refreshed = address_provider(); + if (refreshed.hostname == old_master.hostname && refreshed.port == old_master.port) { + // Heartbeat has not learned about a new master yet. Return the + // original error and let upper layer (stream load executor) decide. + return st.ok() ? Status::Error<ErrorCode::NOT_MASTER>( + "FE returned NOT_MASTER but no new master address available") + : st; + } + LOG(INFO) << "fe RPC failed against " << old_master + << ", heartbeat now reports master=" << refreshed << ", retrying"; + } + + // Second (and only) attempt against the new master address. + Status retry_st = rpc<FrontendServiceClient>(address_provider, callback, timeout_ms); + if (!retry_st.ok()) { + LOG(WARNING) << "fe RPC retry after master refresh still failed: " << retry_st; + } + return retry_st; +} + } // namespace doris diff --git a/be/src/util/thrift_rpc_helper.h b/be/src/util/thrift_rpc_helper.h index 3617967b815..c5f786599ce 100644 --- a/be/src/util/thrift_rpc_helper.h +++ b/be/src/util/thrift_rpc_helper.h @@ -29,6 +29,7 @@ namespace doris { class ExecEnv; +class FrontendServiceClient; template <class T> class ClientConnection; @@ -58,6 +59,33 @@ public: static Status rpc(std::function<TNetworkAddress()> address_provider, std::function<void(ClientConnection<T>&)> callback, int timeout_ms); + // Wrapper around `rpc<FrontendServiceClient>` for the 6 stream-load FE RPCs + // (loadTxnBegin / loadTxnPreCommit / loadTxnCommit / loadTxn2PC / + // loadTxnRollback / streamLoadPut). It transparently retries **once** in + // two narrow situations: + // 1) Transport-layer failure (master FE not reachable, e.g. restarting): + // the first RPC never reached FE, so retrying with the same txn_id is + // side-effect-free. + // 2) The FE returned TStatusCode::NOT_MASTER (BE was still pointing at + // the previous master after a leadership transfer). FE early-returns + // before touching the txn, and now also sets `result.master_address` + // so this helper can refresh `cluster_info()->master_fe_addr` without + // any extra round trip. + // Other failures (ANALYSIS_ERROR, LABEL_ALREADY_EXISTS, ...) are returned + // verbatim — they are not master-related and retrying is unsafe. + // + // `status_extractor` returns the response's TStatus. + // `master_addr_extractor` returns a pointer to the response's + // master_address (or nullptr if not set). Wiring these as callbacks keeps + // this helper free of any compile-time dependency on the specific Resp + // struct. + static Status 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 = config::thrift_rpc_timeout_ms); + static ExecEnv* get_exec_env() { return _s_exec_env; } private: diff --git a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java index b737ed2bc94..6d7fd0d487e 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java +++ b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java @@ -401,6 +401,19 @@ public class FrontendServiceImpl implements FrontendService.Iface { return new TNetworkAddress(masterHost, masterRpcPort); } + // Same as getMasterAddress, but returns null when the master is unknown + // (host empty or rpc port 0). Used by NOT_MASTER early-return paths so that + // BE can distinguish "FE knows the new master" vs "no master yet". + private static TNetworkAddress getMasterAddressOrNull() { + Env env = Env.getCurrentEnv(); + String masterHost = env.getMasterHost(); + int masterRpcPort = env.getMasterRpcPort(); + if (masterHost == null || masterHost.isEmpty() || masterRpcPort == 0) { + return null; + } + return new TNetworkAddress(masterHost, masterRpcPort); + } + public FrontendServiceImpl(ExecuteEnv exeEnv) { masterImpl = new MasterImpl(); this.exeEnv = exeEnv; @@ -1412,6 +1425,10 @@ public class FrontendServiceImpl implements FrontendService.Iface { result.setStatus(status); if (status.getStatusCode() != TStatusCode.OK) { + TNetworkAddress masterAddr = getMasterAddressOrNull(); + if (masterAddr != null) { + result.setMasterAddress(masterAddr); + } return result; } @@ -1634,6 +1651,14 @@ public class FrontendServiceImpl implements FrontendService.Iface { return result; } + if (status.getStatusCode() != TStatusCode.OK) { + TNetworkAddress masterAddr = getMasterAddressOrNull(); + if (masterAddr != null) { + result.setMasterAddress(masterAddr); + } + return result; + } + try { loadTxnPreCommitImpl(request); } catch (UserException e) { @@ -1775,6 +1800,10 @@ public class FrontendServiceImpl implements FrontendService.Iface { result.setStatus(status); if (status.getStatusCode() != TStatusCode.OK) { + TNetworkAddress masterAddr = getMasterAddressOrNull(); + if (masterAddr != null) { + result.setMasterAddress(masterAddr); + } return result; } @@ -1867,6 +1896,10 @@ public class FrontendServiceImpl implements FrontendService.Iface { result.setStatus(status); if (status.getStatusCode() != TStatusCode.OK) { + TNetworkAddress masterAddr = getMasterAddressOrNull(); + if (masterAddr != null) { + result.setMasterAddress(masterAddr); + } return result; } diff --git a/gensrc/thrift/FrontendService.thrift b/gensrc/thrift/FrontendService.thrift index 33c0a117821..2d67bd33dce 100644 --- a/gensrc/thrift/FrontendService.thrift +++ b/gensrc/thrift/FrontendService.thrift @@ -504,6 +504,9 @@ struct TLoadTxnBeginResult { // If use_table_group_commit_mode is true in TLoadTxnBeginRequest, and table group_commit_mode property is // async_mode or sync_mode, return table group_commit_mode (begin_txn is skipped) 5: optional string table_group_commit_mode + // Set when status.status_code == NOT_MASTER, so that BE can refresh its + // cached master_fe_addr without an extra round trip. + 6: optional Types.TNetworkAddress master_address } struct TBeginTxnRequest { @@ -710,6 +713,10 @@ struct TLoadTxnCommitRequest { struct TLoadTxnCommitResult { 1: required Status.TStatus status + // Set when status.status_code == NOT_MASTER, so that BE can refresh its + // cached master_fe_addr without an extra round trip. Shared between + // loadTxnCommit and loadTxnPreCommit. + 2: optional Types.TNetworkAddress master_address } struct TCommitTxnRequest { @@ -756,6 +763,9 @@ struct TLoadTxn2PCRequest { struct TLoadTxn2PCResult { 1: required Status.TStatus status + // Set when status.status_code == NOT_MASTER, so that BE can refresh its + // cached master_fe_addr without an extra round trip. + 2: optional Types.TNetworkAddress master_address } struct TRollbackTxnRequest { @@ -797,6 +807,9 @@ struct TLoadTxnRollbackRequest { struct TLoadTxnRollbackResult { 1: required Status.TStatus status + // Set when status.status_code == NOT_MASTER, so that BE can refresh its + // cached master_fe_addr without an extra round trip. + 2: optional Types.TNetworkAddress master_address } struct TSnapshotLoaderReportRequest { --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
