From 92c5bd7f85b89f10924b4471ced47f758d7a2e48 Mon Sep 17 00:00:00 2001 From: Di Wang Date: Sat, 24 Oct 2026 00:42:55 +0800 Subject: [PATCH] Fix leader distribution and RegionGroup directory assertions Count all DataRegion leaders and require a complete, healthy distribution. Check only the migrated consensus group and preserve unrelated region files. --- .../IoTDBRegionGroupLeaderDistributionIT.java | 147 ++++++------------ ...IoTDBRegionMigrateITFrameworkForRatis.java | 7 + ...RegionOperationReliabilityITFramework.java | 56 +++---- ...DBRegionMigrateNormalForIoTV2StreamIT.java | 74 +++++++++ 4 files changed, 152 insertions(+), 132 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/load/IoTDBRegionGroupLeaderDistributionIT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/load/IoTDBRegionGroupLeaderDistributionIT.java index b8d0018cefab5..42b8f4e8456ea 100644 --- a/integration-test/src/test/java/org/apache/iotdb/confignode/it/load/IoTDBRegionGroupLeaderDistributionIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/confignode/it/load/IoTDBRegionGroupLeaderDistributionIT.java @@ -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; @@ -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; @@ -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}) @@ -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 = @@ -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 countRegionLeaders(final SyncConfigNodeIServiceClient client) - throws TException { - final Map 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 regionIds = new HashSet<>(); + Map leadersPerRegion = new TreeMap<>(); + Map 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. - * - *

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 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 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); } } diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/IoTDBRegionMigrateITFrameworkForRatis.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/IoTDBRegionMigrateITFrameworkForRatis.java index 3b08f555d903d..f689aeefca4b6 100644 --- a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/IoTDBRegionMigrateITFrameworkForRatis.java +++ b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/IoTDBRegionMigrateITFrameworkForRatis.java @@ -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; @@ -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(); + } } diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/IoTDBRegionOperationReliabilityITFramework.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/IoTDBRegionOperationReliabilityITFramework.java index de096a95712b3..f12e8689e49eb 100644 --- a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/IoTDBRegionOperationReliabilityITFramework.java +++ b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/IoTDBRegionOperationReliabilityITFramework.java @@ -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) { @@ -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"); @@ -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() { @@ -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 @@ -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) { diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/commit/stream/IoTDBRegionMigrateNormalForIoTV2StreamIT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/commit/stream/IoTDBRegionMigrateNormalForIoTV2StreamIT.java index aaaa4dadacbde..588ac0615e191 100644 --- a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/commit/stream/IoTDBRegionMigrateNormalForIoTV2StreamIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/commit/stream/IoTDBRegionMigrateNormalForIoTV2StreamIT.java @@ -19,18 +19,30 @@ package org.apache.iotdb.confignode.it.regionmigration.pass.commit.stream; +import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient; import org.apache.iotdb.commons.utils.KillPoint.KillNode; import org.apache.iotdb.confignode.it.regionmigration.IoTDBRegionOperationReliabilityITFramework; +import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema; import org.apache.iotdb.consensus.ConsensusFactory; import org.apache.iotdb.it.env.EnvFactory; import org.apache.iotdb.it.framework.IoTDBTestRunner; import org.apache.iotdb.itbase.category.ClusterIT; +import org.apache.iotdb.rpc.TSStatusCode; +import org.awaitility.Awaitility; +import org.junit.Assert; import org.junit.Before; import org.junit.Test; import org.junit.experimental.categories.Category; import org.junit.runner.RunWith; +import java.sql.Connection; +import java.sql.Statement; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.TimeUnit; + @Category({ClusterIT.class}) @RunWith(IoTDBTestRunner.class) public class IoTDBRegionMigrateNormalForIoTV2StreamIT @@ -55,4 +67,66 @@ public void normal1C2DTest() throws Exception { public void normal3C3DTest() throws Exception { successTest(2, 3, 3, 3, noKillPoints(), noKillPoints(), KillNode.ALL_NODES); } + + @Test + public void migrationShouldPreserveOtherRegionFiles() throws Exception { + EnvFactory.getEnv() + .getConfig() + .getCommonConfig() + .setDataReplicationFactor(2) + .setSchemaReplicationFactor(3); + EnvFactory.getEnv().initClusterEnvironment(1, 3); + + try (Connection connection = EnvFactory.getEnv().getConnection(); + Statement statement = connection.createStatement(); + SyncConfigNodeIServiceClient client = + (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) { + statement.execute(INSERTION1); + Map> regionMap = getDataRegionMap(statement); + Assert.assertEquals(1, regionMap.size()); + int selectedRegion = selectRegion(regionMap); + Set allDataNodes = getAllDataNodes(statement); + int source = selectDataNodeContainsRegion(allDataNodes, regionMap, selectedRegion); + int destination = selectDataNodeNotContainsRegion(allDataNodes, regionMap, selectedRegion); + + // Three replicas ensure that the unrelated database also has a replica on the source DN. + Assert.assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), + client + .setDatabase( + new TDatabaseSchema("root.other") + .setDataReplicationFactor(3) + .setSchemaReplicationFactor(3)) + .getCode()); + statement.execute("INSERT INTO root.other.d1(timestamp,s) values(100, 1)"); + statement.execute(FLUSH_COMMAND); + Map> allRegions = getDataRegionMap(statement); + Set otherRegions = new HashSet<>(allRegions.keySet()); + otherRegions.removeAll(regionMap.keySet()); + Assert.assertFalse(otherRegions.isEmpty()); + for (int regionId : otherRegions) { + Assert.assertEquals(allDataNodes, allRegions.get(regionId)); + checkRegionFileExist(source, regionId); + } + checkRegionFileExist(source, selectedRegion); + + statement.execute( + "migrate region " + selectedRegion + " from " + source + " to " + destination); + Set expectedDataNodes = new HashSet<>(regionMap.get(selectedRegion)); + expectedDataNodes.remove(source); + expectedDataNodes.add(destination); + Awaitility.await() + .atMost(1, TimeUnit.MINUTES) + .untilAsserted( + () -> + Assert.assertEquals( + expectedDataNodes, getDataRegionMap(statement).get(selectedRegion))); + + checkRegionFileClear(source, selectedRegion); + checkRegionFileExist(destination, selectedRegion); + for (int regionId : otherRegions) { + checkRegionFileExist(source, regionId); + } + } + } }