Copilot commented on code in PR #18275:
URL: https://github.com/apache/iotdb/pull/18275#discussion_r3629128887
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeRegionManager.java:
##########
@@ -164,23 +226,122 @@ public TSStatus createDataRegion(TRegionReplicaSet
regionReplicaSet, String stor
peers.add(new Peer(dataRegionId, dataNodeLocation.getDataNodeId(),
endpoint));
}
DataRegionConsensusImpl.getInstance().createLocalPeer(dataRegionId,
peers);
+ dataRegionLockMap.putIfAbsent(dataRegionId, new
ReentrantReadWriteLock(false));
tsStatus = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
} catch (DataRegionException e) {
LOGGER.error(DataNodeMiscMessages.CREATE_DATA_REGION_FAILED,
storageGroup, e.getMessage());
tsStatus = new
TSStatus(TSStatusCode.CREATE_REGION_ERROR.getStatusCode());
tsStatus.setMessage(
String.format(DataNodeMiscMessages.CREATE_DATA_REGION_FAILED_FMT,
e.getMessage()));
} catch (ConsensusGroupAlreadyExistException e) {
+ dataRegionLockMap.putIfAbsent(dataRegionId, new
ReentrantReadWriteLock(false));
tsStatus = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
tsStatus.setMessage(
String.format(DataNodeMiscMessages.DATA_REGION_ALREADY_EXISTS_FMT,
dataRegionId.getId()));
} catch (ConsensusException e) {
+ rollbackDataRegionCreation(
+ dataRegionId, localRegionCreated, consensusGroupExisted,
storageGroup);
tsStatus = new
TSStatus(TSStatusCode.CREATE_REGION_ERROR.getStatusCode());
tsStatus.setMessage(e.getMessage());
+ } catch (RuntimeException | OutOfMemoryError e) {
+ rollbackDataRegionCreation(
+ dataRegionId,
+ localRegionCreated
+ || (!localRegionExisted &&
storageEngine.getDataRegion(dataRegionId) != null),
+ consensusGroupExisted,
+ storageGroup);
+ LOGGER.error(DataNodeMiscMessages.CREATE_DATA_REGION_FAILED,
storageGroup, e.getMessage());
+ tsStatus = new
TSStatus(TSStatusCode.CREATE_REGION_ERROR.getStatusCode());
+ tsStatus.setMessage(
+ String.format(DataNodeMiscMessages.CREATE_DATA_REGION_FAILED_FMT,
e.getMessage()));
+ } finally {
+ creationLock.unlock();
}
return tsStatus;
}
+ private ReentrantLock getRegionCreationLock(ConsensusGroupId regionId) {
+ return regionCreationLocks[
+ (regionId.hashCode() & Integer.MAX_VALUE) %
REGION_CREATION_LOCK_COUNT];
+ }
+
+ /**
+ * Fences delayed create RPCs after a whole RegionGroup starts deletion.
+ *
+ * <p>RegionGroup ids are never reused. The caller acquires the same striped
lock as creation, so
+ * an already-running creation finishes before deletion starts, while every
later creation is
+ * rejected even if it came from an old ConfigNode leader.
+ */
+ public void markRegionGroupDeleted(ConsensusGroupId regionId) {
+ final ReentrantLock creationLock = getRegionCreationLock(regionId);
+ creationLock.lock();
+ try {
+ synchronized (deletedRegionGroupFenceLock) {
+ getDeletedRegionGroupSet(regionId).set(regionId.getId());
+ }
+ } finally {
+ creationLock.unlock();
+ }
+ }
+
+ private boolean isRegionGroupDeleted(ConsensusGroupId regionId) {
+ synchronized (deletedRegionGroupFenceLock) {
+ return getDeletedRegionGroupSet(regionId).get(regionId.getId());
+ }
+ }
+
+ private BitSet getDeletedRegionGroupSet(ConsensusGroupId regionId) {
+ return regionId instanceof DataRegionId ? deletedDataRegionGroups :
deletedSchemaRegionGroups;
+ }
+
+ private void rollbackDataRegionCreation(
+ DataRegionId regionId,
+ boolean localRegionCreated,
+ boolean consensusGroupExisted,
+ String storageGroup) {
+ rollbackConsensusPeer(
+ DataRegionConsensusImpl.getInstance(), regionId,
consensusGroupExisted, storageGroup);
+ if (localRegionCreated && !consensusGroupExisted) {
+ storageEngine.deleteDataRegion(regionId);
+ dataRegionLockMap.remove(regionId);
+ }
+ }
+
+ private void rollbackSchemaRegionCreation(
+ SchemaRegionId regionId,
+ boolean localRegionCreated,
+ boolean consensusGroupExisted,
+ String storageGroup) {
+ rollbackConsensusPeer(
+ SchemaRegionConsensusImpl.getInstance(), regionId,
consensusGroupExisted, storageGroup);
+ if (localRegionCreated && !consensusGroupExisted) {
+ try {
+ schemaEngine.deleteSchemaRegion(regionId);
+ schemaRegionLockMap.remove(regionId);
+ } catch (MetadataException e) {
+ LOGGER.error(
+ DataNodeMiscMessages.CREATE_SCHEMA_REGION_FAILED, storageGroup,
e.getMessage());
+ }
+ }
+ }
+
+ private void rollbackConsensusPeer(
+ org.apache.iotdb.consensus.IConsensus consensus,
+ ConsensusGroupId regionId,
+ boolean consensusGroupExisted,
+ String storageGroup) {
+ if (consensusGroupExisted) {
+ return;
+ }
+ try {
+ if (consensus.getAllConsensusGroupIds().contains(regionId)) {
+ consensus.deleteLocalPeer(regionId);
+ }
+ } catch (ConsensusException | RuntimeException | OutOfMemoryError e) {
+ LOGGER.error(DataNodeMiscMessages.CREATE_DATA_REGION_FAILED,
storageGroup, e.getMessage());
+ }
Review Comment:
rollbackConsensusPeer logs DataNodeMiscMessages.CREATE_DATA_REGION_FAILED
even when rolling back a SchemaRegion consensus peer, which can mislead
operators during SchemaRegion creation failures. Use the schema-specific
message when regionId is a SchemaRegionId (or choose based on regionId type).
--
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]
