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/06/15 09:31:29 UTC
[GitHub] [iotdb] BanDuWuZi commented on a diff in pull request #6236: [IOTDB-3457] start new feature: data node shrink
BanDuWuZi commented on code in PR #6236:
URL: https://github.com/apache/iotdb/pull/6236#discussion_r897756613
##########
server/src/main/java/org/apache/iotdb/db/service/thrift/impl/InternalServiceImpl.java:
##########
@@ -450,5 +493,175 @@ public TSStatus dropFunction(TDropFunctionRequest request) {
}
}
+ @Override
+ public TSStatus migrateRegion(TMigrateRegionReq req) throws TException {
+ TConsensusGroupId tRgId = req.getRegionId();
+ ConsensusGroupId regionId = ConsensusGroupId.Factory.createFromTConsensusGroupId(tRgId);
+ TEndPoint oldNode = getConsensusEndPoint(req.getFromNode(), regionId);
+ TEndPoint newNode = getConsensusEndPoint(req.getToNode(), regionId);
+ LOGGER.debug("start to migrate a region: {}, from {} to {}", regionId, oldNode, newNode);
+
+ TSStatus status = addPeer(regionId, new Peer(regionId, newNode));
+ if (isFailed(status)) {
+ return status;
+ }
+
+ status = removePeer(regionId, new Peer(regionId, oldNode));
+ if (isFailed(status)) {
+ return status;
+ }
+
+ status = removeConsensusGroup(regionId);
+ if (isFailed(status)) {
+ return status;
+ }
+ status.setMessage("Migrate region " + regionId.getId() + "succeed!");
+ LOGGER.debug("succeed to migrate a region: {}, from {} to {}", regionId, oldNode, newNode);
+ return status;
+ }
+
+ private TEndPoint getConsensusEndPoint(
+ TDataNodeLocation nodeLocation, ConsensusGroupId regionId) {
+ if (regionId instanceof DataRegionId) {
+ return nodeLocation.getDataRegionConsensusEndPoint();
+ }
+ return nodeLocation.getSchemaRegionConsensusEndPoint();
+ }
+
+ private boolean isSucceed(TSStatus status) {
+ return status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
+
+ private boolean isFailed(TSStatus status) {
+ return !isSucceed(status);
+ }
+
+ private TSStatus removePeer(ConsensusGroupId regionId, Peer oldPeer) {
+ LOGGER.debug("start to remove peer {} for region {}", oldPeer, regionId);
+ TSStatus status = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ ConsensusGenericResponse resp;
+ if (regionId instanceof DataRegionId) {
+ resp = DataRegionConsensusImpl.getInstance().removePeer(regionId, oldPeer);
+ } else if (regionId instanceof SchemaRegionId) {
+ resp = SchemaRegionConsensusImpl.getInstance().removePeer(regionId, oldPeer);
+ } else {
+ status.setCode(TSStatusCode.REGION_MIGRATE_FAILED.getStatusCode());
+ status.setMessage("Error Region type. region: " + regionId);
+ return status;
+ }
+ if (!resp.isSuccess()) {
+ LOGGER.error(
+ "remove old peer {} for region {} error.", oldPeer, regionId, resp.getException());
+ status.setCode(TSStatusCode.REGION_MIGRATE_FAILED.getStatusCode());
+ status.setMessage(
+ "remove old peer: "
+ + oldPeer
+ + " failed. exception: \n"
+ + resp.getException().getMessage());
+ return status;
+ }
+ LOGGER.debug("succeed to remove peer {} for region {}", oldPeer, regionId);
+ status.setMessage("remove peer " + oldPeer + " for region " + regionId + " succeed");
+ return status;
+ }
+
+ private TSStatus addPeer(ConsensusGroupId regionId, Peer newPeer) {
+ LOGGER.debug("start to add peer {} for region {}", newPeer, regionId);
+ TSStatus status = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ ConsensusGenericResponse resp;
+ if (regionId instanceof DataRegionId) {
+ resp = DataRegionConsensusImpl.getInstance().addPeer(regionId, newPeer);
+ } else if (regionId instanceof SchemaRegionId) {
+ resp = SchemaRegionConsensusImpl.getInstance().addPeer(regionId, newPeer);
+ } else {
+ status.setCode(TSStatusCode.REGION_MIGRATE_FAILED.getStatusCode());
+ status.setMessage("Error Region type. region: " + regionId);
+ return status;
+ }
+ if (!resp.isSuccess()) {
+ LOGGER.error("add peer {} for region {} error.", newPeer, regionId, resp.getException());
+ status.setCode(TSStatusCode.REGION_MIGRATE_FAILED.getStatusCode());
+ status.setMessage(
+ "add peer: " + newPeer + " failed. exception: \n" + resp.getException().getMessage());
+ return status;
+ }
+ LOGGER.debug("succeed to add peer {} for region {}", newPeer, regionId);
+ status.setMessage("add peer " + newPeer + " for region " + regionId + " succeed");
+ return status;
+ }
+
+ private TSStatus removeConsensusGroup(ConsensusGroupId regionId) {
+ LOGGER.debug("start to remove region {} consensus group", regionId);
+ TSStatus status = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ ConsensusGenericResponse resp;
+ if (regionId instanceof DataRegionId) {
+ resp = DataRegionConsensusImpl.getInstance().removeConsensusGroup(regionId);
+ } else if (regionId instanceof SchemaRegionId) {
+ resp = SchemaRegionConsensusImpl.getInstance().removeConsensusGroup(regionId);
+ } else {
+ status.setCode(TSStatusCode.REGION_MIGRATE_FAILED.getStatusCode());
+ status.setMessage("Error Region type. region: " + regionId);
+ return status;
+ }
+ if (!resp.isSuccess()) {
+ LOGGER.error("remove region {} consensus group error", regionId, resp.getException());
+ status.setCode(TSStatusCode.REGION_MIGRATE_FAILED.getStatusCode());
+ status.setMessage(resp.getException().getMessage());
+ return status;
+ }
+ LOGGER.debug("succeed to remove region {} consensus group", regionId);
+ status.setMessage("remove region consensus group " + regionId + "succeed");
+ return status;
+ }
+
+ private TSStatus addConsensusGroup(ConsensusGroupId regionId, List<Peer> peers) {
+ LOGGER.debug("start to add peers {} to region {} consensus group", peers, regionId);
+ TSStatus status = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ ConsensusGenericResponse resp;
+ if (regionId instanceof DataRegionId) {
+ resp = DataRegionConsensusImpl.getInstance().addConsensusGroup(regionId, peers);
+ } else if (regionId instanceof SchemaRegionId) {
+ resp = SchemaRegionConsensusImpl.getInstance().addConsensusGroup(regionId, peers);
+ } else {
+ status.setCode(TSStatusCode.REGION_MIGRATE_FAILED.getStatusCode());
+ status.setMessage("Error Region type. region: " + regionId);
+ return status;
+ }
+ if (!resp.isSuccess()) {
+ LOGGER.error(
+ "add peers {} to region {} consensus group error", peers, regionId, resp.getException());
+ status.setCode(TSStatusCode.REGION_MIGRATE_FAILED.getStatusCode());
+ status.setMessage(resp.getException().getMessage());
+ return status;
+ }
+ LOGGER.debug("succeed to add peers {} to region {} consensus group", peers, regionId);
+ status.setMessage("add peers to region consensus group " + regionId + "succeed");
+ return status;
+ }
+
+ @Override
+ public TSStatus disableDataNode(TDisableDataNodeReq req) throws TException {
+ LOGGER.info("start disable data node in the request: {}", req);
+ TSStatus status = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ status.setMessage("disable datanode succeed");
+ // TODO QSW 清理该节点相关缓存等具体实现逻辑
Review Comment:
OK, fixed
--
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