diff --git a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java index 494201b1e9fec..e159d65c62fd5 100644 --- a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java +++ b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java @@ -665,4 +665,22 @@ private ConfigNodeMessages() {} public static final String EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_MODE_CONSENSUS_DOES_NOT_SUPPORT_TOPIC_ATTRIBUTES_ARG_3C2D0BDA = "Failed to create or alter topic, mode=consensus does not support topic attributes %s"; + public static final String + LOG_REJECT_CREATEREGIONGROUPSPLAN_BECAUSE_DATABASE_ARG_DOES_NOT_EXIST_616E0CDE = + "Reject CreateRegionGroupsPlan because database {} does not exist"; + public static final String + LOG_REJECT_CREATEREGIONGROUPSPLAN_BECAUSE_DATABASE_ARG_IS_BEING_DELETED_C085AC01 = + "Reject CreateRegionGroupsPlan because database {} is being deleted"; + public static final String + LOG_REJECT_CREATEREGIONGROUPSPLAN_BECAUSE_DATABASE_ARG_LIFECYCLE_GENERATION_CHANGED_FROM_ARG_TO_ARG_4306DEC3 = + "Reject CreateRegionGroupsPlan because database {} lifecycle generation changed from {} to {}"; + public static final String + MESSAGE_CREATE_REGIONGROUPS_FAILED_BECAUSE_DATABASE_ARG_DOES_NOT_EXIST_AF0F2440 = + "Create RegionGroups failed because database %s does not exist"; + public static final String + MESSAGE_CREATE_REGIONGROUPS_FAILED_BECAUSE_DATABASE_ARG_IS_BEING_DELETED_651DB780 = + "Create RegionGroups failed because database %s is being deleted"; + public static final String + MESSAGE_CREATE_REGIONGROUPS_FAILED_BECAUSE_DATABASE_ARG_LIFECYCLE_GENERATION_CHANGED_FROM_ARG_TO_ARG_CCDAF444 = + "Create RegionGroups failed because database %s lifecycle generation changed from %d to %d"; } diff --git a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java index 0193e5a04854e..d2a269ba97270 100644 --- a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java +++ b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java @@ -710,4 +710,22 @@ private ConfigNodeMessages() {} public static final String EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_MODE_CONSENSUS_DOES_NOT_SUPPORT_TOPIC_ATTRIBUTES_ARG_3C2D0BDA = "创建或修改 topic 失败，mode=consensus 不支持 topic 属性 %s"; + public static final String + LOG_REJECT_CREATEREGIONGROUPSPLAN_BECAUSE_DATABASE_ARG_DOES_NOT_EXIST_616E0CDE = + "拒绝 CreateRegionGroupsPlan，因为数据库 {} 不存在"; + public static final String + LOG_REJECT_CREATEREGIONGROUPSPLAN_BECAUSE_DATABASE_ARG_IS_BEING_DELETED_C085AC01 = + "拒绝 CreateRegionGroupsPlan，因为数据库 {} 正在删除"; + public static final String + LOG_REJECT_CREATEREGIONGROUPSPLAN_BECAUSE_DATABASE_ARG_LIFECYCLE_GENERATION_CHANGED_FROM_ARG_TO_ARG_4306DEC3 = + "拒绝 CreateRegionGroupsPlan，因为数据库 {} 的生命周期代次已从 {} 变为 {}"; + public static final String + MESSAGE_CREATE_REGIONGROUPS_FAILED_BECAUSE_DATABASE_ARG_DOES_NOT_EXIST_AF0F2440 = + "创建 RegionGroups 失败，因为数据库 %s 不存在"; + public static final String + MESSAGE_CREATE_REGIONGROUPS_FAILED_BECAUSE_DATABASE_ARG_IS_BEING_DELETED_651DB780 = + "创建 RegionGroups 失败，因为数据库 %s 正在删除"; + public static final String + MESSAGE_CREATE_REGIONGROUPS_FAILED_BECAUSE_DATABASE_ARG_LIFECYCLE_GENERATION_CHANGED_FROM_ARG_TO_ARG_CCDAF444 = + "创建 RegionGroups 失败，因为数据库 %s 的生命周期代次已从 %d 变为 %d"; } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/region/CreateRegionGroupsPlan.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/region/CreateRegionGroupsPlan.java index 83b774a543f9a..24692ddd74743 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/region/CreateRegionGroupsPlan.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/region/CreateRegionGroupsPlan.java @@ -43,23 +43,47 @@ /** Create regions for specified Databases. */ public class CreateRegionGroupsPlan extends ConfigPhysicalPlan { + public static final long DATABASE_GENERATION_NOT_SET = -1; + // Map> protected final Map> regionGroupMap; + // Map. It fences a RegionGroup allocation from a later database + // that reuses the same name. + protected final Map databaseGenerationMap; + public CreateRegionGroupsPlan() { super(ConfigPhysicalPlanType.CreateRegionGroups); this.regionGroupMap = new HashMap<>(); + this.databaseGenerationMap = new HashMap<>(); } public CreateRegionGroupsPlan(final ConfigPhysicalPlanType type) { super(type); this.regionGroupMap = new HashMap<>(); + this.databaseGenerationMap = new HashMap<>(); } public Map> getRegionGroupMap() { return regionGroupMap; } + public Map getDatabaseGenerationMap() { + return databaseGenerationMap; + } + + public long getDatabaseGeneration(final String database) { + return databaseGenerationMap.getOrDefault(database, DATABASE_GENERATION_NOT_SET); + } + + public boolean isDatabaseGenerationSet(final String database) { + return databaseGenerationMap.containsKey(database); + } + + public void setDatabaseGeneration(final String database, final long databaseGeneration) { + databaseGenerationMap.put(database, databaseGeneration); + } + public void addRegionGroup(final String database, final TRegionReplicaSet regionReplicaSet) { regionGroupMap .computeIfAbsent(database, regionReplicaSets -> new ArrayList<>()) @@ -84,17 +108,22 @@ public void planLog(final Logger logger) { } public void serializeForProcedure(final DataOutputStream stream) throws IOException { - this.serializeImpl(stream); + serializeRegionGroupMap(stream); } public void deserializeForProcedure(final ByteBuffer buffer) throws IOException { // to remove the planType of ConfigPhysicalPlanType buffer.getShort(); - this.deserializeImpl(buffer); + deserializeRegionGroupMap(buffer); } @Override protected void serializeImpl(final DataOutputStream stream) throws IOException { + serializeRegionGroupMap(stream); + serializeDatabaseGenerationMap(stream); + } + + private void serializeRegionGroupMap(final DataOutputStream stream) throws IOException { stream.writeShort(getType().getPlanType()); stream.writeInt(regionGroupMap.size()); @@ -111,6 +140,13 @@ protected void serializeImpl(final DataOutputStream stream) throws IOException { @Override protected void deserializeImpl(final ByteBuffer buffer) throws IOException { + deserializeRegionGroupMap(buffer); + if (buffer.hasRemaining()) { + deserializeDatabaseGenerationMap(buffer); + } + } + + private void deserializeRegionGroupMap(final ByteBuffer buffer) throws IOException { final int databaseNum = buffer.getInt(); for (int i = 0; i < databaseNum; i++) { final String database = BasicStructureSerDeUtil.readString(buffer); @@ -125,6 +161,21 @@ protected void deserializeImpl(final ByteBuffer buffer) throws IOException { } } + public void serializeDatabaseGenerationMap(final DataOutputStream stream) throws IOException { + stream.writeInt(databaseGenerationMap.size()); + for (final Entry entry : databaseGenerationMap.entrySet()) { + BasicStructureSerDeUtil.write(entry.getKey(), stream); + stream.writeLong(entry.getValue()); + } + } + + public void deserializeDatabaseGenerationMap(final ByteBuffer buffer) { + final int databaseNum = buffer.getInt(); + for (int i = 0; i < databaseNum; i++) { + databaseGenerationMap.put(BasicStructureSerDeUtil.readString(buffer), buffer.getLong()); + } + } + @Override public boolean equals(final Object o) { if (this == o) { @@ -137,11 +188,12 @@ public boolean equals(final Object o) { return false; } final CreateRegionGroupsPlan that = (CreateRegionGroupsPlan) o; - return Objects.equals(regionGroupMap, that.regionGroupMap); + return Objects.equals(regionGroupMap, that.regionGroupMap) + && Objects.equals(databaseGenerationMap, that.databaseGenerationMap); } @Override public int hashCode() { - return Objects.hash(super.hashCode(), regionGroupMap); + return Objects.hash(super.hashCode(), regionGroupMap, databaseGenerationMap); } } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RegionBalancer.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RegionBalancer.java index 73583151f9819..b5e97f6fcae64 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RegionBalancer.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RegionBalancer.java @@ -116,6 +116,8 @@ public CreateRegionGroupsPlan genRegionGroupsAllocationPlan( for (final Map.Entry entry : allotmentMap.entrySet()) { final String database = entry.getKey(); final int allotment = entry.getValue(); + createRegionGroupsPlan.setDatabaseGeneration( + database, getPartitionManager().getDatabaseGeneration(database)); final int replicationFactor = getClusterSchemaManager().getReplicationFactor(database, consensusGroupType); // Only considering the specified Database when doing allocation diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java index 7efdae278db3c..f9b1961504d64 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java @@ -1132,6 +1132,14 @@ public boolean isDatabasePreDeleted(final String database) { return partitionInfo.isDatabasePreDeleted(database); } + public long getDatabaseGeneration(final String database) { + return partitionInfo.getDatabaseGeneration(database); + } + + public TSStatus validateCreateRegionGroups(final CreateRegionGroupsPlan plan) { + return partitionInfo.validateCreateRegionGroups(plan); + } + /** * Get TSeriesPartitionSlot. * diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/DatabasePartitionTable.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/DatabasePartitionTable.java index 1ed35210afcd2..c2284714b7ccb 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/DatabasePartitionTable.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/DatabasePartitionTable.java @@ -29,6 +29,7 @@ import org.apache.iotdb.commons.partition.SchemaPartitionTable; import org.apache.iotdb.commons.utils.CommonDateTimeUtils; import org.apache.iotdb.confignode.consensus.request.read.region.GetRegionInfoListPlan; +import org.apache.iotdb.confignode.consensus.request.write.region.CreateRegionGroupsPlan; import org.apache.iotdb.confignode.i18n.ConfigNodeMessages; import org.apache.iotdb.confignode.rpc.thrift.TRegionInfo; import org.apache.iotdb.confignode.rpc.thrift.TShowRegionReq; @@ -66,6 +67,8 @@ public class DatabasePartitionTable { private volatile boolean preDeleted = false; // The name of database private String databaseName; + // The incarnation of databaseName. A new value is assigned whenever the name is recreated. + private final long databaseGeneration; // RegionGroup private final Map regionGroupMap; @@ -75,7 +78,12 @@ public class DatabasePartitionTable { private final DataPartitionTable dataPartitionTable; public DatabasePartitionTable(String databaseName) { + this(databaseName, CreateRegionGroupsPlan.DATABASE_GENERATION_NOT_SET); + } + + public DatabasePartitionTable(String databaseName, long databaseGeneration) { this.databaseName = databaseName; + this.databaseGeneration = databaseGeneration; this.regionGroupMap = new ConcurrentHashMap<>(); @@ -91,6 +99,10 @@ public void setPreDeleted(boolean preDeleted) { this.preDeleted = preDeleted; } + public long getDatabaseGeneration() { + return databaseGeneration; + } + /** * Update the DataNodeLocation in cached RegionGroups. * @@ -655,7 +667,8 @@ public boolean equals(Object o) { return false; } DatabasePartitionTable that = (DatabasePartitionTable) o; - return databaseName.equals(that.databaseName) + return databaseGeneration == that.databaseGeneration + && databaseName.equals(that.databaseName) && regionGroupMap.equals(that.regionGroupMap) && schemaPartitionTable.equals(that.schemaPartitionTable) && dataPartitionTable.equals(that.dataPartitionTable); @@ -663,6 +676,7 @@ public boolean equals(Object o) { @Override public int hashCode() { - return Objects.hash(databaseName, regionGroupMap, schemaPartitionTable, dataPartitionTable); + return Objects.hash( + databaseName, databaseGeneration, regionGroupMap, schemaPartitionTable, dataPartitionTable); } } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/PartitionInfo.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/PartitionInfo.java index e71b61d7b338c..31c761c306b26 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/PartitionInfo.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/PartitionInfo.java @@ -103,6 +103,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; import java.util.stream.Collectors; /** @@ -123,10 +124,16 @@ public class PartitionInfo implements SnapshotProcessor { // Allocate 8MB buffer for load snapshot of PartitionInfo private static final int PARTITION_TABLE_BUFFER_SIZE = 32 * 1024 * 1024; + // A negative value cannot collide with nextRegionGroupId, whose only negative value is -1. + private static final int SNAPSHOT_WITH_DATABASE_GENERATION_MAGIC = -20260721; + /** For Cluster Partition. */ // For allocating Regions private final AtomicInteger nextRegionGroupId; + // Monotonically identifies different incarnations that reuse the same database name. + private final AtomicLong nextDatabaseGeneration; + // Map // For tree model databases: The databaseName is a partial path's full path with "root." // For table model databases: The databaseName is a full name without "root." @@ -140,6 +147,7 @@ public class PartitionInfo implements SnapshotProcessor { public PartitionInfo() { this.nextRegionGroupId = new AtomicInteger(-1); + this.nextDatabaseGeneration = new AtomicLong(0); this.databasePartitionTables = new ConcurrentHashMap<>(); this.regionMaintainTaskList = Collections.synchronizedList(new ArrayList<>()); @@ -180,7 +188,8 @@ public TSStatus updateDataNode(UpdateDataNodePlan updateDataNodePlan) { */ public TSStatus createDatabase(final DatabaseSchemaPlan plan) { final String databaseName = plan.getSchema().getName(); - final DatabasePartitionTable databasePartitionTable = new DatabasePartitionTable(databaseName); + final DatabasePartitionTable databasePartitionTable = + new DatabasePartitionTable(databaseName, nextDatabaseGeneration.incrementAndGet()); databasePartitionTables.put(databaseName, databasePartitionTable); return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()); } @@ -192,37 +201,97 @@ public TSStatus createDatabase(final DatabaseSchemaPlan plan) { * @return {@link TSStatusCode#SUCCESS_STATUS} */ public TSStatus createRegionGroups(CreateRegionGroupsPlan plan) { - TSStatus result; - AtomicInteger maxRegionId = new AtomicInteger(Integer.MIN_VALUE); + updateNextRegionGroupId(plan); + + final TSStatus validationStatus = validateCreateRegionGroups(plan); + if (validationStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + return validationStatus; + } plan.getRegionGroupMap() .forEach( (database, regionReplicaSets) -> { - if (isDatabasePreDeleted(database)) { - LOGGER.warn( - ConfigNodeMessages - .CREATEREGIONGROUPS_DATABASE_HAS_BEEN_DELETED_CORRESPONDING_REGIONGROUPS, - database); - return; - } databasePartitionTables.get(database).createRegionGroups(regionReplicaSets); - regionReplicaSets.forEach( - regionReplicaSet -> - maxRegionId.set( - Math.max(maxRegionId.get(), regionReplicaSet.getRegionId().getId()))); }); + return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()); + } + + /** Validates all databases before any RegionGroup in a potentially batched plan is persisted. */ + public TSStatus validateCreateRegionGroups(final CreateRegionGroupsPlan plan) { + for (final String database : plan.getRegionGroupMap().keySet()) { + final DatabasePartitionTable databasePartitionTable = databasePartitionTables.get(database); + if (databasePartitionTable == null) { + LOGGER.warn( + ConfigNodeMessages + .LOG_REJECT_CREATEREGIONGROUPSPLAN_BECAUSE_DATABASE_ARG_DOES_NOT_EXIST_616E0CDE, + database); + return new TSStatus(TSStatusCode.DATABASE_NOT_EXIST.getStatusCode()) + .setMessage( + String.format( + ConfigNodeMessages + .MESSAGE_CREATE_REGIONGROUPS_FAILED_BECAUSE_DATABASE_ARG_DOES_NOT_EXIST_AF0F2440, + database)); + } + if (!databasePartitionTable.isNotPreDeleted()) { + LOGGER.warn( + ConfigNodeMessages + .LOG_REJECT_CREATEREGIONGROUPSPLAN_BECAUSE_DATABASE_ARG_IS_BEING_DELETED_C085AC01, + database); + return new TSStatus(TSStatusCode.DATABASE_NOT_EXIST.getStatusCode()) + .setMessage( + String.format( + ConfigNodeMessages + .MESSAGE_CREATE_REGIONGROUPS_FAILED_BECAUSE_DATABASE_ARG_IS_BEING_DELETED_651DB780, + database)); + } + + final long expectedGeneration = plan.getDatabaseGeneration(database); + final long currentGeneration = databasePartitionTable.getDatabaseGeneration(); + if (plan.isDatabaseGenerationSet(database) && expectedGeneration != currentGeneration) { + LOGGER.warn( + ConfigNodeMessages + .LOG_REJECT_CREATEREGIONGROUPSPLAN_BECAUSE_DATABASE_ARG_LIFECYCLE_GENERATION_CHANGED_FROM_ARG_TO_ARG_4306DEC3, + database, + expectedGeneration, + currentGeneration); + return new TSStatus(TSStatusCode.DATABASE_CONFIG_ERROR.getStatusCode()) + .setMessage( + String.format( + ConfigNodeMessages + .MESSAGE_CREATE_REGIONGROUPS_FAILED_BECAUSE_DATABASE_ARG_LIFECYCLE_GENERATION_CHANGED_FROM_ARG_TO_ARG_CCDAF444, + database, + expectedGeneration, + currentGeneration)); + } + } + + return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()); + } + + private void updateNextRegionGroupId(final CreateRegionGroupsPlan plan) { + final int maxRegionId = + plan.getRegionGroupMap().values().stream() + .flatMap(List::stream) + .mapToInt(regionReplicaSet -> regionReplicaSet.getRegionId().getId()) + .max() + .orElse(Integer.MIN_VALUE); + // To ensure that the nextRegionGroupId is updated correctly when // the ConfigNode-followers concurrently processes CreateRegionsPlan, // we need to add a synchronization lock here synchronized (nextRegionGroupId) { - if (nextRegionGroupId.get() < maxRegionId.get()) { - nextRegionGroupId.set(maxRegionId.get()); + if (nextRegionGroupId.get() < maxRegionId) { + nextRegionGroupId.set(maxRegionId); } } + } - result = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()); - return result; + public long getDatabaseGeneration(final String database) { + final DatabasePartitionTable databasePartitionTable = databasePartitionTables.get(database); + return databasePartitionTable == null + ? CreateRegionGroupsPlan.DATABASE_GENERATION_NOT_SET + : databasePartitionTable.getDatabaseGeneration(); } /** @@ -1014,13 +1083,17 @@ public boolean processTakeSnapshot(File snapshotDir) throws TException, IOExcept TProtocol protocol = new TBinaryProtocol(tioStreamTransport); // serialize nextRegionGroupId + ReadWriteIOUtils.write(SNAPSHOT_WITH_DATABASE_GENERATION_MAGIC, bufferedOutputStream); ReadWriteIOUtils.write(nextRegionGroupId.get(), bufferedOutputStream); + ReadWriteIOUtils.write(nextDatabaseGeneration.get(), bufferedOutputStream); // serialize databasePartitionTable ReadWriteIOUtils.write(databasePartitionTables.size(), bufferedOutputStream); for (Map.Entry databasePartitionTableEntry : databasePartitionTables.entrySet()) { ReadWriteIOUtils.write(databasePartitionTableEntry.getKey(), bufferedOutputStream); + ReadWriteIOUtils.write( + databasePartitionTableEntry.getValue().getDatabaseGeneration(), bufferedOutputStream); databasePartitionTableEntry.getValue().serialize(bufferedOutputStream, protocol); } @@ -1072,7 +1145,15 @@ public void processLoadSnapshot(final File snapshotDir) throws TException, IOExc clear(); // start to restore - nextRegionGroupId.set(ReadWriteIOUtils.readInt(fileInputStream)); + final int firstSnapshotValue = ReadWriteIOUtils.readInt(fileInputStream); + final boolean hasDatabaseGeneration = + firstSnapshotValue == SNAPSHOT_WITH_DATABASE_GENERATION_MAGIC; + if (hasDatabaseGeneration) { + nextRegionGroupId.set(ReadWriteIOUtils.readInt(fileInputStream)); + nextDatabaseGeneration.set(ReadWriteIOUtils.readLong(fileInputStream)); + } else { + nextRegionGroupId.set(firstSnapshotValue); + } // restore databasePartitionTable int length = ReadWriteIOUtils.readInt(fileInputStream); @@ -1082,7 +1163,12 @@ public void processLoadSnapshot(final File snapshotDir) throws TException, IOExc throw new IOException( ConfigNodeMessages.FAILED_TO_LOAD_SNAPSHOT_BECAUSE_GET_NULL_DATABASE_NAME); } - final DatabasePartitionTable databasePartitionTable = new DatabasePartitionTable(database); + final long databaseGeneration = + hasDatabaseGeneration + ? ReadWriteIOUtils.readLong(fileInputStream) + : CreateRegionGroupsPlan.DATABASE_GENERATION_NOT_SET; + final DatabasePartitionTable databasePartitionTable = + new DatabasePartitionTable(database, databaseGeneration); databasePartitionTable.deserialize(fileInputStream, protocol); databasePartitionTables.put(database, databasePartitionTable); } @@ -1266,6 +1352,7 @@ public Optional getRegionType(int regionId) { public void clear() { nextRegionGroupId.set(-1); + nextDatabaseGeneration.set(0); databasePartitionTables.clear(); regionMaintainTaskList.clear(); } @@ -1280,12 +1367,14 @@ public boolean equals(Object o) { } PartitionInfo that = (PartitionInfo) o; return nextRegionGroupId.get() == that.nextRegionGroupId.get() + && nextDatabaseGeneration.get() == that.nextDatabaseGeneration.get() && databasePartitionTables.equals(that.databasePartitionTables) && regionMaintainTaskList.equals(that.regionMaintainTaskList); } @Override public int hashCode() { - return Objects.hash(nextRegionGroupId, databasePartitionTables, regionMaintainTaskList); + return Objects.hash( + nextRegionGroupId, nextDatabaseGeneration, databasePartitionTables, regionMaintainTaskList); } } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java index 76bf575681c87..9fbce2ce0e5f3 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java @@ -53,6 +53,7 @@ import org.apache.iotdb.confignode.manager.schema.ClusterSchemaManager; import org.apache.iotdb.confignode.persistence.partition.PartitionInfo; import org.apache.iotdb.confignode.persistence.schema.ClusterSchemaInfo; +import org.apache.iotdb.confignode.procedure.Procedure; import org.apache.iotdb.confignode.procedure.exception.ProcedureException; import org.apache.iotdb.confignode.procedure.impl.schema.SchemaUtils; import org.apache.iotdb.confignode.procedure.scheduler.LockQueue; @@ -102,6 +103,7 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.TreeSet; import java.util.concurrent.TimeUnit; import java.util.concurrent.locks.ReentrantLock; import java.util.stream.Collectors; @@ -115,6 +117,9 @@ public class ConfigNodeProcedureEnv { /** Add or remove node lock. */ private final LockQueue nodeLock = new LockQueue(); + /** Serializes procedures that mutate the lifecycle of the same database. */ + private final Map databaseLockMap = new HashMap<>(); + private final ReentrantLock schedulerLock = new ReentrantLock(true); private final ReentrantLock submitRegionMigrateLock = new ReentrantLock(true); @@ -490,6 +495,10 @@ public TSStatus persistRegionGroup(CreateRegionGroupsPlan createRegionGroupsPlan } } + public TSStatus validateCreateRegionGroups(final CreateRegionGroupsPlan createRegionGroupsPlan) { + return getPartitionManager().validateCreateRegionGroups(createRegionGroupsPlan); + } + /** * Force activating RegionGroup by setting status to Running, therefore the ConfigNode-leader can * select leader for it and use it to allocate new Partitions @@ -1135,6 +1144,67 @@ public LockQueue getNodeLock() { return nodeLock; } + /** + * Atomically tries to lock all databases in lexical order. + * + * @return the first database whose lock is unavailable, or null when all locks are acquired + */ + public String tryLockDatabases(final Procedure procedure, final Set databaseNames) { + schedulerLock.lock(); + try { + final List acquiredDatabases = new ArrayList<>(); + for (final String database : new TreeSet<>(databaseNames)) { + final LockQueue lockQueue = + databaseLockMap.computeIfAbsent(database, key -> new LockQueue()); + if (!lockQueue.tryLock(procedure)) { + acquiredDatabases.forEach( + acquiredDatabase -> { + final LockQueue acquiredLock = databaseLockMap.get(acquiredDatabase); + if (acquiredLock != null && acquiredLock.releaseLock(procedure)) { + acquiredLock.wakeWaitingProcedures(scheduler); + if (acquiredLock.isIdle()) { + databaseLockMap.remove(acquiredDatabase, acquiredLock); + } + } + }); + return database; + } + acquiredDatabases.add(database); + } + return null; + } finally { + schedulerLock.unlock(); + } + } + + public void waitDatabaseLock(final Procedure procedure, final String databaseName) { + schedulerLock.lock(); + try { + databaseLockMap + .computeIfAbsent(databaseName, key -> new LockQueue()) + .waitProcedure(procedure, scheduler); + } finally { + schedulerLock.unlock(); + } + } + + public void releaseDatabaseLocks(final Procedure procedure, final Set databaseNames) { + schedulerLock.lock(); + try { + for (final String database : databaseNames) { + final LockQueue lockQueue = databaseLockMap.get(database); + if (lockQueue != null && lockQueue.releaseLock(procedure)) { + lockQueue.wakeWaitingProcedures(scheduler); + if (lockQueue.isIdle()) { + databaseLockMap.remove(database, lockQueue); + } + } + } + } finally { + schedulerLock.unlock(); + } + } + public ProcedureScheduler getScheduler() { return scheduler; } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/AbstractDatabaseProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/AbstractDatabaseProcedure.java new file mode 100644 index 0000000000000..fc41b9afbc5a3 --- /dev/null +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/AbstractDatabaseProcedure.java @@ -0,0 +1,67 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.confignode.procedure.impl; + +import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv; +import org.apache.iotdb.confignode.procedure.state.ProcedureLockState; + +import java.util.Set; + +/** A procedure that holds exclusive lifecycle locks for its databases until it finishes. */ +public abstract class AbstractDatabaseProcedure + extends StateMachineProcedure { + + private transient String waitingDatabase; + + protected AbstractDatabaseProcedure() { + super(); + } + + protected AbstractDatabaseProcedure(final boolean isGeneratedByPipe) { + super(isGeneratedByPipe); + } + + protected abstract Set getDatabaseNames(); + + @Override + protected ProcedureLockState acquireLock(final ConfigNodeProcedureEnv env) { + waitingDatabase = env.tryLockDatabases(this, getDatabaseNames()); + return waitingDatabase == null + ? ProcedureLockState.LOCK_ACQUIRED + : ProcedureLockState.LOCK_EVENT_WAIT; + } + + @Override + protected void waitForLock(final ConfigNodeProcedureEnv env) { + if (waitingDatabase != null) { + env.waitDatabaseLock(this, waitingDatabase); + } + } + + @Override + protected void releaseLock(final ConfigNodeProcedureEnv env) { + env.releaseDatabaseLocks(this, getDatabaseNames()); + } + + @Override + protected boolean holdLock(final ConfigNodeProcedureEnv env) { + return true; + } +} diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/region/CreateRegionGroupsProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/region/CreateRegionGroupsProcedure.java index 276cdf432d9e6..c266f02c75cf6 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/region/CreateRegionGroupsProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/region/CreateRegionGroupsProcedure.java @@ -38,7 +38,7 @@ import org.apache.iotdb.confignode.persistence.partition.maintainer.RegionCreateTask; import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv; import org.apache.iotdb.confignode.procedure.exception.ProcedureException; -import org.apache.iotdb.confignode.procedure.impl.StateMachineProcedure; +import org.apache.iotdb.confignode.procedure.impl.AbstractDatabaseProcedure; import org.apache.iotdb.confignode.procedure.state.CreateRegionGroupsState; import org.apache.iotdb.confignode.procedure.store.ProcedureType; import org.apache.iotdb.consensus.exception.ConsensusException; @@ -61,7 +61,7 @@ import java.util.stream.Collectors; public class CreateRegionGroupsProcedure - extends StateMachineProcedure { + extends AbstractDatabaseProcedure { private static final Logger LOGGER = LoggerFactory.getLogger(CreateRegionGroupsProcedure.class); @@ -102,11 +102,19 @@ protected Flow executeFromState( final ConfigNodeProcedureEnv env, final CreateRegionGroupsState state) { switch (state) { case CREATE_REGION_GROUPS: + final TSStatus validationStatus = env.validateCreateRegionGroups(createRegionGroupsPlan); + if (validationStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + setFailure(new ProcedureException(new IoTDBException(validationStatus))); + return Flow.NO_MORE_STATE; + } failedRegionReplicaSets = env.doRegionCreation(consensusGroupType, createRegionGroupsPlan); setNextState(CreateRegionGroupsState.SHUNT_REGION_REPLICAS); break; case SHUNT_REGION_REPLICAS: persistPlan = new CreateRegionGroupsPlan(); + createRegionGroupsPlan + .getDatabaseGenerationMap() + .forEach(persistPlan::setDatabaseGeneration); final OfferRegionMaintainTasksPlan offerPlan = new OfferRegionMaintainTasksPlan(); // RegionGroups that failed to reach a serving quorum have their redundant (already-created) // replicas removed via an independent root RemoveRegionGroupProcedure. Submitting them as @@ -191,6 +199,13 @@ protected Flow executeFromState( final TSStatus persistStatus = env.persistRegionGroup(persistPlan); if (persistStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + getCreatedRegionReplicas() + .forEach( + replicaSet -> + env.getConfigManager() + .getProcedureManager() + .getExecutor() + .submitProcedure(new RemoveRegionGroupProcedure(replicaSet))); setFailure(new ProcedureException(new IoTDBException(persistStatus))); return Flow.NO_MORE_STATE; } @@ -323,6 +338,42 @@ protected CreateRegionGroupsState getInitialState() { return CreateRegionGroupsState.CREATE_REGION_GROUPS; } + @Override + protected Set getDatabaseNames() { + return createRegionGroupsPlan.getRegionGroupMap().keySet(); + } + + private List getCreatedRegionReplicas() { + final List createdRegionReplicas = new ArrayList<>(); + createRegionGroupsPlan + .getRegionGroupMap() + .values() + .forEach( + regionReplicaSets -> + regionReplicaSets.forEach( + regionReplicaSet -> { + final TRegionReplicaSet failedRegionReplicas = + failedRegionReplicaSets.get(regionReplicaSet.getRegionId()); + final TRegionReplicaSet createdRegionReplicaSet = + new TRegionReplicaSet().setRegionId(regionReplicaSet.getRegionId()); + regionReplicaSet + .getDataNodeLocations() + .forEach( + dataNodeLocation -> { + if (failedRegionReplicas == null + || !failedRegionReplicas + .getDataNodeLocations() + .contains(dataNodeLocation)) { + createdRegionReplicaSet.addToDataNodeLocations(dataNodeLocation); + } + }); + if (createdRegionReplicaSet.getDataNodeLocationsSize() > 0) { + createdRegionReplicas.add(createdRegionReplicaSet); + } + })); + return createdRegionReplicas; + } + @Override public void serialize(final DataOutputStream stream) throws IOException { // Must serialize CREATE_REGION_GROUPS.getTypeCode() firstly @@ -337,6 +388,8 @@ public void serialize(final DataOutputStream stream) throws IOException { ThriftCommonsSerDeUtils.serializeTRegionReplicaSet(replica, stream); }); persistPlan.serializeForProcedure(stream); + createRegionGroupsPlan.serializeDatabaseGenerationMap(stream); + persistPlan.serializeDatabaseGenerationMap(stream); } @Override @@ -357,6 +410,12 @@ public void deserialize(final ByteBuffer byteBuffer) { if (byteBuffer.hasRemaining()) { persistPlan.deserializeForProcedure(byteBuffer); } + if (byteBuffer.hasRemaining()) { + createRegionGroupsPlan.deserializeDatabaseGenerationMap(byteBuffer); + } + if (byteBuffer.hasRemaining()) { + persistPlan.deserializeDatabaseGenerationMap(byteBuffer); + } } catch (final Exception e) { LOGGER.error(ProcedureMessages.DESERIALIZE_MEETS_ERROR_IN_CREATEREGIONGROUPSPROCEDURE, e); throw new RuntimeException(e); diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeleteDatabaseProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeleteDatabaseProcedure.java index b547e8ebc7d39..21757f581e5cd 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeleteDatabaseProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeleteDatabaseProcedure.java @@ -29,7 +29,7 @@ import org.apache.iotdb.confignode.manager.partition.PartitionMetrics; import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv; import org.apache.iotdb.confignode.procedure.exception.ProcedureException; -import org.apache.iotdb.confignode.procedure.impl.StateMachineProcedure; +import org.apache.iotdb.confignode.procedure.impl.AbstractDatabaseProcedure; import org.apache.iotdb.confignode.procedure.impl.region.RemoveRegionGroupProcedure; import org.apache.iotdb.confignode.procedure.state.schema.DeleteDatabaseState; import org.apache.iotdb.confignode.procedure.store.ProcedureType; @@ -43,11 +43,12 @@ import java.io.DataOutputStream; import java.io.IOException; import java.nio.ByteBuffer; +import java.util.Collections; import java.util.List; import java.util.Objects; +import java.util.Set; -public class DeleteDatabaseProcedure - extends StateMachineProcedure { +public class DeleteDatabaseProcedure extends AbstractDatabaseProcedure { private static final Logger LOG = LoggerFactory.getLogger(DeleteDatabaseProcedure.class); private static final int RETRY_THRESHOLD = 5; @@ -234,6 +235,13 @@ protected DeleteDatabaseState getInitialState() { return DeleteDatabaseState.PRE_DELETE_DATABASE; } + @Override + protected Set getDatabaseNames() { + return deleteDatabaseSchema == null + ? Collections.emptySet() + : Collections.singleton(deleteDatabaseSchema.getName()); + } + public String getDatabase() { return deleteDatabaseSchema.getName(); } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/scheduler/LockQueue.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/scheduler/LockQueue.java index e2f5935a909ab..164f778c1f22c 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/scheduler/LockQueue.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/scheduler/LockQueue.java @@ -64,4 +64,8 @@ public int wakeWaitingProcedures(ProcedureScheduler procedureScheduler) { } return count; } + + public boolean isIdle() { + return lockOwnerProcedure == null && deque.isEmpty(); + } } diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java index e844dcf6910e9..85c67a8d27867 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java @@ -273,6 +273,20 @@ public void CreateDatabasePlanTest() throws IOException { Assert.assertEquals(req0, req1); } + @Test + public void CreateRegionGroupsPlanTest() throws IOException { + final CreateRegionGroupsPlan plan = new CreateRegionGroupsPlan(); + plan.setDatabaseGeneration("root.sg", 7); + plan.addRegionGroup( + "root.sg", + new TRegionReplicaSet( + new TConsensusGroupId(TConsensusGroupType.DataRegion, 1), Collections.emptyList())); + + final CreateRegionGroupsPlan deserializedPlan = + (CreateRegionGroupsPlan) ConfigPhysicalPlan.Factory.create(plan.serializeToByteBuffer()); + Assert.assertEquals(plan, deserializedPlan); + } + @Test public void AlterDatabasePlanTest() throws IOException { DatabaseSchemaPlan req0 = @@ -749,9 +763,13 @@ public void updateProcedureTest() throws IOException { failedRegions.put(dataRegionGroupId, dataRegionSet); failedRegions.put(schemaRegionGroupId, schemaRegionSet); CreateRegionGroupsPlan createRegionGroupsPlan = new CreateRegionGroupsPlan(); + createRegionGroupsPlan.setDatabaseGeneration("root.sg0", 1); + createRegionGroupsPlan.setDatabaseGeneration("root.sg1", 2); createRegionGroupsPlan.addRegionGroup("root.sg0", dataRegionSet); createRegionGroupsPlan.addRegionGroup("root.sg1", schemaRegionSet); CreateRegionGroupsPlan persistPlan = new CreateRegionGroupsPlan(); + persistPlan.setDatabaseGeneration("root.sg0", 1); + persistPlan.setDatabaseGeneration("root.sg1", 2); persistPlan.addRegionGroup("root.sg0", dataRegionSet); persistPlan.addRegionGroup("root.sg1", schemaRegionSet); CreateRegionGroupsProcedure procedure0 = diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/PartitionInfoTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/PartitionInfoTest.java index 8c32378cfcb1a..6045190a4c630 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/PartitionInfoTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/PartitionInfoTest.java @@ -24,6 +24,7 @@ import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation; import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet; +import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot; import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot; import org.apache.iotdb.commons.partition.DataPartitionTable; @@ -32,11 +33,14 @@ import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType; import org.apache.iotdb.confignode.consensus.request.read.region.GetRegionInfoListPlan; import org.apache.iotdb.confignode.consensus.request.write.database.DatabaseSchemaPlan; +import org.apache.iotdb.confignode.consensus.request.write.database.DeleteDatabasePlan; +import org.apache.iotdb.confignode.consensus.request.write.database.PreDeleteDatabasePlan; import org.apache.iotdb.confignode.consensus.request.write.partition.CreateDataPartitionPlan; import org.apache.iotdb.confignode.consensus.request.write.partition.CreateSchemaPartitionPlan; import org.apache.iotdb.confignode.consensus.request.write.region.CreateRegionGroupsPlan; import org.apache.iotdb.confignode.consensus.request.write.region.OfferRegionMaintainTasksPlan; import org.apache.iotdb.confignode.consensus.response.partition.RegionInfoListResp; +import org.apache.iotdb.confignode.exception.DatabaseNotExistsException; import org.apache.iotdb.confignode.persistence.partition.PartitionInfo; import org.apache.iotdb.confignode.persistence.partition.maintainer.RegionCreateTask; import org.apache.iotdb.confignode.persistence.partition.maintainer.RegionDeleteTask; @@ -44,6 +48,7 @@ import org.apache.iotdb.confignode.persistence.partition.maintainer.RegionMaintainType; import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema; import org.apache.iotdb.confignode.rpc.thrift.TShowRegionReq; +import org.apache.iotdb.rpc.TSStatusCode; import org.apache.thrift.TException; import org.apache.tsfile.external.commons.io.FileUtils; @@ -179,6 +184,86 @@ public void testLegacyRegionDeleteTasksAreFiltered() throws TException, IOExcept Assert.assertEquals(2, loaded.getRegionMaintainEntryList().size()); } + @Test + public void testCreateRegionGroupsRejectsPreDeletedAndMissingDatabase() + throws DatabaseNotExistsException { + final String database = "root.lifecycle"; + partitionInfo.createDatabase( + new DatabaseSchemaPlan( + ConfigPhysicalPlanType.CreateDatabase, new TDatabaseSchema(database))); + + final CreateRegionGroupsPlan preDeletedPlan = new CreateRegionGroupsPlan(); + preDeletedPlan.setDatabaseGeneration(database, partitionInfo.getDatabaseGeneration(database)); + preDeletedPlan.addRegionGroup( + database, + generateTRegionReplicaSet(0, new TConsensusGroupId(TConsensusGroupType.DataRegion, 1))); + partitionInfo.preDeleteDatabase( + new PreDeleteDatabasePlan(database, PreDeleteDatabasePlan.PreDeleteType.EXECUTE)); + + TSStatus status = partitionInfo.createRegionGroups(preDeletedPlan); + Assert.assertEquals(TSStatusCode.DATABASE_NOT_EXIST.getStatusCode(), status.getCode()); + Assert.assertTrue( + partitionInfo.getAllReplicaSets(database, TConsensusGroupType.DataRegion).isEmpty()); + + final CreateRegionGroupsPlan missingPlan = new CreateRegionGroupsPlan(); + missingPlan.addRegionGroup( + "root.missing", + generateTRegionReplicaSet(10, new TConsensusGroupId(TConsensusGroupType.SchemaRegion, 2))); + status = partitionInfo.createRegionGroups(missingPlan); + Assert.assertEquals(TSStatusCode.DATABASE_NOT_EXIST.getStatusCode(), status.getCode()); + } + + @Test + public void testOldCreateRegionGroupsPlanCannotPolluteRecreatedDatabase() + throws DatabaseNotExistsException { + final String database = "root.recreated"; + final DatabaseSchemaPlan createDatabasePlan = + new DatabaseSchemaPlan( + ConfigPhysicalPlanType.CreateDatabase, new TDatabaseSchema(database)); + partitionInfo.createDatabase(createDatabasePlan); + + final long oldGeneration = partitionInfo.getDatabaseGeneration(database); + final CreateRegionGroupsPlan oldPlan = new CreateRegionGroupsPlan(); + oldPlan.setDatabaseGeneration(database, oldGeneration); + oldPlan.addRegionGroup( + database, + generateTRegionReplicaSet(0, new TConsensusGroupId(TConsensusGroupType.DataRegion, 3))); + + partitionInfo.deleteDatabase(new DeleteDatabasePlan(database)); + partitionInfo.createDatabase(createDatabasePlan); + Assert.assertNotEquals(oldGeneration, partitionInfo.getDatabaseGeneration(database)); + + final TSStatus status = partitionInfo.createRegionGroups(oldPlan); + Assert.assertEquals(TSStatusCode.DATABASE_CONFIG_ERROR.getStatusCode(), status.getCode()); + Assert.assertEquals( + 0, partitionInfo.getRegionGroupCount(database, TConsensusGroupType.DataRegion)); + } + + @Test + public void testBatchedCreateRegionGroupsPlanIsValidatedAtomically() + throws DatabaseNotExistsException { + final String existingDatabase = "root.existing"; + partitionInfo.createDatabase( + new DatabaseSchemaPlan( + ConfigPhysicalPlanType.CreateDatabase, new TDatabaseSchema(existingDatabase))); + + final CreateRegionGroupsPlan batchedPlan = new CreateRegionGroupsPlan(); + batchedPlan.setDatabaseGeneration( + existingDatabase, partitionInfo.getDatabaseGeneration(existingDatabase)); + batchedPlan.addRegionGroup( + existingDatabase, + generateTRegionReplicaSet(0, new TConsensusGroupId(TConsensusGroupType.DataRegion, 40))); + batchedPlan.addRegionGroup( + "root.missing", + generateTRegionReplicaSet(10, new TConsensusGroupId(TConsensusGroupType.DataRegion, 41))); + + final TSStatus status = partitionInfo.createRegionGroups(batchedPlan); + Assert.assertEquals(TSStatusCode.DATABASE_NOT_EXIST.getStatusCode(), status.getCode()); + Assert.assertEquals( + 0, partitionInfo.getRegionGroupCount(existingDatabase, TConsensusGroupType.DataRegion)); + Assert.assertEquals(42, partitionInfo.generateNextRegionGroupId()); + } + @Test public void testGetRegionType() { diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/CreateRegionGroupsProcedureTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/CreateRegionGroupsProcedureTest.java index 6269953736100..d93d60c68e477 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/CreateRegionGroupsProcedureTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/CreateRegionGroupsProcedureTest.java @@ -24,20 +24,36 @@ import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation; import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet; +import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.confignode.consensus.request.write.region.CreateRegionGroupsPlan; +import org.apache.iotdb.confignode.manager.ConfigManager; +import org.apache.iotdb.confignode.manager.ProcedureManager; +import org.apache.iotdb.confignode.procedure.Procedure; +import org.apache.iotdb.confignode.procedure.ProcedureExecutor; +import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv; import org.apache.iotdb.confignode.procedure.impl.region.CreateRegionGroupsProcedure; +import org.apache.iotdb.confignode.procedure.impl.region.RemoveRegionGroupProcedure; +import org.apache.iotdb.confignode.procedure.impl.schema.DeleteDatabaseProcedure; +import org.apache.iotdb.confignode.procedure.scheduler.ProcedureScheduler; +import org.apache.iotdb.confignode.procedure.state.CreateRegionGroupsState; +import org.apache.iotdb.confignode.procedure.state.ProcedureLockState; import org.apache.iotdb.confignode.procedure.store.ProcedureFactory; import org.apache.iotdb.confignode.procedure.store.ProcedureType; +import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema; +import org.apache.iotdb.rpc.TSStatusCode; import org.apache.tsfile.utils.PublicBAOS; import org.junit.Assert; import org.junit.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.Mockito; import java.io.DataOutputStream; import java.io.IOException; import java.nio.ByteBuffer; import java.util.Collections; import java.util.HashMap; +import java.util.List; import java.util.Map; import static org.apache.iotdb.common.rpc.thrift.TConsensusGroupType.DataRegion; @@ -47,6 +63,44 @@ public class CreateRegionGroupsProcedureTest { + private static class TestCreateRegionGroupsProcedure extends CreateRegionGroupsProcedure { + + private TestCreateRegionGroupsProcedure( + final TConsensusGroupType consensusGroupType, + final CreateRegionGroupsPlan createRegionGroupsPlan, + final CreateRegionGroupsPlan persistPlan, + final Map failedRegionReplicaSets) { + super(consensusGroupType, createRegionGroupsPlan, persistPlan, failedRegionReplicaSets); + } + + private void executeShunt(final ConfigNodeProcedureEnv env) { + executeFromState(env, CreateRegionGroupsState.SHUNT_REGION_REPLICAS); + } + + private ProcedureLockState acquireDatabaseLock(final ConfigNodeProcedureEnv env) { + return acquireLock(env); + } + + private void releaseDatabaseLock(final ConfigNodeProcedureEnv env) { + releaseLock(env); + } + } + + private static class TestDeleteDatabaseProcedure extends DeleteDatabaseProcedure { + + private TestDeleteDatabaseProcedure(final TDatabaseSchema databaseSchema) { + super(databaseSchema, false); + } + + private ProcedureLockState acquireDatabaseLock(final ConfigNodeProcedureEnv env) { + return acquireLock(env); + } + + private void releaseDatabaseLock(final ConfigNodeProcedureEnv env) { + releaseLock(env); + } + } + @Test public void serializeDeserializeTest() { TDataNodeLocation dataNodeLocation0 = new TDataNodeLocation(); @@ -91,10 +145,14 @@ public void serializeDeserializeTest() { assertEquals(failedRegions0, failedRegions1); CreateRegionGroupsPlan createRegionGroupsPlan = new CreateRegionGroupsPlan(); + createRegionGroupsPlan.setDatabaseGeneration("root.sg0", 11); + createRegionGroupsPlan.setDatabaseGeneration("root.sg1", 12); createRegionGroupsPlan.addRegionGroup("root.sg0", dataRegionSet); createRegionGroupsPlan.addRegionGroup("root.sg1", schemaRegionSet); CreateRegionGroupsPlan persistPlan = new CreateRegionGroupsPlan(); + persistPlan.setDatabaseGeneration("root.sg0", 11); + persistPlan.setDatabaseGeneration("root.sg1", 12); persistPlan.addRegionGroup("root.sg0", dataRegionSet); persistPlan.addRegionGroup("root.sg1", schemaRegionSet); @@ -124,4 +182,75 @@ public void serializeDeserializeTest() { fail(); } } + + @Test + public void testPersistRejectionCleansCreatedRegionReplicas() { + final TDataNodeLocation createdDataNode = + new TDataNodeLocation().setDataNodeId(1).setInternalEndPoint(new TEndPoint("0.0.0.1", 1)); + final TDataNodeLocation failedDataNode = + new TDataNodeLocation().setDataNodeId(2).setInternalEndPoint(new TEndPoint("0.0.0.2", 2)); + final TConsensusGroupId regionId = new TConsensusGroupId(DataRegion, 10); + final TRegionReplicaSet allocatedReplicaSet = + new TRegionReplicaSet(regionId, List.of(createdDataNode, failedDataNode)); + final TRegionReplicaSet failedReplicaSet = + new TRegionReplicaSet(regionId, Collections.singletonList(failedDataNode)); + + final CreateRegionGroupsPlan createPlan = new CreateRegionGroupsPlan(); + createPlan.setDatabaseGeneration("root.sg", 1); + createPlan.addRegionGroup("root.sg", allocatedReplicaSet); + final Map failedReplicaSets = new HashMap<>(); + failedReplicaSets.put(regionId, failedReplicaSet); + final TestCreateRegionGroupsProcedure procedure = + new TestCreateRegionGroupsProcedure( + DataRegion, createPlan, new CreateRegionGroupsPlan(), failedReplicaSets); + + final ConfigNodeProcedureEnv env = Mockito.mock(ConfigNodeProcedureEnv.class); + final ConfigManager configManager = Mockito.mock(ConfigManager.class); + final ProcedureManager procedureManager = Mockito.mock(ProcedureManager.class); + @SuppressWarnings("unchecked") + final ProcedureExecutor executor = + Mockito.mock(ProcedureExecutor.class); + Mockito.when(env.persistRegionGroup(Mockito.any())) + .thenReturn(new TSStatus(TSStatusCode.DATABASE_NOT_EXIST.getStatusCode())); + Mockito.when(env.getConfigManager()).thenReturn(configManager); + Mockito.when(configManager.getProcedureManager()).thenReturn(procedureManager); + Mockito.when(procedureManager.getExecutor()).thenReturn(executor); + + procedure.executeShunt(env); + + final ArgumentCaptor> cleanupCaptor = + ArgumentCaptor.forClass(Procedure.class); + Mockito.verify(executor).submitProcedure(cleanupCaptor.capture()); + Assert.assertEquals( + new RemoveRegionGroupProcedure( + new TRegionReplicaSet(regionId, Collections.singletonList(createdDataNode))), + cleanupCaptor.getValue()); + } + + @Test + public void testCreateAndDeleteDatabaseLifecycleAreMutuallyExclusive() { + final String database = "root.sg"; + final CreateRegionGroupsPlan createPlan = new CreateRegionGroupsPlan(); + createPlan.addRegionGroup( + database, + new TRegionReplicaSet(new TConsensusGroupId(DataRegion, 1), Collections.emptyList())); + final TestCreateRegionGroupsProcedure createProcedure = + new TestCreateRegionGroupsProcedure( + DataRegion, createPlan, new CreateRegionGroupsPlan(), Collections.emptyMap()); + createProcedure.setProcId(1); + final TestDeleteDatabaseProcedure deleteProcedure = + new TestDeleteDatabaseProcedure(new TDatabaseSchema(database)); + deleteProcedure.setProcId(2); + + final ConfigNodeProcedureEnv env = + new ConfigNodeProcedureEnv( + Mockito.mock(ConfigManager.class), Mockito.mock(ProcedureScheduler.class)); + Assert.assertEquals(ProcedureLockState.LOCK_ACQUIRED, createProcedure.acquireDatabaseLock(env)); + Assert.assertEquals( + ProcedureLockState.LOCK_EVENT_WAIT, deleteProcedure.acquireDatabaseLock(env)); + + createProcedure.releaseDatabaseLock(env); + Assert.assertEquals(ProcedureLockState.LOCK_ACQUIRED, deleteProcedure.acquireDatabaseLock(env)); + deleteProcedure.releaseDatabaseLock(env); + } }