OneSizeFitsQuorum commented on code in PR #12246:
URL: https://github.com/apache/iotdb/pull/12246#discussion_r1544315602
##########
iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift:
##########
@@ -130,6 +130,16 @@ struct TDataNodeRestartResp {
1: required common.TSStatus status
2: required list<common.TConfigNodeLocation> configNodeList
3: optional TRuntimeConfiguration runtimeConfiguration
+ 4: optional list<common.TConsensusGroupId> consensusGroupIds
+}
+
+struct TDataNodeRemoveRegionReq {
Review Comment:
remove this？
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java:
##########
@@ -400,6 +402,12 @@ public TDataNodeRestartResp
restartDataNode(TDataNodeRestartReq req) throws TExc
() -> client.restartDataNode(req), resp ->
!updateConfigNodeLeader(resp.status));
}
+ @Override
Review Comment:
remove
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java:
##########
@@ -682,6 +684,30 @@ public TSStatus migrateRegion(TMigrateRegionReq
migrateRegionReq) {
return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
}
+ public TSStatus removeRegion(TDataNodeRemoveRegionReq req) {
Review Comment:
remove
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java:
##########
@@ -255,6 +257,16 @@ public TDataNodeRestartResp
restartDataNode(TDataNodeRestartReq req) {
return resp;
}
+ @Override
+ public TDataNodeRemoveRegionResp removeRegion(TDataNodeRemoveRegionReq req) {
Review Comment:
remove
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java:
##########
@@ -450,6 +458,100 @@ private void sendRegisterRequestToConfigNode() throws
StartupException, IOExcept
}
}
+ private List<ConsensusGroupId> getConsensusGroupId() {
+ List<ConsensusGroupId> consensusGroupIds = new ArrayList<>();
+ String dataRegionConsensusDir = config.getDataRegionConsensusDir();
+ try (DirectoryStream<Path> stream =
+ Files.newDirectoryStream(new File(dataRegionConsensusDir).toPath())) {
+ for (Path path : stream) {
+ String[] items = path.getFileName().toString().split("_");
+ ConsensusGroupId consensusGroupId =
+ ConsensusGroupId.Factory.create(Integer.parseInt(items[0]),
Integer.parseInt(items[1]));
+ consensusGroupIds.add(consensusGroupId);
+ }
+ } catch (IOException e) {
+ logger.error("Cannot get consensus group id from {}",
dataRegionConsensusDir, e);
+ }
+ return consensusGroupIds;
+ }
+
+ private void renameInvalidRegionDirs(List<ConsensusGroupId>
invalidConsensusGroupIds) {
Review Comment:
remove for current version, add todo for rename
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java:
##########
@@ -450,6 +458,100 @@ private void sendRegisterRequestToConfigNode() throws
StartupException, IOExcept
}
}
+ private List<ConsensusGroupId> getConsensusGroupId() {
+ List<ConsensusGroupId> consensusGroupIds = new ArrayList<>();
+ String dataRegionConsensusDir = config.getDataRegionConsensusDir();
+ try (DirectoryStream<Path> stream =
+ Files.newDirectoryStream(new File(dataRegionConsensusDir).toPath())) {
+ for (Path path : stream) {
+ String[] items = path.getFileName().toString().split("_");
+ ConsensusGroupId consensusGroupId =
+ ConsensusGroupId.Factory.create(Integer.parseInt(items[0]),
Integer.parseInt(items[1]));
+ consensusGroupIds.add(consensusGroupId);
+ }
+ } catch (IOException e) {
+ logger.error("Cannot get consensus group id from {}",
dataRegionConsensusDir, e);
+ }
+ return consensusGroupIds;
+ }
+
+ private void renameInvalidRegionDirs(List<ConsensusGroupId>
invalidConsensusGroupIds) {
+ for (ConsensusGroupId consensusGroupId : invalidConsensusGroupIds) {
+ File oldDir =
+ new File(
+ IoTConsensus.buildPeerDir(
+ new File(config.getDataRegionConsensusDir()),
consensusGroupId));
+ File newDir =
+ new File(
+ IoTConsensus.buildPeerDir(
+ new File(config.getInvalidDataRegionConsensusDir()),
consensusGroupId));
+ if (oldDir.exists() && !oldDir.renameTo(newDir)) {
+ logger.error("Cannot rename {} to {}", oldDir, newDir);
+ }
+ }
+ }
+
+ private void removeInvalidRegions(List<ConsensusGroupId>
dataNodeConsensusGroupIds)
+ throws StartupException {
+ List<ConsensusGroupId> invalidConsensusGroupIds =
+ getConsensusGroupId().stream()
+ .filter(consensusGroupId ->
!dataNodeConsensusGroupIds.contains(consensusGroupId))
+ .collect(Collectors.toList());
+ renameInvalidRegionDirs(invalidConsensusGroupIds);
+ logger.info("Sending remove invalid region request to
ConfigNode-leader...");
+ long startTime = System.currentTimeMillis();
+ /* Send remove region request */
+ int retry = DEFAULT_RETRY;
+ TDataNodeRemoveRegionReq req = new TDataNodeRemoveRegionReq();
+ req.setDataNodeId(config.getDataNodeId());
+ req.setConsensusGroupIds(
+ invalidConsensusGroupIds.stream()
+ .map(
+ consensusGroupId ->
+ new TConsensusGroupId(consensusGroupId.getType(),
consensusGroupId.getId()))
+ .collect(Collectors.toList()));
+ TDataNodeRemoveRegionResp dataNodeRemoveRegionResp = null;
+ while (retry > 0) {
+ try (ConfigNodeClient configNodeClient =
Review Comment:
remove line
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java:
##########
@@ -450,6 +458,100 @@ private void sendRegisterRequestToConfigNode() throws
StartupException, IOExcept
}
}
+ private List<ConsensusGroupId> getConsensusGroupId() {
Review Comment:
TODO
##########
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java:
##########
@@ -254,6 +256,14 @@ public interface IManager {
*/
TDataNodeRestartResp restartDataNode(TDataNodeRestartReq req);
+ /**
+ * Remove Invalid Region.
+ *
+ * @param req TDataNodeRemoveRegionReq
+ * @return {@link TSStatusCode#SUCCESS_STATUS} if submit
RemoveRegionPeerProcedure success.
+ */
+ TDataNodeRemoveRegionResp removeRegion(TDataNodeRemoveRegionReq req);
Review Comment:
remove
##########
iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java:
##########
@@ -631,9 +634,13 @@ public void recoverConfiguration() {
// recover from split configuration file
Path dirPath = Paths.get(storageDir);
List<Peer> tmpPeerList = getConfiguration(dirPath,
CONFIGURATION_TMP_FILE_NAME);
- tmpConfigurationUpdate(tmpPeerList);
+ configuration.addAll(tmpPeerList);
List<Peer> peerList = getConfiguration(dirPath,
CONFIGURATION_FILE_NAME);
- configuration.addAll(peerList);
+ for (Peer peer : peerList) {
Review Comment:
todo
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
