You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ca...@apache.org on 2023/01/18 09:24:54 UTC
[iotdb] branch master updated: [IOTDB-5287] Added status “Discouraged” to RegionGroup (#8901)
This is an automated email from the ASF dual-hosted git repository.
caogaofei 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 030302c965 [IOTDB-5287] Added status “Discouraged” to RegionGroup (#8901)
030302c965 is described below
commit 030302c96582fe4467a138a3d9e0efdbd2ffb13f
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Wed Jan 18 17:24:45 2023 +0800
[IOTDB-5287] Added status “Discouraged” to RegionGroup (#8901)
---
.../iotdb/confignode/conf/ConfigNodeConfig.java | 11 +
.../confignode/conf/ConfigNodeDescriptor.java | 9 +-
.../confignode/manager/ClusterSchemaManager.java | 28 +-
.../iotdb/confignode/manager/ProcedureManager.java | 11 +
.../manager/partition/PartitionManager.java | 20 +-
.../manager/partition/RegionGroupStatus.java | 11 +-
.../partition/heartbeat/RegionGroupCache.java | 17 +-
.../statemachine/CreateRegionGroupsProcedure.java | 2 +-
.../manager/partition/RegionGroupCacheTest.java | 2 +-
.../partition/IoTDBAutoRegionGroupExtensionIT.java | 3 +-
...urableIT.java => IoTDBPartitionCreationIT.java} | 608 +++++++++++----------
.../it/partition/IoTDBPartitionDurableIT.java | 73 +--
12 files changed, 434 insertions(+), 361 deletions(-)
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConfig.java b/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConfig.java
index 552b052ba4..55ff24d54d 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConfig.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConfig.java
@@ -101,6 +101,9 @@ public class ConfigNodeConfig {
/** The least number of DataRegionGroup for each Database. */
private volatile int leastDataRegionGroupNum = 5;
+ /** Indicate whether the leastDataRegionGroupNum is set by user, if true, lock it. */
+ private volatile boolean leastDataRegionGroupNumSetByUser = false;
+
/** RegionGroup allocate policy. */
private RegionBalancer.RegionGroupAllocatePolicy regionGroupAllocatePolicy =
RegionBalancer.RegionGroupAllocatePolicy.GREEDY;
@@ -540,6 +543,14 @@ public class ConfigNodeConfig {
this.leastDataRegionGroupNum = leastDataRegionGroupNum;
}
+ public boolean isLeastDataRegionGroupNumSetByUser() {
+ return leastDataRegionGroupNumSetByUser;
+ }
+
+ public void setLeastDataRegionGroupNumSetByUser(boolean leastDataRegionGroupNumSetByUser) {
+ this.leastDataRegionGroupNumSetByUser = leastDataRegionGroupNumSetByUser;
+ }
+
public RegionBalancer.RegionGroupAllocatePolicy getRegionGroupAllocatePolicy() {
return regionGroupAllocatePolicy;
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java b/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java
index 4c1873258a..787bc84cbc 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java
@@ -254,10 +254,11 @@ public class ConfigNodeDescriptor {
"data_region_per_processor", String.valueOf(conf.getDataRegionPerProcessor()))
.trim()));
- conf.setLeastDataRegionGroupNum(
- Integer.parseInt(
- properties.getProperty(
- "least_data_region_group_num", String.valueOf(conf.getLeastDataRegionGroupNum()))));
+ if (properties.getProperty("least_data_region_group_num") != null) {
+ conf.setLeastDataRegionGroupNum(
+ Integer.parseInt(properties.getProperty("least_data_region_group_num")));
+ conf.setLeastDataRegionGroupNumSetByUser(true);
+ }
try {
conf.setRegionAllocateStrategy(
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 6259a753f4..7e6a7e57e0 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
@@ -311,7 +311,7 @@ public class ClusterSchemaManager {
/**
* Only leader use this interface. Adjust the maxSchemaRegionGroupNum and maxDataRegionGroupNum of
- * each StorageGroup bases on existing cluster resources
+ * each StorageGroup based on existing cluster resources
*/
public synchronized void adjustMaxRegionGroupNum() {
// Get all StorageGroupSchemas
@@ -329,18 +329,20 @@ public class ClusterSchemaManager {
// Adjust least_data_region_group_num
// TODO: The least_data_region_group_num should be maintained separately by different
// TODO: StorageGroup
- int leastDataRegionGroupNum =
- (int)
- Math.ceil(
- (double) totalCpuCoreNum
- / (double) (storageGroupNum * CONF.getDataReplicationFactor()));
- if (leastDataRegionGroupNum < CONF.getLeastDataRegionGroupNum()) {
- // The leastDataRegionGroupNum should be the maximum integer that satisfy:
- // 1 <= leastDataRegionGroupNum <= 5(default)
- CONF.setLeastDataRegionGroupNum(leastDataRegionGroupNum);
- LOGGER.info(
- "[AdjustRegionGroupNum] The least number of DataRegionGroups per Database is adjusted to: {}",
- leastDataRegionGroupNum);
+ if (!CONF.isLeastDataRegionGroupNumSetByUser()) {
+ int leastDataRegionGroupNum =
+ (int)
+ Math.ceil(
+ (double) totalCpuCoreNum
+ / (double) (storageGroupNum * CONF.getDataReplicationFactor()));
+ if (leastDataRegionGroupNum < CONF.getLeastDataRegionGroupNum()) {
+ // The leastDataRegionGroupNum should be the maximum integer that satisfy:
+ // 1 <= leastDataRegionGroupNum <= 5(default)
+ CONF.setLeastDataRegionGroupNum(leastDataRegionGroupNum);
+ LOGGER.info(
+ "[AdjustRegionGroupNum] The least number of DataRegionGroups per Database is adjusted to: {}",
+ leastDataRegionGroupNum);
+ }
}
AdjustMaxRegionGroupNumPlan adjustMaxRegionGroupNumPlan = new AdjustMaxRegionGroupNumPlan();
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java
index e92c5eb51f..c7f0e6c4c0 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java
@@ -420,6 +420,17 @@ public class ProcedureManager {
.map(TDataNodeConfiguration::getLocation)
.map(TDataNodeLocation::getDataNodeId)
.collect(Collectors.toSet());
+ if (!aliveDataNodes.contains(migrateRegionReq.getFromId())) {
+ LOGGER.warn(
+ "Submit RegionMigrateProcedure failed, because the sourceDataNode {} is Unknown.",
+ migrateRegionReq.getFromId());
+ TSStatus status = new TSStatus(TSStatusCode.MIGRATE_REGION_ERROR.getStatusCode());
+ status.setMessage(
+ "Submit RegionMigrateProcedure failed, because the sourceDataNode "
+ + migrateRegionReq.getFromId()
+ + " is Unknown.");
+ return status;
+ }
dataNodesInRegion.retainAll(aliveDataNodes);
if (dataNodesInRegion.isEmpty()) {
LOGGER.warn(
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java
index 8a0072971d..fafb478be7 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java
@@ -587,9 +587,8 @@ public class PartitionManager {
// Filter RegionGroups that have Disabled status
List<Pair<Long, TConsensusGroupId>> result = new ArrayList<>();
for (Pair<Long, TConsensusGroupId> slotsCounter : regionGroupSlotsCounter) {
- // Use Running or Available RegionGroups
RegionGroupStatus status = getRegionGroupStatus(slotsCounter.getRight());
- if (RegionGroupStatus.Running.equals(status) || RegionGroupStatus.Available.equals(status)) {
+ if (!RegionGroupStatus.Disabled.equals(status)) {
result.add(slotsCounter);
}
}
@@ -598,10 +597,25 @@ public class PartitionManager {
throw new NoAvailableRegionGroupException(type);
}
- result.sort(Comparator.comparingLong(Pair::getLeft));
+ result.sort(new PartitionComparator());
return result;
}
+ class PartitionComparator implements Comparator<Pair<Long, TConsensusGroupId>> {
+
+ @Override
+ public int compare(Pair<Long, TConsensusGroupId> o1, Pair<Long, TConsensusGroupId> o2) {
+ // Use partition number as first priority
+ if (o1.getLeft() < o2.getLeft()) {
+ return -1;
+ } else if (o1.getLeft() > o2.getLeft()) {
+ return 1;
+ } else {
+ // Use RegionGroup status as second priority, Running > Available > Discouraged
+ return getRegionGroupStatus(o1.getRight()).compareTo(getRegionGroupStatus(o2.getRight()));
+ }
+ }
+ }
/**
* Only leader use this interface
*
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/RegionGroupStatus.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/RegionGroupStatus.java
index 73b8f55640..9e58ae6d1c 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/RegionGroupStatus.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/RegionGroupStatus.java
@@ -29,12 +29,19 @@ public enum RegionGroupStatus {
*/
Available("Available"),
+ /**
+ * All Regions in RegionGroup are in the Running, Unknown or ReadOnly status, and at least 1 node
+ * is in ReadOnly status, the number of Regions in the Unknown or ReadOnly status is less than
+ * half
+ */
+ Discouraged("Discouraged"),
+
/**
* The following cases will lead to Disabled RegionGroup:
*
- * <p>1. There is a Region in ReadOnly or Removing status
+ * <p>1. There is a Region in Removing status
*
- * <p>2. More than half of the Regions are in Unknown status
+ * <p>2. More than half of the Regions are in Unknown or ReadOnly status
*/
Disabled("Disabled");
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/heartbeat/RegionGroupCache.java b/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/heartbeat/RegionGroupCache.java
index 7c7ddfdb03..7e490670a9 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/heartbeat/RegionGroupCache.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/heartbeat/RegionGroupCache.java
@@ -123,27 +123,36 @@ public class RegionGroupCache {
private RegionGroupStatus updateRegionGroupStatus(
Map<Integer, RegionStatistics> regionStatisticsMap) {
int unknownCount = 0;
+ int readonlyCount = 0;
for (RegionStatistics regionStatistics : regionStatisticsMap.values()) {
- if (RegionStatus.ReadOnly.equals(regionStatistics.getRegionStatus())
- || RegionStatus.Removing.equals(regionStatistics.getRegionStatus())) {
+ if (RegionStatus.Removing.equals(regionStatistics.getRegionStatus())) {
// The RegionGroup is considered as Disabled when
// at least one Region is in the ReadOnly or Removing status
return RegionGroupStatus.Disabled;
}
unknownCount += RegionStatus.Unknown.equals(regionStatistics.getRegionStatus()) ? 1 : 0;
+ readonlyCount += RegionStatus.ReadOnly.equals(regionStatistics.getRegionStatus()) ? 1 : 0;
}
- if (unknownCount == 0) {
+ if (unknownCount + readonlyCount == 0) {
// The RegionGroup is considered as Running only if
// all Regions are in the Running status
return RegionGroupStatus.Running;
- } else {
+ } else if (readonlyCount == 0) {
return unknownCount <= ((regionCacheMap.size() - 1) / 2)
// The RegionGroup is considered as Available when the number of Unknown Regions is less
// than half
? RegionGroupStatus.Available
// Disabled otherwise
: RegionGroupStatus.Disabled;
+ } else {
+ return unknownCount + readonlyCount <= ((regionCacheMap.size() - 1) / 2)
+ // The RegionGroup is considered as Discouraged when the number of Unknown or ReadOnly
+ // Regions is less
+ // than half, and there are at least 1 ReadOnly Region
+ ? RegionGroupStatus.Discouraged
+ // Disabled otherwise
+ : RegionGroupStatus.Disabled;
}
}
diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/statemachine/CreateRegionGroupsProcedure.java b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/statemachine/CreateRegionGroupsProcedure.java
index 999d6895cc..357ac703ed 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/statemachine/CreateRegionGroupsProcedure.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/statemachine/CreateRegionGroupsProcedure.java
@@ -108,7 +108,7 @@ public class CreateRegionGroupsProcedure
if (failedRegionReplicas.getDataNodeLocationsSize()
<= (regionReplicaSet.getDataNodeLocationsSize() - 1) / 2) {
// A RegionGroup can provide service as long as there are more than
- // half of the RegionReplicas were created successfully
+ // half of the RegionReplicas created successfully
persistPlan.addRegionGroup(storageGroup, regionReplicaSet);
// Build recreate tasks
diff --git a/confignode/src/test/java/org/apache/iotdb/confignode/manager/partition/RegionGroupCacheTest.java b/confignode/src/test/java/org/apache/iotdb/confignode/manager/partition/RegionGroupCacheTest.java
index 69c228e14b..3c3cf4881b 100644
--- a/confignode/src/test/java/org/apache/iotdb/confignode/manager/partition/RegionGroupCacheTest.java
+++ b/confignode/src/test/java/org/apache/iotdb/confignode/manager/partition/RegionGroupCacheTest.java
@@ -90,7 +90,7 @@ public class RegionGroupCacheTest {
2, new RegionHeartbeatSample(currentTime, currentTime, RegionStatus.Running));
Assert.assertTrue(disabledRegionGroup0.periodicUpdate());
Assert.assertEquals(
- RegionGroupStatus.Disabled, disabledRegionGroup0.getStatistics().getRegionGroupStatus());
+ RegionGroupStatus.Discouraged, disabledRegionGroup0.getStatistics().getRegionGroupStatus());
RegionGroupCache disabledRegionGroup1 =
new RegionGroupCache(new TConsensusGroupId(TConsensusGroupType.SchemaRegion, 3));
diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBAutoRegionGroupExtensionIT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBAutoRegionGroupExtensionIT.java
index d8c38cad15..b7e3c88613 100644
--- a/integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBAutoRegionGroupExtensionIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBAutoRegionGroupExtensionIT.java
@@ -73,8 +73,7 @@ public class IoTDBAutoRegionGroupExtensionIT {
.setSchemaReplicationFactor(testReplicationFactor)
.setDataReplicationFactor(testReplicationFactor)
.setDataRegionGroupExtensionPolicy(testDataRegionGroupExtensionPolicy)
- .setTimePartitionInterval(testTimePartitionInterval)
- .setLeastDataRegionGroupNum(testLeastDataRegionGroupNum);
+ .setTimePartitionInterval(testTimePartitionInterval);
// Init 1C3D environment
EnvFactory.getEnv().initClusterEnvironment(1, 3);
}
diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBPartitionDurableIT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBPartitionCreationIT.java
similarity index 53%
copy from integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBPartitionDurableIT.java
copy to integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBPartitionCreationIT.java
index 0fec7296ef..193191d657 100644
--- a/integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBPartitionDurableIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBPartitionCreationIT.java
@@ -18,7 +18,6 @@
*/
package org.apache.iotdb.confignode.it.partition;
-import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
@@ -27,15 +26,11 @@ import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
import org.apache.iotdb.commons.cluster.NodeStatus;
import org.apache.iotdb.commons.cluster.RegionStatus;
import org.apache.iotdb.confignode.it.utils.ConfigNodeTestUtils;
-import org.apache.iotdb.confignode.rpc.thrift.TDataNodeInfo;
import org.apache.iotdb.confignode.rpc.thrift.TDataPartitionReq;
import org.apache.iotdb.confignode.rpc.thrift.TDataPartitionTableResp;
import org.apache.iotdb.confignode.rpc.thrift.TRegionInfo;
-import org.apache.iotdb.confignode.rpc.thrift.TSchemaPartitionReq;
-import org.apache.iotdb.confignode.rpc.thrift.TSchemaPartitionTableResp;
import org.apache.iotdb.confignode.rpc.thrift.TSetDataNodeStatusReq;
import org.apache.iotdb.confignode.rpc.thrift.TSetStorageGroupReq;
-import org.apache.iotdb.confignode.rpc.thrift.TShowClusterResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowDataNodesResp;
import org.apache.iotdb.confignode.rpc.thrift.TShowRegionReq;
import org.apache.iotdb.confignode.rpc.thrift.TShowRegionResp;
@@ -57,29 +52,23 @@ import org.junit.runner.RunWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.Collections;
-import java.util.HashSet;
import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
@RunWith(IoTDBTestRunner.class)
@Category({ClusterIT.class})
-public class IoTDBPartitionDurableIT {
+public class IoTDBPartitionCreationIT {
private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBPartitionDurableIT.class);
private static final String testConsensusProtocolClass = ConsensusFactory.RATIS_CONSENSUS;
+ private static final String IoTConsensusProtocolClass = ConsensusFactory.IOT_CONSENSUS;
private static final int testReplicationFactor = 3;
private static final long testTimePartitionInterval = 604800000;
- private static final int testDataNodeId = 0;
private static final String sg = "root.sg";
- final String d0 = sg + ".d0.s";
- final String d1 = sg + ".d1.s";
private static final int testSeriesPartitionBatchSize = 1;
private static final int testTimePartitionBatchSize = 1;
+ private static final int testLeastDataRegionGroupNum = 4;
private static final TEndPoint defaultEndPoint = new TEndPoint("-1", -1);
private static final TDataNodeLocation defaultDataNode =
new TDataNodeLocation(
@@ -97,10 +86,11 @@ public class IoTDBPartitionDurableIT {
.getCommonConfig()
.setConfigNodeConsensusProtocolClass(testConsensusProtocolClass)
.setSchemaRegionConsensusProtocolClass(testConsensusProtocolClass)
- .setDataRegionConsensusProtocolClass(testConsensusProtocolClass)
+ .setDataRegionConsensusProtocolClass(IoTConsensusProtocolClass)
.setSchemaReplicationFactor(testReplicationFactor)
.setDataReplicationFactor(testReplicationFactor)
- .setTimePartitionInterval(testTimePartitionInterval);
+ .setTimePartitionInterval(testTimePartitionInterval)
+ .setLeastDataRegionGroupNum(testLeastDataRegionGroupNum);
// Init 1C3D environment
EnvFactory.getEnv().initClusterEnvironment(1, 3);
@@ -123,41 +113,51 @@ public class IoTDBPartitionDurableIT {
}
@Test
- public void testRemovingDataNode() throws Exception {
+ public void testPartitionAllocation() throws Exception {
try (SyncConfigNodeIServiceClient client =
(SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
-
- /* Test getOrCreateSchemaPartition, ConfigNode should create SchemaPartition and return */
- TSchemaPartitionReq schemaPartitionReq =
- new TSchemaPartitionReq()
- .setPathPatternTree(ConfigNodeTestUtils.generatePatternTreeBuffer(new String[] {d0}));
- TSchemaPartitionTableResp schemaPartitionTableResp =
- client.getOrCreateSchemaPartitionTable(schemaPartitionReq);
+ // Status: Running, Running, Running, Region: [0], [0], [0]
+ Map<String, Map<TSeriesPartitionSlot, TTimeSlotList>> partitionSlotsMap =
+ ConfigNodeTestUtils.constructPartitionSlotsMap(
+ sg,
+ 0,
+ testSeriesPartitionBatchSize,
+ 0,
+ testTimePartitionBatchSize,
+ testTimePartitionInterval);
+ TDataPartitionReq dataPartitionReq = new TDataPartitionReq(partitionSlotsMap);
+ TDataPartitionTableResp dataPartitionTableResp = null;
+ for (int retry = 0; retry < 5; retry++) {
+ // Build new Client since it's unstable in Win8 environment
+ try (SyncConfigNodeIServiceClient configNodeClient =
+ (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
+ dataPartitionTableResp = configNodeClient.getOrCreateDataPartitionTable(dataPartitionReq);
+ if (dataPartitionTableResp != null) {
+ break;
+ }
+ } catch (Exception e) {
+ // Retry sometimes in order to avoid request timeout
+ LOGGER.error(e.getMessage());
+ TimeUnit.SECONDS.sleep(1);
+ }
+ }
+ Assert.assertNotNull(dataPartitionTableResp);
Assert.assertEquals(
TSStatusCode.SUCCESS_STATUS.getStatusCode(),
- schemaPartitionTableResp.getStatus().getCode());
- Map<String, Map<TSeriesPartitionSlot, TConsensusGroupId>> schemaPartitionTable =
- schemaPartitionTableResp.getSchemaPartitionTable();
- // Successfully create a SchemaPartition
- Assert.assertTrue(schemaPartitionTable.containsKey(sg));
- Assert.assertEquals(1, schemaPartitionTable.get(sg).size());
-
- /* Check Region distribution */
- TShowRegionResp showRegionResp = client.showRegion(new TShowRegionReq());
- Assert.assertEquals(
- TSStatusCode.SUCCESS_STATUS.getStatusCode(), showRegionResp.getStatus().getCode());
- // Create exactly one RegionGroup
- Assert.assertEquals(3, showRegionResp.getRegionInfoListSize());
- // Each DataNode has exactly one Region
- Set<Integer> dataNodeIdSet = new HashSet<>();
- showRegionResp
- .getRegionInfoList()
- .forEach(regionInfo -> dataNodeIdSet.add(regionInfo.getDataNodeId()));
- Assert.assertEquals(3, dataNodeIdSet.size());
+ dataPartitionTableResp.getStatus().getCode());
+ Assert.assertNotNull(dataPartitionTableResp.getDataPartitionTable());
+ ConfigNodeTestUtils.checkDataPartitionTable(
+ sg,
+ 0,
+ testSeriesPartitionBatchSize,
+ 0,
+ testTimePartitionBatchSize,
+ testTimePartitionInterval,
+ dataPartitionTableResp.getDataPartitionTable());
- /* Change the NodeStatus of the test DataNode to Removing */
+ // Status: Running, Running, Removing, Region: [0], [0], [0]
TSetDataNodeStatusReq setDataNodeStatusReq = new TSetDataNodeStatusReq();
- DataNodeWrapper dataNodeWrapper = EnvFactory.getEnv().getDataNodeWrapper(testDataNodeId);
+ DataNodeWrapper dataNodeWrapper = EnvFactory.getEnv().getDataNodeWrapper(2);
setDataNodeStatusReq.setTargetDataNode(
new TDataNodeLocation(defaultDataNode)
.setInternalEndPoint(
@@ -185,114 +185,49 @@ public class IoTDBPartitionDurableIT {
TimeUnit.SECONDS.sleep(1);
}
- /* Test getOrCreateSchemaPartition, the result should be NO_ENOUGH_DATANODE */
- schemaPartitionReq =
- new TSchemaPartitionReq()
- .setPathPatternTree(ConfigNodeTestUtils.generatePatternTreeBuffer(new String[] {d1}));
- schemaPartitionTableResp = client.getOrCreateSchemaPartitionTable(schemaPartitionReq);
- Assert.assertEquals(
- TSStatusCode.NO_ENOUGH_DATANODE.getStatusCode(),
- schemaPartitionTableResp.getStatus().getCode());
-
- /* Register a new DataNode */
+ // Status: Running, Running, Removing, Running, RegionGroup: [0, 1], [0, 1], [0], [1]
EnvFactory.getEnv().registerNewDataNode(true);
-
- /* Test getOrCreateSchemaPartition, ConfigNode should create SchemaPartition and return */
- schemaPartitionReq =
- new TSchemaPartitionReq()
- .setPathPatternTree(ConfigNodeTestUtils.generatePatternTreeBuffer(new String[] {d1}));
- schemaPartitionTableResp = client.getOrCreateSchemaPartitionTable(schemaPartitionReq);
- Assert.assertEquals(
- TSStatusCode.SUCCESS_STATUS.getStatusCode(),
- schemaPartitionTableResp.getStatus().getCode());
- schemaPartitionTable = schemaPartitionTableResp.getSchemaPartitionTable();
- // Successfully create a SchemaPartition
- Assert.assertTrue(schemaPartitionTable.containsKey(sg));
- Assert.assertEquals(1, schemaPartitionTable.get(sg).size());
-
- /* Check Region distribution */
- showRegionResp = client.showRegion(new TShowRegionReq());
- Assert.assertEquals(
- TSStatusCode.SUCCESS_STATUS.getStatusCode(), showRegionResp.getStatus().getCode());
- // There should be 2 RegionGroups
- Assert.assertEquals(6, showRegionResp.getRegionInfoListSize());
- // The new RegionGroup should keep away from the Removing DataNode
- Map<Integer, AtomicInteger> regionCounter = new ConcurrentHashMap<>();
- showRegionResp
- .getRegionInfoList()
- .forEach(
- regionInfo ->
- regionCounter
- .computeIfAbsent(regionInfo.getDataNodeId(), empty -> new AtomicInteger(0))
- .getAndIncrement());
- dataNodeIdSet.forEach(dataNodeId -> regionCounter.get(dataNodeId).getAndDecrement());
- TShowDataNodesResp showDataNodesResp = client.showDataNodes();
- Assert.assertEquals(
- TSStatusCode.SUCCESS_STATUS.getStatusCode(), showDataNodesResp.getStatus().getCode());
- regionCounter.forEach(
- (dataNodeId, regionCount) -> {
- String nodeStatus =
- showDataNodesResp.getDataNodesInfoList().stream()
- .filter(dataNodeInfo -> dataNodeInfo.getDataNodeId() == dataNodeId)
- .findFirst()
- .orElse(new TDataNodeInfo().setStatus("ERROR"))
- .getStatus();
- if (NodeStatus.Removing.getStatus().equals(nodeStatus)) {
- Assert.assertEquals(0, regionCount.get());
- } else if (NodeStatus.Running.getStatus().equals(nodeStatus)) {
- Assert.assertEquals(1, regionCount.get());
- } else {
- Assert.fail();
- }
- });
- }
- }
-
- @Test
- public void testReadOnlyDataNode() throws Exception {
- try (SyncConfigNodeIServiceClient client =
- (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
-
- /* Test getOrCreateDataPartition, ConfigNode should create DataPartition and return */
- Map<String, Map<TSeriesPartitionSlot, TTimeSlotList>> partitionSlotsMap =
+ partitionSlotsMap =
ConfigNodeTestUtils.constructPartitionSlotsMap(
sg,
- 0,
- testSeriesPartitionBatchSize,
- 0,
- testTimePartitionBatchSize,
+ 1,
+ 1 + testSeriesPartitionBatchSize,
+ 1,
+ 1 + testTimePartitionBatchSize,
testTimePartitionInterval);
- TDataPartitionReq dataPartitionReq = new TDataPartitionReq(partitionSlotsMap);
- TDataPartitionTableResp dataPartitionTableResp =
- client.getOrCreateDataPartitionTable(dataPartitionReq);
+ dataPartitionReq = new TDataPartitionReq(partitionSlotsMap);
+ for (int retry = 0; retry < 5; retry++) {
+ // Build new Client since it's unstable in Win8 environment
+ try (SyncConfigNodeIServiceClient configNodeClient =
+ (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
+ dataPartitionTableResp = configNodeClient.getOrCreateDataPartitionTable(dataPartitionReq);
+ if (dataPartitionTableResp != null) {
+ break;
+ }
+ } catch (Exception e) {
+ // Retry sometimes in order to avoid request timeout
+ LOGGER.error(e.getMessage());
+ TimeUnit.SECONDS.sleep(1);
+ }
+ }
+ Assert.assertNotNull(dataPartitionTableResp);
Assert.assertEquals(
TSStatusCode.SUCCESS_STATUS.getStatusCode(),
dataPartitionTableResp.getStatus().getCode());
+ Assert.assertNotNull(dataPartitionTableResp.getDataPartitionTable());
ConfigNodeTestUtils.checkDataPartitionTable(
sg,
- 0,
- testSeriesPartitionBatchSize,
- 0,
- testTimePartitionBatchSize,
+ 1,
+ 1 + testSeriesPartitionBatchSize,
+ 1,
+ 1 + testTimePartitionBatchSize,
testTimePartitionInterval,
dataPartitionTableResp.getDataPartitionTable());
- /* Check Region distribution */
- TShowRegionResp showRegionResp = client.showRegion(new TShowRegionReq());
- Assert.assertEquals(
- TSStatusCode.SUCCESS_STATUS.getStatusCode(), showRegionResp.getStatus().getCode());
- // Create exactly one RegionGroup
- Assert.assertEquals(3, showRegionResp.getRegionInfoListSize());
- // Each DataNode has exactly one Region
- Set<Integer> dataNodeIdSet = new HashSet<>();
- showRegionResp
- .getRegionInfoList()
- .forEach(regionInfo -> dataNodeIdSet.add(regionInfo.getDataNodeId()));
- Assert.assertEquals(3, dataNodeIdSet.size());
-
- /* Change the NodeStatus of the test DataNode to ReadOnly */
- TSetDataNodeStatusReq setDataNodeStatusReq = new TSetDataNodeStatusReq();
- DataNodeWrapper dataNodeWrapper = EnvFactory.getEnv().getDataNodeWrapper(testDataNodeId);
+ // Status: Running, Running, Removing, ReadOnly, RegionGroup: [0, 1], [0, 1],
+ // [0], [1]
+ setDataNodeStatusReq = new TSetDataNodeStatusReq();
+ dataNodeWrapper = EnvFactory.getEnv().getDataNodeWrapper(3);
setDataNodeStatusReq.setTargetDataNode(
new TDataNodeLocation(defaultDataNode)
.setInternalEndPoint(
@@ -320,107 +255,145 @@ public class IoTDBPartitionDurableIT {
TimeUnit.SECONDS.sleep(1);
}
- /* Test getOrCreateDataPartition, the result should be NO_ENOUGH_DATANODE */
+ // Status: Running, Running, Removing, ReadOnly, Running, RegionGroup: [0, 1, 2], [0, 1, 2],
+ // [0], [1], [2]
+ EnvFactory.getEnv().registerNewDataNode(true);
partitionSlotsMap =
ConfigNodeTestUtils.constructPartitionSlotsMap(
sg,
- 1,
- 1 + testSeriesPartitionBatchSize,
- 1,
- 1 + testTimePartitionBatchSize,
+ 2,
+ 2 + testSeriesPartitionBatchSize,
+ 2,
+ 2 + testTimePartitionBatchSize,
testTimePartitionInterval);
dataPartitionReq = new TDataPartitionReq(partitionSlotsMap);
- dataPartitionTableResp = client.getOrCreateDataPartitionTable(dataPartitionReq);
+ for (int retry = 0; retry < 5; retry++) {
+ // Build new Client since it's unstable in Win8 environment
+ try (SyncConfigNodeIServiceClient configNodeClient =
+ (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
+ dataPartitionTableResp = configNodeClient.getOrCreateDataPartitionTable(dataPartitionReq);
+ if (dataPartitionTableResp != null) {
+ break;
+ }
+ } catch (Exception e) {
+ // Retry sometimes in order to avoid request timeout
+ LOGGER.error(e.getMessage());
+ TimeUnit.SECONDS.sleep(1);
+ }
+ }
+ Assert.assertNotNull(dataPartitionTableResp);
Assert.assertEquals(
- TSStatusCode.NO_ENOUGH_DATANODE.getStatusCode(),
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
dataPartitionTableResp.getStatus().getCode());
+ Assert.assertNotNull(dataPartitionTableResp.getDataPartitionTable());
+ ConfigNodeTestUtils.checkDataPartitionTable(
+ sg,
+ 2,
+ 2 + testSeriesPartitionBatchSize,
+ 2,
+ 2 + testTimePartitionBatchSize,
+ testTimePartitionInterval,
+ dataPartitionTableResp.getDataPartitionTable());
- /* Register a new DataNode */
- EnvFactory.getEnv().registerNewDataNode(true);
+ // Status: Running, Running, Removing, ReadOnly, Unknown, RegionGroup:[0, 1, 2], [0, 1, 2],
+ // [0], [1], [2]
+ EnvFactory.getEnv().shutdownDataNode(4);
+ // Wait for shutdown check
+ while (true) {
+ AtomicBoolean containUnknown = new AtomicBoolean(false);
+ TShowDataNodesResp showDataNodesResp = client.showDataNodes();
+ showDataNodesResp
+ .getDataNodesInfoList()
+ .forEach(
+ dataNodeInfo -> {
+ if (NodeStatus.Unknown.getStatus().equals(dataNodeInfo.getStatus())) {
+ containUnknown.set(true);
+ }
+ });
+
+ if (containUnknown.get()) {
+ break;
+ }
+ TimeUnit.SECONDS.sleep(1);
+ }
+
+ // Status: Running, Running, Removing, ReadOnly, Unknown, Running,
+ // RegionGroup: [0, 1, 2, 3], [0, 1, 2, 3], [0], [1], [2], [3]
+ EnvFactory.getEnv().registerNewDataNode(false);
- /* Test getOrCreateDataPartition, ConfigNode should create DataPartition and return */
+ // Use thread sleep to replace verifying because the Unknown DataNode can not pass the
+ // connection check
+ TimeUnit.SECONDS.sleep(25);
partitionSlotsMap =
ConfigNodeTestUtils.constructPartitionSlotsMap(
sg,
- 1,
- 1 + testSeriesPartitionBatchSize,
- 1,
- 1 + testTimePartitionBatchSize,
+ 3,
+ 3 + testSeriesPartitionBatchSize,
+ 3,
+ 3 + testTimePartitionBatchSize,
testTimePartitionInterval);
dataPartitionReq = new TDataPartitionReq(partitionSlotsMap);
- dataPartitionTableResp = client.getOrCreateDataPartitionTable(dataPartitionReq);
+ for (int retry = 0; retry < 5; retry++) {
+ // Build new Client since it's unstable in Win8 environment
+ try (SyncConfigNodeIServiceClient configNodeClient =
+ (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
+ dataPartitionTableResp = configNodeClient.getOrCreateDataPartitionTable(dataPartitionReq);
+ if (dataPartitionTableResp != null) {
+ break;
+ }
+ } catch (Exception e) {
+ // Retry sometimes in order to avoid request timeout
+ LOGGER.error(e.getMessage());
+ TimeUnit.SECONDS.sleep(1);
+ }
+ }
+ Assert.assertNotNull(dataPartitionTableResp);
Assert.assertEquals(
TSStatusCode.SUCCESS_STATUS.getStatusCode(),
dataPartitionTableResp.getStatus().getCode());
+ Assert.assertNotNull(dataPartitionTableResp.getDataPartitionTable());
ConfigNodeTestUtils.checkDataPartitionTable(
sg,
- 1,
- 1 + testSeriesPartitionBatchSize,
- 1,
- 1 + testTimePartitionBatchSize,
+ 3,
+ 3 + testSeriesPartitionBatchSize,
+ 3,
+ 3 + testTimePartitionBatchSize,
testTimePartitionInterval,
dataPartitionTableResp.getDataPartitionTable());
- /* Check Region distribution */
- showRegionResp = client.showRegion(new TShowRegionReq());
+ // Check Region count and status
+ int runningCnt = 0;
+ int unknownCnt = 0;
+ int readOnlyCnt = 0;
+ int removingCnt = 0;
+ TShowRegionResp showRegionResp = client.showRegion(new TShowRegionReq());
Assert.assertEquals(
TSStatusCode.SUCCESS_STATUS.getStatusCode(), showRegionResp.getStatus().getCode());
- // There should be 2 RegionGroups
- Assert.assertEquals(6, showRegionResp.getRegionInfoListSize());
- // The new RegionGroup should keep away from the ReadOnly DataNode
- Map<Integer, AtomicInteger> regionCounter = new ConcurrentHashMap<>();
- showRegionResp
- .getRegionInfoList()
- .forEach(
- regionInfo ->
- regionCounter
- .computeIfAbsent(regionInfo.getDataNodeId(), empty -> new AtomicInteger(0))
- .getAndIncrement());
- dataNodeIdSet.forEach(dataNodeId -> regionCounter.get(dataNodeId).getAndDecrement());
- TShowDataNodesResp showDataNodesResp = client.showDataNodes();
- Assert.assertEquals(
- TSStatusCode.SUCCESS_STATUS.getStatusCode(), showDataNodesResp.getStatus().getCode());
- regionCounter.forEach(
- (dataNodeId, regionCount) -> {
- String nodeStatus =
- showDataNodesResp.getDataNodesInfoList().stream()
- .filter(dataNodeInfo -> dataNodeInfo.getDataNodeId() == dataNodeId)
- .findFirst()
- .orElse(new TDataNodeInfo().setStatus("ERROR"))
- .getStatus();
- if (NodeStatus.ReadOnly.getStatus().equals(nodeStatus)) {
- Assert.assertEquals(0, regionCount.get());
- } else if (NodeStatus.Running.getStatus().equals(nodeStatus)) {
- Assert.assertEquals(1, regionCount.get());
- } else {
- Assert.fail();
- }
- });
- }
- }
-
- @Test
- public void testUnknownDataNode() throws Exception {
- // Shutdown a DataNode, the ConfigNode should still be able to create RegionGroup
- EnvFactory.getEnv().shutdownDataNode(testDataNodeId);
- EnvFactory.getEnv()
- .ensureNodeStatus(
- Collections.singletonList(EnvFactory.getEnv().getDataNodeWrapper(testDataNodeId)),
- Collections.singletonList(NodeStatus.Unknown));
+ for (TRegionInfo regionInfo : showRegionResp.getRegionInfoList()) {
+ if (RegionStatus.Running.getStatus().equals(regionInfo.getStatus())) {
+ runningCnt += 1;
+ } else if (RegionStatus.Unknown.getStatus().equals(regionInfo.getStatus())) {
+ unknownCnt += 1;
+ } else if (RegionStatus.Removing.getStatus().equals(regionInfo.getStatus())) {
+ removingCnt += 1;
+ } else if (RegionStatus.ReadOnly.getStatus().equals(regionInfo.getStatus())) {
+ readOnlyCnt += 1;
+ }
+ }
+ Assert.assertEquals(9, runningCnt);
+ Assert.assertEquals(1, removingCnt);
+ Assert.assertEquals(1, readOnlyCnt);
+ Assert.assertEquals(1, unknownCnt);
- try (SyncConfigNodeIServiceClient client =
- (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
- // Test getOrCreateDataPartition, ConfigNode should create DataPartition and return
- Map<String, Map<TSeriesPartitionSlot, TTimeSlotList>> partitionSlotsMap =
+ partitionSlotsMap =
ConfigNodeTestUtils.constructPartitionSlotsMap(
sg,
- 0,
- testSeriesPartitionBatchSize,
- 0,
- testTimePartitionBatchSize,
+ 4,
+ 4 + testSeriesPartitionBatchSize,
+ 4,
+ 4 + testTimePartitionBatchSize,
testTimePartitionInterval);
- TDataPartitionReq dataPartitionReq = new TDataPartitionReq(partitionSlotsMap);
- TDataPartitionTableResp dataPartitionTableResp = null;
+ dataPartitionReq = new TDataPartitionReq(partitionSlotsMap);
for (int retry = 0; retry < 5; retry++) {
// Build new Client since it's unstable in Win8 environment
try (SyncConfigNodeIServiceClient configNodeClient =
@@ -442,72 +415,89 @@ public class IoTDBPartitionDurableIT {
Assert.assertNotNull(dataPartitionTableResp.getDataPartitionTable());
ConfigNodeTestUtils.checkDataPartitionTable(
sg,
- 0,
- testSeriesPartitionBatchSize,
- 0,
- testTimePartitionBatchSize,
+ 4,
+ 4 + testSeriesPartitionBatchSize,
+ 4,
+ 4 + testTimePartitionBatchSize,
testTimePartitionInterval,
dataPartitionTableResp.getDataPartitionTable());
- // Check Region count
- int runningCnt = 0;
- int unknownCnt = 0;
- TShowRegionResp showRegionResp = client.showRegion(new TShowRegionReq());
+ // RegionGroup statistics:
+ // 0: 1 Removing, 1 partition
+ // 1: 1 ReadOnly, 1 partition
+ // 2: 1 Unknown, 1 partition
+ // 3: All Running, 1 partition
+ // Least Region Group number per storageGroup = 4, match the current Region Group number
+ // Will allocate the new partition to Running RegionGroup 3, DataNodes: [1, 2, 6]
+ showRegionResp = client.showRegion(new TShowRegionReq());
Assert.assertEquals(
TSStatusCode.SUCCESS_STATUS.getStatusCode(), showRegionResp.getStatus().getCode());
for (TRegionInfo regionInfo : showRegionResp.getRegionInfoList()) {
- if (RegionStatus.Running.getStatus().equals(regionInfo.getStatus())) {
- runningCnt += 1;
- } else if (RegionStatus.Unknown.getStatus().equals(regionInfo.getStatus())) {
- unknownCnt += 1;
+ if (regionInfo.getDataNodeId() == 6) {
+ Assert.assertEquals(regionInfo.getTimeSlots(), 2);
}
}
- // The runningCnt should be exactly twice as the unknownCnt
- // since there exists one DataNode is shutdown
- Assert.assertEquals(unknownCnt * 2, runningCnt);
- // Wait for shutdown check
- TShowClusterResp showClusterResp;
- while (true) {
- AtomicBoolean containUnknown = new AtomicBoolean(false);
- TShowDataNodesResp showDataNodesResp = client.showDataNodes();
- showDataNodesResp
- .getDataNodesInfoList()
- .forEach(
- dataNodeInfo -> {
- if (NodeStatus.Unknown.getStatus().equals(dataNodeInfo.getStatus())) {
- containUnknown.set(true);
- }
- });
-
- if (containUnknown.get()) {
- break;
+ partitionSlotsMap =
+ ConfigNodeTestUtils.constructPartitionSlotsMap(
+ sg,
+ 5,
+ 5 + testSeriesPartitionBatchSize,
+ 5,
+ 5 + testTimePartitionBatchSize,
+ testTimePartitionInterval);
+ dataPartitionReq = new TDataPartitionReq(partitionSlotsMap);
+ for (int retry = 0; retry < 5; retry++) {
+ // Build new Client since it's unstable in Win8 environment
+ try (SyncConfigNodeIServiceClient configNodeClient =
+ (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
+ dataPartitionTableResp = configNodeClient.getOrCreateDataPartitionTable(dataPartitionReq);
+ if (dataPartitionTableResp != null) {
+ break;
+ }
+ } catch (Exception e) {
+ // Retry sometimes in order to avoid request timeout
+ LOGGER.error(e.getMessage());
+ TimeUnit.SECONDS.sleep(1);
}
- TimeUnit.SECONDS.sleep(1);
}
- runningCnt = 0;
- unknownCnt = 0;
- showClusterResp = client.showCluster();
- for (TDataNodeLocation dataNodeLocation : showClusterResp.getDataNodeList()) {
- if (NodeStatus.Running.getStatus()
- .equals(showClusterResp.getNodeStatus().get(dataNodeLocation.getDataNodeId()))) {
- runningCnt += 1;
- } else if (NodeStatus.Unknown.getStatus()
- .equals(showClusterResp.getNodeStatus().get(dataNodeLocation.getDataNodeId()))) {
- unknownCnt += 1;
+ Assert.assertNotNull(dataPartitionTableResp);
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ dataPartitionTableResp.getStatus().getCode());
+ Assert.assertNotNull(dataPartitionTableResp.getDataPartitionTable());
+ ConfigNodeTestUtils.checkDataPartitionTable(
+ sg,
+ 5,
+ 5 + testSeriesPartitionBatchSize,
+ 5,
+ 5 + testTimePartitionBatchSize,
+ testTimePartitionInterval,
+ dataPartitionTableResp.getDataPartitionTable());
+
+ // RegionGroup statistics:
+ // 0: 1 Removing, 1 partition
+ // 1: 1 ReadOnly, 1 partition
+ // 2: 1 Unknown, 1 partition
+ // 3: All Running, 2 partition
+ // Least Region Group number per storageGroup = 4, match the current Region Group number
+ // Will allocate the new partition to available RegionGroup 2, DataNodes: [1, 2, 5]
+ showRegionResp = client.showRegion(new TShowRegionReq());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), showRegionResp.getStatus().getCode());
+ for (TRegionInfo regionInfo : showRegionResp.getRegionInfoList()) {
+ if (regionInfo.getDataNodeId() == 5) {
+ Assert.assertEquals(regionInfo.getTimeSlots(), 2);
}
}
- Assert.assertEquals(2, runningCnt);
- Assert.assertEquals(1, unknownCnt);
- // Test getOrCreateDataPartition, ConfigNode should create DataPartition and return
partitionSlotsMap =
ConfigNodeTestUtils.constructPartitionSlotsMap(
sg,
- 1,
- 1 + testSeriesPartitionBatchSize,
- 1,
- 1 + testTimePartitionBatchSize,
+ 6,
+ 6 + testSeriesPartitionBatchSize,
+ 6,
+ 6 + testTimePartitionBatchSize,
testTimePartitionInterval);
dataPartitionReq = new TDataPartitionReq(partitionSlotsMap);
for (int retry = 0; retry < 5; retry++) {
@@ -531,54 +521,82 @@ public class IoTDBPartitionDurableIT {
Assert.assertNotNull(dataPartitionTableResp.getDataPartitionTable());
ConfigNodeTestUtils.checkDataPartitionTable(
sg,
- 1,
- 1 + testSeriesPartitionBatchSize,
- 1,
- 1 + testTimePartitionBatchSize,
+ 6,
+ 6 + testSeriesPartitionBatchSize,
+ 6,
+ 6 + testTimePartitionBatchSize,
testTimePartitionInterval,
dataPartitionTableResp.getDataPartitionTable());
- // Check Region count and status
- runningCnt = 0;
- unknownCnt = 0;
+ // RegionGroup statistics:
+ // 0: 1 Removing, 1 partition
+ // 1: 1 ReadOnly, 1 partition
+ // 2: 1 Unknown, 2 partition
+ // 3: All Running, 2 partition
+ // Least Region Group number per storageGroup = 4, match the current Region Group number
+ // Will allocate the new partition to Discouraged RegionGroup 2, DataNodes: [1, 2, 4]
showRegionResp = client.showRegion(new TShowRegionReq());
Assert.assertEquals(
TSStatusCode.SUCCESS_STATUS.getStatusCode(), showRegionResp.getStatus().getCode());
for (TRegionInfo regionInfo : showRegionResp.getRegionInfoList()) {
- if (RegionStatus.Running.getStatus().equals(regionInfo.getStatus())) {
- runningCnt += 1;
- } else if (RegionStatus.Unknown.getStatus().equals(regionInfo.getStatus())) {
- unknownCnt += 1;
+ if (regionInfo.getDataNodeId() == 4) {
+ Assert.assertEquals(regionInfo.getTimeSlots(), 2);
}
}
- // The runningCnt should be exactly twice as the unknownCnt
- // since there exists one DataNode is shutdown
- Assert.assertEquals(unknownCnt * 2, runningCnt);
-
- EnvFactory.getEnv().startDataNode(testDataNodeId);
- EnvFactory.getEnv()
- .ensureNodeStatus(
- Collections.singletonList(EnvFactory.getEnv().getDataNodeWrapper(testDataNodeId)),
- Collections.singletonList(NodeStatus.Running));
- // All Regions should alive after the testDataNode is restarted
- boolean allRunning = true;
- for (int retry = 0; retry < 30; retry++) {
- allRunning = true;
- showRegionResp = client.showRegion(new TShowRegionReq());
- for (TRegionInfo regionInfo : showRegionResp.getRegionInfoList()) {
- if (!RegionStatus.Running.getStatus().equals(regionInfo.getStatus())) {
- allRunning = false;
+ partitionSlotsMap =
+ ConfigNodeTestUtils.constructPartitionSlotsMap(
+ sg,
+ 7,
+ 7 + testSeriesPartitionBatchSize,
+ 7,
+ 7 + testTimePartitionBatchSize,
+ testTimePartitionInterval);
+ dataPartitionReq = new TDataPartitionReq(partitionSlotsMap);
+ for (int retry = 0; retry < 5; retry++) {
+ // Build new Client since it's unstable in Win8 environment
+ try (SyncConfigNodeIServiceClient configNodeClient =
+ (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
+ dataPartitionTableResp = configNodeClient.getOrCreateDataPartitionTable(dataPartitionReq);
+ if (dataPartitionTableResp != null) {
break;
}
+ } catch (Exception e) {
+ // Retry sometimes in order to avoid request timeout
+ LOGGER.error(e.getMessage());
+ TimeUnit.SECONDS.sleep(1);
}
+ }
+ Assert.assertNotNull(dataPartitionTableResp);
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ dataPartitionTableResp.getStatus().getCode());
+ Assert.assertNotNull(dataPartitionTableResp.getDataPartitionTable());
+ ConfigNodeTestUtils.checkDataPartitionTable(
+ sg,
+ 7,
+ 7 + testSeriesPartitionBatchSize,
+ 7,
+ 7 + testTimePartitionBatchSize,
+ testTimePartitionInterval,
+ dataPartitionTableResp.getDataPartitionTable());
- if (allRunning) {
- break;
+ // RegionGroup statistics:
+ // 0: 1 Removing, 1 partition
+ // 1: 1 ReadOnly, 2 partition
+ // 2: 1 Unknown, 2 partition
+ // 3: All Running, 2 partition
+ // Least Region Group number per storageGroup = 4, match the current Region Group number
+ // Will allocate the new partition to Running RegionGroup 3, DataNodes: [0, 1, 5]
+ // Because RegionGroup 1 is Disabled
+ showRegionResp = client.showRegion(new TShowRegionReq());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(), showRegionResp.getStatus().getCode());
+ for (TRegionInfo regionInfo : showRegionResp.getRegionInfoList()) {
+ if (regionInfo.getDataNodeId() == 6) {
+ Assert.assertEquals(regionInfo.getTimeSlots(), 3);
}
- TimeUnit.SECONDS.sleep(1);
}
- Assert.assertTrue(allRunning);
}
}
}
diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBPartitionDurableIT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBPartitionDurableIT.java
index 0fec7296ef..ad03ec53e1 100644
--- a/integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBPartitionDurableIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBPartitionDurableIT.java
@@ -122,6 +122,7 @@ public class IoTDBPartitionDurableIT {
EnvFactory.getEnv().cleanClusterEnvironment();
}
+ // TODO: Fix this when replica completion is supported
@Test
public void testRemovingDataNode() throws Exception {
try (SyncConfigNodeIServiceClient client =
@@ -248,6 +249,7 @@ public class IoTDBPartitionDurableIT {
}
}
+ // TODO: Fix this when replica completion is supported
@Test
public void testReadOnlyDataNode() throws Exception {
try (SyncConfigNodeIServiceClient client =
@@ -410,6 +412,39 @@ public class IoTDBPartitionDurableIT {
try (SyncConfigNodeIServiceClient client =
(SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
+ // Wait for shutdown check
+ TShowClusterResp showClusterResp;
+ while (true) {
+ AtomicBoolean containUnknown = new AtomicBoolean(false);
+ TShowDataNodesResp showDataNodesResp = client.showDataNodes();
+ showDataNodesResp
+ .getDataNodesInfoList()
+ .forEach(
+ dataNodeInfo -> {
+ if (NodeStatus.Unknown.getStatus().equals(dataNodeInfo.getStatus())) {
+ containUnknown.set(true);
+ }
+ });
+
+ if (containUnknown.get()) {
+ break;
+ }
+ TimeUnit.SECONDS.sleep(1);
+ }
+ int runningCnt = 0;
+ int unknownCnt = 0;
+ showClusterResp = client.showCluster();
+ for (TDataNodeLocation dataNodeLocation : showClusterResp.getDataNodeList()) {
+ if (NodeStatus.Running.getStatus()
+ .equals(showClusterResp.getNodeStatus().get(dataNodeLocation.getDataNodeId()))) {
+ runningCnt += 1;
+ } else if (NodeStatus.Unknown.getStatus()
+ .equals(showClusterResp.getNodeStatus().get(dataNodeLocation.getDataNodeId()))) {
+ unknownCnt += 1;
+ }
+ }
+ Assert.assertEquals(2, runningCnt);
+ Assert.assertEquals(1, unknownCnt);
// Test getOrCreateDataPartition, ConfigNode should create DataPartition and return
Map<String, Map<TSeriesPartitionSlot, TTimeSlotList>> partitionSlotsMap =
ConfigNodeTestUtils.constructPartitionSlotsMap(
@@ -450,8 +485,8 @@ public class IoTDBPartitionDurableIT {
dataPartitionTableResp.getDataPartitionTable());
// Check Region count
- int runningCnt = 0;
- int unknownCnt = 0;
+ runningCnt = 0;
+ unknownCnt = 0;
TShowRegionResp showRegionResp = client.showRegion(new TShowRegionReq());
Assert.assertEquals(
TSStatusCode.SUCCESS_STATUS.getStatusCode(), showRegionResp.getStatus().getCode());
@@ -466,40 +501,6 @@ public class IoTDBPartitionDurableIT {
// since there exists one DataNode is shutdown
Assert.assertEquals(unknownCnt * 2, runningCnt);
- // Wait for shutdown check
- TShowClusterResp showClusterResp;
- while (true) {
- AtomicBoolean containUnknown = new AtomicBoolean(false);
- TShowDataNodesResp showDataNodesResp = client.showDataNodes();
- showDataNodesResp
- .getDataNodesInfoList()
- .forEach(
- dataNodeInfo -> {
- if (NodeStatus.Unknown.getStatus().equals(dataNodeInfo.getStatus())) {
- containUnknown.set(true);
- }
- });
-
- if (containUnknown.get()) {
- break;
- }
- TimeUnit.SECONDS.sleep(1);
- }
- runningCnt = 0;
- unknownCnt = 0;
- showClusterResp = client.showCluster();
- for (TDataNodeLocation dataNodeLocation : showClusterResp.getDataNodeList()) {
- if (NodeStatus.Running.getStatus()
- .equals(showClusterResp.getNodeStatus().get(dataNodeLocation.getDataNodeId()))) {
- runningCnt += 1;
- } else if (NodeStatus.Unknown.getStatus()
- .equals(showClusterResp.getNodeStatus().get(dataNodeLocation.getDataNodeId()))) {
- unknownCnt += 1;
- }
- }
- Assert.assertEquals(2, runningCnt);
- Assert.assertEquals(1, unknownCnt);
-
// Test getOrCreateDataPartition, ConfigNode should create DataPartition and return
partitionSlotsMap =
ConfigNodeTestUtils.constructPartitionSlotsMap(