Skip to content

Fix leader distribution and RegionGroup directory assertions #19006

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
d-wang-commit wants to merge 1 commit into
base: master
Choose a base branch
from
+152 −132
Open

Fix leader distribution and RegionGroup directory assertions #19006

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
147 changes: 47 additions & 100 deletions ...c/test/java/org/apache/iotdb/confignode/it/load/IoTDBRegionGroupLeaderDistributionIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,16 +19,17 @@

package org.apache.iotdb.confignode.it.load;

import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
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.client.sync.SyncConfigNodeIServiceClient;
import org.apache.iotdb.commons.cluster.NodeStatus;
import org.apache.iotdb.commons.cluster.RegionRoleType;
import org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
import org.apache.iotdb.confignode.rpc.thrift.TDataPartitionReq;
import org.apache.iotdb.confignode.rpc.thrift.TDataPartitionTableResp;
import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema;
import org.apache.iotdb.confignode.rpc.thrift.TRegionInfo;
import org.apache.iotdb.confignode.rpc.thrift.TShowRegionReq;
import org.apache.iotdb.confignode.rpc.thrift.TShowRegionResp;
import org.apache.iotdb.confignode.rpc.thrift.TTimeSlotList;
Expand All @@ -39,7 +40,7 @@
import org.apache.iotdb.itbase.category.ClusterIT;
import org.apache.iotdb.rpc.TSStatusCode;

import org.apache.thrift.TException;
import org.awaitility.Awaitility;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
Expand All @@ -51,10 +52,11 @@
import java.sql.Statement;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

