This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 071cde31ad [IOTDB-3261][IOTDB-3332][IOTDB-3372] Ensure the concurrency
security of Region alloction (#6156)
071cde31ad is described below
commit 071cde31adf4bf040c78b49d89661560b5c1d6d3
Author: YongzaoDan <[email protected]>
AuthorDate: Mon Jun 6 12:25:13 2022 +0800
[IOTDB-3261][IOTDB-3332][IOTDB-3372] Ensure the concurrency security of
Region alloction (#6156)
---
.../confignode/exception/TimeoutException.java | 26 +++++
.../confignode/manager/ClusterSchemaManager.java | 24 ++++
.../iotdb/confignode/manager/PartitionManager.java | 102 ++++++++++++++--
.../iotdb/confignode/manager/load/LoadManager.java | 14 +--
.../manager/load/balancer/RegionBalancer.java | 9 +-
.../balancer/allocator/CopySetRegionAllocator.java | 13 +--
.../load/balancer/allocator/IRegionAllocator.java | 4 +
.../confignode/persistence/ClusterSchemaInfo.java | 128 ++++++++++++++++++++-
.../thrift/ConfigNodeRPCServiceProcessor.java | 4 +-
.../persistence/ClusterSchemaInfoTest.java | 14 ++-
10 files changed, 302 insertions(+), 36 deletions(-)
diff --git
a/confignode/src/main/java/org/apache/iotdb/confignode/exception/TimeoutException.java
b/confignode/src/main/java/org/apache/iotdb/confignode/exception/TimeoutException.java
new file mode 100644
index 0000000000..f8e895d191
--- /dev/null
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/exception/TimeoutException.java
@@ -0,0 +1,26 @@
+/*
+ * 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.exception;
+
+public class TimeoutException extends ConfigNodeException {
+
+ public TimeoutException(String message) {
+ super(message);
+ }
+}
diff --git
a/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterSchemaManager.java
b/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterSchemaManager.java
index baf67a015a..e888075a84 100644
---
a/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterSchemaManager.java
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/manager/ClusterSchemaManager.java
@@ -95,6 +95,18 @@ public class ClusterSchemaManager {
return clusterSchemaInfo.getRegionGroupIds(storageGroup, type);
}
+ /**
+ * Only leader use this interface.
+ *
+ * @param storageGroup StorageGroupName
+ * @param type SchemaRegion or DataRegion
+ * @return Number of Regions currently owned by the specific StorageGroup
+ */
+ public int getRegionGroupCount(String storageGroup, TConsensusGroupType type)
+ throws MetadataException {
+ return clusterSchemaInfo.getRegionGroupCount(storageGroup, type);
+ }
+
/**
* Only leader use this interface.
*
@@ -118,6 +130,18 @@ public class ClusterSchemaManager {
return clusterSchemaInfo.getMatchedStorageGroupSchemasByName(rawPathList);
}
+ /**
+ * Only leader use this interface. Contending the Region allocation particle
+ *
+ * @param storageGroup The specific StorageGroup
+ * @param consensusGroupType SchemaRegion or DataRegion
+ * @return True if successfully get the Region allocation particle, false
otherwise.
+ */
+ public boolean getRegionAllocationParticle(
+ String storageGroup, TConsensusGroupType consensusGroupType) {
+ return clusterSchemaInfo.getRegionAllocationParticle(storageGroup,
consensusGroupType);
+ }
+
public TSStatus setTTL(SetTTLReq setTTLReq) {
// TODO: Inform DataNodes
return getConsensusManager().write(setTTLReq).getStatus();
diff --git
a/confignode/src/main/java/org/apache/iotdb/confignode/manager/PartitionManager.java
b/confignode/src/main/java/org/apache/iotdb/confignode/manager/PartitionManager.java
index 638f78acdf..d6e77c4de2 100644
---
a/confignode/src/main/java/org/apache/iotdb/confignode/manager/PartitionManager.java
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/manager/PartitionManager.java
@@ -25,6 +25,7 @@ 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.concurrent.IoTDBThreadPoolFactory;
+import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.commons.partition.executor.SeriesPartitionExecutor;
import org.apache.iotdb.confignode.client.SyncDataNodeClientPool;
import org.apache.iotdb.confignode.conf.ConfigNodeConf;
@@ -41,8 +42,10 @@ import
org.apache.iotdb.confignode.consensus.response.DataPartitionResp;
import org.apache.iotdb.confignode.consensus.response.SchemaNodeManagementResp;
import org.apache.iotdb.confignode.consensus.response.SchemaPartitionResp;
import org.apache.iotdb.confignode.exception.NotEnoughDataNodeException;
+import org.apache.iotdb.confignode.exception.TimeoutException;
import org.apache.iotdb.confignode.manager.load.LoadManager;
import org.apache.iotdb.confignode.persistence.PartitionInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TStorageGroupSchema;
import org.apache.iotdb.consensus.common.DataSet;
import org.apache.iotdb.consensus.common.response.ConsensusReadResponse;
import org.apache.iotdb.rpc.TSStatusCode;
@@ -125,6 +128,23 @@ public class PartitionManager {
} catch (NotEnoughDataNodeException e) {
SchemaPartitionResp resp = new SchemaPartitionResp();
resp.setStatus(new
TSStatus(TSStatusCode.NOT_ENOUGH_DATA_NODE.getStatusCode()));
+ resp.getStatus()
+ .setMessage(
+ "ConfigNode failed to allocate DataPartition because there are
not enough DataNodes");
+ return resp;
+ } catch (TimeoutException e) {
+ SchemaPartitionResp resp = new SchemaPartitionResp();
+ resp.setStatus(new TSStatus(TSStatusCode.TIME_OUT.getStatusCode()));
+ resp.getStatus()
+ .setMessage(
+ "ConfigNode failed to allocate DataPartition because waiting
for another thread's Region allocation timeout.");
+ return resp;
+ } catch (MetadataException e) {
+ SchemaPartitionResp resp = new SchemaPartitionResp();
+ resp.setStatus(new
TSStatus(TSStatusCode.STORAGE_GROUP_NOT_EXIST.getStatusCode()));
+ resp.getStatus()
+ .setMessage(
+ "ConfigNode failed to allocate DataPartition because some
StorageGroup doesn't exist.");
return resp;
}
@@ -224,6 +244,23 @@ public class PartitionManager {
} catch (NotEnoughDataNodeException e) {
DataPartitionResp resp = new DataPartitionResp();
resp.setStatus(new
TSStatus(TSStatusCode.NOT_ENOUGH_DATA_NODE.getStatusCode()));
+ resp.getStatus()
+ .setMessage(
+ "ConfigNode failed to allocate DataPartition because there are
not enough DataNodes");
+ return resp;
+ } catch (TimeoutException e) {
+ DataPartitionResp resp = new DataPartitionResp();
+ resp.setStatus(new TSStatus(TSStatusCode.TIME_OUT.getStatusCode()));
+ resp.getStatus()
+ .setMessage(
+ "ConfigNode failed to allocate DataPartition because waiting
for another thread's Region allocation timeout.");
+ return resp;
+ } catch (MetadataException e) {
+ SchemaPartitionResp resp = new SchemaPartitionResp();
+ resp.setStatus(new
TSStatus(TSStatusCode.STORAGE_GROUP_NOT_EXIST.getStatusCode()));
+ resp.getStatus()
+ .setMessage(
+ "ConfigNode failed to allocate DataPartition because some
StorageGroup doesn't exist.");
return resp;
}
@@ -303,13 +340,39 @@ public class PartitionManager {
private void checkAndAllocateRegionsIfNecessary(
List<String> storageGroups, TConsensusGroupType consensusGroupType)
- throws NotEnoughDataNodeException {
- List<String> storageGroupWithoutRegion = new ArrayList<>();
+ throws NotEnoughDataNodeException, TimeoutException, MetadataException {
+
+ int leastDataNode = 0;
+ List<String> unreadyStorageGroups = new ArrayList<>();
for (String storageGroup : storageGroups) {
- List<TConsensusGroupId> groupIds =
- getClusterSchemaManager().getRegionGroupIds(storageGroup,
consensusGroupType);
- if (groupIds.size() == 0) {
- storageGroupWithoutRegion.add(storageGroup);
+ if (getClusterSchemaManager().getRegionGroupCount(storageGroup,
consensusGroupType) == 0) {
+ TStorageGroupSchema storageGroupSchema =
+
getClusterSchemaManager().getStorageGroupSchemaByName(storageGroup);
+ switch (consensusGroupType) {
+ case SchemaRegion:
+ leastDataNode =
+ Math.max(leastDataNode,
storageGroupSchema.getSchemaReplicationFactor());
+ break;
+ case DataRegion:
+ leastDataNode = Math.max(leastDataNode,
storageGroupSchema.getDataReplicationFactor());
+ }
+ // Recording StorageGroups without Region
+ unreadyStorageGroups.add(storageGroup);
+ }
+ }
+ if (getNodeManager().getOnlineDataNodeCount() < leastDataNode) {
+ // Make sure DataNode enough
+ throw new NotEnoughDataNodeException();
+ }
+
+ List<String> storageGroupsNeedAllocation = new ArrayList<>();
+ List<String> storageGroupsInAllocation = new ArrayList<>();
+ for (String storageGroup : unreadyStorageGroups) {
+ // Try to get the allocation particle
+ if (getClusterSchemaManager().getRegionAllocationParticle(storageGroup,
consensusGroupType)) {
+ storageGroupsNeedAllocation.add(storageGroup);
+ } else {
+ storageGroupsInAllocation.add(storageGroup);
}
}
@@ -332,7 +395,32 @@ public class PartitionManager {
/
ConfigNodeDescriptor.getInstance().getConf().getDataReplicationFactor()));
}
- getLoadManager().initializeRegions(storageGroupWithoutRegion,
consensusGroupType, regionNum);
+ // Do Region allocation for those StorageGroups who get the particle
+ // TODO: Put back Region allocation particles when creation process
failed. We can use Procedure
+ // to protect this
+ getLoadManager().initializeRegions(storageGroupsNeedAllocation,
consensusGroupType, regionNum);
+
+ // Waiting allocation for those StorageGroups who don't get the particle
+ for (int retry = 0; retry < 10; retry++) {
+ boolean allocationFinished = true;
+ for (String storageGroup : storageGroupsInAllocation) {
+ if (getClusterSchemaManager().getRegionGroupCount(storageGroup,
consensusGroupType) > 0) {
+ allocationFinished = false;
+ break;
+ }
+ }
+ if (allocationFinished) {
+ return;
+ }
+
+ try {
+ // Sleep 1s to wait Region allocation
+ TimeUnit.SECONDS.sleep(1);
+ } catch (InterruptedException e) {
+ LOGGER.warn("The PartitionManager is interrupted.", e);
+ }
+ }
+ throw new TimeoutException("");
}
/** Get all allocated RegionReplicaSets */
diff --git
a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/LoadManager.java
b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/LoadManager.java
index daa37f7a24..a780b52109 100644
---
a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/LoadManager.java
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/LoadManager.java
@@ -86,16 +86,10 @@ public class LoadManager implements Runnable {
*/
public void initializeRegions(
List<String> storageGroups, TConsensusGroupType consensusGroupType, int
regionNum)
- throws NotEnoughDataNodeException {
- CreateRegionsReq createRegionsReq = null;
-
- try {
- createRegionsReq =
- regionBalancer.genRegionsAllocationPlan(storageGroups,
consensusGroupType, regionNum);
- createRegionsOnDataNodes(createRegionsReq);
- } catch (MetadataException e) {
- LOGGER.error("Meet error when create Regions", e);
- }
+ throws NotEnoughDataNodeException, MetadataException {
+ CreateRegionsReq createRegionsReq =
+ regionBalancer.genRegionsAllocationPlan(storageGroups,
consensusGroupType, regionNum);
+ createRegionsOnDataNodes(createRegionsReq);
getConsensusManager().write(createRegionsReq);
}
diff --git
a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RegionBalancer.java
b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RegionBalancer.java
index cfc23769da..40d66f01a1 100644
---
a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RegionBalancer.java
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RegionBalancer.java
@@ -41,12 +41,9 @@ import java.util.List;
public class RegionBalancer {
private final Manager configManager;
- private final IRegionAllocator regionAllocator;
public RegionBalancer(Manager configManager) {
this.configManager = configManager;
- // TODO: The RegionAllocator should be configurable
- this.regionAllocator = new CopySetRegionAllocator();
}
/**
@@ -63,6 +60,7 @@ public class RegionBalancer {
List<String> storageGroups, TConsensusGroupType consensusGroupType, int
regionNum)
throws NotEnoughDataNodeException, MetadataException {
CreateRegionsReq createRegionsReq = new CreateRegionsReq();
+ IRegionAllocator regionAllocator = genRegionAllocator();
List<TDataNodeInfo> onlineDataNodes =
getNodeManager().getOnlineDataNodes(-1);
List<TRegionReplicaSet> allocatedRegions =
getPartitionManager().getAllocatedRegions();
@@ -99,6 +97,11 @@ public class RegionBalancer {
return createRegionsReq;
}
+ private IRegionAllocator genRegionAllocator() {
+ // TODO: The RegionAllocator should be configurable
+ return new CopySetRegionAllocator();
+ }
+
private NodeManager getNodeManager() {
return configManager.getNodeManager();
}
diff --git
a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/allocator/CopySetRegionAllocator.java
b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/allocator/CopySetRegionAllocator.java
index c762b833bd..b3e5cb30fc 100644
---
a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/allocator/CopySetRegionAllocator.java
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/allocator/CopySetRegionAllocator.java
@@ -42,10 +42,10 @@ public class CopySetRegionAllocator implements
IRegionAllocator {
private int maxId = 0;
private int intersectionSize = 0;
- private List<TDataNodeLocation> weightList;
+ private final List<TDataNodeLocation> weightList;
public CopySetRegionAllocator() {
- // Empty constructor
+ this.weightList = new ArrayList<>();
}
@Override
@@ -74,7 +74,6 @@ public class CopySetRegionAllocator implements
IRegionAllocator {
intersectionSize += 1;
}
- clear();
result.setRegionId(consensusGroupId);
return result;
}
@@ -97,7 +96,6 @@ public class CopySetRegionAllocator implements
IRegionAllocator {
}
}
- weightList = new ArrayList<>();
for (Map.Entry<TDataNodeLocation, Integer> countEntry :
countMap.entrySet()) {
int weight = maximumRegionNum - countEntry.getValue() + 1;
// Repeatedly add DataNode copies equal to the number of their weights
@@ -158,11 +156,4 @@ public class CopySetRegionAllocator implements
IRegionAllocator {
}
return true;
}
-
- private void clear() {
- maxId = 0;
- intersectionSize = 0;
- weightList.clear();
- weightList = null;
- }
}
diff --git
a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/allocator/IRegionAllocator.java
b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/allocator/IRegionAllocator.java
index a702d9c2e1..605de85cb4 100644
---
a/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/allocator/IRegionAllocator.java
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/allocator/IRegionAllocator.java
@@ -24,6 +24,10 @@ import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import java.util.List;
+/**
+ * The IRegionAllocator is a functional interface, which means a new
functional class who implements
+ * the IRegionAllocator must be created for each Region allocation.
+ */
public interface IRegionAllocator {
/**
diff --git
a/confignode/src/main/java/org/apache/iotdb/confignode/persistence/ClusterSchemaInfo.java
b/confignode/src/main/java/org/apache/iotdb/confignode/persistence/ClusterSchemaInfo.java
index 569789b0ec..9043d202f1 100644
---
a/confignode/src/main/java/org/apache/iotdb/confignode/persistence/ClusterSchemaInfo.java
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/persistence/ClusterSchemaInfo.java
@@ -43,6 +43,7 @@ import org.apache.iotdb.db.metadata.mnode.StorageGroupMNode;
import org.apache.iotdb.db.metadata.mtree.MTreeAboveSG;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.tsfile.utils.Pair;
+import org.apache.iotdb.tsfile.utils.ReadWriteIOUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -54,6 +55,7 @@ import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -72,18 +74,24 @@ public class ClusterSchemaInfo implements SnapshotProcessor
{
// StorageGroup read write lock
private final ReentrantReadWriteLock storageGroupReadWriteLock;
+ private final MTreeAboveSG mTree;
- private MTreeAboveSG mTree;
+ // Particle for allocating the first Region
+ private final Set<String> schemaRegionParticles;
+ private final Set<String> dataRegionParticles;
private final String snapshotFileName = "cluster_schema.bin";
- public ClusterSchemaInfo() {
+ public ClusterSchemaInfo() throws IOException {
storageGroupReadWriteLock = new ReentrantReadWriteLock();
+ schemaRegionParticles = Collections.synchronizedSet(new HashSet<>());
+ dataRegionParticles = Collections.synchronizedSet(new HashSet<>());
try {
mTree = new MTreeAboveSG();
} catch (MetadataException e) {
LOGGER.error("Can't construct StorageGroupInfo", e);
+ throw new IOException(e);
}
}
@@ -108,6 +116,10 @@ public class ClusterSchemaInfo implements
SnapshotProcessor {
.getStorageGroupNodeByStorageGroupPath(partialPathName)
.setStorageGroupSchema(storageGroupSchema);
+ // Mark the new StorageGroup as without Region
+ schemaRegionParticles.add(storageGroupSchema.getName());
+ dataRegionParticles.add(storageGroupSchema.getName());
+
result.setCode(TSStatusCode.SUCCESS_STATUS.getStatusCode());
LOGGER.info("Successfully set StorageGroup: {}", storageGroupSchema);
@@ -136,6 +148,11 @@ public class ClusterSchemaInfo implements
SnapshotProcessor {
TStorageGroupSchema storageGroupSchema = req.getStorageGroup();
PartialPath partialPathName = new
PartialPath(storageGroupSchema.getName());
mTree.deleteStorageGroup(partialPathName);
+
+ // Clean the unready set
+ schemaRegionParticles.remove(storageGroupSchema.getName());
+ dataRegionParticles.remove(storageGroupSchema.getName());
+
result.setCode(TSStatusCode.SUCCESS_STATUS.getStatusCode());
} catch (MetadataException e) {
LOGGER.warn("Storage group not exist", e);
@@ -443,6 +460,66 @@ public class ClusterSchemaInfo implements
SnapshotProcessor {
return result;
}
+ /**
+ * Return the number of Regions currently owned by the specific StorageGroup
+ *
+ * @param storageGroup StorageGroupName
+ * @param type SchemaRegion or DataRegion
+ * @return Number of Regions currently owned by the specific StorageGroup
+ */
+ public int getRegionGroupCount(String storageGroup, TConsensusGroupType type)
+ throws MetadataException {
+ storageGroupReadWriteLock.readLock().lock();
+ try {
+ StorageGroupMNode mNode =
+ (StorageGroupMNode)
+ mTree.getStorageGroupNodeByStorageGroupPath(new
PartialPath(storageGroup));
+ switch (type) {
+ case SchemaRegion:
+ return mNode.getStorageGroupSchema().getSchemaRegionGroupIdsSize();
+ case DataRegion:
+ return mNode.getStorageGroupSchema().getDataRegionGroupIdsSize();
+ default:
+ return 0;
+ }
+ } finally {
+ storageGroupReadWriteLock.readLock().unlock();
+ }
+ }
+
+ /**
+ * Contending the Region allocation particle
+ *
+ * @param storageGroup The specific StorageGroup
+ * @param consensusGroupType SchemaRegion or DataRegion
+ * @return True if successfully get the Region allocation particle, false
otherwise.
+ */
+ public boolean getRegionAllocationParticle(
+ String storageGroup, TConsensusGroupType consensusGroupType) {
+ switch (consensusGroupType) {
+ case SchemaRegion:
+ synchronized (schemaRegionParticles) {
+ if (schemaRegionParticles.contains(storageGroup)) {
+ schemaRegionParticles.remove(storageGroup);
+ return true;
+ } else {
+ return false;
+ }
+ }
+ case DataRegion:
+ synchronized (dataRegionParticles) {
+ if (dataRegionParticles.contains(storageGroup)) {
+ dataRegionParticles.remove(storageGroup);
+ return true;
+ } else {
+ return false;
+ }
+ }
+ default:
+ return false;
+ }
+ }
+
@Override
public boolean processTakeSnapshot(File snapshotDir) throws IOException {
@@ -460,9 +537,27 @@ public class ClusterSchemaInfo implements
SnapshotProcessor {
try {
try (FileOutputStream fileOutputStream = new FileOutputStream(tmpFile);
BufferedOutputStream outputStream = new
BufferedOutputStream(fileOutputStream)) {
+
+ // Take snapshot for particles
+ synchronized (schemaRegionParticles) {
+ synchronized (dataRegionParticles) {
+ ReadWriteIOUtils.write(schemaRegionParticles.size(), outputStream);
+ for (String schemaRegionParticle : schemaRegionParticles) {
+ ReadWriteIOUtils.write(schemaRegionParticle, outputStream);
+ }
+
+ ReadWriteIOUtils.write(dataRegionParticles.size(), outputStream);
+ for (String dataRegionParticle : dataRegionParticles) {
+ ReadWriteIOUtils.write(dataRegionParticle, outputStream);
+ }
+ }
+ }
+
+ // Take snapshot for MTree
mTree.serialize(outputStream);
outputStream.flush();
}
+
return tmpFile.renameTo(snapshotFile);
} finally {
for (int retry = 0; retry < 5; retry++) {
@@ -490,6 +585,25 @@ public class ClusterSchemaInfo implements
SnapshotProcessor {
storageGroupReadWriteLock.writeLock().lock();
try (FileInputStream fileInputStream = new FileInputStream(snapshotFile);
BufferedInputStream bufferedInputStream = new
BufferedInputStream(fileInputStream)) {
+
+ // Load snapshot of particles
+ synchronized (schemaRegionParticles) {
+ synchronized (dataRegionParticles) {
+ int particleNum = ReadWriteIOUtils.readInt(bufferedInputStream);
+ schemaRegionParticles.clear();
+ for (int i = 0; i < particleNum; i++) {
+
schemaRegionParticles.add(ReadWriteIOUtils.readString(bufferedInputStream));
+ }
+
+ particleNum = ReadWriteIOUtils.readInt(bufferedInputStream);
+ dataRegionParticles.clear();
+ for (int i = 0; i < particleNum; i++) {
+
dataRegionParticles.add(ReadWriteIOUtils.readString(bufferedInputStream));
+ }
+ }
+ }
+
+ // Load snapshot of MTree
mTree.clear();
mTree.deserialize(bufferedInputStream);
} finally {
@@ -540,6 +654,16 @@ public class ClusterSchemaInfo implements
SnapshotProcessor {
return matchedNamesInNextLevel;
}
+ @TestOnly
+ public Set<String> getSchemaRegionParticles() {
+ return schemaRegionParticles;
+ }
+
+ @TestOnly
+ public Set<String> getDataRegionParticles() {
+ return dataRegionParticles;
+ }
+
@TestOnly
public void clear() {
mTree.clear();
diff --git
a/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
b/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
index 5a457a3da9..03de05999e 100644
---
a/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
+++
b/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
@@ -179,8 +179,8 @@ public class ConfigNodeRPCServiceProcessor implements
ConfigIService.Iface {
}
// Mark the StorageGroup as SchemaRegions and DataRegions not yet created
- storageGroupSchema.setMaximumSchemaRegionCount(-1);
- storageGroupSchema.setMaximumDataRegionCount(-1);
+ storageGroupSchema.setMaximumSchemaRegionCount(0);
+ storageGroupSchema.setMaximumDataRegionCount(0);
// Initialize RegionGroupId List
storageGroupSchema.setSchemaRegionGroupIds(new ArrayList<>());
diff --git
a/confignode/src/test/java/org/apache/iotdb/confignode/persistence/ClusterSchemaInfoTest.java
b/confignode/src/test/java/org/apache/iotdb/confignode/persistence/ClusterSchemaInfoTest.java
index a2eb84a6ec..514a38664c 100644
---
a/confignode/src/test/java/org/apache/iotdb/confignode/persistence/ClusterSchemaInfoTest.java
+++
b/confignode/src/test/java/org/apache/iotdb/confignode/persistence/ClusterSchemaInfoTest.java
@@ -47,7 +47,7 @@ public class ClusterSchemaInfoTest {
private static final File snapshotDir = new File(BASE_OUTPUT_PATH,
"snapshot");
@BeforeClass
- public static void setup() {
+ public static void setup() throws IOException {
clusterSchemaInfo = new ClusterSchemaInfo();
if (!snapshotDir.exists()) {
snapshotDir.mkdirs();
@@ -95,5 +95,17 @@ public class ClusterSchemaInfoTest {
Map<String, TStorageGroupSchema> reloadResult =
clusterSchemaInfo.getMatchedStorageGroupSchemas(getStorageGroupReq).getSchemaMap();
Assert.assertEquals(testMap, reloadResult);
+
+ Assert.assertEquals(
+ clusterSchemaInfo.getSchemaRegionParticles().size(),
storageGroupPathList.size());
+ storageGroupPathList.forEach(
+ storageGroup ->
+
Assert.assertTrue(clusterSchemaInfo.getSchemaRegionParticles().contains(storageGroup)));
+
+ Assert.assertEquals(
+ clusterSchemaInfo.getDataRegionParticles().size(),
storageGroupPathList.size());
+ storageGroupPathList.forEach(
+ storageGroup ->
+
Assert.assertTrue(clusterSchemaInfo.getDataRegionParticles().contains(storageGroup)));
}
}
