You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by kx...@apache.org on 2023/10/13 17:25:24 UTC
[doris] 03/14: [feature](thrift) Add FE thrift rpc redirect master address (#25371)
This is an automated email from the ASF dual-hosted git repository.
kxiao pushed a commit to branch branch-2.0
in repository https://gitbox.apache.org/repos/asf/doris.git
commit c7cba573d489fc2b114602590869175d5c761170
Author: Jack Drogon <ja...@gmail.com>
AuthorDate: Thu Oct 12 22:17:46 2023 -0500
[feature](thrift) Add FE thrift rpc redirect master address (#25371)
Signed-off-by: Jack Drogon <ja...@gmail.com>
---
.../java/org/apache/doris/service/FrontendServiceImpl.java | 13 +++++++++++++
gensrc/thrift/FrontendService.thrift | 6 ++++++
2 files changed, 19 insertions(+)
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 d8d69583696..8db840b5bf3 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
@@ -228,6 +228,13 @@ public class FrontendServiceImpl implements FrontendService.Iface {
private ConcurrentHashMap<Long, Integer> multiTableFragmentInstanceIdIndexMap =
new ConcurrentHashMap<>(64);
+ private static TNetworkAddress getMasterAddress() {
+ Env env = Env.getCurrentEnv();
+ String masterHost = env.getMasterHost();
+ int masterRpcPort = env.getMasterRpcPort();
+ return new TNetworkAddress(masterHost, masterRpcPort);
+ }
+
public FrontendServiceImpl(ExecuteEnv exeEnv) {
masterImpl = new MasterImpl();
this.exeEnv = exeEnv;
@@ -1188,6 +1195,7 @@ public class FrontendServiceImpl implements FrontendService.Iface {
if (!Env.getCurrentEnv().isMaster()) {
status.setStatusCode(TStatusCode.NOT_MASTER);
status.addToErrorMsgs(NOT_MASTER_ERR_MSG);
+ result.setMasterAddress(getMasterAddress());
LOG.error("failed to get beginTxn: {}", NOT_MASTER_ERR_MSG);
return result;
}
@@ -1566,6 +1574,7 @@ public class FrontendServiceImpl implements FrontendService.Iface {
if (!Env.getCurrentEnv().isMaster()) {
status.setStatusCode(TStatusCode.NOT_MASTER);
status.addToErrorMsgs(NOT_MASTER_ERR_MSG);
+ result.setMasterAddress(getMasterAddress());
LOG.error("failed to get commitTxn: {}", NOT_MASTER_ERR_MSG);
return result;
}
@@ -1753,6 +1762,7 @@ public class FrontendServiceImpl implements FrontendService.Iface {
if (!Env.getCurrentEnv().isMaster()) {
status.setStatusCode(TStatusCode.NOT_MASTER);
status.addToErrorMsgs(NOT_MASTER_ERR_MSG);
+ result.setMasterAddress(getMasterAddress());
LOG.error("failed to get rollbackTxn: {}", NOT_MASTER_ERR_MSG);
return result;
}
@@ -2639,6 +2649,7 @@ public class FrontendServiceImpl implements FrontendService.Iface {
if (!Env.getCurrentEnv().isMaster()) {
status.setStatusCode(TStatusCode.NOT_MASTER);
status.addToErrorMsgs(NOT_MASTER_ERR_MSG);
+ result.setMasterAddress(getMasterAddress());
LOG.error("failed to get getSnapshot: {}", NOT_MASTER_ERR_MSG);
return result;
}
@@ -2726,6 +2737,7 @@ public class FrontendServiceImpl implements FrontendService.Iface {
if (!Env.getCurrentEnv().isMaster()) {
status.setStatusCode(TStatusCode.NOT_MASTER);
status.addToErrorMsgs(NOT_MASTER_ERR_MSG);
+ result.setMasterAddress(getMasterAddress());
LOG.error("failed to get restoreSnapshot: {}", NOT_MASTER_ERR_MSG);
return result;
}
@@ -2847,6 +2859,7 @@ public class FrontendServiceImpl implements FrontendService.Iface {
if (!Env.getCurrentEnv().isMaster()) {
status.setStatusCode(TStatusCode.NOT_MASTER);
status.addToErrorMsgs(NOT_MASTER_ERR_MSG);
+ result.setMasterAddress(getMasterAddress());
LOG.error("failed to get getMasterToken: {}", NOT_MASTER_ERR_MSG);
return result;
}
diff --git a/gensrc/thrift/FrontendService.thrift b/gensrc/thrift/FrontendService.thrift
index 5a3cf88db8f..834e6bcb46c 100644
--- a/gensrc/thrift/FrontendService.thrift
+++ b/gensrc/thrift/FrontendService.thrift
@@ -558,6 +558,7 @@ struct TBeginTxnResult {
2: optional i64 txn_id
3: optional string job_status // if label already used, set status of existing job
4: optional i64 db_id
+ 5: optional Types.TNetworkAddress master_address
}
// StreamLoad request, used to load a streaming to engine
@@ -706,6 +707,7 @@ struct TCommitTxnRequest {
struct TCommitTxnResult {
1: optional Status.TStatus status
+ 2: optional Types.TNetworkAddress master_address
}
struct TLoadTxn2PCRequest {
@@ -741,6 +743,7 @@ struct TRollbackTxnRequest {
struct TRollbackTxnResult {
1: optional Status.TStatus status
+ 2: optional Types.TNetworkAddress master_address
}
struct TLoadTxnRollbackRequest {
@@ -1045,6 +1048,7 @@ struct TGetSnapshotResult {
1: optional Status.TStatus status
2: optional binary meta
3: optional binary job_info
+ 4: optional Types.TNetworkAddress master_address
}
struct TTableRef {
@@ -1069,6 +1073,7 @@ struct TRestoreSnapshotRequest {
struct TRestoreSnapshotResult {
1: optional Status.TStatus status
+ 2: optional Types.TNetworkAddress master_address
}
struct TGetMasterTokenRequest {
@@ -1080,6 +1085,7 @@ struct TGetMasterTokenRequest {
struct TGetMasterTokenResult {
1: optional Status.TStatus status
2: optional string token
+ 3: optional Types.TNetworkAddress master_address
}
typedef TGetBinlogRequest TGetBinlogLagRequest
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org