This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new c4b22f26ac [IOTDB-3477] Send two phase message to ConfigNode when
starting DataNode (#6269)
c4b22f26ac is described below
commit c4b22f26ac86b7de0d0a3d53d4785973af3da019
Author: Chen YZ <[email protected]>
AuthorDate: Tue Jun 14 14:57:01 2022 +0800
[IOTDB-3477] Send two phase message to ConfigNode when starting DataNode
(#6269)
---
.../thrift/ConfigNodeRPCServiceProcessor.java | 7 ++
.../apache/iotdb/db/client/ConfigNodeClient.java | 17 +++++
.../java/org/apache/iotdb/db/service/DataNode.java | 75 +++++++++++++++++++---
.../src/main/thrift/confignode.thrift | 7 ++
4 files changed, 96 insertions(+), 10 deletions(-)
diff --git
a/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
b/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
index aabffbed77..4d0cf0e5c0 100644
---
a/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
@@ -62,6 +62,7 @@ import
org.apache.iotdb.confignode.rpc.thrift.TConfigNodeRegisterReq;
import org.apache.iotdb.confignode.rpc.thrift.TConfigNodeRegisterResp;
import org.apache.iotdb.confignode.rpc.thrift.TCountStorageGroupResp;
import org.apache.iotdb.confignode.rpc.thrift.TCreateFunctionReq;
+import org.apache.iotdb.confignode.rpc.thrift.TDataNodeActiveReq;
import org.apache.iotdb.confignode.rpc.thrift.TDataNodeInfoResp;
import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRegisterReq;
import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRegisterResp;
@@ -134,6 +135,12 @@ public class ConfigNodeRPCServiceProcessor implements
ConfigIService.Iface {
return resp;
}
+ @Override
+ public TSStatus activeDataNode(TDataNodeActiveReq req) throws TException {
+ // TODO: implement active data node
+ return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ }
+
@Override
public TDataNodeInfoResp getDataNodeInfo(int dataNodeID) throws TException {
GetDataNodeInfoReq queryReq = new GetDataNodeInfoReq(dataNodeID);
diff --git
a/server/src/main/java/org/apache/iotdb/db/client/ConfigNodeClient.java
b/server/src/main/java/org/apache/iotdb/db/client/ConfigNodeClient.java
index f7e42bc075..f295613b4e 100644
--- a/server/src/main/java/org/apache/iotdb/db/client/ConfigNodeClient.java
+++ b/server/src/main/java/org/apache/iotdb/db/client/ConfigNodeClient.java
@@ -39,6 +39,7 @@ import
org.apache.iotdb.confignode.rpc.thrift.TConfigNodeRegisterReq;
import org.apache.iotdb.confignode.rpc.thrift.TConfigNodeRegisterResp;
import org.apache.iotdb.confignode.rpc.thrift.TCountStorageGroupResp;
import org.apache.iotdb.confignode.rpc.thrift.TCreateFunctionReq;
+import org.apache.iotdb.confignode.rpc.thrift.TDataNodeActiveReq;
import org.apache.iotdb.confignode.rpc.thrift.TDataNodeInfoResp;
import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRegisterReq;
import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRegisterResp;
@@ -251,6 +252,22 @@ public class ConfigNodeClient implements
ConfigIService.Iface, SyncThriftClient,
throw new TException(MSG_RECONNECTION_FAIL);
}
+ @Override
+ public TSStatus activeDataNode(TDataNodeActiveReq req) throws TException {
+ for (int i = 0; i < RETRY_NUM; i++) {
+ try {
+ TSStatus status = client.activeDataNode(req);
+ if (!updateConfigNodeLeader(status)) {
+ return status;
+ }
+ } catch (TException e) {
+ configLeader = null;
+ }
+ reconnect();
+ }
+ throw new TException(MSG_RECONNECTION_FAIL);
+ }
+
@Override
public TDataNodeInfoResp getDataNodeInfo(int dataNodeId) throws TException {
for (int i = 0; i < RETRY_NUM; i++) {
diff --git a/server/src/main/java/org/apache/iotdb/db/service/DataNode.java
b/server/src/main/java/org/apache/iotdb/db/service/DataNode.java
index 629672dbe8..1e8d500d15 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/DataNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/DataNode.java
@@ -22,6 +22,7 @@ import org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TDataNodeInfo;
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.concurrent.IoTDBDefaultThreadExceptionHandler;
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.exception.ConfigurationException;
@@ -32,6 +33,7 @@ import org.apache.iotdb.commons.service.StartupChecks;
import org.apache.iotdb.commons.udf.service.UDFClassLoaderManager;
import org.apache.iotdb.commons.udf.service.UDFExecutableManager;
import org.apache.iotdb.commons.udf.service.UDFRegistrationService;
+import org.apache.iotdb.confignode.rpc.thrift.TDataNodeActiveReq;
import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRegisterReq;
import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRegisterResp;
import org.apache.iotdb.db.client.ConfigNodeClient;
@@ -119,10 +121,12 @@ public class DataNode implements DataNodeMBean {
protected void doAddNode(String[] args) {
try {
// setup InternalService
+ setUpInternalService();
+ // contact with config node to join into the cluster
prepareJoinCluster();
// setup DataNode
active();
- // contact with config node to join into the cluster
+ // send message to config node stating that data node is ready
joinCluster();
// setup rpc service
setUpRPCService();
@@ -144,7 +148,7 @@ public class DataNode implements DataNodeMBean {
}
/** prepare iotdb and start InternalService */
- private void prepareJoinCluster() throws StartupException {
+ private void setUpInternalService() throws StartupException {
// check iotdb server first
StartupChecks checks = new StartupChecks().withDefaultTest();
checks.verify();
@@ -161,13 +165,13 @@ public class DataNode implements DataNodeMBean {
}
/** register DataNode with ConfigNode */
- private void joinCluster() throws StartupException {
+ private void prepareJoinCluster() throws StartupException {
int retry = DEFAULT_JOIN_RETRY;
ConfigNodeInfo.getInstance()
.updateConfigNodeList(IoTDBDescriptor.getInstance().getConfig().getConfigNodeList());
while (retry > 0) {
- logger.info("start joining the cluster.");
+ logger.info("start registering to the cluster.");
try (ConfigNodeClient configNodeClient = new ConfigNodeClient()) {
IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
@@ -212,14 +216,14 @@ public class DataNode implements DataNodeMBean {
config.setDataNodeId(dataNodeID);
}
IoTDBDescriptor.getInstance().loadGlobalConfig(dataNodeRegisterResp.globalConfig);
- logger.info("Joined the cluster successfully");
+ logger.info("Register to the cluster successfully");
return;
}
} catch (IOException e) {
- logger.warn("Cannot join the cluster, because: {}", e.getMessage());
+ logger.warn("Cannot register to the cluster, because: {}",
e.getMessage());
} catch (TException e) {
// read config nodes from system.properties
- logger.warn("Cannot join the cluster, because: {}", e.getMessage());
+ logger.warn("Cannot register to the cluster, because: {}",
e.getMessage());
ConfigNodeInfo.getInstance().loadConfigNodeList();
}
@@ -228,7 +232,7 @@ public class DataNode implements DataNodeMBean {
Thread.sleep(IoTDBDescriptor.getInstance().getConfig().getJoinClusterTimeOutMs());
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
- logger.warn("Unexpected interruption when waiting to join the
cluster", e);
+ logger.warn("Unexpected interruption when waiting to register to the
cluster", e);
break;
}
@@ -236,8 +240,8 @@ public class DataNode implements DataNodeMBean {
retry--;
}
// all tries failed
- logger.error("Cannot join the cluster after {} retries",
DEFAULT_JOIN_RETRY);
- throw new StartupException("Cannot join the cluster.");
+ logger.error("Cannot register to the cluster after {} retries",
DEFAULT_JOIN_RETRY);
+ throw new StartupException("Cannot register to the cluster.");
}
/** register services and set up DataNode */
@@ -314,6 +318,57 @@ public class DataNode implements DataNodeMBean {
MetricsService.getInstance().startAllReporter();
}
+ /** send a message to ConfigNode after DataNode is available */
+ private void joinCluster() throws StartupException {
+ int retry = DEFAULT_JOIN_RETRY;
+
+ ConfigNodeInfo.getInstance()
+
.updateConfigNodeList(IoTDBDescriptor.getInstance().getConfig().getConfigNodeList());
+ while (retry > 0) {
+ logger.info("start joining the cluster.");
+ try (ConfigNodeClient configNodeClient = new ConfigNodeClient()) {
+ IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
+
+ // Set DataNodeLocation
+ TDataNodeLocation location = new TDataNodeLocation();
+ location.setDataNodeId(config.getDataNodeId());
+ location.setExternalEndPoint(new TEndPoint(config.getRpcAddress(),
config.getRpcPort()));
+ location.setInternalEndPoint(
+ new TEndPoint(config.getInternalIp(), config.getInternalPort()));
+ location.setDataBlockManagerEndPoint(
+ new TEndPoint(config.getInternalIp(),
config.getDataBlockManagerPort()));
+ location.setDataRegionConsensusEndPoint(
+ new TEndPoint(config.getInternalIp(),
config.getDataRegionConsensusPort()));
+ location.setSchemaRegionConsensusEndPoint(
+ new TEndPoint(config.getInternalIp(),
config.getSchemaRegionConsensusPort()));
+ TDataNodeActiveReq req = new TDataNodeActiveReq();
+ req.setLocation(location);
+ req.setDataNodeId(config.getDataNodeId());
+ TSStatus status = configNodeClient.activeDataNode(req);
+ if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ logger.info("Joined the cluster successfully");
+ return;
+ }
+ } catch (TException e) {
+ logger.warn("Cannot join the cluster, because: {}", e.getMessage());
+ }
+
+ try {
+ // wait 5s to start the next try
+
Thread.sleep(IoTDBDescriptor.getInstance().getConfig().getJoinClusterTimeOutMs());
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ logger.warn("Unexpected interruption when waiting to join the
cluster", e);
+ break;
+ }
+ // start the next try
+ retry--;
+ }
+ // all tries failed
+ logger.error("Cannot join the cluster after {} retries",
DEFAULT_JOIN_RETRY);
+ throw new StartupException("Cannot join the cluster.");
+ }
+
/** set up RPC and protocols after DataNode is available */
private void setUpRPCService() throws StartupException {
// init rpc service
diff --git a/thrift-confignode/src/main/thrift/confignode.thrift
b/thrift-confignode/src/main/thrift/confignode.thrift
index f1bd0e0818..05a1c548ea 100644
--- a/thrift-confignode/src/main/thrift/confignode.thrift
+++ b/thrift-confignode/src/main/thrift/confignode.thrift
@@ -29,6 +29,11 @@ struct TDataNodeRegisterReq {
2: optional map<string, TStorageGroupSchema> statusMap
}
+struct TDataNodeActiveReq {
+ 1: required common.TDataNodeLocation location
+ 2: required i32 dataNodeId
+}
+
struct TGlobalConfig {
1: required string dataRegionConsensusProtocolClass
2: required string schemaRegionConsensusProtocolClass
@@ -230,6 +235,8 @@ service ConfigIService {
TDataNodeRegisterResp registerDataNode(TDataNodeRegisterReq req)
+ common.TSStatus activeDataNode(TDataNodeActiveReq req)
+
TDataNodeInfoResp getDataNodeInfo(i32 dataNodeId)
/* Show Cluster */
