You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by GitBox <gi...@apache.org> on 2022/10/25 02:11:40 UTC
[GitHub] [iotdb] CRZbulabula commented on a diff in pull request #7620: [IOTDB-4644] Support modifying DataNode's IP:Port
CRZbulabula commented on code in PR #7620:
URL: https://github.com/apache/iotdb/pull/7620#discussion_r1003908169
##########
confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java:
##########
@@ -293,6 +294,35 @@ public DataSet removeDataNode(RemoveDataNodePlan removeDataNodePlan) {
return dataSet;
}
+ /**
+ * Update DataNode
+ *
+ * @param updateDataNodePlan UpdateDataNodePlan
+ * @return TSStatus. The TSStatus will be set to SUCCESS_STATUS when update success, and
+ * DATANODE_NOT_EXIST when some datanode is not exist, UPDATE_DATANODE_FAILED when update
+ * failed.
+ */
+ public TSStatus updateDataNode(UpdateDataNodePlan updateDataNodePlan) {
+ LOGGER.info("NodeManager start to update DataNode {}", updateDataNodePlan);
+
+ // check if node is already exist
+ boolean found = false;
+ List<TDataNodeConfiguration> configurationList = getRegisteredDataNodes();
+ for (TDataNodeConfiguration configuration : configurationList) {
+ if (configuration.getLocation().getDataNodeId()
+ == updateDataNodePlan.getDataNodeLocation().getDataNodeId()) {
+ found = true;
+ break;
+ }
+ }
+ if (found) {
+ getConsensusManager().write(updateDataNodePlan);
+ return nodeInfo.updateDataNode(updateDataNodePlan);
+ } else {
+ return new TSStatus(TSStatusCode.DATANODE_NOT_EXIST.getStatusCode());
Review Comment:
```suggestion
return new TSStatus(TSStatusCode.DATANODE_NOT_EXIST.getStatusCode()).setMessage("The specified DataNode(nodeId={}) doesn't exist", updateDataNodePlan.getDataNodeLocation().getDataNodeId());
```
##########
integration-test/src/test/java/org/apache/iotdb/confignode/it/IoTDBClusterRestartIT.java:
##########
@@ -83,5 +138,137 @@ public void clusterRestartTest() throws InterruptedException {
((AbstractEnv) EnvFactory.getEnv()).testWorking();
}
+ @Test
+ public void clusterRestartAfterUpdateDataNodeTest() throws InterruptedException {
+ TShowClusterResp clusterNodes;
+ final String sg = "root.sg";
+ final String sg0 = "root.sg0";
+ final String sg1 = "root.sg1";
+
+ final String d00 = sg0 + ".d0.s";
+ final String d01 = sg0 + ".d1.s";
+ final String d10 = sg1 + ".d0.s";
+ final String d11 = sg1 + ".d1.s";
+
+ final String allPaths = "root.**";
+ final String allSg0 = "root.sg0.**";
+ final String allSg1 = "root.sg1.**";
+ // Shutdown all data nodes
+ for (int i = 0; i < testDataNodeNum; i++) {
+ EnvFactory.getEnv().shutdownDataNode(i);
+ }
+
+ // Sleep 1s before restart
+ TimeUnit.SECONDS.sleep(1);
+
+ // Modify data node config
+ List<DataNodeWrapper> dataNodeWrapperList = EnvFactory.getEnv().getDataNodeWrapperList();
+ List<ConfigNodeWrapper> configNodeWrappersList = EnvFactory.getEnv().getConfigNodeWrapperList();
+ for (int i = 0; i < testDataNodeNum; i++) {
+ int[] portList = EnvUtils.searchAvailablePorts();
+ dataNodeWrapperList.get(i).setPort(portList[0]);
+ dataNodeWrapperList.get(i).setInternalPort(portList[1]);
+ dataNodeWrapperList.get(i).setMppDataExchangePort(portList[2]);
+
+ // update data node files'names
+ dataNodeWrapperList.get(i).renameFile();
+ }
+
+ for (int i = 0; i < testDataNodeNum; i++) {
+ dataNodeWrapperList.get(i).changeConfig(ConfigFactory.getConfig().getEngineProperties());
+ EnvFactory.getEnv().startDataNode(i);
+ }
+
+ ((AbstractEnv) EnvFactory.getEnv()).testWorking();
+
+ // check nodeInfo in cluster
+ try (SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
+ // check the number and status of nodes
+ clusterNodes = getClusterNodeInfos(client, testConfigNodeNum, testDataNodeNum);
+ assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), clusterNodes.getStatus().getCode());
+
+ // check the configuration of nodes
+ List<TConfigNodeLocation> configNodeLocationList = clusterNodes.getConfigNodeList();
+ List<TDataNodeLocation> dataNodeLocationList = clusterNodes.getDataNodeList();
+ checkNodeConfig(
+ configNodeLocationList,
+ dataNodeLocationList,
+ configNodeWrappersList,
+ dataNodeWrapperList);
+
+ // check whether the cluster is working by testing GetAndCreateSchemaPartition
+ TSStatus status;
+ ByteBuffer buffer;
+ TSchemaPartitionReq schemaPartitionReq;
+ TSchemaPartitionTableResp schemaPartitionTableResp;
+ Map<String, Map<TSeriesPartitionSlot, TConsensusGroupId>> schemaPartitionTable;
+
+ // Set StorageGroups
+ status = client.setStorageGroup(new TSetStorageGroupReq(new TStorageGroupSchema(sg0)));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+ status = client.setStorageGroup(new TSetStorageGroupReq(new TStorageGroupSchema(sg1)));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
+
+ // Test getSchemaPartition, the result should be empty
+ buffer = generatePatternTreeBuffer(new String[] {d00, d01, allSg1});
+ schemaPartitionReq = new TSchemaPartitionReq(buffer);
+ schemaPartitionTableResp = client.getSchemaPartitionTable(schemaPartitionReq);
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ schemaPartitionTableResp.getStatus().getCode());
+ Assert.assertEquals(0, schemaPartitionTableResp.getSchemaPartitionTableSize());
+
+ // Test getOrCreateSchemaPartition, ConfigNode should create SchemaPartitions and return
+ buffer = generatePatternTreeBuffer(new String[] {d00, d01, d10, d11});
+ schemaPartitionReq.setPathPatternTree(buffer);
+ schemaPartitionTableResp = client.getOrCreateSchemaPartitionTable(schemaPartitionReq);
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ schemaPartitionTableResp.getStatus().getCode());
+ Assert.assertEquals(2, schemaPartitionTableResp.getSchemaPartitionTableSize());
+ schemaPartitionTable = schemaPartitionTableResp.getSchemaPartitionTable();
+ for (int i = 0; i < 2; i++) {
+ Assert.assertTrue(schemaPartitionTable.containsKey(sg + i));
+ Assert.assertEquals(2, schemaPartitionTable.get(sg + i).size());
+ }
+
+ // Test getSchemaPartition, when a device path doesn't match any StorageGroup and including
+ // "**", ConfigNode will return all the SchemaPartitions
+ buffer = generatePatternTreeBuffer(new String[] {allPaths});
+ schemaPartitionReq.setPathPatternTree(buffer);
+ schemaPartitionTableResp = client.getSchemaPartitionTable(schemaPartitionReq);
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ schemaPartitionTableResp.getStatus().getCode());
+ Assert.assertEquals(2, schemaPartitionTableResp.getSchemaPartitionTableSize());
+ schemaPartitionTable = schemaPartitionTableResp.getSchemaPartitionTable();
+ for (int i = 0; i < 2; i++) {
+ Assert.assertTrue(schemaPartitionTable.containsKey(sg + i));
+ Assert.assertEquals(2, schemaPartitionTable.get(sg + i).size());
+ }
+
+ // Test getSchemaPartition, when a device path matches with a StorageGroup and end with "*",
+ // ConfigNode will return all the SchemaPartitions in this StorageGroup
+ buffer = generatePatternTreeBuffer(new String[] {allSg0, d11});
+ schemaPartitionReq.setPathPatternTree(buffer);
+ schemaPartitionTableResp = client.getSchemaPartitionTable(schemaPartitionReq);
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ schemaPartitionTableResp.getStatus().getCode());
+ Assert.assertEquals(2, schemaPartitionTableResp.getSchemaPartitionTableSize());
+ schemaPartitionTable = schemaPartitionTableResp.getSchemaPartitionTable();
+ // Check "root.sg0"
+ Assert.assertTrue(schemaPartitionTable.containsKey(sg0));
+ Assert.assertEquals(2, schemaPartitionTable.get(sg0).size());
+ // Check "root.sg1"
+ Assert.assertTrue(schemaPartitionTable.containsKey(sg1));
+ Assert.assertEquals(1, schemaPartitionTable.get(sg1).size());
+ } catch (Exception e) {
+ e.printStackTrace();
Review Comment:
Use LOG instead of printStackTrace()
##########
confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java:
##########
@@ -293,6 +294,35 @@ public DataSet removeDataNode(RemoveDataNodePlan removeDataNodePlan) {
return dataSet;
}
+ /**
+ * Update DataNode
Review Comment:
```suggestion
* Update the specified DataNode‘s location
```
##########
integration-test/src/test/java/org/apache/iotdb/confignode/it/IoTDBClusterRestartIT.java:
##########
@@ -83,5 +138,137 @@ public void clusterRestartTest() throws InterruptedException {
((AbstractEnv) EnvFactory.getEnv()).testWorking();
}
+ @Test
+ public void clusterRestartAfterUpdateDataNodeTest() throws InterruptedException {
+ TShowClusterResp clusterNodes;
+ final String sg = "root.sg";
+ final String sg0 = "root.sg0";
+ final String sg1 = "root.sg1";
+
+ final String d00 = sg0 + ".d0.s";
+ final String d01 = sg0 + ".d1.s";
+ final String d10 = sg1 + ".d0.s";
+ final String d11 = sg1 + ".d1.s";
+
+ final String allPaths = "root.**";
+ final String allSg0 = "root.sg0.**";
+ final String allSg1 = "root.sg1.**";
+ // Shutdown all data nodes
+ for (int i = 0; i < testDataNodeNum; i++) {
+ EnvFactory.getEnv().shutdownDataNode(i);
+ }
+
+ // Sleep 1s before restart
+ TimeUnit.SECONDS.sleep(1);
+
+ // Modify data node config
+ List<DataNodeWrapper> dataNodeWrapperList = EnvFactory.getEnv().getDataNodeWrapperList();
+ List<ConfigNodeWrapper> configNodeWrappersList = EnvFactory.getEnv().getConfigNodeWrapperList();
+ for (int i = 0; i < testDataNodeNum; i++) {
+ int[] portList = EnvUtils.searchAvailablePorts();
+ dataNodeWrapperList.get(i).setPort(portList[0]);
+ dataNodeWrapperList.get(i).setInternalPort(portList[1]);
+ dataNodeWrapperList.get(i).setMppDataExchangePort(portList[2]);
+
+ // update data node files'names
+ dataNodeWrapperList.get(i).renameFile();
+ }
+
+ for (int i = 0; i < testDataNodeNum; i++) {
+ dataNodeWrapperList.get(i).changeConfig(ConfigFactory.getConfig().getEngineProperties());
+ EnvFactory.getEnv().startDataNode(i);
+ }
+
+ ((AbstractEnv) EnvFactory.getEnv()).testWorking();
+
+ // check nodeInfo in cluster
+ try (SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
+ // check the number and status of nodes
+ clusterNodes = getClusterNodeInfos(client, testConfigNodeNum, testDataNodeNum);
+ assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), clusterNodes.getStatus().getCode());
+
+ // check the configuration of nodes
+ List<TConfigNodeLocation> configNodeLocationList = clusterNodes.getConfigNodeList();
+ List<TDataNodeLocation> dataNodeLocationList = clusterNodes.getDataNodeList();
+ checkNodeConfig(
+ configNodeLocationList,
+ dataNodeLocationList,
+ configNodeWrappersList,
+ dataNodeWrapperList);
+
+ // check whether the cluster is working by testing GetAndCreateSchemaPartition
Review Comment:
The following test cases can be simplified?
##########
thrift-confignode/src/main/thrift/confignode.thrift:
##########
@@ -551,6 +559,8 @@ service IConfigNodeRPCService {
*/
TDataNodeRemoveResp removeDataNode(TDataNodeRemoveReq req)
+ TDataNodeUpdateResp updateDataNode(TDataNodeUpdateReq req)
Review Comment:
The response of updateDataNode should also be TDataNodeRegisterResp. Because there are configuration parameters in TDataNodeRegisterResp.
##########
thrift-confignode/src/main/thrift/confignode.thrift:
##########
@@ -551,6 +559,8 @@ service IConfigNodeRPCService {
*/
TDataNodeRemoveResp removeDataNode(TDataNodeRemoveReq req)
+ TDataNodeUpdateResp updateDataNode(TDataNodeUpdateReq req)
Review Comment:
Add interface annotation
##########
integration-test/src/test/java/org/apache/iotdb/confignode/it/IoTDBClusterRestartIT.java:
##########
@@ -83,5 +138,137 @@ public void clusterRestartTest() throws InterruptedException {
((AbstractEnv) EnvFactory.getEnv()).testWorking();
}
+ @Test
+ public void clusterRestartAfterUpdateDataNodeTest() throws InterruptedException {
+ TShowClusterResp clusterNodes;
+ final String sg = "root.sg";
+ final String sg0 = "root.sg0";
+ final String sg1 = "root.sg1";
+
+ final String d00 = sg0 + ".d0.s";
+ final String d01 = sg0 + ".d1.s";
+ final String d10 = sg1 + ".d0.s";
+ final String d11 = sg1 + ".d1.s";
+
+ final String allPaths = "root.**";
+ final String allSg0 = "root.sg0.**";
+ final String allSg1 = "root.sg1.**";
+ // Shutdown all data nodes
+ for (int i = 0; i < testDataNodeNum; i++) {
+ EnvFactory.getEnv().shutdownDataNode(i);
+ }
+
+ // Sleep 1s before restart
+ TimeUnit.SECONDS.sleep(1);
+
+ // Modify data node config
+ List<DataNodeWrapper> dataNodeWrapperList = EnvFactory.getEnv().getDataNodeWrapperList();
+ List<ConfigNodeWrapper> configNodeWrappersList = EnvFactory.getEnv().getConfigNodeWrapperList();
+ for (int i = 0; i < testDataNodeNum; i++) {
+ int[] portList = EnvUtils.searchAvailablePorts();
+ dataNodeWrapperList.get(i).setPort(portList[0]);
+ dataNodeWrapperList.get(i).setInternalPort(portList[1]);
+ dataNodeWrapperList.get(i).setMppDataExchangePort(portList[2]);
+
+ // update data node files'names
+ dataNodeWrapperList.get(i).renameFile();
+ }
+
+ for (int i = 0; i < testDataNodeNum; i++) {
+ dataNodeWrapperList.get(i).changeConfig(ConfigFactory.getConfig().getEngineProperties());
+ EnvFactory.getEnv().startDataNode(i);
+ }
+
+ ((AbstractEnv) EnvFactory.getEnv()).testWorking();
+
+ // check nodeInfo in cluster
+ try (SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
+ // check the number and status of nodes
+ clusterNodes = getClusterNodeInfos(client, testConfigNodeNum, testDataNodeNum);
+ assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), clusterNodes.getStatus().getCode());
+
+ // check the configuration of nodes
+ List<TConfigNodeLocation> configNodeLocationList = clusterNodes.getConfigNodeList();
+ List<TDataNodeLocation> dataNodeLocationList = clusterNodes.getDataNodeList();
+ checkNodeConfig(
+ configNodeLocationList,
+ dataNodeLocationList,
+ configNodeWrappersList,
+ dataNodeWrapperList);
Review Comment:
We should also check the ip:port that shown by `show cluster`
##########
server/src/main/java/org/apache/iotdb/db/conf/IoTDBStartCheck.java:
##########
@@ -500,6 +517,37 @@ public void serializeGlobalConfig(TGlobalConfig globalConfig) throws IOException
FileUtils.moveFile(tmpPropertiesFile, propertiesFile);
}
+ public void serializeNewDataNode(TDataNodeLocation dataNodeLocation) throws IOException {
+ // create an empty tmpPropertiesFile
+ if (tmpPropertiesFile.createNewFile()) {
+ logger.info("Create system.properties.tmp {}.", tmpPropertiesFile);
+ } else {
+ logger.error("Create system.properties.tmp {} failed.", tmpPropertiesFile);
+ System.exit(-1);
+ }
+
+ reloadProperties();
+
+ try (FileOutputStream tmpFOS = new FileOutputStream(tmpPropertiesFile.toString())) {
+ properties.setProperty(
+ INTERNAL_PORT, String.valueOf(dataNodeLocation.getInternalEndPoint().getPort()));
+ properties.setProperty(
+ RPC_ADDRESS, String.valueOf(dataNodeLocation.getClientRpcEndPoint().getIp()));
+ properties.setProperty(
+ RPC_PORT, String.valueOf(dataNodeLocation.getClientRpcEndPoint().getPort()));
+ properties.setProperty(
+ MPP_DATA_EXCHANGE_PORT,
+ String.valueOf(dataNodeLocation.getMPPDataExchangeEndPoint().getPort()));
Review Comment:
We should also record the internal_address, schema_region_port and data_region_port
##########
server/src/main/java/org/apache/iotdb/db/service/DataNode.java:
##########
@@ -182,6 +191,50 @@ private void prepareDataNode() throws StartupException {
config.setClusterMode(true);
}
+ private void updateDataNodeLocation() throws StartupException {
Review Comment:
Is the updateDataNodeLocation interface can be merged into registerInConfigNode?
##########
confignode/src/main/java/org/apache/iotdb/confignode/persistence/node/NodeInfo.java:
##########
@@ -191,6 +192,27 @@ public TSStatus removeDataNode(RemoveDataNodePlan req) {
return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
}
+ /**
+ * Persist data node info
Review Comment:
```suggestion
* Update the specified DataNode‘s location
```
##########
confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java:
##########
@@ -293,6 +294,35 @@ public DataSet removeDataNode(RemoveDataNodePlan removeDataNodePlan) {
return dataSet;
}
+ /**
+ * Update DataNode
+ *
+ * @param updateDataNodePlan UpdateDataNodePlan
+ * @return TSStatus. The TSStatus will be set to SUCCESS_STATUS when update success, and
+ * DATANODE_NOT_EXIST when some datanode is not exist, UPDATE_DATANODE_FAILED when update
+ * failed.
+ */
+ public TSStatus updateDataNode(UpdateDataNodePlan updateDataNodePlan) {
+ LOGGER.info("NodeManager start to update DataNode {}", updateDataNodePlan);
+
+ // check if node is already exist
+ boolean found = false;
+ List<TDataNodeConfiguration> configurationList = getRegisteredDataNodes();
+ for (TDataNodeConfiguration configuration : configurationList) {
+ if (configuration.getLocation().getDataNodeId()
+ == updateDataNodePlan.getDataNodeLocation().getDataNodeId()) {
+ found = true;
+ break;
+ }
+ }
+ if (found) {
+ getConsensusManager().write(updateDataNodePlan);
+ return nodeInfo.updateDataNode(updateDataNodePlan);
Review Comment:
Why update twice?
##########
server/src/main/java/org/apache/iotdb/db/conf/IoTDBStartCheck.java:
##########
@@ -510,4 +558,20 @@ public boolean checkConsensusProtocolExists(TConsensusGroupType type) {
logger.error("Unexpected consensus group type");
return false;
}
+
+ public boolean isUpdate() {
+ if (!(properties.getProperty(INTERNAL_PORT).equals(internalPort))) {
+ return true;
+ }
+ if (!(properties.getProperty(RPC_ADDRESS).equals(rpcAddress))) {
+ return true;
+ }
+ if (!(properties.getProperty(RPC_PORT).equals(rpcPort))) {
+ return true;
+ }
+ if (!(properties.getProperty(MPP_DATA_EXCHANGE_PORT).equals(mppDataExchangePort))) {
+ return true;
+ }
+ return false;
Review Comment:
We should reject restart if the internal_address, schema_region_port or data_region_port is updated since we don't support them yet.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org