@RunWith(IoTDBTestRunner.class)
@Category({ClusterIT.class})
Expand Down Expand Up @@ -166,8 +168,6 @@ public void testCFDWithUnknownStatus() throws Exception {

@Test
public void testCFDWithReadOnlyStatus() throws Exception {
final int retryNum = 50;

TSStatus status;
final int databaseNum = 3;
try (SyncConfigNodeIServiceClient client =
Expand Down Expand Up @@ -213,107 +213,54 @@ public void testCFDWithReadOnlyStatus() throws Exception {
Collections.singletonList(EnvFactory.getEnv().getDataNodeWrapper(0)),
Collections.singletonList(NodeStatus.ReadOnly));

// Check leader distribution
TShowRegionResp showRegionResp;
AtomicBoolean isDistributionBalanced = new AtomicBoolean();
for (int retry = 0; retry < retryNum; retry++) {
isDistributionBalanced.set(true);
showRegionResp = client.showRegion(new TShowRegionReq());
showRegionResp
.getRegionInfoList()
.removeIf(r -> r.database.startsWith(SystemConstant.SYSTEM_DATABASE));
showRegionResp
.getRegionInfoList()
.removeIf(r -> r.database.startsWith(SystemConstant.AUDIT_DATABASE));
showRegionResp
.getRegionInfoList()
.forEach(
regionInfo -> {
if (RegionRoleType.Leader.getRoleType().equals(regionInfo.getRoleType())
&& NodeStatus.ReadOnly.getStatus().equals(regionInfo.getStatus())) {
// ReadOnly DataNode couldn't have Region-leader
isDistributionBalanced.set(false);
}
});

if (isDistributionBalanced.get()) {
break;
} else {
TimeUnit.SECONDS.sleep(1);
}
}
Assert.assertTrue(isDistributionBalanced.get());
// Only Running DataNodes may hold leaders after the ReadOnly transition.
awaitBalancedLeaderDistribution(client, TEST_DATA_NODE_NUM - 1);
}
}

/**
* Counts the RegionGroup leaders held by each DataNode, ignoring the system and audit databases.
* Checks the same population as the DataRegion leader balancer: all databases, including internal
* databases. SchemaRegion leaders are balanced separately and must not be mixed into these
* counts.
*/
private static Map<Integer, Integer> countRegionLeaders(final SyncConfigNodeIServiceClient client)
throws TException {
final Map<Integer, Integer> leaderCounter = new TreeMap<>();
final TShowRegionResp showRegionResp = client.showRegion(new TShowRegionReq());
showRegionResp
.getRegionInfoList()
.removeIf(r -> r.database.startsWith(SystemConstant.SYSTEM_DATABASE));
showRegionResp
.getRegionInfoList()
.removeIf(r -> r.database.startsWith(SystemConstant.AUDIT_DATABASE));
showRegionResp
.getRegionInfoList()
.forEach(
regionInfo -> {
if (RegionRoleType.Leader.getRoleType().equals(regionInfo.getRoleType())) {
leaderCounter.merge(regionInfo.getDataNodeId(), 1, Integer::sum);
private static void awaitBalancedLeaderDistribution(
final SyncConfigNodeIServiceClient client, final int expectedDataNodeNum) {
Awaitility.await()
.atMost(RETRY_NUM, TimeUnit.SECONDS)
.pollInterval(1, TimeUnit.SECONDS)
.untilAsserted(
() -> {
TShowRegionResp response =
client.showRegion(
new TShowRegionReq().setConsensusGroupType(TConsensusGroupType.DataRegion));
Assert.assertEquals(
TSStatusCode.SUCCESS_STATUS.getStatusCode(), response.getStatus().getCode());
Set<Integer> regionIds = new HashSet<>();
Map<Integer, Integer> leadersPerRegion = new TreeMap<>();
Map<Integer, Integer> leadersPerNode = new TreeMap<>();
for (TRegionInfo region : response.getRegionInfoList()) {
regionIds.add(region.getConsensusGroupId().getId());
if (RegionRoleType.Leader.getRoleType().equals(region.getRoleType())) {
Assert.assertEquals(NodeStatus.Running.getStatus(), region.getStatus());
leadersPerRegion.merge(region.getConsensusGroupId().getId(), 1, Integer::sum);
leadersPerNode.merge(region.getDataNodeId(), 1, Integer::sum);
}
}
// A partial heartbeat result must not look balanced just because leaders are missing.
Assert.assertFalse("No DataRegionGroups were reported", regionIds.isEmpty());
Assert.assertEquals(
"Some DataRegionGroups have no leader", regionIds, leadersPerRegion.keySet());
for (int leaderCount : leadersPerRegion.values()) {
Assert.assertEquals(
"Each DataRegionGroup must have exactly one leader", 1, leaderCount);
}
Assert.assertEquals(
leadersPerNode.toString(), expectedDataNodeNum, leadersPerNode.size());
Assert.assertTrue(
leadersPerNode.toString(),
Collections.max(leadersPerNode.values())
- Collections.min(leadersPerNode.values())
<= 1);
});
return leaderCounter;
}

/**
* Returns whether every expected DataNode holds at least one RegionGroup leader and no two
* DataNodes hold a different number of leaders by more than one.
*
* <p>A fixed per-DataNode leader count cannot be asserted: the number of RegionGroups is not
* necessarily divisible by the number of DataNodes, and the leader balancer also balances the
* RegionGroups of the system and audit databases, which are not counted here.
*/
private static boolean isLeaderDistributionBalanced(
final Map<Integer, Integer> leaderCounter, final int expectedDataNodeNum) {
if (leaderCounter.size() != expectedDataNodeNum) {
return false;
}
int minLeaderCount = Integer.MAX_VALUE;
int maxLeaderCount = Integer.MIN_VALUE;
for (final int leaderCount : leaderCounter.values()) {
minLeaderCount = Math.min(minLeaderCount, leaderCount);
maxLeaderCount = Math.max(maxLeaderCount, leaderCount);
}
return maxLeaderCount - minLeaderCount <= 1;
}

/**
* Waits until the leader balancer has converged and fails with the observed distribution
* otherwise. Polling is required because a RegionGroup only reports its leader once the election
* finishes, and leaders keep being migrated while the RegionGroups are being created.
*/
private static void awaitBalancedLeaderDistribution(
final SyncConfigNodeIServiceClient client, final int expectedDataNodeNum)
throws TException, InterruptedException {
Map<Integer, Integer> leaderCounter = new TreeMap<>();
for (int retry = 0; retry < RETRY_NUM; retry++) {
leaderCounter = countRegionLeaders(client);
if (isLeaderDistributionBalanced(leaderCounter, expectedDataNodeNum)) {
return;
}
TimeUnit.SECONDS.sleep(1);
}
Assert.fail(
"The RegionGroup leaders were not balanced after "
+ RETRY_NUM
+ " retries: expected "
+ expectedDataNodeNum
+ " DataNodes holding leaders within one of each other, but got "
+ leaderCounter);
}
}
7 changes: 7 additions & 0 deletions ...org/apache/iotdb/confignode/it/regionmigration/IoTDBRegionMigrateITFrameworkForRatis.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,9 @@

package org.apache.iotdb.confignode.it.regionmigration;

import org.apache.iotdb.commons.consensus.DataRegionId;
import org.apache.iotdb.consensus.ConsensusFactory;
import org.apache.iotdb.consensus.ratis.utils.Utils;
import org.apache.iotdb.it.env.EnvFactory;

import org.junit.Before;
Expand All @@ -39,4 +41,9 @@ public void setUp() throws Exception {
.setDataRegionRatisReconfigurationMaxRetryAttempts(10)
.setSchemaRegionRatisReconfigurationMaxRetryAttempts(10);
}

@Override
protected String getRegionDirName(int regionId) {
return Utils.fromConsensusGroupIdToRaftGroupId(new DataRegionId(regionId)).getUuid().toString();
}
}
56 changes: 24 additions & 32 deletions ...pache/iotdb/confignode/it/regionmigration/IoTDBRegionOperationReliabilityITFramework.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -321,7 +321,7 @@ private void generalTestWithAllOptions(
final int originalDataNode = selectOriginalDataNode(regionMap, selectedRegion);
final int destDataNode =
selectDataNodeNotContainsRegion(allDataNodeId, regionMap, selectedRegion);
checkRegionFileExist(originalDataNode);
checkRegionFileExist(originalDataNode, selectedRegion);

// set kill points
if (killNode == KillNode.ORIGINAL_DATANODE) {
Expand Down Expand Up @@ -415,13 +415,13 @@ private void generalTestWithAllOptions(

// check the remaining file
if (success) {
checkRegionFileClearIfNodeAlive(originalDataNode);
checkRegionFileExistIfNodeAlive(destDataNode);
checkRegionFileClearIfNodeAlive(originalDataNode, selectedRegion);
checkRegionFileExistIfNodeAlive(destDataNode, selectedRegion);
// TODO: @YongzaoDan enable this check after the __system database is refactored!!!
// checkClusterStillWritable();
} else {
checkRegionFileClearIfNodeAlive(destDataNode);
checkRegionFileExistIfNodeAlive(originalDataNode);
checkRegionFileClearIfNodeAlive(destDataNode, selectedRegion);
checkRegionFileExistIfNodeAlive(originalDataNode, selectedRegion);
}

LOGGER.info("test pass");
Expand Down Expand Up @@ -788,41 +788,27 @@ protected static void awaitUntilSuccess(
LOGGER.info("DataNode set has been successfully changed to {}", lastTimeDataNodes.get());
}

private static void checkRegionFileExistIfNodeAlive(int dataNode) {
private void checkRegionFileExistIfNodeAlive(int dataNode, int regionId) {
if (EnvFactory.getEnv().dataNodeIdToWrapper(dataNode).get().isAlive()) {
checkRegionFileExist(dataNode);
checkRegionFileExist(dataNode, regionId);
}
}

private static void checkRegionFileExist(int dataNode) {
File originalRegionDir = new File(buildRegionDirPath(dataNode));
Assert.assertTrue(originalRegionDir.isDirectory());
Assert.assertNotEquals(0, Objects.requireNonNull(originalRegionDir.listFiles()).length);
protected void checkRegionFileExist(int dataNode, int regionId) {
File regionDir = new File(buildRegionDirPath(dataNode, regionId));
Assert.assertTrue(regionDir.getAbsolutePath(), regionDir.isDirectory());
}

private static void checkRegionFileClearIfNodeAlive(int dataNode) {
private void checkRegionFileClearIfNodeAlive(int dataNode, int regionId) {
if (EnvFactory.getEnv().dataNodeIdToWrapper(dataNode).get().isAlive()) {
checkRegionFileClear(dataNode);
checkRegionFileClear(dataNode, regionId);
}
}

/** Check whether the original DataNode's region file has been deleted. */
private static void checkRegionFileClear(int dataNode) {
File originalRegionDir = new File(buildRegionDirPath(dataNode));
Assert.assertTrue(originalRegionDir.isDirectory());
File[] files = originalRegionDir.listFiles();
try {
int length = Objects.requireNonNull(files).length;
// the node may still have a region of the system database
Assert.assertTrue(length == 0 || length == 1 && files[0].getName().equals("1_1"));
} catch (AssertionError e) {
LOGGER.error(
"Original DataNode {} region file not clear, these files is still remain: {}",
dataNode,
Arrays.toString(files));
throw e;
}
LOGGER.info("Original DataNode {} region file clear", dataNode);
/** Other RegionGroups, including internal databases, may still have replicas on this DataNode. */
protected void checkRegionFileClear(int dataNode, int regionId) {
File regionDir = new File(buildRegionDirPath(dataNode, regionId));
Assert.assertFalse(regionDir.getAbsolutePath(), regionDir.exists());
}

private void checkClusterStillWritable() {
Expand All @@ -847,7 +833,7 @@ private void checkClusterStillWritable() {
}
}

private static String buildRegionDirPath(int dataNode) {
private String buildRegionDirPath(int dataNode, int regionId) {
String nodePath = EnvFactory.getEnv().dataNodeIdToWrapper(dataNode).get().getNodePath();
return nodePath
+ File.separator
Expand All @@ -857,7 +843,13 @@ private static String buildRegionDirPath(int dataNode) {
+ File.separator
+ IoTDBConstant.CONSENSUS_FOLDER_NAME
+ File.separator
+ IoTDBConstant.DATA_REGION_FOLDER_NAME;
+ IoTDBConstant.DATA_REGION_FOLDER_NAME
+ File.separator
+ getRegionDirName(regionId);
}

protected String getRegionDirName(int regionId) {
return TConsensusGroupType.DataRegion.getValue() + "_" + regionId;
}

private static String buildDataPath(int dataNode, boolean isSequence) {
Expand Down
Loading
Loading