This is an automated email from the ASF dual-hosted git repository.
CRZbulabula 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 5b9e8868c47 [ConfigNode] Skip non-existent and duplicate region IDs
during reconstruction (#18664)
5b9e8868c47 is described below
commit 5b9e8868c475ad2699a3252862ece5dde3607d92
Author: Yongzao <[email protected]>
AuthorDate: Thu Sep 17 15:25:43 2026 +0800
[ConfigNode] Skip non-existent and duplicate region IDs during
reconstruction (#18664)
* Avoid duplicate procedures for repeated RECONSTRUCT REGION IDs
* Skip unusable reconstruct region IDs
---
.../iotdb/confignode/i18n/ManagerMessages.java | 6 +
.../iotdb/confignode/i18n/ManagerMessages.java | 6 +
.../iotdb/confignode/manager/ProcedureManager.java | 31 ++--
.../manager/partition/PartitionManager.java | 21 ++-
.../ProcedureManagerReconstructRegionTest.java | 169 +++++++++++++++++++++
5 files changed, 218 insertions(+), 15 deletions(-)
diff --git
a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java
b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java
index a61e8e2aa5b..03ec8003251 100644
---
a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java
+++
b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java
@@ -289,6 +289,12 @@ public final class ManagerMessages {
"LoadStatistics service is started successfully.";
public static final String LOADSTATISTICS_SERVICE_IS_STOPPED_SUCCESSFULLY =
"LoadStatistics service is stopped successfully.";
+ public static final String
+
LOG_SKIP_DUPLICATE_REGION_ID_ARG_IN_RECONSTRUCTREGION_REQUEST_TO_DATANODE_ARG_ED195F69
=
+ "Skip duplicate Region ID {} in ReconstructRegion request to
DataNode {}.";
+ public static final String
+
LOG_SKIP_NON_EXISTENT_REGION_ID_ARG_IN_RECONSTRUCTREGION_REQUEST_TO_DATANODE_ARG_7F76D789
=
+ "Skip non-existent Region ID {} in ReconstructRegion request to
DataNode {}.";
public static final String
MIGRATEREGION_SUBMIT_REGIONMIGRATEPROCEDURE_SUCCESSFULLY_REGION_ORIGIN_DATANODE
=
"[MigrateRegion] Submit RegionMigrateProcedure successfully, Region: {},
Origin DataNode: {}, Dest DataNode: {}, Add Coordinator: {}, Remove
Coordinator: {}";
public static final String
SUBMIT_REGIONMIGRATEPROCEDURE_FAILED_BECAUSE_REGIONGROUP_DOESN_T_EXIST =
diff --git
a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java
b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java
index fa8fc8e3217..f145e04052e 100644
---
a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java
+++
b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java
@@ -287,6 +287,12 @@ public final class ManagerMessages {
"LoadStatistics 服务已成功启动。";
public static final String LOADSTATISTICS_SERVICE_IS_STOPPED_SUCCESSFULLY =
"LoadStatistics 服务已成功停止。";
+ public static final String
+
LOG_SKIP_DUPLICATE_REGION_ID_ARG_IN_RECONSTRUCTREGION_REQUEST_TO_DATANODE_ARG_ED195F69
=
+ "跳过 ReconstructRegion 请求中重复的 Region ID {}，目标 DataNode 为 {}。";
+ public static final String
+
LOG_SKIP_NON_EXISTENT_REGION_ID_ARG_IN_RECONSTRUCTREGION_REQUEST_TO_DATANODE_ARG_7F76D789
=
+ "跳过 ReconstructRegion 请求中不存在的 Region ID {}，目标 DataNode 为 {}。";
public static final String
MIGRATEREGION_SUBMIT_REGIONMIGRATEPROCEDURE_SUCCESSFULLY_REGION_ORIGIN_DATANODE
=
"[MigrateRegion] 成功提交 RegionMigrateProcedure，Region：{}，原 DataNode：{}，目标
DataNode：{}，新增 Coordinator：{}，移除 Coordinator：{}";
public static final String
SUBMIT_REGIONMIGRATEPROCEDURE_FAILED_BECAUSE_REGIONGROUP_DOESN_T_EXIST =
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java
index 50b003d0494..ac1a53539f6 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java
@@ -187,6 +187,7 @@ import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
@@ -1295,17 +1296,27 @@ public class ProcedureManager {
try (AutoCloseableLock ignoredLock =
AutoCloseableLock.acquire(env.getSubmitRegionMigrateLock())) {
List<ReconstructRegionProcedure> procedures = new ArrayList<>();
+ Set<Integer> seenRegionIds = new HashSet<>();
for (int x : req.getRegionIds()) {
- TConsensusGroupId regionId =
- configManager
- .getPartitionManager()
- .generateTConsensusGroupIdByRegionId(x)
- .orElseThrow(
- () ->
- new IllegalArgumentException(
- ManagerMessages.REGION_ID
- + x
- + ManagerMessages.EXCEPTION_INVALID_2928F475));
+ if (!seenRegionIds.add(x)) {
+ LOGGER.info(
+ ManagerMessages
+
.LOG_SKIP_DUPLICATE_REGION_ID_ARG_IN_RECONSTRUCTREGION_REQUEST_TO_DATANODE_ARG_ED195F69,
+ x,
+ req.getDataNodeId());
+ continue;
+ }
+ Optional<TConsensusGroupId> regionIdOptional =
+
configManager.getPartitionManager().findTConsensusGroupIdByRegionId(x);
+ if (!regionIdOptional.isPresent()) {
+ LOGGER.info(
+ ManagerMessages
+
.LOG_SKIP_NON_EXISTENT_REGION_ID_ARG_IN_RECONSTRUCTREGION_REQUEST_TO_DATANODE_ARG_7F76D789,
+ x,
+ req.getDataNodeId());
+ continue;
+ }
+ TConsensusGroupId regionId = regionIdOptional.get();
final TDataNodeLocation coordinator =
handler
.filterDataNodeWithOtherRegionReplica(
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java
index 28655923cca..2ee0bcdb60b 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java
@@ -1084,11 +1084,9 @@ public class PartitionManager {
}
public Optional<TConsensusGroupId> generateTConsensusGroupIdByRegionId(final
int regionId) {
- if (isRegionGroupExists(new
TConsensusGroupId(TConsensusGroupType.SchemaRegion, regionId))) {
- return Optional.of(new
TConsensusGroupId(TConsensusGroupType.SchemaRegion, regionId));
- }
- if (isRegionGroupExists(new
TConsensusGroupId(TConsensusGroupType.DataRegion, regionId))) {
- return Optional.of(new TConsensusGroupId(TConsensusGroupType.DataRegion,
regionId));
+ Optional<TConsensusGroupId> result =
findTConsensusGroupIdByRegionId(regionId);
+ if (result.isPresent()) {
+ return result;
}
String msg =
String.format(
@@ -1098,6 +1096,19 @@ public class PartitionManager {
return Optional.empty();
}
+ /**
+ * Returns the RegionGroup with the specified numeric id without logging
when it does not exist.
+ */
+ public Optional<TConsensusGroupId> findTConsensusGroupIdByRegionId(final int
regionId) {
+ if (isRegionGroupExists(new
TConsensusGroupId(TConsensusGroupType.SchemaRegion, regionId))) {
+ return Optional.of(new
TConsensusGroupId(TConsensusGroupType.SchemaRegion, regionId));
+ }
+ if (isRegionGroupExists(new
TConsensusGroupId(TConsensusGroupType.DataRegion, regionId))) {
+ return Optional.of(new TConsensusGroupId(TConsensusGroupType.DataRegion,
regionId));
+ }
+ return Optional.empty();
+ }
+
/**
* GetNodePathsPartition.
*
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/ProcedureManagerReconstructRegionTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/ProcedureManagerReconstructRegionTest.java
new file mode 100644
index 00000000000..a21a5da7083
--- /dev/null
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/ProcedureManagerReconstructRegionTest.java
@@ -0,0 +1,169 @@
+/*
+ * 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.manager;
+
+import org.apache.iotdb.common.rpc.thrift.Model;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
+import org.apache.iotdb.common.rpc.thrift.TDataNodeConfiguration;
+import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.cluster.NodeStatus;
+import org.apache.iotdb.confignode.manager.node.NodeManager;
+import org.apache.iotdb.confignode.manager.partition.PartitionManager;
+import org.apache.iotdb.confignode.persistence.ProcedureInfo;
+import org.apache.iotdb.confignode.procedure.Procedure;
+import org.apache.iotdb.confignode.procedure.ProcedureExecutor;
+import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
+import org.apache.iotdb.confignode.procedure.env.RegionMaintainHandler;
+import
org.apache.iotdb.confignode.procedure.impl.region.ReconstructRegionProcedure;
+import org.apache.iotdb.confignode.rpc.thrift.TReconstructRegionReq;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+
+import java.lang.reflect.Field;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.locks.ReentrantLock;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class ProcedureManagerReconstructRegionTest {
+
+ private final TConsensusGroupId firstRegion =
+ new TConsensusGroupId(TConsensusGroupType.DataRegion, 12);
+ private final TConsensusGroupId secondRegion =
+ new TConsensusGroupId(TConsensusGroupType.DataRegion, 14);
+ private final TDataNodeLocation target = new
TDataNodeLocation().setDataNodeId(7);
+ private final TDataNodeLocation coordinator = new
TDataNodeLocation().setDataNodeId(8);
+
+ private ProcedureManager manager;
+ private ProcedureExecutor<ConfigNodeProcedureEnv> executor;
+ private PartitionManager partitionManager;
+ private final ConcurrentHashMap<Long, Procedure<ConfigNodeProcedureEnv>>
procedures =
+ new ConcurrentHashMap<>();
+
+ @Before
+ public void setUp() throws Exception {
+ ConfigManager configManager = mock(ConfigManager.class);
+ NodeManager nodeManager = mock(NodeManager.class);
+ partitionManager = mock(PartitionManager.class);
+ ConfigNodeProcedureEnv env = mock(ConfigNodeProcedureEnv.class);
+ RegionMaintainHandler handler = mock(RegionMaintainHandler.class);
+ executor = mock(ProcedureExecutor.class);
+
+ when(configManager.getNodeManager()).thenReturn(nodeManager);
+ when(configManager.getPartitionManager()).thenReturn(partitionManager);
+ when(nodeManager.getRegisteredDataNode(target.getDataNodeId()))
+ .thenReturn(new TDataNodeConfiguration().setLocation(target));
+ when(nodeManager.filterDataNodeThroughStatus(NodeStatus.Running))
+ .thenReturn(Collections.singletonList(new
TDataNodeConfiguration().setLocation(target)));
+
when(partitionManager.findTConsensusGroupIdByRegionId(12)).thenReturn(Optional.of(firstRegion));
+ when(partitionManager.findTConsensusGroupIdByRegionId(14))
+ .thenReturn(Optional.of(secondRegion));
+
when(partitionManager.findTConsensusGroupIdByRegionId(99)).thenReturn(Optional.empty());
+
when(partitionManager.getRegionDatabase(any(TConsensusGroupId.class))).thenReturn("root.sg");
+
+ Map<TConsensusGroupId, TRegionReplicaSet> replicaSets = new HashMap<>();
+ replicaSets.put(
+ firstRegion, new TRegionReplicaSet(firstRegion, Arrays.asList(target,
coordinator)));
+ replicaSets.put(
+ secondRegion, new TRegionReplicaSet(secondRegion,
Arrays.asList(target, coordinator)));
+ when(partitionManager.getAllReplicaSetsMap(TConsensusGroupType.DataRegion))
+ .thenReturn(replicaSets);
+ when(partitionManager.getAllReplicaSets(target.getDataNodeId()))
+ .thenReturn(Arrays.asList(replicaSets.get(firstRegion),
replicaSets.get(secondRegion)));
+
+ when(env.getSubmitRegionMigrateLock()).thenReturn(new ReentrantLock());
+ when(env.getRegionMaintainHandler()).thenReturn(handler);
+ when(handler.filterDataNodeWithOtherRegionReplica(
+ any(TConsensusGroupId.class),
+ eq(target),
+ eq(NodeStatus.Running),
+ eq(NodeStatus.Removing),
+ eq(NodeStatus.ReadOnly)))
+ .thenReturn(Optional.of(coordinator));
+ when(executor.getProcedures()).thenReturn(procedures);
+
+ manager = new ProcedureManager(configManager, mock(ProcedureInfo.class));
+ Field envField = ProcedureManager.class.getDeclaredField("env");
+ envField.setAccessible(true);
+ envField.set(manager, env);
+ Field executorField = ProcedureManager.class.getDeclaredField("executor");
+ executorField.setAccessible(true);
+ executorField.set(manager, executor);
+ }
+
+ @Test
+ public void testDuplicateAndNonExistentRegionIdsAreSkippedInInputOrder() {
+ TReconstructRegionReq request =
+ new TReconstructRegionReq(Arrays.asList(12, 99, 14, 12, 99, 14), 7,
Model.TREE);
+
+ assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
manager.reconstructRegion(request).getCode());
+
+ ArgumentCaptor<ReconstructRegionProcedure> captor =
+ ArgumentCaptor.forClass(ReconstructRegionProcedure.class);
+ verify(executor, times(2)).submitProcedure(captor.capture());
+ assertEquals(firstRegion, captor.getAllValues().get(0).getRegionId());
+ assertEquals(secondRegion, captor.getAllValues().get(1).getRegionId());
+ verify(partitionManager, times(1)).findTConsensusGroupIdByRegionId(12);
+ verify(partitionManager, times(1)).findTConsensusGroupIdByRegionId(14);
+ verify(partitionManager, times(1)).findTConsensusGroupIdByRegionId(99);
+ }
+
+ @Test
+ public void
testRequestWithNoUsableRegionIdsSucceedsWithoutSubmittingProcedure() {
+ TReconstructRegionReq request = new
TReconstructRegionReq(Arrays.asList(99, 99), 7, Model.TREE);
+
+ assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
manager.reconstructRegion(request).getCode());
+ verify(executor, times(0)).submitProcedure(any());
+ }
+
+ @Test
+ public void testAnotherRequestCannotReconstructRegionWithActiveProcedure() {
+ TReconstructRegionReq request =
+ new TReconstructRegionReq(Collections.singletonList(12), 7,
Model.TREE);
+ ReconstructRegionProcedure activeProcedure =
+ new ReconstructRegionProcedure(firstRegion, target, coordinator);
+ procedures.put(1L, activeProcedure);
+
+ TSStatus status = manager.reconstructRegion(request);
+ assertEquals(TSStatusCode.RECONSTRUCT_REGION_ERROR.getStatusCode(),
status.getCode());
+ assertTrue(status.getMessage().contains("in progress"));
+ verify(executor, times(0)).submitProcedure(any());
+ }
+}
