You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ta...@apache.org on 2024/03/26 12:53:15 UTC
(iotdb) branch master updated: Region migration improvement (#12165)
This is an automated email from the ASF dual-hosted git repository.
tanxinyu 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 a5df27c75cf Region migration improvement (#12165)
a5df27c75cf is described below
commit a5df27c75cf4ad095b46bf7ad7149310ec767010
Author: Li Yu Heng <li...@126.com>
AuthorDate: Tue Mar 26 20:53:08 2024 +0800
Region migration improvement (#12165)
---
.../iotdb/it/env/cluster/env/AbstractEnv.java | 20 +
.../it/env/cluster/node/AbstractNodeWrapper.java | 2 +-
.../iotdb/it/env/remote/env/RemoteServerEnv.java | 6 +
.../java/org/apache/iotdb/itbase/env/BaseEnv.java | 3 +
.../it/IoTDBRegionMigrateReliabilityIT.java | 428 +++++++++++++++++++++
.../confignode/it/procedure/IoTDBProcedureIT.java | 2 +
.../confignode/client/DataNodeRequestType.java | 1 +
.../client/sync/SyncDataNodeClientPool.java | 5 +-
.../consensus/request/ConfigPhysicalPlan.java | 14 +-
.../consensus/request/ConfigPhysicalPlanType.java | 4 +-
...ocationPlan.java => AddRegionLocationPlan.java} | 40 +-
...tionPlan.java => RemoveRegionLocationPlan.java} | 46 +--
.../write/partition/UpdateRegionLocationPlan.java | 1 +
.../iotdb/confignode/manager/ConfigManager.java | 10 -
.../apache/iotdb/confignode/manager/IManager.java | 9 -
.../iotdb/confignode/manager/ProcedureManager.java | 234 ++++++-----
.../manager/load/service/StatisticsService.java | 26 +-
.../iotdb/confignode/manager/node/NodeManager.java | 8 +-
.../manager/partition/PartitionManager.java | 46 ++-
.../persistence/executor/ConfigPlanExecutor.java | 6 +
.../partition/DatabasePartitionTable.java | 16 +-
.../persistence/partition/PartitionInfo.java | 41 +-
.../procedure/env/ConfigNodeProcedureEnv.java | 32 +-
...moveHandler.java => RegionMaintainHandler.java} | 181 +++++----
.../procedure/impl/StateMachineProcedure.java | 14 +-
.../impl/node/RemoveDataNodeProcedure.java | 22 +-
.../impl/region/AddRegionPeerProcedure.java | 203 ++++++++++
.../impl/region/RegionMigrateProcedure.java | 182 +++------
.../impl/region/RemoveRegionPeerProcedure.java | 188 +++++++++
...ransitionState.java => AddRegionPeerState.java} | 10 +-
.../procedure/state/RegionTransitionState.java | 5 +-
...sitionState.java => RemoveRegionPeerState.java} | 8 +-
.../procedure/store/ProcedureFactory.java | 12 +
.../confignode/procedure/store/ProcedureType.java | 2 +
.../thrift/ConfigNodeRPCServiceProcessor.java | 6 -
.../request/ConfigPhysicalPlanSerDeTest.java | 36 ++
...reTest.java => AddRegionPeerProcedureTest.java} | 16 +-
.../impl/region/RegionMigrateProcedureTest.java | 16 +-
...est.java => RemoveRegionPeerProcedureTest.java} | 16 +-
.../org/apache/iotdb/consensus/IConsensus.java | 12 +
.../org/apache/iotdb/consensus/common/Peer.java | 21 +
.../apache/iotdb/consensus/iot/IoTConsensus.java | 24 ++
.../consensus/iot/IoTConsensusServerImpl.java | 102 +++--
.../iotdb/consensus/ratis/RatisConsensus.java | 17 +
.../iotdb/consensus/simple/SimpleConsensus.java | 5 +
.../iotdb/db/protocol/client/ConfigNodeClient.java | 7 -
.../impl/DataNodeInternalRPCServiceImpl.java | 13 +
.../iotdb/db/service/RegionMigrateService.java | 169 +++++---
.../iotdb/commons/utils/DataNodeKillPoints.java} | 16 +-
.../org/apache/iotdb/commons/utils/FileUtils.java | 10 +
.../thrift-commons/src/main/thrift/common.thrift | 17 +-
.../src/main/thrift/confignode.thrift | 9 -
.../src/main/thrift/datanode.thrift | 16 +
53 files changed, 1727 insertions(+), 628 deletions(-)
diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/env/AbstractEnv.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/env/AbstractEnv.java
index 17086de3cb2..766f2e0c617 100644
--- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/env/AbstractEnv.java
+++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/env/AbstractEnv.java
@@ -29,8 +29,10 @@ import org.apache.iotdb.commons.client.exception.ClientManagerException;
import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
import org.apache.iotdb.commons.cluster.NodeStatus;
import org.apache.iotdb.confignode.rpc.thrift.IConfigNodeRPCService;
+import org.apache.iotdb.confignode.rpc.thrift.TDataNodeInfo;
import org.apache.iotdb.confignode.rpc.thrift.TRegionInfo;
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;
import org.apache.iotdb.isession.ISession;
@@ -1012,4 +1014,22 @@ public abstract class AbstractEnv implements BaseEnv {
public String getLibPath() {
return TEMPLATE_NODE_LIB_PATH;
}
+
+ @Override
+ public Optional<DataNodeWrapper> dataNodeIdToWrapper(int nodeId) {
+ try (SyncConfigNodeIServiceClient leaderClient =
+ (SyncConfigNodeIServiceClient) getLeaderConfigNodeConnection()) {
+ TShowDataNodesResp resp = leaderClient.showDataNodes();
+ for (TDataNodeInfo dataNodeInfo : resp.getDataNodesInfoList()) {
+ if (dataNodeInfo.getDataNodeId() == nodeId) {
+ return dataNodeWrapperList.stream()
+ .filter(dataNodeWrapper -> dataNodeWrapper.getPort() == dataNodeInfo.getRpcPort())
+ .findAny();
+ }
+ }
+ return Optional.empty();
+ } catch (Exception e) {
+ return Optional.empty();
+ }
+ }
}
diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/node/AbstractNodeWrapper.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/node/AbstractNodeWrapper.java
index 550635a69d9..4e6372d5b8d 100644
--- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/node/AbstractNodeWrapper.java
+++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/node/AbstractNodeWrapper.java
@@ -553,7 +553,7 @@ public abstract class AbstractNodeWrapper implements BaseNodeWrapper {
return clusterIndex + HYPHEN + outputCommonConfig.getClusterConfigStr();
}
- protected String getNodePath() {
+ public String getNodePath() {
return System.getProperty(USER_DIR) + File.separator + TARGET + File.separator + getId();
}
diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/env/RemoteServerEnv.java b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/env/RemoteServerEnv.java
index 2b3dcee92cb..9dd357dbae6 100644
--- a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/env/RemoteServerEnv.java
+++ b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/env/RemoteServerEnv.java
@@ -49,6 +49,7 @@ import java.sql.Statement;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.Optional;
import static org.apache.iotdb.jdbc.Config.VERSION;
@@ -366,4 +367,9 @@ public class RemoteServerEnv implements BaseEnv {
public String getLibPath() {
throw new UnsupportedOperationException();
}
+
+ @Override
+ public Optional<DataNodeWrapper> dataNodeIdToWrapper(int nodeId) {
+ throw new UnsupportedOperationException();
+ }
}
diff --git a/integration-test/src/main/java/org/apache/iotdb/itbase/env/BaseEnv.java b/integration-test/src/main/java/org/apache/iotdb/itbase/env/BaseEnv.java
index b74f78ee615..38b2383baa8 100644
--- a/integration-test/src/main/java/org/apache/iotdb/itbase/env/BaseEnv.java
+++ b/integration-test/src/main/java/org/apache/iotdb/itbase/env/BaseEnv.java
@@ -39,6 +39,7 @@ import java.net.URL;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.List;
+import java.util.Optional;
public interface BaseEnv {
@@ -252,4 +253,6 @@ public interface BaseEnv {
String getToolsPath();
String getLibPath();
+
+ Optional<DataNodeWrapper> dataNodeIdToWrapper(int nodeId);
}
diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/IoTDBRegionMigrateReliabilityIT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/IoTDBRegionMigrateReliabilityIT.java
new file mode 100644
index 00000000000..9073bcc1617
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/confignode/it/IoTDBRegionMigrateReliabilityIT.java
@@ -0,0 +1,428 @@
+/*
+ * 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.it;
+
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
+import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
+import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
+import org.apache.iotdb.commons.conf.IoTDBConstant;
+import org.apache.iotdb.commons.utils.DataNodeKillPoints;
+import org.apache.iotdb.confignode.procedure.state.AddRegionPeerState;
+import org.apache.iotdb.confignode.procedure.state.RegionTransitionState;
+import org.apache.iotdb.confignode.procedure.state.RemoveRegionPeerState;
+import org.apache.iotdb.consensus.ConsensusFactory;
+import org.apache.iotdb.db.queryengine.common.header.ColumnHeaderConstant;
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.env.cluster.node.AbstractNodeWrapper;
+import org.apache.iotdb.it.env.cluster.node.ConfigNodeWrapper;
+import org.apache.iotdb.itbase.exception.InconsistentDataException;
+
+import org.awaitility.Awaitility;
+import org.awaitility.core.ConditionTimeoutException;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentHashMap.KeySetView;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+
+public class IoTDBRegionMigrateReliabilityIT {
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(IoTDBRegionMigrateReliabilityIT.class);
+ private static final String INSERTION =
+ "INSERT INTO root.sg.d1(timestamp,speed,temperature) values(100, 10.1, 20.7)";
+ private static final String SHOW_REGIONS = "show regions";
+ private static final String SHOW_DATANODES = "show datanodes";
+ private static final String REGION_MIGRATE_COMMAND_FORMAT = "migrate region %d from %d to %d";
+
+ @Before
+ public void setUp() throws Exception {
+ EnvFactory.getEnv()
+ .getConfig()
+ .getCommonConfig()
+ .setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS)
+ .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS);
+ }
+
+ @After
+ public void tearDown() throws InterruptedException {
+ EnvFactory.getEnv().cleanClusterEnvironment();
+ }
+
+ // region Normal tests
+
+ @Test
+ public void normal1C2DTest() throws Exception {
+ generalTest(1, 1, 1, 2, buildSet(), buildSet());
+ }
+
+ @Test
+ public void normal3C3DTest() throws Exception {
+ generalTest(2, 3, 3, 3, buildSet(), buildSet());
+ }
+
+ // endregion
+
+ // region ConfigNode crash tests
+ @Test
+ public void cnCrashDuringPreCheck() throws Exception {
+ generalTest(
+ 1, 1, 1, 2, buildSet(RegionTransitionState.REGION_MIGRATE_PREPARE.toString()), buildSet());
+ }
+
+ @Test
+ public void cnCrashDuringCreatePeer() throws Exception {
+ generalTest(
+ 1, 1, 1, 2, buildSet(AddRegionPeerState.CREATE_NEW_REGION_PEER.toString()), buildSet());
+ }
+
+ @Test
+ public void cnCrashDuringDoAddPeer() throws Exception {
+ generalTest(1, 1, 1, 2, buildSet(AddRegionPeerState.DO_ADD_REGION_PEER.toString()), buildSet());
+ }
+
+ @Test
+ public void cnCrashDuringUpdateCache() throws Exception {
+ generalTest(
+ 1,
+ 1,
+ 1,
+ 2,
+ buildSet(AddRegionPeerState.UPDATE_REGION_LOCATION_CACHE.toString()),
+ buildSet());
+ }
+
+ @Test
+ public void cnCrashDuringChangeRegionLeader() throws Exception {
+ generalTest(
+ 1, 1, 1, 2, buildSet(RegionTransitionState.CHANGE_REGION_LEADER.toString()), buildSet());
+ }
+
+ @Test
+ public void cnCrashDuringRemoveRegionPeer() throws Exception {
+ generalTest(
+ 1, 1, 1, 2, buildSet(RemoveRegionPeerState.REMOVE_REGION_PEER.toString()), buildSet());
+ }
+
+ @Test
+ public void cnCrashDuringDeleteOldRegionPeer() throws Exception {
+ generalTest(
+ 1, 1, 1, 2, buildSet(RemoveRegionPeerState.DELETE_OLD_REGION_PEER.toString()), buildSet());
+ }
+
+ @Test
+ public void cnCrashDuringRemoveRegionLocationCache() throws Exception {
+ generalTest(
+ 1,
+ 1,
+ 1,
+ 2,
+ buildSet(RemoveRegionPeerState.REMOVE_REGION_LOCATION_CACHE.toString()),
+ buildSet());
+ }
+
+ @Test
+ public void cnCrashTest() throws Exception {
+ KeySetView<String, Boolean> killConfigNodeKeywords = buildSet();
+ killConfigNodeKeywords.addAll(
+ Arrays.stream(AddRegionPeerState.values())
+ .map(Enum::toString)
+ .collect(Collectors.toList()));
+ killConfigNodeKeywords.addAll(
+ Arrays.stream(RemoveRegionPeerState.values())
+ .map(Enum::toString)
+ .collect(Collectors.toList()));
+ generalTest(1, 1, 1, 2, killConfigNodeKeywords, buildSet());
+ }
+
+ @Ignore
+ @Test
+ public void badKillPoint() throws Exception {
+ generalTest(1, 1, 1, 2, buildSet("??"), buildSet());
+ }
+
+ // endregion
+
+ // region coordinator DataNode crash tests
+
+ @Test
+ public void coordinatorCrashDuringRemovePeer() throws Exception {
+ generalTest(1, 1, 1, 2, buildSet(), buildSet(DataNodeKillPoints.CoordinatorRemovePeer.name()));
+ }
+
+ // endregion
+
+ // region original DataNode crash tests
+
+ @Test
+ public void originalCrashDuringRemovePeer() throws Exception {
+ generalTest(1, 1, 1, 2, buildSet(), buildSet(DataNodeKillPoints.OriginalRemovePeer.name()));
+ }
+
+ @Test
+ public void originalCrashDuringDeleteLocalPeer() throws Exception {
+ generalTest(
+ 1, 1, 1, 2, buildSet(), buildSet(DataNodeKillPoints.OriginalDeleteOldRegionPeer.name()));
+ }
+
+ // region Helpers
+
+ public void generalTest(
+ final int dataReplicateFactor,
+ final int schemaReplicationFactor,
+ final int configNodeNum,
+ final int dataNodeNum,
+ KeySetView<String, Boolean> killConfigNodeKeywords,
+ KeySetView<String, Boolean> killDataNodeKeywords)
+ throws Exception {
+ // prepare env
+ EnvFactory.getEnv()
+ .getConfig()
+ .getCommonConfig()
+ .setDataReplicationFactor(dataReplicateFactor)
+ .setSchemaReplicationFactor(schemaReplicationFactor);
+ EnvFactory.getEnv().initClusterEnvironment(configNodeNum, dataNodeNum);
+
+ ExecutorService service = IoTDBThreadPoolFactory.newCachedThreadPool("regionMigrateIT");
+ EnvFactory.getEnv()
+ .getConfigNodeWrapperList()
+ .forEach(
+ configNodeWrapper ->
+ service.submit(() -> nodeLogKillPoint(configNodeWrapper, killConfigNodeKeywords)));
+ EnvFactory.getEnv()
+ .getDataNodeWrapperList()
+ .forEach(
+ dataNodeWrapper ->
+ service.submit(() -> nodeLogKillPoint(dataNodeWrapper, killDataNodeKeywords)));
+
+ try (final Connection connection = EnvFactory.getEnv().getConnection();
+ final Statement statement = connection.createStatement();
+ final SyncConfigNodeIServiceClient configClient =
+ (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
+
+ statement.execute(INSERTION);
+
+ ResultSet result = statement.executeQuery(SHOW_REGIONS);
+ Map<Integer, Set<Integer>> regionMap = getRegionMap(result);
+
+ result = statement.executeQuery(SHOW_DATANODES);
+ Set<Integer> dataNodeSet = new HashSet<>();
+ while (result.next()) {
+ dataNodeSet.add(result.getInt(ColumnHeaderConstant.NODE_ID));
+ }
+
+ final int selectedRegion = selectRegion(regionMap);
+ final int originalDataNode = selectOriginalDataNode(regionMap, selectedRegion);
+ final int destDataNode = selectDestDataNode(dataNodeSet, regionMap, selectedRegion);
+
+ statement.execute(regionMigrateCommand(selectedRegion, originalDataNode, destDataNode));
+
+ awaitUntilSuccess(statement, selectedRegion, originalDataNode, destDataNode);
+
+ // make sure all kill points have been triggered
+ Assert.assertTrue(killConfigNodeKeywords.isEmpty());
+ Assert.assertTrue(killDataNodeKeywords.isEmpty());
+
+ checkRegionFileClear(originalDataNode);
+ } catch (InconsistentDataException ignore) {
+
+ }
+ LOGGER.info("test pass");
+ }
+
+ /**
+ * Monitor the node's log and do something.
+ *
+ * @param nodeWrapper Easy to understand
+ * @param killNodeKeywords When detect these keywords in node's log, stop the node forcibly
+ */
+ private static void nodeLogKillPoint(
+ AbstractNodeWrapper nodeWrapper, KeySetView<String, Boolean> killNodeKeywords) {
+ if (killNodeKeywords.isEmpty()) {
+ return;
+ }
+ final String logFileName;
+ if (nodeWrapper instanceof ConfigNodeWrapper) {
+ logFileName = "log_confignode_all.log";
+ } else {
+ logFileName = "log_datanode_all.log";
+ }
+ ProcessBuilder builder =
+ new ProcessBuilder(
+ "tail",
+ "-f",
+ nodeWrapper.getNodePath() + File.separator + "logs" + File.separator + logFileName);
+ builder.redirectErrorStream(true);
+
+ try {
+ Process process = builder.start();
+ try (BufferedReader reader =
+ new BufferedReader(new InputStreamReader(process.getInputStream()))) {
+ String line;
+ while ((line = reader.readLine()) != null) {
+ // if trigger more than one keyword at a same time, test code may have mistakes
+ Assert.assertTrue(killNodeKeywords.stream().filter(line::contains).count() <= 1);
+ String finalLine = line;
+ Optional<String> detectedKeyword =
+ killNodeKeywords.stream()
+ .filter(keyword -> finalLine.contains("breakpoint:" + keyword))
+ .findAny();
+ if (detectedKeyword.isPresent()) {
+ // each keyword only trigger once
+ killNodeKeywords.remove(detectedKeyword.get());
+ // reboot the node
+ nodeWrapper.stopForcibly();
+ nodeWrapper.start();
+ }
+ if (killNodeKeywords.isEmpty()) {
+ break;
+ }
+ }
+ }
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static String regionMigrateCommand(int who, int from, int to) {
+ return String.format(REGION_MIGRATE_COMMAND_FORMAT, who, from, to);
+ }
+
+ private static Map<Integer, Set<Integer>> getRegionMap(ResultSet showRegionsResult)
+ throws SQLException {
+ Map<Integer, Set<Integer>> regionMap = new HashMap<>();
+ while (showRegionsResult.next()) {
+ if (String.valueOf(TConsensusGroupType.DataRegion)
+ .equals(showRegionsResult.getString(ColumnHeaderConstant.TYPE))) {
+ int region = showRegionsResult.getInt(ColumnHeaderConstant.REGION_ID);
+ int dataNode = showRegionsResult.getInt(ColumnHeaderConstant.DATA_NODE_ID);
+ regionMap.putIfAbsent(region, new HashSet<>());
+ regionMap.get(region).add(dataNode);
+ }
+ }
+ return regionMap;
+ }
+
+ private static int selectRegion(Map<Integer, Set<Integer>> regionMap) {
+ return regionMap.keySet().stream().findAny().orElseThrow(() -> new RuntimeException("gg"));
+ }
+
+ private static int selectOriginalDataNode(
+ Map<Integer, Set<Integer>> regionMap, int selectedRegion) {
+ return regionMap.get(selectedRegion).stream()
+ .findAny()
+ .orElseThrow(() -> new RuntimeException("cannot find original DataNode"));
+ }
+
+ private static int selectDestDataNode(
+ Set<Integer> dataNodeSet, Map<Integer, Set<Integer>> regionMap, int selectedRegion) {
+ return dataNodeSet.stream()
+ .filter(dataNodeId -> !regionMap.get(selectedRegion).contains(dataNodeId))
+ .findAny()
+ .orElseThrow(() -> new RuntimeException("cannot find dest DataNode"));
+ }
+
+ private static void awaitUntilSuccess(
+ Statement statement, int selectedRegion, int originalDataNode, int destDataNode) {
+ AtomicReference<Set<Integer>> lastTimeDataNodes = new AtomicReference<>();
+ AtomicReference<Exception> lastException = new AtomicReference<>();
+ try {
+ Awaitility.await()
+ .atMost(1, TimeUnit.MINUTES)
+ .until(
+ () -> {
+ try {
+ Map<Integer, Set<Integer>> newRegionMap =
+ getRegionMap(statement.executeQuery(SHOW_REGIONS));
+ Set<Integer> dataNodes = newRegionMap.get(selectedRegion);
+ lastTimeDataNodes.set(dataNodes);
+ return !dataNodes.contains(originalDataNode) && dataNodes.contains(destDataNode);
+ } catch (Exception e) {
+ // Any exception can be ignored
+ lastException.set(e);
+ return false;
+ }
+ });
+ } catch (ConditionTimeoutException e) {
+ if (lastTimeDataNodes.get() == null) {
+ LOGGER.error(
+ "maybe show regions fail, lastTimeDataNodes is null, last Exception:",
+ lastException.get());
+ throw e;
+ }
+ String actualSetStr = lastTimeDataNodes.get().toString();
+ lastTimeDataNodes.get().remove(originalDataNode);
+ lastTimeDataNodes.get().add(destDataNode);
+ String expectSetStr = lastTimeDataNodes.toString();
+ LOGGER.info("DataNode Set {} is unexpected, expect {}", actualSetStr, expectSetStr);
+ throw e;
+ }
+ }
+
+ /** Check whether the original DataNode's region file has been deleted. */
+ private static void checkRegionFileClear(int dataNode) {
+ String nodePath = EnvFactory.getEnv().dataNodeIdToWrapper(dataNode).get().getNodePath();
+ File originalRegionDir =
+ new File(
+ nodePath
+ + File.separator
+ + IoTDBConstant.DATA_FOLDER_NAME
+ + File.separator
+ + "datanode"
+ + File.separator
+ + IoTDBConstant.CONSENSUS_FOLDER_NAME
+ + File.separator
+ + "data_region");
+ Assert.assertTrue(originalRegionDir.isDirectory());
+ Assert.assertEquals(0, Objects.requireNonNull(originalRegionDir.listFiles()).length);
+ }
+
+ private static KeySetView<String, Boolean> buildSet(String... keywords) {
+ KeySetView<String, Boolean> result = ConcurrentHashMap.newKeySet();
+ result.addAll(Arrays.asList(keywords));
+ return result;
+ }
+
+ // endregion
+}
diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/procedure/IoTDBProcedureIT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/procedure/IoTDBProcedureIT.java
index 6b92ca8c64f..0722dde7489 100644
--- a/integration-test/src/test/java/org/apache/iotdb/confignode/it/procedure/IoTDBProcedureIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/confignode/it/procedure/IoTDBProcedureIT.java
@@ -100,11 +100,13 @@ public class IoTDBProcedureIT {
@Test
public void procedureRecoverAtAnotherConfigNodeTest() throws Exception {
recoverTest(3, false);
+ LOGGER.info("test pass");
}
@Test
public void procedureRecoverAtTheSameConfigNodeTest() throws Exception {
recoverTest(1, true);
+ LOGGER.info("test pass");
}
private void recoverTest(int configNodeNum, boolean needRestartLeader) throws Exception {
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/DataNodeRequestType.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/DataNodeRequestType.java
index db0007f1a95..70b1a14a88e 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/DataNodeRequestType.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/DataNodeRequestType.java
@@ -42,6 +42,7 @@ public enum DataNodeRequestType {
ADD_REGION_PEER,
REMOVE_REGION_PEER,
DELETE_OLD_REGION_PEER,
+ RESET_PEER_LIST,
UPDATE_REGION_ROUTE_MAP,
CHANGE_REGION_LEADER,
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/SyncDataNodeClientPool.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/SyncDataNodeClientPool.java
index 7b1549d44cf..32d09ca5191 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/SyncDataNodeClientPool.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/SyncDataNodeClientPool.java
@@ -36,6 +36,7 @@ import org.apache.iotdb.mpp.rpc.thrift.TInvalidateCacheReq;
import org.apache.iotdb.mpp.rpc.thrift.TInvalidatePermissionCacheReq;
import org.apache.iotdb.mpp.rpc.thrift.TMaintainPeerReq;
import org.apache.iotdb.mpp.rpc.thrift.TRegionLeaderChangeReq;
+import org.apache.iotdb.mpp.rpc.thrift.TResetPeerListReq;
import org.apache.iotdb.mpp.rpc.thrift.TUpdateTemplateReq;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
@@ -134,6 +135,8 @@ public class SyncDataNodeClientPool {
return client.removeRegionPeer((TMaintainPeerReq) req);
case DELETE_OLD_REGION_PEER:
return client.deleteOldRegionPeer((TMaintainPeerReq) req);
+ case RESET_PEER_LIST:
+ return client.resetPeerList((TResetPeerListReq) req);
default:
return RpcUtils.getStatus(
TSStatusCode.EXECUTE_STATEMENT_ERROR, "Unknown request type: " + requestType);
@@ -150,7 +153,7 @@ public class SyncDataNodeClientPool {
TimeUnit.MILLISECONDS.sleep(3200L);
}
} catch (InterruptedException e) {
- LOGGER.error("Retry wait failed.", e);
+ LOGGER.warn("Retry wait failed.", e);
Thread.currentThread().interrupt();
}
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java
index ee944226022..0d5536f72c5 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java
@@ -73,8 +73,10 @@ import org.apache.iotdb.confignode.consensus.request.write.datanode.RemoveDataNo
import org.apache.iotdb.confignode.consensus.request.write.datanode.UpdateDataNodePlan;
import org.apache.iotdb.confignode.consensus.request.write.function.CreateFunctionPlan;
import org.apache.iotdb.confignode.consensus.request.write.function.DropFunctionPlan;
+import org.apache.iotdb.confignode.consensus.request.write.partition.AddRegionLocationPlan;
import org.apache.iotdb.confignode.consensus.request.write.partition.CreateDataPartitionPlan;
import org.apache.iotdb.confignode.consensus.request.write.partition.CreateSchemaPartitionPlan;
+import org.apache.iotdb.confignode.consensus.request.write.partition.RemoveRegionLocationPlan;
import org.apache.iotdb.confignode.consensus.request.write.partition.UpdateRegionLocationPlan;
import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeDeactivateTemplatePlan;
import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeDeleteLogicalViewPlan;
@@ -224,6 +226,15 @@ public abstract class ConfigPhysicalPlan implements IConsensusRequest {
case CreateRegionGroups:
plan = new CreateRegionGroupsPlan();
break;
+ case UpdateRegionLocation:
+ plan = new UpdateRegionLocationPlan();
+ break;
+ case AddRegionLocation:
+ plan = new AddRegionLocationPlan();
+ break;
+ case RemoveRegionLocation:
+ plan = new RemoveRegionLocationPlan();
+ break;
case OfferRegionMaintainTasks:
plan = new OfferRegionMaintainTasksPlan();
break;
@@ -386,9 +397,6 @@ public abstract class ConfigPhysicalPlan implements IConsensusRequest {
case GetRegionInfoList:
plan = new GetRegionInfoListPlan();
break;
- case UpdateRegionLocation:
- plan = new UpdateRegionLocationPlan();
- break;
case CreatePipeSinkV1:
plan = new CreatePipeSinkPlanV1();
break;
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanType.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanType.java
index cc963606c61..65dcce67828 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanType.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanType.java
@@ -53,6 +53,7 @@ public enum ConfigPhysicalPlanType {
CreateRegionGroups((short) 300),
DeleteRegionGroups((short) 301),
GetRegionInfoList((short) 302),
+ @Deprecated
UpdateRegionLocation((short) 303),
OfferRegionMaintainTasks((short) 304),
PollRegionMaintainTask((short) 305),
@@ -60,8 +61,9 @@ public enum ConfigPhysicalPlanType {
GetSeriesSlotList((short) 307),
GetTimeSlotList((short) 308),
PollSpecificRegionMaintainTask((short) 309),
-
CountTimeSlotList((short) 310),
+ AddRegionLocation((short) 311),
+ RemoveRegionLocation((short) 312),
/** Partition. */
GetSchemaPartition((short) 400),
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/partition/UpdateRegionLocationPlan.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/partition/AddRegionLocationPlan.java
similarity index 62%
copy from iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/partition/UpdateRegionLocationPlan.java
copy to iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/partition/AddRegionLocationPlan.java
index 06c94b07394..7f056bc37e7 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/partition/UpdateRegionLocationPlan.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/partition/AddRegionLocationPlan.java
@@ -29,59 +29,39 @@ import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
-public class UpdateRegionLocationPlan extends ConfigPhysicalPlan {
- /*which region*/
+public class AddRegionLocationPlan extends ConfigPhysicalPlan {
TConsensusGroupId regionId;
- /*remove it from the region's location*/
- TDataNodeLocation oldNode;
+ TDataNodeLocation newLocation;
- /*add it to the region's location*/
- TDataNodeLocation newNode;
-
- public UpdateRegionLocationPlan() {
- super(ConfigPhysicalPlanType.UpdateRegionLocation);
+ public AddRegionLocationPlan() {
+ super(ConfigPhysicalPlanType.AddRegionLocation);
}
- /**
- * Constructor.
- *
- * @param regionId update the region location
- * @param oldNode remove the old location
- * @param newNode add the new location
- */
- public UpdateRegionLocationPlan(
- TConsensusGroupId regionId, TDataNodeLocation oldNode, TDataNodeLocation newNode) {
+ public AddRegionLocationPlan(TConsensusGroupId regionId, TDataNodeLocation newLocation) {
this();
this.regionId = regionId;
- this.oldNode = oldNode;
- this.newNode = newNode;
+ this.newLocation = newLocation;
}
@Override
protected void serializeImpl(DataOutputStream stream) throws IOException {
stream.writeShort(getType().getPlanType());
ThriftCommonsSerDeUtils.serializeTConsensusGroupId(regionId, stream);
- ThriftCommonsSerDeUtils.serializeTDataNodeLocation(oldNode, stream);
- ThriftCommonsSerDeUtils.serializeTDataNodeLocation(newNode, stream);
+ ThriftCommonsSerDeUtils.serializeTDataNodeLocation(newLocation, stream);
}
@Override
protected void deserializeImpl(ByteBuffer buffer) throws IOException {
regionId = ThriftCommonsSerDeUtils.deserializeTConsensusGroupId(buffer);
- oldNode = ThriftCommonsSerDeUtils.deserializeTDataNodeLocation(buffer);
- newNode = ThriftCommonsSerDeUtils.deserializeTDataNodeLocation(buffer);
+ newLocation = ThriftCommonsSerDeUtils.deserializeTDataNodeLocation(buffer);
}
public TConsensusGroupId getRegionId() {
return regionId;
}
- public TDataNodeLocation getOldNode() {
- return oldNode;
- }
-
- public TDataNodeLocation getNewNode() {
- return newNode;
+ public TDataNodeLocation getNewLocation() {
+ return newLocation;
}
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/partition/UpdateRegionLocationPlan.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/partition/RemoveRegionLocationPlan.java
similarity index 63%
copy from iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/partition/UpdateRegionLocationPlan.java
copy to iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/partition/RemoveRegionLocationPlan.java
index 06c94b07394..9dc6a007052 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/partition/UpdateRegionLocationPlan.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/partition/RemoveRegionLocationPlan.java
@@ -29,59 +29,45 @@ import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
-public class UpdateRegionLocationPlan extends ConfigPhysicalPlan {
- /*which region*/
+public class RemoveRegionLocationPlan extends ConfigPhysicalPlan {
TConsensusGroupId regionId;
- /*remove it from the region's location*/
- TDataNodeLocation oldNode;
+ TDataNodeLocation deprecatedLocation;
- /*add it to the region's location*/
- TDataNodeLocation newNode;
-
- public UpdateRegionLocationPlan() {
- super(ConfigPhysicalPlanType.UpdateRegionLocation);
+ public RemoveRegionLocationPlan() {
+ super(ConfigPhysicalPlanType.RemoveRegionLocation);
}
- /**
- * Constructor.
- *
- * @param regionId update the region location
- * @param oldNode remove the old location
- * @param newNode add the new location
- */
- public UpdateRegionLocationPlan(
- TConsensusGroupId regionId, TDataNodeLocation oldNode, TDataNodeLocation newNode) {
+ public RemoveRegionLocationPlan(
+ TConsensusGroupId regionId, TDataNodeLocation deprecatedLocation) {
this();
this.regionId = regionId;
- this.oldNode = oldNode;
- this.newNode = newNode;
+ this.deprecatedLocation = deprecatedLocation;
}
@Override
protected void serializeImpl(DataOutputStream stream) throws IOException {
stream.writeShort(getType().getPlanType());
ThriftCommonsSerDeUtils.serializeTConsensusGroupId(regionId, stream);
- ThriftCommonsSerDeUtils.serializeTDataNodeLocation(oldNode, stream);
- ThriftCommonsSerDeUtils.serializeTDataNodeLocation(newNode, stream);
+ ThriftCommonsSerDeUtils.serializeTDataNodeLocation(deprecatedLocation, stream);
}
@Override
protected void deserializeImpl(ByteBuffer buffer) throws IOException {
regionId = ThriftCommonsSerDeUtils.deserializeTConsensusGroupId(buffer);
- oldNode = ThriftCommonsSerDeUtils.deserializeTDataNodeLocation(buffer);
- newNode = ThriftCommonsSerDeUtils.deserializeTDataNodeLocation(buffer);
+ deprecatedLocation = ThriftCommonsSerDeUtils.deserializeTDataNodeLocation(buffer);
}
- public TConsensusGroupId getRegionId() {
- return regionId;
+ @Override
+ public ConfigPhysicalPlanType getType() {
+ return super.getType();
}
- public TDataNodeLocation getOldNode() {
- return oldNode;
+ public TConsensusGroupId getRegionId() {
+ return regionId;
}
- public TDataNodeLocation getNewNode() {
- return newNode;
+ public TDataNodeLocation getDeprecatedLocation() {
+ return deprecatedLocation;
}
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/partition/UpdateRegionLocationPlan.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/partition/UpdateRegionLocationPlan.java
index 06c94b07394..91b47d06200 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/partition/UpdateRegionLocationPlan.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/partition/UpdateRegionLocationPlan.java
@@ -29,6 +29,7 @@ import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
+@Deprecated
public class UpdateRegionLocationPlan extends ConfigPhysicalPlan {
/*which region*/
TConsensusGroupId regionId;
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
index 812b232340b..a72e48a874b 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
@@ -163,7 +163,6 @@ import org.apache.iotdb.confignode.rpc.thrift.TNodeVersionInfo;
import org.apache.iotdb.confignode.rpc.thrift.TPermissionInfoResp;
import org.apache.iotdb.confignode.rpc.thrift.TPipeConfigTransferReq;
import org.apache.iotdb.confignode.rpc.thrift.TPipeConfigTransferResp;
-import org.apache.iotdb.confignode.rpc.thrift.TRegionMigrateResultReportReq;
import org.apache.iotdb.confignode.rpc.thrift.TRegionRouteMapResp;
import org.apache.iotdb.confignode.rpc.thrift.TSchemaNodeManagementResp;
import org.apache.iotdb.confignode.rpc.thrift.TSchemaPartitionTableResp;
@@ -440,15 +439,6 @@ public class ConfigManager implements IManager {
return status;
}
- @Override
- public TSStatus reportRegionMigrateResult(TRegionMigrateResultReportReq req) {
- TSStatus status = confirmLeader();
- if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- procedureManager.reportRegionMigrateResult(req);
- }
- return status;
- }
-
@Override
public DataSet getDataNodeConfiguration(
GetDataNodeConfigurationPlan getDataNodeConfigurationPlan) {
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java
index 0e0c1f41af3..05288fc46a0 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java
@@ -100,7 +100,6 @@ import org.apache.iotdb.confignode.rpc.thrift.TMigrateRegionReq;
import org.apache.iotdb.confignode.rpc.thrift.TPermissionInfoResp;
import org.apache.iotdb.confignode.rpc.thrift.TPipeConfigTransferReq;
import org.apache.iotdb.confignode.rpc.thrift.TPipeConfigTransferResp;
-import org.apache.iotdb.confignode.rpc.thrift.TRegionMigrateResultReportReq;
import org.apache.iotdb.confignode.rpc.thrift.TRegionRouteMapResp;
import org.apache.iotdb.confignode.rpc.thrift.TSchemaNodeManagementResp;
import org.apache.iotdb.confignode.rpc.thrift.TSchemaPartitionTableResp;
@@ -272,14 +271,6 @@ public interface IManager {
*/
TSStatus reportDataNodeShutdown(TDataNodeLocation dataNodeLocation);
- /**
- * DataNode report region migrate result to ConfigNode when remove DataNode.
- *
- * @param req TRegionMigrateResultReportReq
- * @return TSStatus
- */
- TSStatus reportRegionMigrateResult(TRegionMigrateResultReportReq req);
-
/**
* Get DataNode info.
*
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 ad3ae6b5b92..9f8a390d5d7 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
@@ -51,6 +51,8 @@ import org.apache.iotdb.confignode.procedure.Procedure;
import org.apache.iotdb.confignode.procedure.ProcedureExecutor;
import org.apache.iotdb.confignode.procedure.ProcedureMetrics;
import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
+import org.apache.iotdb.confignode.procedure.env.RegionMaintainHandler;
+import org.apache.iotdb.confignode.procedure.exception.ProcedureException;
import org.apache.iotdb.confignode.procedure.impl.cq.CreateCQProcedure;
import org.apache.iotdb.confignode.procedure.impl.node.AddConfigNodeProcedure;
import org.apache.iotdb.confignode.procedure.impl.node.RemoveConfigNodeProcedure;
@@ -102,7 +104,6 @@ import org.apache.iotdb.confignode.rpc.thrift.TCreateTopicReq;
import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema;
import org.apache.iotdb.confignode.rpc.thrift.TDeleteLogicalViewReq;
import org.apache.iotdb.confignode.rpc.thrift.TMigrateRegionReq;
-import org.apache.iotdb.confignode.rpc.thrift.TRegionMigrateResultReportReq;
import org.apache.iotdb.confignode.rpc.thrift.TSubscribeReq;
import org.apache.iotdb.confignode.rpc.thrift.TUnsubscribeReq;
import org.apache.iotdb.db.exception.BatchProcessException;
@@ -124,10 +125,12 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
-import java.util.Set;
+import java.util.Optional;
import java.util.concurrent.ScheduledExecutorService;
import java.util.stream.Collectors;
+import static org.apache.iotdb.confignode.conf.ConfigNodeConstant.REGION_MIGRATE_PROCESS;
+
public class ProcedureManager {
private static final Logger LOGGER = LoggerFactory.getLogger(ProcedureManager.class);
@@ -540,35 +543,102 @@ public class ProcedureManager {
return true;
}
- public TSStatus migrateRegion(TMigrateRegionReq migrateRegionReq) {
- final TConsensusGroupId regionGroupId;
- final TSStatus status;
+ // region region migration
+
+ private TConsensusGroupId regionIdToTConsensusGroupId(final int regionId)
+ throws ProcedureException {
if (configManager
.getPartitionManager()
- .isRegionGroupExists(
- new TConsensusGroupId(
- TConsensusGroupType.SchemaRegion, migrateRegionReq.getRegionId()))) {
- regionGroupId =
- new TConsensusGroupId(TConsensusGroupType.SchemaRegion, migrateRegionReq.getRegionId());
- } else if (configManager
+ .isRegionGroupExists(new TConsensusGroupId(TConsensusGroupType.SchemaRegion, regionId))) {
+ return new TConsensusGroupId(TConsensusGroupType.SchemaRegion, regionId);
+ }
+ if (configManager
.getPartitionManager()
- .isRegionGroupExists(
- new TConsensusGroupId(
- TConsensusGroupType.DataRegion, migrateRegionReq.getRegionId()))) {
- regionGroupId =
- new TConsensusGroupId(TConsensusGroupType.DataRegion, migrateRegionReq.getRegionId());
- } else {
- LOGGER.warn(
- "Submit RegionMigrateProcedure failed, because RegionGroup: {} doesn't exist",
- migrateRegionReq.getRegionId());
- status = new TSStatus(TSStatusCode.MIGRATE_REGION_ERROR.getStatusCode());
- status.setMessage(
+ .isRegionGroupExists(new TConsensusGroupId(TConsensusGroupType.DataRegion, regionId))) {
+ return new TConsensusGroupId(TConsensusGroupType.DataRegion, regionId);
+ }
+ String msg =
+ String.format(
+ "Submit RegionMigrateProcedure failed, because RegionGroup: %s doesn't exist",
+ regionId);
+ LOGGER.warn(msg);
+ throw new ProcedureException(msg);
+ }
+
+ private TSStatus checkRegionMigrate(
+ TMigrateRegionReq migrateRegionReq,
+ TConsensusGroupId regionGroupId,
+ TDataNodeLocation originalDataNode,
+ TDataNodeLocation destDataNode,
+ TDataNodeLocation coordinatorForAddPeer) {
+ String failMessage = null;
+ if (originalDataNode == null) {
+ failMessage =
String.format(
- "Submit RegionMigrateProcedure failed, because RegionGroup: %s doesn't exist",
- migrateRegionReq.getRegionId()));
- return status;
+ "Submit RegionMigrateProcedure failed, because no original DataNode %d",
+ migrateRegionReq.getFromId());
+ } else if (destDataNode == null) {
+ failMessage =
+ String.format(
+ "Submit RegionMigrateProcedure failed, because no target DataNode %s",
+ migrateRegionReq.getToId());
+ } else if (coordinatorForAddPeer == null) {
+ failMessage =
+ String.format(
+ "%s, There are no other DataNodes could be selected to perform the add peer process, "
+ + "please check RegionGroup: %s by show regions sql command",
+ REGION_MIGRATE_PROCESS, regionGroupId);
+ } else if (configManager.getPartitionManager()
+ .getAllReplicaSets(originalDataNode.getDataNodeId()).stream()
+ .noneMatch(replicaSet -> replicaSet.getRegionId().equals(regionGroupId))) {
+ failMessage =
+ String.format(
+ "Submit RegionMigrateProcedure failed, because the original DataNode %s doesn't contain Region %s",
+ migrateRegionReq.getFromId(), migrateRegionReq.getRegionId());
+ } else if (configManager.getPartitionManager().getAllReplicaSets(destDataNode.getDataNodeId())
+ .stream()
+ .anyMatch(replicaSet -> replicaSet.getRegionId().equals(regionGroupId))) {
+ failMessage =
+ String.format(
+ "Submit RegionMigrateProcedure failed, because the target DataNode %s already contains Region %s",
+ migrateRegionReq.getToId(), migrateRegionReq.getRegionId());
+ } else if (!configManager.getNodeManager().filterDataNodeThroughStatus(NodeStatus.Running)
+ .stream()
+ .map(TDataNodeConfiguration::getLocation)
+ .map(TDataNodeLocation::getDataNodeId)
+ .collect(Collectors.toSet())
+ .contains(migrateRegionReq.getToId())) {
+ // Here we only check Running DataNode to implement migration, because removing nodes may not
+ // exist when add peer is performing
+ failMessage =
+ String.format(
+ "Submit RegionMigrateProcedure failed, because the destDataNode %s is ReadOnly or Unknown.",
+ migrateRegionReq.getToId());
+ }
+ if (failMessage != null) {
+ LOGGER.warn(failMessage);
+ TSStatus failStatus = new TSStatus(TSStatusCode.MIGRATE_REGION_ERROR.getStatusCode());
+ failStatus.setMessage(failMessage);
+ return failStatus;
+ }
+ return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ }
+
+ public TSStatus migrateRegion(TMigrateRegionReq migrateRegionReq) {
+ TConsensusGroupId regionGroupId;
+ Optional<TConsensusGroupId> optional =
+ configManager
+ .getPartitionManager()
+ .generateTConsensusGroupIdByRegionId(migrateRegionReq.getRegionId());
+ if (optional.isPresent()) {
+ regionGroupId = optional.get();
+ } else {
+ LOGGER.error("get region group id fail");
+ return new TSStatus(TSStatusCode.MIGRATE_REGION_ERROR.getStatusCode())
+ .setMessage("get region group id fail");
}
+ // find original dn and dest dn
final TDataNodeLocation originalDataNode =
configManager
.getNodeManager()
@@ -579,87 +649,29 @@ public class ProcedureManager {
.getNodeManager()
.getRegisteredDataNode(migrateRegionReq.getToId())
.getLocation();
-
- if (originalDataNode == null) {
- LOGGER.warn(
- "Submit RegionMigrateProcedure failed, because no original DataNode {}",
- migrateRegionReq.getFromId());
- status = new TSStatus(TSStatusCode.MIGRATE_REGION_ERROR.getStatusCode());
- status.setMessage(
- "Submit RegionMigrateProcedure failed, because no original DataNode "
- + migrateRegionReq.getFromId());
- return status;
- } else if (destDataNode == null) {
- LOGGER.warn(
- "Submit RegionMigrateProcedure failed, because no target DataNode {}",
- migrateRegionReq.getToId());
- status = new TSStatus(TSStatusCode.MIGRATE_REGION_ERROR.getStatusCode());
- status.setMessage(
- "Submit RegionMigrateProcedure failed, because no target DataNode "
- + migrateRegionReq.getToId());
- return status;
- } else if (configManager.getPartitionManager()
- .getAllReplicaSets(originalDataNode.getDataNodeId()).stream()
- .noneMatch(replicaSet -> replicaSet.getRegionId().equals(regionGroupId))) {
- LOGGER.warn(
- "Submit RegionMigrateProcedure failed, because the original DataNode {} doesn't contain Region {}",
- migrateRegionReq.getFromId(),
- migrateRegionReq.getRegionId());
- status = new TSStatus(TSStatusCode.MIGRATE_REGION_ERROR.getStatusCode());
- status.setMessage(
- "Submit RegionMigrateProcedure failed, because the original DataNode "
- + migrateRegionReq.getFromId()
- + " doesn't contain Region "
- + migrateRegionReq.getRegionId());
- return status;
- } else if (configManager.getPartitionManager().getAllReplicaSets(destDataNode.getDataNodeId())
- .stream()
- .anyMatch(replicaSet -> replicaSet.getRegionId().equals(regionGroupId))) {
- LOGGER.warn(
- "Submit RegionMigrateProcedure failed, because the target DataNode {} already contains Region {}",
- migrateRegionReq.getToId(),
- migrateRegionReq.getRegionId());
- status = new TSStatus(TSStatusCode.MIGRATE_REGION_ERROR.getStatusCode());
- status.setMessage(
- "Submit RegionMigrateProcedure failed, because the target DataNode "
- + migrateRegionReq.getToId()
- + " already contains Region "
- + migrateRegionReq.getRegionId());
- return status;
- }
- // Here we only check Running DataNode to implement migration, because removing nodes may not
- // exist when add peer is performing
- final Set<Integer> aliveDataNodes =
- configManager.getNodeManager().filterDataNodeThroughStatus(NodeStatus.Running).stream()
- .map(TDataNodeConfiguration::getLocation)
- .map(TDataNodeLocation::getDataNodeId)
- .collect(Collectors.toSet());
- if (NodeStatus.Unknown.equals(
- configManager.getLoadManager().getNodeStatus(migrateRegionReq.getFromId()))) {
- LOGGER.warn(
- "Submit RegionMigrateProcedure failed, because the sourceDataNode {} is Unknown.",
- migrateRegionReq.getFromId());
- status = new TSStatus(TSStatusCode.MIGRATE_REGION_ERROR.getStatusCode());
- status.setMessage(
- "Submit RegionMigrateProcedure failed, because the sourceDataNode "
- + migrateRegionReq.getFromId()
- + " is Unknown.");
+ // select coordinator for adding peer
+ RegionMaintainHandler handler = new RegionMaintainHandler(configManager);
+ final TDataNodeLocation coordinatorForAddPeer =
+ handler.filterDataNodeWithOtherRegionReplica(regionGroupId, destDataNode).orElse(null);
+ // Select coordinator for removing peer
+ // For now, destDataNode temporarily acts as the coordinatorForRemovePeer
+ final TDataNodeLocation coordinatorForRemovePeer = destDataNode;
+
+ TSStatus status =
+ checkRegionMigrate(
+ migrateRegionReq, regionGroupId, originalDataNode, destDataNode, coordinatorForAddPeer);
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
return status;
}
- if (!aliveDataNodes.contains(migrateRegionReq.getToId())) {
- LOGGER.warn(
- "Submit RegionMigrateProcedure failed, because the destDataNode {} is ReadOnly or Unknown.",
- migrateRegionReq.getToId());
- status = new TSStatus(TSStatusCode.MIGRATE_REGION_ERROR.getStatusCode());
- status.setMessage(
- "Submit RegionMigrateProcedure failed, because the destDataNode "
- + migrateRegionReq.getToId()
- + " is ReadOnly or Unknown.");
- return status;
- }
+ // finally, submit procedure
this.executor.submitProcedure(
- new RegionMigrateProcedure(regionGroupId, originalDataNode, destDataNode));
+ new RegionMigrateProcedure(
+ regionGroupId,
+ originalDataNode,
+ destDataNode,
+ coordinatorForAddPeer,
+ coordinatorForRemovePeer));
LOGGER.info(
"Submit RegionMigrateProcedure successfully, Region: {}, From: {}, To: {}",
migrateRegionReq.getRegionId(),
@@ -668,6 +680,8 @@ public class ProcedureManager {
return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
}
+ // endregion
+
/**
* Generate {@link CreateRegionGroupsProcedure} and wait until it finished.
*
@@ -1181,22 +1195,6 @@ public class ProcedureManager {
this.env = env;
}
- public void reportRegionMigrateResult(TRegionMigrateResultReportReq req) {
-
- this.executor
- .getProcedures()
- .values()
- .forEach(
- procedure -> {
- if (procedure instanceof RegionMigrateProcedure) {
- RegionMigrateProcedure regionMigrateProcedure = (RegionMigrateProcedure) procedure;
- if (regionMigrateProcedure.getConsensusGroupId().equals(req.getRegionId())) {
- regionMigrateProcedure.notifyTheRegionMigrateFinished(req);
- }
- }
- });
- }
-
public void addMetrics() {
MetricService.getInstance().addMetricSet(this.procedureMetrics);
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/StatisticsService.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/StatisticsService.java
index 4d98a66106b..03379d8b097 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/StatisticsService.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/StatisticsService.java
@@ -279,17 +279,21 @@ public class StatisticsService implements IClusterStatusSubscriber {
regionPriorityEntry : priorityMap.entrySet()) {
if (!Objects.equals(
regionPriorityEntry.getValue().getRight(), regionPriorityEntry.getValue().getLeft())) {
- LOGGER.info(
- "[RegionPriority]\t {}: {}->{}",
- regionPriorityEntry.getKey(),
- regionPriorityEntry.getValue().getLeft() == null
- ? "null"
- : regionPriorityEntry.getValue().getLeft().getDataNodeLocations().stream()
- .map(TDataNodeLocation::getDataNodeId)
- .collect(Collectors.toList()),
- regionPriorityEntry.getValue().getRight().getDataNodeLocations().stream()
- .map(TDataNodeLocation::getDataNodeId)
- .collect(Collectors.toList()));
+ try {
+ LOGGER.info(
+ "[RegionPriority]\t {}: {}->{}",
+ regionPriorityEntry.getKey(),
+ regionPriorityEntry.getValue().getLeft() == null
+ ? "null"
+ : regionPriorityEntry.getValue().getLeft().getDataNodeLocations().stream()
+ .map(TDataNodeLocation::getDataNodeId)
+ .collect(Collectors.toList()),
+ regionPriorityEntry.getValue().getRight().getDataNodeLocations().stream()
+ .map(TDataNodeLocation::getDataNodeId)
+ .collect(Collectors.toList()));
+ } catch (Exception e) {
+ LOGGER.error("unexcepted exception", e);
+ }
}
}
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
index 7eac6c23c5f..008882b67cd 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
@@ -62,7 +62,7 @@ import org.apache.iotdb.confignode.manager.partition.PartitionMetrics;
import org.apache.iotdb.confignode.manager.pipe.coordinator.PipeManager;
import org.apache.iotdb.confignode.manager.schema.ClusterSchemaManager;
import org.apache.iotdb.confignode.persistence.node.NodeInfo;
-import org.apache.iotdb.confignode.procedure.env.DataNodeRemoveHandler;
+import org.apache.iotdb.confignode.procedure.env.RegionMaintainHandler;
import org.apache.iotdb.confignode.rpc.thrift.TCQConfig;
import org.apache.iotdb.confignode.rpc.thrift.TConfigNodeInfo;
import org.apache.iotdb.confignode.rpc.thrift.TConfigNodeRegisterReq;
@@ -366,10 +366,8 @@ public class NodeManager {
public DataSet removeDataNode(RemoveDataNodePlan removeDataNodePlan) {
LOGGER.info("NodeManager start to remove DataNode {}", removeDataNodePlan);
- DataNodeRemoveHandler dataNodeRemoveHandler =
- new DataNodeRemoveHandler((ConfigManager) configManager);
- DataNodeToStatusResp preCheckStatus =
- dataNodeRemoveHandler.checkRemoveDataNodeRequest(removeDataNodePlan);
+ RegionMaintainHandler handler = new RegionMaintainHandler((ConfigManager) configManager);
+ DataNodeToStatusResp preCheckStatus = handler.checkRemoveDataNodeRequest(removeDataNodePlan);
if (preCheckStatus.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
LOGGER.error(
"The remove DataNode request check failed. req: {}, check result: {}",
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 c61f13043a1..2172087be0d 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
@@ -51,9 +51,10 @@ import org.apache.iotdb.confignode.consensus.request.read.partition.GetTimeSlotL
import org.apache.iotdb.confignode.consensus.request.read.region.GetRegionIdPlan;
import org.apache.iotdb.confignode.consensus.request.read.region.GetRegionInfoListPlan;
import org.apache.iotdb.confignode.consensus.request.write.database.PreDeleteDatabasePlan;
+import org.apache.iotdb.confignode.consensus.request.write.partition.AddRegionLocationPlan;
import org.apache.iotdb.confignode.consensus.request.write.partition.CreateDataPartitionPlan;
import org.apache.iotdb.confignode.consensus.request.write.partition.CreateSchemaPartitionPlan;
-import org.apache.iotdb.confignode.consensus.request.write.partition.UpdateRegionLocationPlan;
+import org.apache.iotdb.confignode.consensus.request.write.partition.RemoveRegionLocationPlan;
import org.apache.iotdb.confignode.consensus.request.write.region.CreateRegionGroupsPlan;
import org.apache.iotdb.confignode.consensus.request.write.region.PollSpecificRegionMaintainTaskPlan;
import org.apache.iotdb.confignode.consensus.response.partition.CountTimeSlotListResp;
@@ -732,6 +733,11 @@ public class PartitionManager {
return partitionInfo.getReplicaSets(database, regionGroupIds);
}
+ public boolean isDataNodeContainsRegion(int dataNodeId, TConsensusGroupId regionId) {
+ return getAllReplicaSets(dataNodeId).stream()
+ .anyMatch(tRegionReplicaSet -> tRegionReplicaSet.getRegionId().equals(regionId));
+ }
+
/**
* Only leader use this interface.
*
@@ -910,6 +916,25 @@ public class PartitionManager {
return partitionInfo.generateNextRegionGroupId();
}
+ public Optional<TConsensusGroupId> generateTConsensusGroupIdByRegionId(final int regionId) {
+ if (configManager
+ .getPartitionManager()
+ .isRegionGroupExists(new TConsensusGroupId(TConsensusGroupType.SchemaRegion, regionId))) {
+ return Optional.of(new TConsensusGroupId(TConsensusGroupType.SchemaRegion, regionId));
+ }
+ if (configManager
+ .getPartitionManager()
+ .isRegionGroupExists(new TConsensusGroupId(TConsensusGroupType.DataRegion, regionId))) {
+ return Optional.of(new TConsensusGroupId(TConsensusGroupType.DataRegion, regionId));
+ }
+ String msg =
+ String.format(
+ "Submit RegionMigrateProcedure failed, because RegionGroup: %s doesn't exist",
+ regionId);
+ LOGGER.warn(msg);
+ return Optional.empty();
+ }
+
/**
* GetNodePathsPartition.
*
@@ -1000,13 +1025,18 @@ public class PartitionManager {
return partitionInfo.isRegionGroupExisted(regionGroupId);
}
- /**
- * Update region location.
- *
- * @param req UpdateRegionLocationReq
- * @return TSStatus
- */
- public TSStatus updateRegionLocation(UpdateRegionLocationPlan req) {
+ public TSStatus addRegionLocation(AddRegionLocationPlan req) {
+ try {
+ return getConsensusManager().write(req);
+ } catch (ConsensusException e) {
+ LOGGER.warn(CONSENSUS_WRITE_ERROR, e);
+ TSStatus res = new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
+ res.setMessage(e.getMessage());
+ return res;
+ }
+ }
+
+ public TSStatus removeRegionLocation(RemoveRegionLocationPlan req) {
try {
return getConsensusManager().write(req);
} catch (ConsensusException e) {
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java
index 14f81103771..3899c091e1f 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/executor/ConfigPlanExecutor.java
@@ -68,8 +68,10 @@ import org.apache.iotdb.confignode.consensus.request.write.datanode.RemoveDataNo
import org.apache.iotdb.confignode.consensus.request.write.datanode.UpdateDataNodePlan;
import org.apache.iotdb.confignode.consensus.request.write.function.CreateFunctionPlan;
import org.apache.iotdb.confignode.consensus.request.write.function.DropFunctionPlan;
+import org.apache.iotdb.confignode.consensus.request.write.partition.AddRegionLocationPlan;
import org.apache.iotdb.confignode.consensus.request.write.partition.CreateDataPartitionPlan;
import org.apache.iotdb.confignode.consensus.request.write.partition.CreateSchemaPartitionPlan;
+import org.apache.iotdb.confignode.consensus.request.write.partition.RemoveRegionLocationPlan;
import org.apache.iotdb.confignode.consensus.request.write.partition.UpdateRegionLocationPlan;
import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeEnrichedPlan;
import org.apache.iotdb.confignode.consensus.request.write.pipe.plugin.CreatePipePluginPlan;
@@ -415,6 +417,10 @@ public class ConfigPlanExecutor {
return clusterSchemaInfo.createSchemaTemplate((CreateSchemaTemplatePlan) physicalPlan);
case UpdateRegionLocation:
return partitionInfo.updateRegionLocation((UpdateRegionLocationPlan) physicalPlan);
+ case AddRegionLocation:
+ return partitionInfo.addRegionLocation((AddRegionLocationPlan) physicalPlan);
+ case RemoveRegionLocation:
+ return partitionInfo.removeRegionLocation((RemoveRegionLocationPlan) physicalPlan);
case SetSchemaTemplate:
return clusterSchemaInfo.setSchemaTemplate((SetSchemaTemplatePlan) physicalPlan);
case PreSetSchemaTemplate:
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/DatabasePartitionTable.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/DatabasePartitionTable.java
index 5a4971b9699..a2f07df8e20 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/DatabasePartitionTable.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/DatabasePartitionTable.java
@@ -519,20 +519,8 @@ public class DatabasePartitionTable {
return schemaPartitionTable.getSeriesSlotList();
}
}
- /**
- * update region location.
- *
- * @param regionId regionId
- * @param oldNode old location, will remove it
- * @param newNode new location, will add it
- */
- public void updateRegionLocation(
- TConsensusGroupId regionId, TDataNodeLocation oldNode, TDataNodeLocation newNode) {
- addRegionNewLocation(regionId, newNode);
- removeRegionOldLocation(regionId, oldNode);
- }
- private void addRegionNewLocation(TConsensusGroupId regionId, TDataNodeLocation node) {
+ void addRegionNewLocation(TConsensusGroupId regionId, TDataNodeLocation node) {
RegionGroup regionGroup = regionGroupMap.get(regionId);
if (regionGroup == null) {
LOGGER.warn(
@@ -553,7 +541,7 @@ public class DatabasePartitionTable {
regionGroup.addRegionLocation(node);
}
- private void removeRegionOldLocation(TConsensusGroupId regionId, TDataNodeLocation node) {
+ void removeRegionLocation(TConsensusGroupId regionId, TDataNodeLocation node) {
RegionGroup regionGroup = regionGroupMap.get(regionId);
if (regionGroup == null) {
LOGGER.warn(
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/PartitionInfo.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/PartitionInfo.java
index 16317152dee..e26082401fc 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/PartitionInfo.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/PartitionInfo.java
@@ -40,8 +40,10 @@ import org.apache.iotdb.confignode.consensus.request.write.database.DatabaseSche
import org.apache.iotdb.confignode.consensus.request.write.database.DeleteDatabasePlan;
import org.apache.iotdb.confignode.consensus.request.write.database.PreDeleteDatabasePlan;
import org.apache.iotdb.confignode.consensus.request.write.datanode.UpdateDataNodePlan;
+import org.apache.iotdb.confignode.consensus.request.write.partition.AddRegionLocationPlan;
import org.apache.iotdb.confignode.consensus.request.write.partition.CreateDataPartitionPlan;
import org.apache.iotdb.confignode.consensus.request.write.partition.CreateSchemaPartitionPlan;
+import org.apache.iotdb.confignode.consensus.request.write.partition.RemoveRegionLocationPlan;
import org.apache.iotdb.confignode.consensus.request.write.partition.UpdateRegionLocationPlan;
import org.apache.iotdb.confignode.consensus.request.write.region.CreateRegionGroupsPlan;
import org.apache.iotdb.confignode.consensus.request.write.region.OfferRegionMaintainTasksPlan;
@@ -559,24 +561,37 @@ public class PartitionInfo implements SnapshotProcessor {
databasePartitionTable -> databasePartitionTable.containRegionGroup(regionGroupId));
}
- /**
- * Update the location info of given regionId.
- *
- * @param req UpdateRegionLocationReq
- * @return {@link TSStatus}
- */
public TSStatus updateRegionLocation(UpdateRegionLocationPlan req) {
- TSStatus status = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
- TConsensusGroupId regionId = req.getRegionId();
- TDataNodeLocation oldNode = req.getOldNode();
- TDataNodeLocation newNode = req.getNewNode();
+ TSStatus addStatus =
+ addRegionLocation(new AddRegionLocationPlan(req.getRegionId(), req.getNewNode()));
+ if (addStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ return addStatus;
+ }
+ return removeRegionLocation(new RemoveRegionLocationPlan(req.getRegionId(), req.getOldNode()));
+ }
+
+ /** The region has expanded to a new DataNode, now update the databasePartitionTable. */
+ public TSStatus addRegionLocation(AddRegionLocationPlan req) {
databasePartitionTables.values().stream()
- .filter(databasePartitionTable -> databasePartitionTable.containRegionGroup(regionId))
+ .filter(
+ databasePartitionTable -> databasePartitionTable.containRegionGroup(req.getRegionId()))
.forEach(
databasePartitionTable ->
- databasePartitionTable.updateRegionLocation(regionId, oldNode, newNode));
+ databasePartitionTable.addRegionNewLocation(
+ req.getRegionId(), req.getNewLocation()));
+ return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ }
- return status;
+ /** The region is no longer located on a DataNode, now update the databasePartitionTable. */
+ public TSStatus removeRegionLocation(RemoveRegionLocationPlan req) {
+ databasePartitionTables.values().stream()
+ .filter(
+ databasePartitionTable -> databasePartitionTable.containRegionGroup(req.getRegionId()))
+ .forEach(
+ databasePartitionTable ->
+ databasePartitionTable.removeRegionLocation(
+ req.getRegionId(), req.getDeprecatedLocation()));
+ return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
}
/**
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java
index 93990bb04d5..0f558bd8f55 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java
@@ -110,14 +110,14 @@ public class ConfigNodeProcedureEnv {
private final ProcedureScheduler scheduler;
- private final DataNodeRemoveHandler dataNodeRemoveHandler;
+ private final RegionMaintainHandler regionMaintainHandler;
private final ReentrantLock removeConfigNodeLock;
public ConfigNodeProcedureEnv(ConfigManager configManager, ProcedureScheduler scheduler) {
this.configManager = configManager;
this.scheduler = scheduler;
- this.dataNodeRemoveHandler = new DataNodeRemoveHandler(configManager);
+ this.regionMaintainHandler = new RegionMaintainHandler(configManager);
this.removeConfigNodeLock = new ReentrantLock();
}
@@ -211,15 +211,19 @@ public class ConfigNodeProcedureEnv {
.allMatch(tsStatus -> tsStatus.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode());
}
- public boolean doubleCheckReplica(TDataNodeLocation removedDatanode) {
- return getNodeManager()
- .filterDataNodeThroughStatus(NodeStatus.Running, NodeStatus.ReadOnly)
- .size()
- - Boolean.compare(
- getLoadManager().getNodeStatus(removedDatanode.getDataNodeId())
- != NodeStatus.Unknown,
- false)
- >= NodeInfo.getMinimumDataNode();
+ public boolean checkEnoughDataNodeAfterRemoving(TDataNodeLocation removedDatanode) {
+ final int existedDataNodeNum =
+ getNodeManager()
+ .filterDataNodeThroughStatus(
+ NodeStatus.Running, NodeStatus.ReadOnly, NodeStatus.Removing)
+ .size();
+ int dataNodeNumAfterRemoving;
+ if (getLoadManager().getNodeStatus(removedDatanode.getDataNodeId()) != NodeStatus.Unknown) {
+ dataNodeNumAfterRemoving = existedDataNodeNum - 1;
+ } else {
+ dataNodeNumAfterRemoving = existedDataNodeNum;
+ }
+ return dataNodeNumAfterRemoving >= NodeInfo.getMinimumDataNode();
}
/**
@@ -773,15 +777,15 @@ public class ConfigNodeProcedureEnv {
}
public LockQueue getRegionMigrateLock() {
- return dataNodeRemoveHandler.getRegionMigrateLock();
+ return regionMaintainHandler.getRegionMigrateLock();
}
public ReentrantLock getSchedulerLock() {
return schedulerLock;
}
- public DataNodeRemoveHandler getDataNodeRemoveHandler() {
- return dataNodeRemoveHandler;
+ public RegionMaintainHandler getRegionMaintainHandler() {
+ return regionMaintainHandler;
}
private ConsensusManager getConsensusManager() {
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/DataNodeRemoveHandler.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java
similarity index 82%
rename from iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/DataNodeRemoveHandler.java
rename to iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java
index d7029d3c5bf..ba17d45890d 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/DataNodeRemoveHandler.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java
@@ -22,8 +22,15 @@ 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.TEndPoint;
+import org.apache.iotdb.common.rpc.thrift.TRegionMaintainTaskStatus;
+import org.apache.iotdb.common.rpc.thrift.TRegionMigrateFailedType;
+import org.apache.iotdb.common.rpc.thrift.TRegionMigrateResult;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.ClientPoolFactory;
+import org.apache.iotdb.commons.client.IClientManager;
+import org.apache.iotdb.commons.client.sync.SyncDataNodeInternalServiceClient;
import org.apache.iotdb.commons.cluster.NodeStatus;
import org.apache.iotdb.commons.service.metric.MetricService;
import org.apache.iotdb.commons.utils.NodeUrlUtils;
@@ -32,7 +39,8 @@ import org.apache.iotdb.confignode.client.sync.SyncDataNodeClientPool;
import org.apache.iotdb.confignode.conf.ConfigNodeConfig;
import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
import org.apache.iotdb.confignode.consensus.request.write.datanode.RemoveDataNodePlan;
-import org.apache.iotdb.confignode.consensus.request.write.partition.UpdateRegionLocationPlan;
+import org.apache.iotdb.confignode.consensus.request.write.partition.AddRegionLocationPlan;
+import org.apache.iotdb.confignode.consensus.request.write.partition.RemoveRegionLocationPlan;
import org.apache.iotdb.confignode.consensus.response.datanode.DataNodeToStatusResp;
import org.apache.iotdb.confignode.manager.ConfigManager;
import org.apache.iotdb.confignode.manager.partition.PartitionMetrics;
@@ -42,6 +50,7 @@ import org.apache.iotdb.consensus.exception.ConsensusException;
import org.apache.iotdb.mpp.rpc.thrift.TCreatePeerReq;
import org.apache.iotdb.mpp.rpc.thrift.TDisableDataNodeReq;
import org.apache.iotdb.mpp.rpc.thrift.TMaintainPeerReq;
+import org.apache.iotdb.mpp.rpc.thrift.TResetPeerListReq;
import org.apache.iotdb.rpc.TSStatusCode;
import org.slf4j.Logger;
@@ -49,8 +58,10 @@ import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
import java.util.Optional;
+import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import static org.apache.iotdb.confignode.conf.ConfigNodeConstant.REGION_MIGRATE_PROCESS;
@@ -58,9 +69,9 @@ import static org.apache.iotdb.confignode.conf.ConfigNodeConstant.REMOVE_DATANOD
import static org.apache.iotdb.consensus.ConsensusFactory.IOT_CONSENSUS;
import static org.apache.iotdb.consensus.ConsensusFactory.SIMPLE_CONSENSUS;
-public class DataNodeRemoveHandler {
+public class RegionMaintainHandler {
- private static final Logger LOGGER = LoggerFactory.getLogger(DataNodeRemoveHandler.class);
+ private static final Logger LOGGER = LoggerFactory.getLogger(RegionMaintainHandler.class);
private static final ConfigNodeConfig CONF = ConfigNodeDescriptor.getInstance().getConf();
@@ -69,8 +80,14 @@ public class DataNodeRemoveHandler {
/** region migrate lock */
private final LockQueue regionMigrateLock = new LockQueue();
- public DataNodeRemoveHandler(ConfigManager configManager) {
+ private final IClientManager<TEndPoint, SyncDataNodeInternalServiceClient> dataNodeClientManager;
+
+ public RegionMaintainHandler(ConfigManager configManager) {
this.configManager = configManager;
+ dataNodeClientManager =
+ new IClientManager.Factory<TEndPoint, SyncDataNodeInternalServiceClient>()
+ .createClientManager(
+ new ClientPoolFactory.SyncDataNodeInternalServiceClientPoolFactory());
}
public static String getIdWithRpcEndpoint(TDataNodeLocation location) {
@@ -232,41 +249,27 @@ public class DataNodeRemoveHandler {
* @param regionId region id
* @return TSStatus
*/
- public TSStatus addRegionPeer(TDataNodeLocation destDataNode, TConsensusGroupId regionId) {
+ public TSStatus addRegionPeer(
+ long procedureId,
+ TDataNodeLocation destDataNode,
+ TConsensusGroupId regionId,
+ TDataNodeLocation coordinator) {
TSStatus status;
- // Here we pick the DataNode who contains one of the RegionReplica of the specified
- // ConsensusGroup except the new one
- // in order to notify the origin ConsensusGroup that another peer is created and demand to join
- Optional<TDataNodeLocation> selectedDataNode =
- filterDataNodeWithOtherRegionReplica(regionId, destDataNode);
- if (!selectedDataNode.isPresent()) {
- LOGGER.warn(
- "{}, There are no other DataNodes could be selected to perform the add peer process, "
- + "please check RegionGroup: {} by show regions sql command",
- REGION_MIGRATE_PROCESS,
- regionId);
- status = new TSStatus(TSStatusCode.MIGRATE_REGION_ERROR.getStatusCode());
- status.setMessage(
- "There are no other DataNodes could be selected to perform the add peer process, "
- + "please check by show regions sql command");
- return status;
- }
-
// Send addRegionPeer request to the selected DataNode,
// destDataNode is where the new RegionReplica is created
- TMaintainPeerReq maintainPeerReq = new TMaintainPeerReq(regionId, destDataNode);
+ TMaintainPeerReq maintainPeerReq = new TMaintainPeerReq(regionId, destDataNode, procedureId);
status =
SyncDataNodeClientPool.getInstance()
.sendSyncRequestToDataNodeWithRetry(
- selectedDataNode.get().getInternalEndPoint(),
+ coordinator.getInternalEndPoint(),
maintainPeerReq,
DataNodeRequestType.ADD_REGION_PEER);
LOGGER.info(
"{}, Send action addRegionPeer finished, regionId: {}, rpcDataNode: {}, destDataNode: {}",
REGION_MIGRATE_PROCESS,
regionId,
- getIdWithRpcEndpoint(selectedDataNode.get()),
+ getIdWithRpcEndpoint(coordinator),
getIdWithRpcEndpoint(destDataNode));
return status;
}
@@ -282,34 +285,26 @@ public class DataNodeRemoveHandler {
* @return TSStatus
*/
public TSStatus removeRegionPeer(
+ long procedureId,
TDataNodeLocation originalDataNode,
- TDataNodeLocation destDataNode,
- TConsensusGroupId regionId) {
+ TConsensusGroupId regionId,
+ TDataNodeLocation coordinator) {
TSStatus status;
- TDataNodeLocation rpcClientDataNode;
-
- // Here we pick the DataNode who contains one of the RegionReplica of the specified
- // ConsensusGroup except the origin one
- // in order to notify the new ConsensusGroup that the origin peer should secede now
- // If the selectedDataNode equals null, we choose the destDataNode to execute the method
- Optional<TDataNodeLocation> selectedDataNode =
- filterDataNodeWithOtherRegionReplica(regionId, originalDataNode);
- rpcClientDataNode = selectedDataNode.orElse(destDataNode);
-
// Send removeRegionPeer request to the rpcClientDataNode
- TMaintainPeerReq maintainPeerReq = new TMaintainPeerReq(regionId, originalDataNode);
+ TMaintainPeerReq maintainPeerReq =
+ new TMaintainPeerReq(regionId, originalDataNode, procedureId);
status =
SyncDataNodeClientPool.getInstance()
.sendSyncRequestToDataNodeWithRetry(
- rpcClientDataNode.getInternalEndPoint(),
+ coordinator.getInternalEndPoint(),
maintainPeerReq,
DataNodeRequestType.REMOVE_REGION_PEER);
LOGGER.info(
"{}, Send action removeRegionPeer finished, regionId: {}, rpcDataNode: {}",
REGION_MIGRATE_PROCESS,
regionId,
- getIdWithRpcEndpoint(rpcClientDataNode));
+ getIdWithRpcEndpoint(coordinator));
return status;
}
@@ -324,10 +319,11 @@ public class DataNodeRemoveHandler {
* @return TSStatus
*/
public TSStatus deleteOldRegionPeer(
- TDataNodeLocation originalDataNode, TConsensusGroupId regionId) {
+ long procedureId, TDataNodeLocation originalDataNode, TConsensusGroupId regionId) {
TSStatus status;
- TMaintainPeerReq maintainPeerReq = new TMaintainPeerReq(regionId, originalDataNode);
+ TMaintainPeerReq maintainPeerReq =
+ new TMaintainPeerReq(regionId, originalDataNode, procedureId);
status =
configManager.getLoadManager().getNodeStatus(originalDataNode.getDataNodeId())
@@ -351,31 +347,86 @@ public class DataNodeRemoveHandler {
return status;
}
- /**
- * Update region location cache
- *
- * @param regionId region id
- * @param originalDataNode old location data node
- * @param destDataNode dest data node
- */
- public void updateRegionLocationCache(
- TConsensusGroupId regionId,
- TDataNodeLocation originalDataNode,
- TDataNodeLocation destDataNode) {
+ public TSStatus resetPeerList(
+ TConsensusGroupId regionId, List<TDataNodeLocation> correctDataNodeLocations) {
+ Optional<TDataNodeLocation> optional = filterDataNodeWithOtherRegionReplica(regionId, null);
+ TDataNodeLocation selectDataNode = optional.get();
+ TSStatus status =
+ SyncDataNodeClientPool.getInstance()
+ .sendSyncRequestToDataNodeWithRetry(
+ selectDataNode.getInternalEndPoint(),
+ new TResetPeerListReq(regionId, correctDataNodeLocations),
+ DataNodeRequestType.RESET_PEER_LIST);
+ return status;
+ }
+
+ // TODO: will use 'procedure yield' to refactor later
+ public TRegionMigrateResult waitTaskFinish(long taskId, TDataNodeLocation dataNodeLocation) {
+ long lastTimeConnectDataNode = System.currentTimeMillis();
+ while (configManager.getLoadManager().getNodeStatus(dataNodeLocation.getDataNodeId())
+ != NodeStatus.Unknown) {
+ try (SyncDataNodeInternalServiceClient dataNodeClient =
+ dataNodeClientManager.borrowClient(dataNodeLocation.getInternalEndPoint())) {
+ TRegionMigrateResult report = dataNodeClient.getRegionMaintainResult(taskId);
+ lastTimeConnectDataNode = System.currentTimeMillis();
+ if (report.getTaskStatus() != TRegionMaintainTaskStatus.PROCESSING) {
+ return report;
+ }
+ } catch (Exception ignore) {
+
+ } finally {
+ try {
+ TimeUnit.SECONDS.sleep(1);
+ } catch (InterruptedException ignore) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+ LOGGER.warn(
+ "{} task {} cannot contact to DataNode {}",
+ REGION_MIGRATE_PROCESS,
+ taskId,
+ dataNodeLocation);
+ TRegionMigrateResult report = new TRegionMigrateResult();
+ report.setTaskStatus(TRegionMaintainTaskStatus.FAIL);
+ report.setFailedNodeAndReason(new HashMap<>());
+ report.getFailedNodeAndReason().put(dataNodeLocation, TRegionMigrateFailedType.Disconnect);
+ return report;
+ }
+
+ public void addRegionLocation(TConsensusGroupId regionId, TDataNodeLocation newLocation) {
LOGGER.info(
- "Start to updateRegionLocationCache {} location from {} to {} when it migrate succeed",
+ "AddRegionLocation started, add region {} to {}",
regionId,
- getIdWithRpcEndpoint(originalDataNode),
- getIdWithRpcEndpoint(destDataNode));
- UpdateRegionLocationPlan req =
- new UpdateRegionLocationPlan(regionId, originalDataNode, destDataNode);
- TSStatus status = configManager.getPartitionManager().updateRegionLocation(req);
+ getIdWithRpcEndpoint(newLocation));
+ AddRegionLocationPlan req = new AddRegionLocationPlan(regionId, newLocation);
+ TSStatus status = configManager.getPartitionManager().addRegionLocation(req);
LOGGER.info(
- "UpdateRegionLocationCache finished, region:{}, result:{}, old:{}, new:{}",
+ "AddRegionLocation finished, add region {} to {}, result is {}",
regionId,
- status,
- getIdWithRpcEndpoint(originalDataNode),
- getIdWithRpcEndpoint(destDataNode));
+ getIdWithRpcEndpoint(newLocation),
+ status);
+
+ // Remove the RegionGroupCache of the regionId
+ configManager.getLoadManager().removeRegionGroupCache(regionId);
+
+ // Broadcast the latest RegionRouteMap when Region migration finished
+ configManager.getLoadManager().broadcastLatestRegionRouteMap();
+ }
+
+ public void removeRegionLocation(
+ TConsensusGroupId regionId, TDataNodeLocation deprecatedLocation) {
+ LOGGER.info(
+ "RemoveRegionLocation started, remove region {} from DataNode {}",
+ regionId,
+ getIdWithRpcEndpoint(deprecatedLocation));
+ RemoveRegionLocationPlan req = new RemoveRegionLocationPlan(regionId, deprecatedLocation);
+ TSStatus status = configManager.getPartitionManager().removeRegionLocation(req);
+ LOGGER.info(
+ "RemoveRegionLocation finished, remove region {} from DataNode {}, result is {}",
+ regionId,
+ getIdWithRpcEndpoint(deprecatedLocation),
+ status);
// Remove the RegionGroupCache of the regionId
configManager.getLoadManager().removeRegionGroupCache(regionId);
@@ -646,7 +697,7 @@ public class DataNodeRemoveHandler {
* @return A DataNodeLocation that contains other RegionReplica and different from the
* filterLocation
*/
- private Optional<TDataNodeLocation> filterDataNodeWithOtherRegionReplica(
+ public Optional<TDataNodeLocation> filterDataNodeWithOtherRegionReplica(
TConsensusGroupId regionId, TDataNodeLocation filterLocation) {
List<TDataNodeLocation> regionLocations = findRegionLocations(regionId);
if (regionLocations.isEmpty()) {
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedure.java
index e6ce3127017..b29d5034915 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedure.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/StateMachineProcedure.java
@@ -152,20 +152,16 @@ public abstract class StateMachineProcedure<Env, TState> extends Procedure<Env>
/**
* Add a child procedure to execute.
*
- * @param subProcedure the child procedure
+ * @param childProcedure the child procedure
*/
- protected <T extends Procedure<Env>> void addChildProcedure(T... subProcedure) {
- if (subProcedure == null) {
- return;
- }
- final int len = subProcedure.length;
- if (len == 0) {
+ protected void addChildProcedure(Procedure<Env> childProcedure) {
+ if (childProcedure == null) {
return;
}
if (subProcList == null) {
- subProcList = new ArrayList<>(len);
+ subProcList = new ArrayList<>();
}
- subProcList.addAll(Arrays.asList(subProcedure).subList(0, len));
+ subProcList.add(childProcedure);
}
@Override
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveDataNodeProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveDataNodeProcedure.java
index c3efd35f0ff..80af94a8321 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveDataNodeProcedure.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveDataNodeProcedure.java
@@ -25,7 +25,7 @@ import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import org.apache.iotdb.commons.exception.runtime.ThriftSerDeException;
import org.apache.iotdb.commons.utils.ThriftCommonsSerDeUtils;
import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
-import org.apache.iotdb.confignode.procedure.env.DataNodeRemoveHandler;
+import org.apache.iotdb.confignode.procedure.env.RegionMaintainHandler;
import org.apache.iotdb.confignode.procedure.exception.ProcedureException;
import org.apache.iotdb.confignode.procedure.impl.region.RegionMigrateProcedure;
import org.apache.iotdb.confignode.procedure.state.RemoveDataNodeState;
@@ -68,11 +68,11 @@ public class RemoveDataNodeProcedure extends AbstractNodeProcedure<RemoveDataNod
return Flow.NO_MORE_STATE;
}
- DataNodeRemoveHandler handler = env.getDataNodeRemoveHandler();
+ RegionMaintainHandler handler = env.getRegionMaintainHandler();
try {
switch (state) {
case REGION_REPLICA_CHECK:
- if (env.doubleCheckReplica(removedDataNode)) {
+ if (env.checkEnoughDataNodeAfterRemoving(removedDataNode)) {
setNextState(RemoveDataNodeState.REMOVE_DATA_NODE_PREPARE);
} else {
LOG.error(
@@ -126,10 +126,22 @@ public class RemoveDataNodeProcedure extends AbstractNodeProcedure<RemoveDataNod
migratedDataNodeRegions.forEach(
regionId -> {
TDataNodeLocation destDataNode =
- env.getDataNodeRemoveHandler().findDestDataNode(regionId);
+ env.getRegionMaintainHandler().findDestDataNode(regionId);
+ // TODO: need to improve the coordinator selection method here, maybe through load
+ // balancing and other means.
+ final TDataNodeLocation coordinatorForAddPeer =
+ env.getRegionMaintainHandler()
+ .filterDataNodeWithOtherRegionReplica(regionId, destDataNode)
+ .orElse(removedDataNode);
+ final TDataNodeLocation coordinatorForRemovePeer = destDataNode;
if (destDataNode != null) {
RegionMigrateProcedure regionMigrateProcedure =
- new RegionMigrateProcedure(regionId, removedDataNode, destDataNode);
+ new RegionMigrateProcedure(
+ regionId,
+ removedDataNode,
+ destDataNode,
+ coordinatorForAddPeer,
+ coordinatorForRemovePeer);
addChildProcedure(regionMigrateProcedure);
LOG.info("Submit child procedure {} for regionId {}", regionMigrateProcedure, regionId);
} else {
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/region/AddRegionPeerProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/region/AddRegionPeerProcedure.java
new file mode 100644
index 00000000000..ba242cf5c07
--- /dev/null
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/region/AddRegionPeerProcedure.java
@@ -0,0 +1,203 @@
+/*
+ * 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.procedure.impl.region;
+
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TRegionMigrateResult;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.exception.runtime.ThriftSerDeException;
+import org.apache.iotdb.commons.utils.ThriftCommonsSerDeUtils;
+import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
+import org.apache.iotdb.confignode.procedure.env.RegionMaintainHandler;
+import org.apache.iotdb.confignode.procedure.exception.ProcedureException;
+import org.apache.iotdb.confignode.procedure.exception.ProcedureSuspendedException;
+import org.apache.iotdb.confignode.procedure.exception.ProcedureYieldException;
+import org.apache.iotdb.confignode.procedure.impl.StateMachineProcedure;
+import org.apache.iotdb.confignode.procedure.state.AddRegionPeerState;
+import org.apache.iotdb.confignode.procedure.store.ProcedureType;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+import static org.apache.iotdb.confignode.procedure.state.AddRegionPeerState.UPDATE_REGION_LOCATION_CACHE;
+import static org.apache.iotdb.rpc.TSStatusCode.SUCCESS_STATUS;
+
+public class AddRegionPeerProcedure
+ extends StateMachineProcedure<ConfigNodeProcedureEnv, AddRegionPeerState> {
+ private static final Logger LOGGER = LoggerFactory.getLogger(AddRegionPeerProcedure.class);
+ private TConsensusGroupId consensusGroupId;
+
+ private TDataNodeLocation coordinator;
+
+ private TDataNodeLocation destDataNode;
+
+ public AddRegionPeerProcedure() {
+ super();
+ }
+
+ public AddRegionPeerProcedure(
+ TConsensusGroupId consensusGroupId,
+ TDataNodeLocation coordinator,
+ TDataNodeLocation destDataNode) {
+ super();
+ this.consensusGroupId = consensusGroupId;
+ this.coordinator = coordinator;
+ this.destDataNode = destDataNode;
+ }
+
+ @Override
+ protected Flow executeFromState(ConfigNodeProcedureEnv env, AddRegionPeerState state)
+ throws ProcedureSuspendedException, ProcedureYieldException, InterruptedException {
+ if (consensusGroupId == null) {
+ return Flow.NO_MORE_STATE;
+ }
+ RegionMaintainHandler handler = env.getRegionMaintainHandler();
+ try {
+ outerSwitch:
+ switch (state) {
+ case CREATE_NEW_REGION_PEER:
+ handler.createNewRegionPeer(consensusGroupId, destDataNode);
+ setNextState(AddRegionPeerState.DO_ADD_REGION_PEER);
+ break;
+ case DO_ADD_REGION_PEER:
+ TSStatus tsStatus =
+ handler.addRegionPeer(this.getProcId(), destDataNode, consensusGroupId, coordinator);
+ TRegionMigrateResult result;
+ if (tsStatus.getCode() == SUCCESS_STATUS.getStatusCode()) {
+ result = handler.waitTaskFinish(this.getProcId(), coordinator);
+ } else {
+ throw new ProcedureException("ADD_REGION_PEER executed failed in DataNode");
+ }
+ switch (result.getTaskStatus()) {
+ case TASK_NOT_EXIST:
+ // coordinator crashed and lost its task table
+ case FAIL:
+ // maybe some DataNode crash
+ LOGGER.warn(
+ "result is {}, will use resetPeerList to clean in the future",
+ result.getTaskStatus());
+ // List<TDataNodeLocation> correctDataNodeLocations =
+ //
+ // env.getConfigManager().getPartitionManager().getAllReplicaSets().stream()
+ // .filter(
+ // tRegionReplicaSet ->
+ //
+ // tRegionReplicaSet.getRegionId().equals(consensusGroupId))
+ // .findAny()
+ // .get()
+ // .getDataNodeLocations();
+ // handler.resetPeerList(consensusGroupId, correctDataNodeLocations);
+ return Flow.NO_MORE_STATE;
+ case PROCESSING:
+ // should never happen
+ LOGGER.error("should never happen");
+ throw new UnsupportedOperationException("should never happen");
+ case SUCCESS:
+ setNextState(UPDATE_REGION_LOCATION_CACHE);
+ break outerSwitch;
+ default:
+ String msg = String.format("status %s is unsupported", result.getTaskStatus());
+ LOGGER.error(msg);
+ throw new UnsupportedOperationException(msg);
+ }
+ case UPDATE_REGION_LOCATION_CACHE:
+ handler.addRegionLocation(consensusGroupId, destDataNode);
+ LOGGER.info("AddRegionPeer state {} complete", state);
+ LOGGER.info(
+ "AddRegionPeerProcedure success, region {} has been added to DataNode {}",
+ consensusGroupId.getId(),
+ destDataNode.getDataNodeId());
+ return Flow.NO_MORE_STATE;
+ default:
+ throw new ProcedureException("Unsupported state: " + state.name());
+ }
+ } catch (Exception e) {
+ LOGGER.error("AddRegionPeer state {} failed", state, e);
+ return Flow.NO_MORE_STATE;
+ }
+ LOGGER.info("AddRegionPeer state {} complete", state);
+ return Flow.HAS_MORE_STATE;
+ }
+
+ @Override
+ protected void rollbackState(
+ ConfigNodeProcedureEnv configNodeProcedureEnv, AddRegionPeerState addRegionPeerState)
+ throws IOException, InterruptedException, ProcedureException {}
+
+ @Override
+ protected AddRegionPeerState getState(int stateId) {
+ return AddRegionPeerState.values()[stateId];
+ }
+
+ @Override
+ protected int getStateId(AddRegionPeerState addRegionPeerState) {
+ return addRegionPeerState.ordinal();
+ }
+
+ @Override
+ protected AddRegionPeerState getInitialState() {
+ return AddRegionPeerState.CREATE_NEW_REGION_PEER;
+ }
+
+ @Override
+ public void serialize(DataOutputStream stream) throws IOException {
+ stream.writeShort(ProcedureType.ADD_REGION_PEER_PROCEDURE.getTypeCode());
+ super.serialize(stream);
+ ThriftCommonsSerDeUtils.serializeTConsensusGroupId(consensusGroupId, stream);
+ ThriftCommonsSerDeUtils.serializeTDataNodeLocation(destDataNode, stream);
+ ThriftCommonsSerDeUtils.serializeTDataNodeLocation(coordinator, stream);
+ }
+
+ @Override
+ public void deserialize(ByteBuffer byteBuffer) {
+ super.deserialize(byteBuffer);
+ try {
+ consensusGroupId = ThriftCommonsSerDeUtils.deserializeTConsensusGroupId(byteBuffer);
+ destDataNode = ThriftCommonsSerDeUtils.deserializeTDataNodeLocation(byteBuffer);
+ coordinator = ThriftCommonsSerDeUtils.deserializeTDataNodeLocation(byteBuffer);
+ } catch (ThriftSerDeException e) {
+ LOGGER.error("Error in deserialize {}", this.getClass(), e);
+ }
+ }
+
+ public TConsensusGroupId getConsensusGroupId() {
+ return consensusGroupId;
+ }
+
+ public TDataNodeLocation getCoordinator() {
+ return coordinator;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (!(obj instanceof AddRegionPeerProcedure)) {
+ return false;
+ }
+ AddRegionPeerProcedure procedure = (AddRegionPeerProcedure) obj;
+ return this.consensusGroupId.equals(procedure.consensusGroupId)
+ && this.destDataNode.equals(procedure.destDataNode)
+ && this.coordinator.equals(procedure.coordinator);
+ }
+}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/region/RegionMigrateProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/region/RegionMigrateProcedure.java
index 2043eccbf98..b6f418f0dc9 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/region/RegionMigrateProcedure.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/region/RegionMigrateProcedure.java
@@ -21,18 +21,15 @@ package org.apache.iotdb.confignode.procedure.impl.region;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
-import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.exception.runtime.ThriftSerDeException;
import org.apache.iotdb.commons.utils.ThriftCommonsSerDeUtils;
import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
-import org.apache.iotdb.confignode.procedure.env.DataNodeRemoveHandler;
+import org.apache.iotdb.confignode.procedure.env.RegionMaintainHandler;
import org.apache.iotdb.confignode.procedure.exception.ProcedureException;
import org.apache.iotdb.confignode.procedure.impl.StateMachineProcedure;
import org.apache.iotdb.confignode.procedure.state.ProcedureLockState;
import org.apache.iotdb.confignode.procedure.state.RegionTransitionState;
import org.apache.iotdb.confignode.procedure.store.ProcedureType;
-import org.apache.iotdb.confignode.rpc.thrift.TRegionMigrateResultReportReq;
-import org.apache.iotdb.rpc.TSStatusCode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -42,29 +39,20 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Objects;
-import static org.apache.iotdb.confignode.conf.ConfigNodeConstant.REGION_MIGRATE_PROCESS;
-import static org.apache.iotdb.confignode.procedure.env.DataNodeRemoveHandler.getIdWithRpcEndpoint;
-import static org.apache.iotdb.rpc.TSStatusCode.SUCCESS_STATUS;
-
/** Region migrate procedure */
public class RegionMigrateProcedure
extends StateMachineProcedure<ConfigNodeProcedureEnv, RegionTransitionState> {
- // TODO: Reach an agreement on RegionMigrateProcedure
-
- private static final Logger LOG = LoggerFactory.getLogger(RegionMigrateProcedure.class);
+ private static final Logger LOGGER = LoggerFactory.getLogger(RegionMigrateProcedure.class);
private static final int RETRY_THRESHOLD = 5;
/** Wait region migrate finished */
- private final Object regionMigrateLock = new Object();
-
private TConsensusGroupId consensusGroupId;
private TDataNodeLocation originalDataNode;
-
private TDataNodeLocation destDataNode;
-
- private boolean migrateSuccess = true;
+ private TDataNodeLocation coordinatorForAddPeer;
+ private TDataNodeLocation coordinatorForRemovePeer;
private String migrateResult = "";
@@ -75,11 +63,15 @@ public class RegionMigrateProcedure
public RegionMigrateProcedure(
TConsensusGroupId consensusGroupId,
TDataNodeLocation originalDataNode,
- TDataNodeLocation destDataNode) {
+ TDataNodeLocation destDataNode,
+ TDataNodeLocation coordinatorForAddPeer,
+ TDataNodeLocation coordinatorForRemovePeer) {
super();
this.consensusGroupId = consensusGroupId;
this.originalDataNode = originalDataNode;
this.destDataNode = destDataNode;
+ this.coordinatorForAddPeer = coordinatorForAddPeer;
+ this.coordinatorForRemovePeer = coordinatorForRemovePeer;
}
@Override
@@ -87,24 +79,26 @@ public class RegionMigrateProcedure
if (consensusGroupId == null) {
return Flow.NO_MORE_STATE;
}
- TSStatus tsStatus;
- DataNodeRemoveHandler handler = env.getDataNodeRemoveHandler();
+ RegionMaintainHandler handler = env.getRegionMaintainHandler();
try {
switch (state) {
case REGION_MIGRATE_PREPARE:
- setNextState(RegionTransitionState.CREATE_NEW_REGION_PEER);
- break;
- case CREATE_NEW_REGION_PEER:
- handler.createNewRegionPeer(consensusGroupId, destDataNode);
setNextState(RegionTransitionState.ADD_REGION_PEER);
break;
case ADD_REGION_PEER:
- tsStatus = handler.addRegionPeer(destDataNode, consensusGroupId);
- if (tsStatus.getCode() == SUCCESS_STATUS.getStatusCode()) {
- waitForOneMigrationStepFinished(consensusGroupId, state);
- } else {
- throw new ProcedureException("ADD_REGION_PEER executed failed in DataNode");
+ addChildProcedure(
+ new AddRegionPeerProcedure(consensusGroupId, coordinatorForAddPeer, destDataNode));
+ setNextState(RegionTransitionState.CHECK_ADD_REGION_PEER);
+ break;
+ case CHECK_ADD_REGION_PEER:
+ if (!env.getConfigManager()
+ .getPartitionManager()
+ .isDataNodeContainsRegion(destDataNode.getDataNodeId(), consensusGroupId)) {
+ LOGGER.warn(
+ "sub-procedure AddRegionPeerProcedure fail, RegionMigrateProcedure will not continue");
+ return Flow.NO_MORE_STATE;
}
+ LOGGER.info("sub-procedure AddRegionPeerProcedure success");
setNextState(RegionTransitionState.CHANGE_REGION_LEADER);
break;
case CHANGE_REGION_LEADER:
@@ -112,53 +106,34 @@ public class RegionMigrateProcedure
setNextState(RegionTransitionState.REMOVE_REGION_PEER);
break;
case REMOVE_REGION_PEER:
- tsStatus = handler.removeRegionPeer(originalDataNode, destDataNode, consensusGroupId);
- if (tsStatus.getCode() == SUCCESS_STATUS.getStatusCode()) {
- waitForOneMigrationStepFinished(consensusGroupId, state);
- } else {
- throw new ProcedureException("REMOVE_REGION_PEER executed failed in DataNode");
- }
- setNextState(RegionTransitionState.DELETE_OLD_REGION_PEER);
+ addChildProcedure(
+ new RemoveRegionPeerProcedure(
+ consensusGroupId, coordinatorForRemovePeer, originalDataNode));
+ setNextState(RegionTransitionState.CHECK_REMOVE_REGION_PEER);
break;
- case DELETE_OLD_REGION_PEER:
- tsStatus = handler.deleteOldRegionPeer(originalDataNode, consensusGroupId);
- if (tsStatus.getCode() == SUCCESS_STATUS.getStatusCode()) {
- waitForOneMigrationStepFinished(consensusGroupId, state);
+ case CHECK_REMOVE_REGION_PEER:
+ if (env.getConfigManager()
+ .getPartitionManager()
+ .isDataNodeContainsRegion(originalDataNode.getDataNodeId(), consensusGroupId)) {
+ LOGGER.warn(
+ "RegionMigrateProcedure success, but you may need to manually clean the old region to make everything works fine");
+ } else {
+ LOGGER.info(
+ "RegionMigrateProcedure success, region {} has been migrated from DataNode {} to {}",
+ consensusGroupId.getId(),
+ originalDataNode.getDataNodeId(),
+ destDataNode.getDataNodeId());
}
- // Remove consensus group after a node stop, which will be failed, but we will
- // continuously execute.
- setNextState(RegionTransitionState.UPDATE_REGION_LOCATION_CACHE);
- break;
- case UPDATE_REGION_LOCATION_CACHE:
- handler.updateRegionLocationCache(consensusGroupId, originalDataNode, destDataNode);
return Flow.NO_MORE_STATE;
+ default:
+ throw new ProcedureException("Unsupported state: " + state.name());
}
} catch (Exception e) {
- LOG.error(
- "{}, Meets error in region migrate state, "
- + "please do the rollback operation yourself manually according to the error message!!! "
- + "error state: {}, migrateResult: {}",
- REGION_MIGRATE_PROCESS,
- state,
- migrateResult);
- if (isRollbackSupported(state)) {
- setFailure(new ProcedureException("Region migrate failed at state: " + state));
- } else {
- LOG.error(
- "{}, Failed state [{}] is not support rollback, originalDataNode: {}",
- REGION_MIGRATE_PROCESS,
- state,
- getIdWithRpcEndpoint(originalDataNode));
- if (getCycles() > RETRY_THRESHOLD) {
- setFailure(
- new ProcedureException(
- "Procedure retried failed exceed 5 times, state stuck at " + state));
- }
-
- // meets exception in region migrate process terminate the process
- return Flow.NO_MORE_STATE;
- }
+ LOGGER.error("RegionMigrateProcedure state {} fail", state, e);
+ // meets exception in region migrate process terminate the process
+ return Flow.NO_MORE_STATE;
}
+ LOGGER.info("RegionMigrateProcedure state {} complete", state);
return Flow.HAS_MORE_STATE;
}
@@ -176,12 +151,12 @@ public class RegionMigrateProcedure
configNodeProcedureEnv.getSchedulerLock().lock();
try {
if (configNodeProcedureEnv.getRegionMigrateLock().tryLock(this)) {
- LOG.info("procedureId {} acquire lock.", getProcId());
+ LOGGER.info("procedureId {} acquire lock.", getProcId());
return ProcedureLockState.LOCK_ACQUIRED;
}
configNodeProcedureEnv.getRegionMigrateLock().waitProcedure(this);
- LOG.info("procedureId {} wait for lock.", getProcId());
+ LOGGER.info("procedureId {} wait for lock.", getProcId());
return ProcedureLockState.LOCK_EVENT_WAIT;
} finally {
configNodeProcedureEnv.getSchedulerLock().unlock();
@@ -192,7 +167,7 @@ public class RegionMigrateProcedure
protected void releaseLock(ConfigNodeProcedureEnv configNodeProcedureEnv) {
configNodeProcedureEnv.getSchedulerLock().lock();
try {
- LOG.info("procedureId {} release lock.", getProcId());
+ LOGGER.info("procedureId {} release lock.", getProcId());
if (configNodeProcedureEnv.getRegionMigrateLock().releaseLock(this)) {
configNodeProcedureEnv
.getRegionMigrateLock()
@@ -225,6 +200,8 @@ public class RegionMigrateProcedure
ThriftCommonsSerDeUtils.serializeTDataNodeLocation(originalDataNode, stream);
ThriftCommonsSerDeUtils.serializeTDataNodeLocation(destDataNode, stream);
ThriftCommonsSerDeUtils.serializeTConsensusGroupId(consensusGroupId, stream);
+ ThriftCommonsSerDeUtils.serializeTDataNodeLocation(coordinatorForAddPeer, stream);
+ ThriftCommonsSerDeUtils.serializeTDataNodeLocation(coordinatorForRemovePeer, stream);
}
@Override
@@ -234,8 +211,14 @@ public class RegionMigrateProcedure
originalDataNode = ThriftCommonsSerDeUtils.deserializeTDataNodeLocation(byteBuffer);
destDataNode = ThriftCommonsSerDeUtils.deserializeTDataNodeLocation(byteBuffer);
consensusGroupId = ThriftCommonsSerDeUtils.deserializeTConsensusGroupId(byteBuffer);
+ coordinatorForAddPeer = ThriftCommonsSerDeUtils.deserializeTDataNodeLocation(byteBuffer);
+ coordinatorForRemovePeer = ThriftCommonsSerDeUtils.deserializeTDataNodeLocation(byteBuffer);
} catch (ThriftSerDeException e) {
- LOG.error("Error in deserialize RemoveConfigNodeProcedure", e);
+ LOGGER.error(
+ "Error in deserialize {} (procID {}), this procedure may belong to old version and already cannot be used.",
+ this.getClass(),
+ this.getProcId(),
+ e);
}
}
@@ -257,59 +240,6 @@ public class RegionMigrateProcedure
return Objects.hash(this.originalDataNode, this.destDataNode, this.consensusGroupId);
}
- public TSStatus waitForOneMigrationStepFinished(
- TConsensusGroupId consensusGroupId, RegionTransitionState state) throws Exception {
-
- LOG.info(
- "{}, Wait for state {} finished, regionId: {}",
- REGION_MIGRATE_PROCESS,
- state,
- consensusGroupId);
-
- TSStatus status = new TSStatus(SUCCESS_STATUS.getStatusCode());
- synchronized (regionMigrateLock) {
- try {
- // TODO set timeOut?
- regionMigrateLock.wait();
-
- if (!migrateSuccess) {
- throw new ProcedureException(
- String.format("Region migration failed, regionId: %s", consensusGroupId));
- }
- } catch (InterruptedException e) {
- LOG.error("{}, region migration {} interrupt", REGION_MIGRATE_PROCESS, consensusGroupId, e);
- Thread.currentThread().interrupt();
- status.setCode(TSStatusCode.MIGRATE_REGION_ERROR.getStatusCode());
- status.setMessage("Waiting for region migration interruption," + e.getMessage());
- }
- }
- return status;
- }
-
- /** DataNode report region migrate result to ConfigNode, and continue */
- public void notifyTheRegionMigrateFinished(TRegionMigrateResultReportReq req) {
-
- LOG.info(
- "{}, ConfigNode received region migration result reported by DataNode: {}",
- REGION_MIGRATE_PROCESS,
- req);
-
- // TODO the req is used in roll back
- synchronized (regionMigrateLock) {
- TSStatus migrateStatus = req.getMigrateResult();
- // Migration failed
- if (migrateStatus.getCode() != SUCCESS_STATUS.getStatusCode()) {
- LOG.info(
- "{}, Region migration failed in DataNode, migrateStatus: {}",
- REGION_MIGRATE_PROCESS,
- migrateStatus);
- migrateSuccess = false;
- migrateResult = migrateStatus.toString();
- }
- regionMigrateLock.notifyAll();
- }
- }
-
public TConsensusGroupId getConsensusGroupId() {
return consensusGroupId;
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/region/RemoveRegionPeerProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/region/RemoveRegionPeerProcedure.java
new file mode 100644
index 00000000000..b29aaf4387b
--- /dev/null
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/region/RemoveRegionPeerProcedure.java
@@ -0,0 +1,188 @@
+/*
+ * 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.procedure.impl.region;
+
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TRegionMaintainTaskStatus;
+import org.apache.iotdb.common.rpc.thrift.TRegionMigrateResult;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.exception.runtime.ThriftSerDeException;
+import org.apache.iotdb.commons.utils.ThriftCommonsSerDeUtils;
+import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
+import org.apache.iotdb.confignode.procedure.env.RegionMaintainHandler;
+import org.apache.iotdb.confignode.procedure.exception.ProcedureException;
+import org.apache.iotdb.confignode.procedure.exception.ProcedureSuspendedException;
+import org.apache.iotdb.confignode.procedure.exception.ProcedureYieldException;
+import org.apache.iotdb.confignode.procedure.impl.StateMachineProcedure;
+import org.apache.iotdb.confignode.procedure.state.RemoveRegionPeerState;
+import org.apache.iotdb.confignode.procedure.store.ProcedureType;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+import static org.apache.iotdb.confignode.procedure.state.RemoveRegionPeerState.DELETE_OLD_REGION_PEER;
+import static org.apache.iotdb.confignode.procedure.state.RemoveRegionPeerState.REMOVE_REGION_LOCATION_CACHE;
+import static org.apache.iotdb.rpc.TSStatusCode.SUCCESS_STATUS;
+
+public class RemoveRegionPeerProcedure
+ extends StateMachineProcedure<ConfigNodeProcedureEnv, RemoveRegionPeerState> {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RemoveRegionPeerProcedure.class);
+ private TConsensusGroupId consensusGroupId;
+ private TDataNodeLocation coordinator;
+ private TDataNodeLocation targetDataNode;
+
+ public RemoveRegionPeerProcedure() {
+ super();
+ }
+
+ public RemoveRegionPeerProcedure(
+ TConsensusGroupId consensusGroupId,
+ TDataNodeLocation coordinator,
+ TDataNodeLocation targetDataNode) {
+ super();
+ this.consensusGroupId = consensusGroupId;
+ this.coordinator = coordinator;
+ this.targetDataNode = targetDataNode;
+ }
+
+ @Override
+ protected Flow executeFromState(ConfigNodeProcedureEnv env, RemoveRegionPeerState state)
+ throws ProcedureSuspendedException, ProcedureYieldException, InterruptedException {
+ if (consensusGroupId == null) {
+ return Flow.NO_MORE_STATE;
+ }
+ TSStatus tsStatus;
+ RegionMaintainHandler handler = env.getRegionMaintainHandler();
+ try {
+ switch (state) {
+ case REMOVE_REGION_PEER:
+ tsStatus =
+ handler.removeRegionPeer(
+ this.getProcId(), targetDataNode, consensusGroupId, coordinator);
+ if (tsStatus.getCode() != SUCCESS_STATUS.getStatusCode()) {
+ throw new ProcedureException("REMOVE_REGION_PEER executed failed in DataNode");
+ }
+ TRegionMigrateResult removeRegionPeerResult =
+ handler.waitTaskFinish(this.getProcId(), coordinator);
+ if (removeRegionPeerResult.getTaskStatus() != TRegionMaintainTaskStatus.SUCCESS) {
+ throw new ProcedureException("REMOVE_REGION_PEER executed failed in DataNode");
+ }
+ setNextState(DELETE_OLD_REGION_PEER);
+ break;
+ case DELETE_OLD_REGION_PEER:
+ tsStatus =
+ handler.deleteOldRegionPeer(this.getProcId(), targetDataNode, consensusGroupId);
+ if (tsStatus.getCode() != SUCCESS_STATUS.getStatusCode()) {
+ throw new ProcedureException("DELETE_OLD_REGION_PEER executed failed in DataNode");
+ }
+ TRegionMigrateResult deleteOldRegionPeerResult =
+ handler.waitTaskFinish(this.getProcId(), targetDataNode);
+ if (deleteOldRegionPeerResult.getTaskStatus() != TRegionMaintainTaskStatus.SUCCESS) {
+ throw new ProcedureException("DELETE_OLD_REGION_PEER executed failed in DataNode");
+ }
+ setNextState(REMOVE_REGION_LOCATION_CACHE);
+ break;
+ case REMOVE_REGION_LOCATION_CACHE:
+ handler.removeRegionLocation(consensusGroupId, targetDataNode);
+ LOGGER.info("RemoveRegionPeer state {} success", state);
+ LOGGER.info(
+ "RemoveRegionPeerProcedure success, region {} has been removed from DataNode {}",
+ consensusGroupId.getId(),
+ targetDataNode.getDataNodeId());
+ return Flow.NO_MORE_STATE;
+ default:
+ throw new ProcedureException("Unsupported state: " + state.name());
+ }
+ } catch (Exception e) {
+ LOGGER.error("RemoveRegionPeer state {} failed", state, e);
+ return Flow.NO_MORE_STATE;
+ }
+ LOGGER.info("RemoveRegionPeer state {} success", state);
+ return Flow.HAS_MORE_STATE;
+ }
+
+ @Override
+ protected void rollbackState(ConfigNodeProcedureEnv env, RemoveRegionPeerState state)
+ throws IOException, InterruptedException, ProcedureException {}
+
+ @Override
+ protected RemoveRegionPeerState getState(int stateId) {
+ return RemoveRegionPeerState.values()[stateId];
+ }
+
+ @Override
+ protected int getStateId(RemoveRegionPeerState RemoveRegionPeerState) {
+ return RemoveRegionPeerState.ordinal();
+ }
+
+ @Override
+ protected RemoveRegionPeerState getInitialState() {
+ return RemoveRegionPeerState.REMOVE_REGION_PEER;
+ }
+
+ @Override
+ public void serialize(DataOutputStream stream) throws IOException {
+ stream.writeShort(ProcedureType.REMOVE_REGION_PEER_PROCEDURE.getTypeCode());
+ super.serialize(stream);
+ ThriftCommonsSerDeUtils.serializeTConsensusGroupId(consensusGroupId, stream);
+ ThriftCommonsSerDeUtils.serializeTDataNodeLocation(targetDataNode, stream);
+ ThriftCommonsSerDeUtils.serializeTDataNodeLocation(coordinator, stream);
+ }
+
+ @Override
+ public void deserialize(ByteBuffer byteBuffer) {
+ super.deserialize(byteBuffer);
+ try {
+ consensusGroupId = ThriftCommonsSerDeUtils.deserializeTConsensusGroupId(byteBuffer);
+ targetDataNode = ThriftCommonsSerDeUtils.deserializeTDataNodeLocation(byteBuffer);
+ coordinator = ThriftCommonsSerDeUtils.deserializeTDataNodeLocation(byteBuffer);
+ } catch (ThriftSerDeException e) {
+ LOGGER.error("Error in deserialize {}", this.getClass(), e);
+ }
+ }
+
+ public TConsensusGroupId getConsensusGroupId() {
+ return consensusGroupId;
+ }
+
+ public TDataNodeLocation getCoordinator() {
+ return coordinator;
+ }
+
+ public TDataNodeLocation getTargetDataNode() {
+ return targetDataNode;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (!(obj instanceof RemoveRegionPeerProcedure)) {
+ return false;
+ }
+ RemoveRegionPeerProcedure procedure = (RemoveRegionPeerProcedure) obj;
+ return this.consensusGroupId.equals(procedure.consensusGroupId)
+ && this.targetDataNode.equals(procedure.targetDataNode)
+ && this.coordinator.equals(procedure.coordinator);
+ }
+}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/RegionTransitionState.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/AddRegionPeerState.java
similarity index 82%
copy from iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/RegionTransitionState.java
copy to iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/AddRegionPeerState.java
index a21a6af941b..0e3477626b6 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/RegionTransitionState.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/AddRegionPeerState.java
@@ -19,12 +19,8 @@
package org.apache.iotdb.confignode.procedure.state;
-public enum RegionTransitionState {
- REGION_MIGRATE_PREPARE,
+public enum AddRegionPeerState {
CREATE_NEW_REGION_PEER,
- ADD_REGION_PEER,
- CHANGE_REGION_LEADER,
- REMOVE_REGION_PEER,
- DELETE_OLD_REGION_PEER,
- UPDATE_REGION_LOCATION_CACHE
+ DO_ADD_REGION_PEER,
+ UPDATE_REGION_LOCATION_CACHE,
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/RegionTransitionState.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/RegionTransitionState.java
index a21a6af941b..5c8f7bf8c57 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/RegionTransitionState.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/RegionTransitionState.java
@@ -21,10 +21,9 @@ package org.apache.iotdb.confignode.procedure.state;
public enum RegionTransitionState {
REGION_MIGRATE_PREPARE,
- CREATE_NEW_REGION_PEER,
ADD_REGION_PEER,
+ CHECK_ADD_REGION_PEER,
CHANGE_REGION_LEADER,
REMOVE_REGION_PEER,
- DELETE_OLD_REGION_PEER,
- UPDATE_REGION_LOCATION_CACHE
+ CHECK_REMOVE_REGION_PEER,
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/RegionTransitionState.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/RemoveRegionPeerState.java
similarity index 84%
copy from iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/RegionTransitionState.java
copy to iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/RemoveRegionPeerState.java
index a21a6af941b..f4b9002193a 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/RegionTransitionState.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/RemoveRegionPeerState.java
@@ -19,12 +19,8 @@
package org.apache.iotdb.confignode.procedure.state;
-public enum RegionTransitionState {
- REGION_MIGRATE_PREPARE,
- CREATE_NEW_REGION_PEER,
- ADD_REGION_PEER,
- CHANGE_REGION_LEADER,
+public enum RemoveRegionPeerState {
REMOVE_REGION_PEER,
DELETE_OLD_REGION_PEER,
- UPDATE_REGION_LOCATION_CACHE
+ REMOVE_REGION_LOCATION_CACHE,
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureFactory.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureFactory.java
index 0172e39ea16..4fb23e4f987 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureFactory.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureFactory.java
@@ -34,8 +34,10 @@ import org.apache.iotdb.confignode.procedure.impl.pipe.task.CreatePipeProcedureV
import org.apache.iotdb.confignode.procedure.impl.pipe.task.DropPipeProcedureV2;
import org.apache.iotdb.confignode.procedure.impl.pipe.task.StartPipeProcedureV2;
import org.apache.iotdb.confignode.procedure.impl.pipe.task.StopPipeProcedureV2;
+import org.apache.iotdb.confignode.procedure.impl.region.AddRegionPeerProcedure;
import org.apache.iotdb.confignode.procedure.impl.region.CreateRegionGroupsProcedure;
import org.apache.iotdb.confignode.procedure.impl.region.RegionMigrateProcedure;
+import org.apache.iotdb.confignode.procedure.impl.region.RemoveRegionPeerProcedure;
import org.apache.iotdb.confignode.procedure.impl.schema.AlterLogicalViewProcedure;
import org.apache.iotdb.confignode.procedure.impl.schema.DeactivateTemplateProcedure;
import org.apache.iotdb.confignode.procedure.impl.schema.DeleteDatabaseProcedure;
@@ -99,6 +101,12 @@ public class ProcedureFactory implements IProcedureFactory {
case REGION_MIGRATE_PROCEDURE:
procedure = new RegionMigrateProcedure();
break;
+ case ADD_REGION_PEER_PROCEDURE:
+ procedure = new AddRegionPeerProcedure();
+ break;
+ case REMOVE_REGION_PEER_PROCEDURE:
+ procedure = new RemoveRegionPeerProcedure();
+ break;
case CREATE_REGION_GROUPS:
procedure = new CreateRegionGroupsProcedure();
break;
@@ -258,6 +266,10 @@ public class ProcedureFactory implements IProcedureFactory {
return ProcedureType.REMOVE_DATA_NODE_PROCEDURE;
} else if (procedure instanceof RegionMigrateProcedure) {
return ProcedureType.REGION_MIGRATE_PROCEDURE;
+ } else if (procedure instanceof AddRegionPeerProcedure) {
+ return ProcedureType.ADD_REGION_PEER_PROCEDURE;
+ } else if (procedure instanceof RemoveRegionPeerProcedure) {
+ return ProcedureType.REMOVE_REGION_PEER_PROCEDURE;
} else if (procedure instanceof CreateRegionGroupsProcedure) {
return ProcedureType.CREATE_REGION_GROUPS;
} else if (procedure instanceof DeleteTimeSeriesProcedure) {
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureType.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureType.java
index 0edbae8ba13..bf6417f4250 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureType.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureType.java
@@ -38,6 +38,8 @@ public enum ProcedureType {
CREATE_REGION_GROUPS((short) 202),
@TestOnly
CREATE_MANY_DATABASES_PROCEDURE((short) 203),
+ ADD_REGION_PEER_PROCEDURE((short) 204),
+ REMOVE_REGION_PEER_PROCEDURE((short) 205),
/** Timeseries */
DELETE_TIMESERIES_PROCEDURE((short) 300),
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
index 03d8857ea5e..d8fe871ebef 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
@@ -136,7 +136,6 @@ import org.apache.iotdb.confignode.rpc.thrift.TMigrateRegionReq;
import org.apache.iotdb.confignode.rpc.thrift.TPermissionInfoResp;
import org.apache.iotdb.confignode.rpc.thrift.TPipeConfigTransferReq;
import org.apache.iotdb.confignode.rpc.thrift.TPipeConfigTransferResp;
-import org.apache.iotdb.confignode.rpc.thrift.TRegionMigrateResultReportReq;
import org.apache.iotdb.confignode.rpc.thrift.TRegionRouteMapResp;
import org.apache.iotdb.confignode.rpc.thrift.TSchemaNodeManagementReq;
import org.apache.iotdb.confignode.rpc.thrift.TSchemaNodeManagementResp;
@@ -284,11 +283,6 @@ public class ConfigNodeRPCServiceProcessor implements IConfigNodeRPCService.Ifac
return resp;
}
- @Override
- public TSStatus reportRegionMigrateResult(TRegionMigrateResultReportReq req) {
- return configManager.reportRegionMigrateResult(req);
- }
-
@Override
public TShowClusterResp showCluster() {
return configManager.showCluster();
diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java
index 9cce75c03f3..f195768e44e 100644
--- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanSerDeTest.java
@@ -103,8 +103,10 @@ import org.apache.iotdb.confignode.consensus.request.write.datanode.RemoveDataNo
import org.apache.iotdb.confignode.consensus.request.write.datanode.UpdateDataNodePlan;
import org.apache.iotdb.confignode.consensus.request.write.function.CreateFunctionPlan;
import org.apache.iotdb.confignode.consensus.request.write.function.DropFunctionPlan;
+import org.apache.iotdb.confignode.consensus.request.write.partition.AddRegionLocationPlan;
import org.apache.iotdb.confignode.consensus.request.write.partition.CreateDataPartitionPlan;
import org.apache.iotdb.confignode.consensus.request.write.partition.CreateSchemaPartitionPlan;
+import org.apache.iotdb.confignode.consensus.request.write.partition.RemoveRegionLocationPlan;
import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeDeactivateTemplatePlan;
import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeDeleteLogicalViewPlan;
import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeDeleteTimeSeriesPlan;
@@ -1832,4 +1834,38 @@ public class ConfigPhysicalPlanSerDeTest {
ConfigPhysicalPlan.Factory.create(pipeDeactivateTemplatePlan.serializeToByteBuffer());
Assert.assertEquals(pipeDeactivateTemplatePlan, deserializedPlan);
}
+
+ @Test
+ public void addRegionLocationPlanTest() throws IOException {
+ AddRegionLocationPlan plan =
+ new AddRegionLocationPlan(
+ new TConsensusGroupId(DataRegion, 1),
+ new TDataNodeLocation(
+ 10000,
+ new TEndPoint("127.0.0.1", 6600),
+ new TEndPoint("127.0.0.1", 7700),
+ new TEndPoint("127.0.0.1", 8800),
+ new TEndPoint("127.0.0.1", 9900),
+ new TEndPoint("127.0.0.1", 11000)));
+ AddRegionLocationPlan dePlan =
+ (AddRegionLocationPlan) ConfigPhysicalPlan.Factory.create(plan.serializeToByteBuffer());
+ Assert.assertEquals(plan, dePlan);
+ }
+
+ @Test
+ public void removeRegionLocationPlanTest() throws IOException {
+ RemoveRegionLocationPlan plan =
+ new RemoveRegionLocationPlan(
+ new TConsensusGroupId(DataRegion, 1),
+ new TDataNodeLocation(
+ 10000,
+ new TEndPoint("127.0.0.1", 6600),
+ new TEndPoint("127.0.0.1", 7700),
+ new TEndPoint("127.0.0.1", 8800),
+ new TEndPoint("127.0.0.1", 9900),
+ new TEndPoint("127.0.0.1", 11000)));
+ RemoveRegionLocationPlan dePlan =
+ (RemoveRegionLocationPlan) ConfigPhysicalPlan.Factory.create(plan.serializeToByteBuffer());
+ Assert.assertEquals(plan, dePlan);
+ }
}
diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/RegionMigrateProcedureTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/AddRegionPeerProcedureTest.java
similarity index 87%
copy from iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/RegionMigrateProcedureTest.java
copy to iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/AddRegionPeerProcedureTest.java
index 012140241e9..7836fb03217 100644
--- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/RegionMigrateProcedureTest.java
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/AddRegionPeerProcedureTest.java
@@ -16,6 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.iotdb.confignode.procedure.impl.region;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
@@ -29,15 +30,13 @@ import org.junit.Assert;
import org.junit.Test;
import java.io.DataOutputStream;
-import java.io.IOException;
import java.nio.ByteBuffer;
-public class RegionMigrateProcedureTest {
-
+public class AddRegionPeerProcedureTest {
@Test
- public void serDeTest() throws IOException {
- RegionMigrateProcedure procedure0 =
- new RegionMigrateProcedure(
+ public void serDeTest() throws Exception {
+ AddRegionPeerProcedure procedure =
+ new AddRegionPeerProcedure(
new TConsensusGroupId(TConsensusGroupType.DataRegion, 10),
new TDataNodeLocation(
1,
@@ -53,13 +52,12 @@ public class RegionMigrateProcedureTest {
new TEndPoint("127.0.0.1", 7),
new TEndPoint("127.0.0.1", 8),
new TEndPoint("127.0.0.1", 9)));
-
try (PublicBAOS byteArrayOutputStream = new PublicBAOS();
DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream)) {
- procedure0.serialize(outputStream);
+ procedure.serialize(outputStream);
ByteBuffer buffer =
ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0, byteArrayOutputStream.size());
- Assert.assertEquals(procedure0, ProcedureFactory.getInstance().create(buffer));
+ Assert.assertEquals(procedure, ProcedureFactory.getInstance().create(buffer));
}
}
}
diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/RegionMigrateProcedureTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/RegionMigrateProcedureTest.java
index 012140241e9..ec432e692c8 100644
--- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/RegionMigrateProcedureTest.java
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/RegionMigrateProcedureTest.java
@@ -52,7 +52,21 @@ public class RegionMigrateProcedureTest {
new TEndPoint("127.0.0.1", 6),
new TEndPoint("127.0.0.1", 7),
new TEndPoint("127.0.0.1", 8),
- new TEndPoint("127.0.0.1", 9)));
+ new TEndPoint("127.0.0.1", 9)),
+ new TDataNodeLocation(
+ 11,
+ new TEndPoint("127.0.0.1", 10),
+ new TEndPoint("127.0.0.1", 11),
+ new TEndPoint("127.0.0.1", 12),
+ new TEndPoint("127.0.0.1", 13),
+ new TEndPoint("127.0.0.1", 14)),
+ new TDataNodeLocation(
+ 15,
+ new TEndPoint("127.0.0.1", 15),
+ new TEndPoint("127.0.0.1", 16),
+ new TEndPoint("127.0.0.1", 17),
+ new TEndPoint("127.0.0.1", 18),
+ new TEndPoint("127.0.0.1", 19)));
try (PublicBAOS byteArrayOutputStream = new PublicBAOS();
DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream)) {
diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/RegionMigrateProcedureTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/RemoveRegionPeerProcedureTest.java
similarity index 87%
copy from iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/RegionMigrateProcedureTest.java
copy to iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/RemoveRegionPeerProcedureTest.java
index 012140241e9..63c6133cb31 100644
--- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/RegionMigrateProcedureTest.java
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/RemoveRegionPeerProcedureTest.java
@@ -16,6 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.iotdb.confignode.procedure.impl.region;
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
@@ -29,15 +30,13 @@ import org.junit.Assert;
import org.junit.Test;
import java.io.DataOutputStream;
-import java.io.IOException;
import java.nio.ByteBuffer;
-public class RegionMigrateProcedureTest {
-
+public class RemoveRegionPeerProcedureTest {
@Test
- public void serDeTest() throws IOException {
- RegionMigrateProcedure procedure0 =
- new RegionMigrateProcedure(
+ public void serDeTest() throws Exception {
+ RemoveRegionPeerProcedure procedure =
+ new RemoveRegionPeerProcedure(
new TConsensusGroupId(TConsensusGroupType.DataRegion, 10),
new TDataNodeLocation(
1,
@@ -53,13 +52,12 @@ public class RegionMigrateProcedureTest {
new TEndPoint("127.0.0.1", 7),
new TEndPoint("127.0.0.1", 8),
new TEndPoint("127.0.0.1", 9)));
-
try (PublicBAOS byteArrayOutputStream = new PublicBAOS();
DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream)) {
- procedure0.serialize(outputStream);
+ procedure.serialize(outputStream);
ByteBuffer buffer =
ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0, byteArrayOutputStream.size());
- Assert.assertEquals(procedure0, ProcedureFactory.getInstance().create(buffer));
+ Assert.assertEquals(procedure, ProcedureFactory.getInstance().create(buffer));
}
}
}
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/IConsensus.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/IConsensus.java
index 4ee908a7af8..e9860379ed4 100644
--- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/IConsensus.java
+++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/IConsensus.java
@@ -143,6 +143,18 @@ public interface IConsensus {
*/
void removeRemotePeer(ConsensusGroupId groupId, Peer peer) throws ConsensusException;
+ /**
+ * Reset the peer list of the corresponding consensus group. Currently only used in the automatic
+ * cleanup of region migration as a rollback for {@link #addRemotePeer(ConsensusGroupId, Peer)},
+ * so it will only be less but not more.
+ *
+ * @param groupId the consensus group
+ * @param peers the new peer list
+ * @throws ConsensusException when resetPeerList doesn't success with other reasons
+ * @throws ConsensusGroupNotExistException when the specified consensus group doesn't exist
+ */
+ void resetPeerList(ConsensusGroupId groupId, List<Peer> peers) throws ConsensusException;
+
// management API
/**
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/common/Peer.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/common/Peer.java
index 11ad7b5bf51..e60efc6565f 100644
--- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/common/Peer.java
+++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/common/Peer.java
@@ -19,8 +19,13 @@
package org.apache.iotdb.consensus.common;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
+import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
+import org.apache.iotdb.commons.consensus.DataRegionId;
+import org.apache.iotdb.commons.consensus.SchemaRegionId;
import org.apache.iotdb.commons.utils.BasicStructureSerDeUtil;
import org.apache.iotdb.commons.utils.ThriftCommonsSerDeUtils;
@@ -99,4 +104,20 @@ public class Peer {
public String toString() {
return "Peer{" + "groupId=" + groupId + ", endpoint=" + endpoint + ", nodeId=" + nodeId + '}';
}
+
+ public static Peer valueOf(
+ TConsensusGroupId consensusGroupId, TDataNodeLocation dataNodeLocation) {
+ if (consensusGroupId.getType() == TConsensusGroupType.SchemaRegion) {
+ return new Peer(
+ new SchemaRegionId(consensusGroupId.getId()),
+ dataNodeLocation.getDataNodeId(),
+ dataNodeLocation.getSchemaRegionConsensusEndPoint());
+ } else if (consensusGroupId.getType() == TConsensusGroupType.DataRegion) {
+ return new Peer(
+ new DataRegionId(consensusGroupId.getId()),
+ dataNodeLocation.getDataNodeId(),
+ dataNodeLocation.getDataRegionConsensusEndPoint());
+ }
+ throw new UnsupportedOperationException();
+ }
}
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java
index babe939393f..a722a3c1d23 100644
--- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java
+++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java
@@ -392,6 +392,30 @@ public class IoTConsensus implements IConsensus {
return new ArrayList<>(stateMachineMap.keySet());
}
+ public void resetPeerList(ConsensusGroupId groupId, List<Peer> peers) throws ConsensusException {
+ IoTConsensusServerImpl impl =
+ Optional.ofNullable(stateMachineMap.get(groupId))
+ .orElseThrow(() -> new ConsensusGroupNotExistException(groupId));
+ if (impl.isReadOnly()) {
+ throw new ConsensusException("system is in read-only status now");
+ } else if (!impl.isActive()) {
+ throw new ConsensusException(
+ "peer is inactive and not ready to receive reset configuration request.");
+ } else {
+ for (Peer peer : impl.getConfiguration()) {
+ if (!peers.contains(peer)) {
+ try {
+ removeRemotePeer(groupId, peer);
+ } catch (ConsensusException e) {
+ logger.error("Failed to remove peer {} from group {}", peer, groupId, e);
+ throw e;
+ }
+ }
+ }
+ impl.resetConfiguration(peers);
+ }
+ }
+
public IoTConsensusServerImpl getImpl(ConsensusGroupId groupId) {
return stateMachineMap.get(groupId);
}
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java
index 800754ddba6..ebee4982058 100644
--- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java
+++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java
@@ -605,27 +605,14 @@ public class IoTConsensusServerImpl {
public void persistConfigurationUpdate() throws ConsensusGroupModifyPeerException {
try {
serializeConfigurationAndFsyncToDisk(CONFIGURATION_TMP_FILE_NAME);
- Path tmpConfigurationPath =
- Paths.get(new File(storageDir, CONFIGURATION_TMP_FILE_NAME).getAbsolutePath());
- Path configurationPath =
- Paths.get(new File(storageDir, CONFIGURATION_FILE_NAME).getAbsolutePath());
- Files.deleteIfExists(configurationPath);
- Files.move(tmpConfigurationPath, configurationPath);
+ tmpConfigurationUpdate(configuration);
} catch (IOException e) {
throw new ConsensusGroupModifyPeerException(
"Unexpected error occurs when update configuration", e);
}
}
- private void serializeConfigurationTo(DataOutputStream outputStream) throws IOException {
- outputStream.writeInt(configuration.size());
- for (Peer peer : configuration) {
- peer.serialize(outputStream);
- }
- }
-
public void recoverConfiguration() {
- ByteBuffer buffer;
try {
Path tmpConfigurationPath =
Paths.get(new File(storageDir, CONFIGURATION_TMP_FILE_NAME).getAbsolutePath());
@@ -635,15 +622,18 @@ public class IoTConsensusServerImpl {
// interrupted
// unexpectedly, we need substitute configuration with tmpConfiguration file
if (Files.exists(tmpConfigurationPath)) {
- if (Files.exists(configurationPath)) {
- Files.delete(configurationPath);
- }
+ Files.deleteIfExists(configurationPath);
Files.move(tmpConfigurationPath, configurationPath);
}
- buffer = ByteBuffer.wrap(Files.readAllBytes(configurationPath));
- int size = buffer.getInt();
- for (int i = 0; i < size; i++) {
- configuration.add(Peer.deserialize(buffer));
+ if (Files.exists(configurationPath)) {
+ recoverFromOldConfigurationFile(configurationPath);
+ } else {
+ // recover from split configuration file
+ Path dirPath = Paths.get(storageDir);
+ List<Peer> tmpPeerList = getConfiguration(dirPath, CONFIGURATION_TMP_FILE_NAME);
+ tmpConfigurationUpdate(tmpPeerList);
+ List<Peer> peerList = getConfiguration(dirPath, CONFIGURATION_FILE_NAME);
+ configuration.addAll(peerList);
}
logger.info("Recover IoTConsensus server Impl, configuration: {}", configuration);
} catch (IOException e) {
@@ -651,6 +641,53 @@ public class IoTConsensusServerImpl {
}
}
+ // @Compatibility
+ private void recoverFromOldConfigurationFile(Path oldConfigurationPath) throws IOException {
+ ByteBuffer buffer = ByteBuffer.wrap(Files.readAllBytes(oldConfigurationPath));
+ int size = buffer.getInt();
+ for (int i = 0; i < size; i++) {
+ configuration.add(Peer.deserialize(buffer));
+ }
+ persistConfiguration();
+ Files.delete(oldConfigurationPath);
+ }
+
+ private void tmpConfigurationUpdate(List<Peer> tmpPeerList) throws IOException {
+ for (Peer peer : tmpPeerList) {
+ Path tmpConfigurationPath =
+ Paths.get(
+ new File(storageDir, peer.getNodeId() + "_" + CONFIGURATION_TMP_FILE_NAME)
+ .getAbsolutePath());
+ Path configurationPath =
+ Paths.get(
+ new File(storageDir, peer.getNodeId() + "_" + CONFIGURATION_FILE_NAME)
+ .getAbsolutePath());
+ Files.deleteIfExists(configurationPath);
+ Files.move(tmpConfigurationPath, configurationPath);
+ }
+ }
+
+ private List<Peer> getConfiguration(Path dirPath, String configurationFileName)
+ throws IOException {
+ ByteBuffer buffer;
+ List<Peer> tmpConfiguration = new ArrayList<>();
+ Path[] files =
+ Files.walk(dirPath)
+ .filter(Files::isRegularFile)
+ .filter(filePath -> filePath.getFileName().toString().contains(configurationFileName))
+ .toArray(Path[]::new);
+ for (Path file : files) {
+ buffer = ByteBuffer.wrap(Files.readAllBytes(file));
+ tmpConfiguration.add(Peer.deserialize(buffer));
+ }
+ return tmpConfiguration;
+ }
+
+ public void resetConfiguration(List<Peer> newConfiguration) {
+ configuration.clear();
+ configuration.addAll(newConfiguration);
+ }
+
public IndexedConsensusRequest buildIndexedConsensusRequestForLocalRequest(
IConsensusRequest request) {
if (request instanceof ComparableConsensusRequest) {
@@ -841,15 +878,20 @@ public class IoTConsensusServerImpl {
private void serializeConfigurationAndFsyncToDisk(String configurationFileName)
throws IOException {
- FileOutputStream fileOutputStream =
- new FileOutputStream(new File(storageDir, configurationFileName));
- DataOutputStream outputStream = new DataOutputStream(fileOutputStream);
- try {
- serializeConfigurationTo(outputStream);
- } finally {
- fileOutputStream.flush();
- fileOutputStream.getFD().sync();
- outputStream.close();
+ for (Peer peer : configuration) {
+ String peerConfigurationFileName = peer.getNodeId() + "_" + configurationFileName;
+ FileOutputStream fileOutputStream =
+ new FileOutputStream(new File(storageDir, peerConfigurationFileName));
+ try (DataOutputStream outputStream = new DataOutputStream(fileOutputStream)) {
+ peer.serialize(outputStream);
+ } finally {
+ try {
+ fileOutputStream.flush();
+ fileOutputStream.getFD().sync();
+ } catch (IOException e) {
+ logger.error("Failed to fsync the configuration file {}", peerConfigurationFileName, e);
+ }
+ }
}
}
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java
index 0a798bee793..7c8d7c8b74e 100644
--- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java
+++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java
@@ -537,6 +537,23 @@ class RatisConsensus implements IConsensus {
sendReconfiguration(RaftGroup.valueOf(raftGroupId, newConfig));
}
+ @Override
+ public void resetPeerList(ConsensusGroupId groupId, List<Peer> peers) throws ConsensusException {
+ final RaftGroupId raftGroupId = Utils.fromConsensusGroupIdToRaftGroupId(groupId);
+ final RaftGroup group = getGroupInfo(raftGroupId);
+
+ // pre-conditions: group exists and myself in this group
+ if (group == null || !group.getPeers().contains(myself)) {
+ throw new ConsensusGroupNotExistException(groupId);
+ }
+
+ final List<RaftPeer> newGroupPeers =
+ Utils.fromPeersAndPriorityToRaftPeers(peers, DEFAULT_PRIORITY);
+ final RaftGroup newGroup = RaftGroup.valueOf(raftGroupId, newGroupPeers);
+
+ sendReconfiguration(newGroup);
+ }
+
/** NOTICE: transferLeader *does not guarantee* the leader be transferred to newLeader. */
@Override
public void transferLeader(ConsensusGroupId groupId, Peer newLeader) throws ConsensusException {
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/simple/SimpleConsensus.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/simple/SimpleConsensus.java
index c7d6141ceba..884d5da9b8c 100644
--- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/simple/SimpleConsensus.java
+++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/simple/SimpleConsensus.java
@@ -243,6 +243,11 @@ class SimpleConsensus implements IConsensus {
return new ArrayList<>(stateMachineMap.keySet());
}
+ @Override
+ public void resetPeerList(ConsensusGroupId groupId, List<Peer> peers) throws ConsensusException {
+ throw new ConsensusException("SimpleConsensus does not support reset peer list");
+ }
+
private String buildPeerDir(ConsensusGroupId groupId) {
return storageDir + File.separator + groupId.getType().getValue() + "_" + groupId.getId();
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java
index 20133990811..60a7fc68392 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java
@@ -105,7 +105,6 @@ import org.apache.iotdb.confignode.rpc.thrift.TMigrateRegionReq;
import org.apache.iotdb.confignode.rpc.thrift.TPermissionInfoResp;
import org.apache.iotdb.confignode.rpc.thrift.TPipeConfigTransferReq;
import org.apache.iotdb.confignode.rpc.thrift.TPipeConfigTransferResp;
-import org.apache.iotdb.confignode.rpc.thrift.TRegionMigrateResultReportReq;
import org.apache.iotdb.confignode.rpc.thrift.TRegionRouteMapResp;
import org.apache.iotdb.confignode.rpc.thrift.TSchemaNodeManagementReq;
import org.apache.iotdb.confignode.rpc.thrift.TSchemaNodeManagementResp;
@@ -421,12 +420,6 @@ public class ConfigNodeClient implements IConfigNodeRPCService.Iface, ThriftClie
resp -> !updateConfigNodeLeader(resp.status));
}
- @Override
- public TSStatus reportRegionMigrateResult(TRegionMigrateResultReportReq req) throws TException {
- return executeRemoteCallWithRetry(
- () -> client.reportRegionMigrateResult(req), status -> !updateConfigNodeLeader(status));
- }
-
@Override
public TShowClusterResp showCluster() throws TException {
return executeRemoteCallWithRetry(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
index d2e57c6bbb6..1ef0692afe3 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
@@ -23,6 +23,7 @@ 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.TFlushReq;
+import org.apache.iotdb.common.rpc.thrift.TRegionMigrateResult;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.common.rpc.thrift.TSetSpaceQuotaReq;
import org.apache.iotdb.common.rpc.thrift.TSetTTLReq;
@@ -200,6 +201,7 @@ import org.apache.iotdb.mpp.rpc.thrift.TPushTopicMetaResp;
import org.apache.iotdb.mpp.rpc.thrift.TPushTopicRespExceptionMessage;
import org.apache.iotdb.mpp.rpc.thrift.TRegionLeaderChangeReq;
import org.apache.iotdb.mpp.rpc.thrift.TRegionRouteReq;
+import org.apache.iotdb.mpp.rpc.thrift.TResetPeerListReq;
import org.apache.iotdb.mpp.rpc.thrift.TRollbackSchemaBlackListReq;
import org.apache.iotdb.mpp.rpc.thrift.TRollbackSchemaBlackListWithTemplateReq;
import org.apache.iotdb.mpp.rpc.thrift.TRollbackViewSchemaBlackListReq;
@@ -1737,6 +1739,17 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
return status;
}
+ // TODO: return which DataNode fail
+ @Override
+ public TSStatus resetPeerList(TResetPeerListReq req) throws TException {
+ return RegionMigrateService.getInstance().resetPeerList(req);
+ }
+
+ @Override
+ public TRegionMigrateResult getRegionMaintainResult(long taskId) throws TException {
+ return RegionMigrateService.getInstance().getRegionMaintainResult(taskId);
+ }
+
private TSStatus createNewRegion(ConsensusGroupId regionId, String storageGroup, long ttl) {
return regionManager.createNewRegion(regionId, storageGroup, ttl);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/RegionMigrateService.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/RegionMigrateService.java
index 092394c9bf2..8fcd602af5e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/RegionMigrateService.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/RegionMigrateService.java
@@ -21,9 +21,10 @@ package org.apache.iotdb.db.service;
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.TRegionMaintainTaskStatus;
import org.apache.iotdb.common.rpc.thrift.TRegionMigrateFailedType;
+import org.apache.iotdb.common.rpc.thrift.TRegionMigrateResult;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
-import org.apache.iotdb.commons.client.exception.ClientManagerException;
import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
import org.apache.iotdb.commons.concurrent.ThreadName;
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
@@ -32,28 +33,27 @@ import org.apache.iotdb.commons.consensus.SchemaRegionId;
import org.apache.iotdb.commons.exception.StartupException;
import org.apache.iotdb.commons.service.IService;
import org.apache.iotdb.commons.service.ServiceType;
-import org.apache.iotdb.confignode.rpc.thrift.TRegionMigrateResultReportReq;
import org.apache.iotdb.consensus.common.Peer;
import org.apache.iotdb.consensus.exception.ConsensusException;
import org.apache.iotdb.consensus.exception.PeerAlreadyInConsensusGroupException;
import org.apache.iotdb.consensus.exception.PeerNotInConsensusGroupException;
import org.apache.iotdb.db.consensus.DataRegionConsensusImpl;
import org.apache.iotdb.db.consensus.SchemaRegionConsensusImpl;
-import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
-import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager;
-import org.apache.iotdb.db.protocol.client.ConfigNodeInfo;
import org.apache.iotdb.db.schemaengine.SchemaEngine;
import org.apache.iotdb.db.storageengine.StorageEngine;
import org.apache.iotdb.db.storageengine.rescon.memory.AbstractPoolManager;
import org.apache.iotdb.mpp.rpc.thrift.TMaintainPeerReq;
+import org.apache.iotdb.mpp.rpc.thrift.TResetPeerListReq;
import org.apache.iotdb.rpc.TSStatusCode;
-import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
public class RegionMigrateService implements IService {
@@ -67,6 +67,13 @@ public class RegionMigrateService implements IService {
private RegionMigratePool regionMigratePool;
+ // Map<taskId, taskStatus>
+ // TODO: Due to the use of procedureId as taskId, it is currently unable to handle the situation
+ // where different asynchronous tasks are submitted to the same datanode within a single procedure
+ private static final ConcurrentHashMap<Long, TRegionMigrateResult> taskResultMap =
+ new ConcurrentHashMap<>();
+ private static final TRegionMigrateResult unfinishedResult = new TRegionMigrateResult();
+
private RegionMigrateService() {}
public static RegionMigrateService getInstance() {
@@ -80,10 +87,17 @@ public class RegionMigrateService implements IService {
* @return if the submit task succeed
*/
public synchronized boolean submitAddRegionPeerTask(TMaintainPeerReq req) {
-
boolean submitSucceed = true;
try {
- regionMigratePool.submit(new AddRegionPeerTask(req.getRegionId(), req.getDestNode()));
+ if (!addToTaskResultMap(req.getTaskId())) {
+ LOGGER.warn(
+ "{} The AddRegionPeerTask {} has already been submitted and will not be submitted again.",
+ REGION_MIGRATE_PROCESS,
+ req.getTaskId());
+ return true;
+ }
+ regionMigratePool.submit(
+ new AddRegionPeerTask(req.getTaskId(), req.getRegionId(), req.getDestNode()));
} catch (Exception e) {
LOGGER.error(
"{}, Submit AddRegionPeerTask error for Region: {}",
@@ -105,7 +119,15 @@ public class RegionMigrateService implements IService {
boolean submitSucceed = true;
try {
- regionMigratePool.submit(new RemoveRegionPeerTask(req.getRegionId(), req.getDestNode()));
+ if (!addToTaskResultMap(req.getTaskId())) {
+ LOGGER.warn(
+ "{} The RemoveRegionPeer {} has already been submitted and will not be submitted again.",
+ REGION_MIGRATE_PROCESS,
+ req.getTaskId());
+ return true;
+ }
+ regionMigratePool.submit(
+ new RemoveRegionPeerTask(req.getTaskId(), req.getRegionId(), req.getDestNode()));
} catch (Exception e) {
LOGGER.error(
"{}, Submit RemoveRegionPeer task error for Region: {}",
@@ -124,10 +146,17 @@ public class RegionMigrateService implements IService {
* @return if the submit task succeed
*/
public synchronized boolean submitDeleteOldRegionPeerTask(TMaintainPeerReq req) {
-
boolean submitSucceed = true;
try {
- regionMigratePool.submit(new DeleteOldRegionPeerTask(req.getRegionId(), req.getDestNode()));
+ if (!addToTaskResultMap(req.getTaskId())) {
+ LOGGER.warn(
+ "{} The DeleteOldRegionPeerTask {} has already been submitted and will not be submitted again.",
+ REGION_MIGRATE_PROCESS,
+ req.getTaskId());
+ return true;
+ }
+ regionMigratePool.submit(
+ new DeleteOldRegionPeerTask(req.getTaskId(), req.getRegionId(), req.getDestNode()));
} catch (Exception e) {
LOGGER.error(
"{}, Submit DeleteOldRegionPeerTask error for Region: {}",
@@ -139,6 +168,30 @@ public class RegionMigrateService implements IService {
return submitSucceed;
}
+ public synchronized TSStatus resetPeerList(TResetPeerListReq req) {
+ List<Peer> correctPeers =
+ req.getCorrectLocations().stream()
+ .map(location -> Peer.valueOf(req.getRegionId(), location))
+ .collect(Collectors.toList());
+ ConsensusGroupId regionId =
+ ConsensusGroupId.Factory.createFromTConsensusGroupId(req.getRegionId());
+ try {
+ if (regionId instanceof DataRegionId) {
+ DataRegionConsensusImpl.getInstance().resetPeerList(regionId, correctPeers);
+ } else {
+ SchemaRegionConsensusImpl.getInstance().resetPeerList(regionId, correctPeers);
+ }
+ } catch (ConsensusException e) {
+ LOGGER.error("reset peer list fail", e);
+ return new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
+ }
+ return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ }
+
+ private boolean addToTaskResultMap(long taskId) {
+ return taskResultMap.putIfAbsent(taskId, unfinishedResult) == null;
+ }
+
@Override
public void start() throws StartupException {
regionMigratePool = new RegionMigratePool();
@@ -190,13 +243,17 @@ public class RegionMigrateService implements IService {
private static final Logger taskLogger = LoggerFactory.getLogger(AddRegionPeerTask.class);
+ private final long taskId;
+
// The RegionGroup that shall perform the add peer process
private final TConsensusGroupId tRegionId;
// The new DataNode to be added in the RegionGroup
private final TDataNodeLocation destDataNode;
- public AddRegionPeerTask(TConsensusGroupId tRegionId, TDataNodeLocation destDataNode) {
+ public AddRegionPeerTask(
+ long taskId, TConsensusGroupId tRegionId, TDataNodeLocation destDataNode) {
+ this.taskId = taskId;
this.tRegionId = tRegionId;
this.destDataNode = destDataNode;
}
@@ -205,11 +262,12 @@ public class RegionMigrateService implements IService {
public void run() {
TSStatus runResult = addPeer();
if (isFailed(runResult)) {
- reportFailed(tRegionId, destDataNode, TRegionMigrateFailedType.AddPeerFailed, runResult);
+ taskFail(
+ taskId, tRegionId, destDataNode, TRegionMigrateFailedType.AddPeerFailed, runResult);
return;
}
- reportSucceed(tRegionId, "AddPeer");
+ taskSucceed(taskId, tRegionId, "AddPeer");
}
private TSStatus addPeer() {
@@ -282,11 +340,15 @@ public class RegionMigrateService implements IService {
private static final Logger taskLogger = LoggerFactory.getLogger(RemoveRegionPeerTask.class);
+ private final long taskId;
+
private final TConsensusGroupId tRegionId;
private final TDataNodeLocation destDataNode;
- public RemoveRegionPeerTask(TConsensusGroupId tRegionId, TDataNodeLocation destDataNode) {
+ public RemoveRegionPeerTask(
+ long taskId, TConsensusGroupId tRegionId, TDataNodeLocation destDataNode) {
+ this.taskId = taskId;
this.tRegionId = tRegionId;
this.destDataNode = destDataNode;
}
@@ -295,9 +357,10 @@ public class RegionMigrateService implements IService {
public void run() {
TSStatus runResult = removePeer();
if (isSucceed(runResult)) {
- reportSucceed(tRegionId, "RemovePeer");
+ taskSucceed(taskId, tRegionId, "RemovePeer");
} else {
- reportFailed(tRegionId, destDataNode, TRegionMigrateFailedType.RemovePeerFailed, runResult);
+ taskFail(
+ taskId, tRegionId, destDataNode, TRegionMigrateFailedType.RemovePeerFailed, runResult);
}
}
@@ -375,13 +438,15 @@ public class RegionMigrateService implements IService {
private static class DeleteOldRegionPeerTask implements Runnable {
private static final Logger taskLogger = LoggerFactory.getLogger(DeleteOldRegionPeerTask.class);
+ private final long taskId;
private final TConsensusGroupId tRegionId;
private final TDataNodeLocation originalDataNode;
public DeleteOldRegionPeerTask(
- TConsensusGroupId tRegionId, TDataNodeLocation originalDataNode) {
+ long taskId, TConsensusGroupId tRegionId, TDataNodeLocation originalDataNode) {
+ this.taskId = taskId;
this.tRegionId = tRegionId;
this.originalDataNode = originalDataNode;
}
@@ -391,7 +456,8 @@ public class RegionMigrateService implements IService {
// deletePeer: remove the peer from the consensus group
TSStatus runResult = deletePeer();
if (isFailed(runResult)) {
- reportFailed(
+ taskFail(
+ taskId,
tRegionId,
originalDataNode,
TRegionMigrateFailedType.RemoveConsensusGroupFailed,
@@ -400,12 +466,17 @@ public class RegionMigrateService implements IService {
// deleteRegion: delete region data
runResult = deleteRegion();
+
if (isFailed(runResult)) {
- reportFailed(
- tRegionId, originalDataNode, TRegionMigrateFailedType.DeleteRegionFailed, runResult);
+ taskFail(
+ taskId,
+ tRegionId,
+ originalDataNode,
+ TRegionMigrateFailedType.DeleteRegionFailed,
+ runResult);
}
- reportSucceed(tRegionId, "DeletePeer");
+ taskSucceed(taskId, tRegionId, "DeletePeer");
}
private TSStatus deletePeer() {
@@ -476,57 +547,39 @@ public class RegionMigrateService implements IService {
private Holder() {}
}
- private static void reportSucceed(TConsensusGroupId tRegionId, String migrateState) {
+ private static void taskSucceed(long taskId, TConsensusGroupId tRegionId, String migrateState) {
TSStatus status = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
status.setMessage(
String.format("Region: %s, state: %s, executed succeed", tRegionId, migrateState));
- TRegionMigrateResultReportReq req = new TRegionMigrateResultReportReq(tRegionId, status);
- try {
- reportRegionMigrateResultToConfigNode(req);
- } catch (Exception e) {
- LOGGER.error(
- "{}, Report region {} migrate result error in reportSucceed, result: {}",
- REGION_MIGRATE_PROCESS,
- tRegionId,
- req,
- e);
- }
+ TRegionMigrateResult req = new TRegionMigrateResult(TRegionMaintainTaskStatus.SUCCESS);
+ req.setRegionId(tRegionId).setMigrateResult(status);
+ taskResultMap.put(taskId, req);
}
- private static void reportFailed(
+ private static void taskFail(
+ long taskId,
TConsensusGroupId tRegionId,
TDataNodeLocation failedNode,
TRegionMigrateFailedType failedType,
TSStatus status) {
Map<TDataNodeLocation, TRegionMigrateFailedType> failedNodeAndReason = new HashMap<>();
failedNodeAndReason.put(failedNode, failedType);
- TRegionMigrateResultReportReq req = new TRegionMigrateResultReportReq(tRegionId, status);
+ TRegionMigrateResult req = new TRegionMigrateResult(TRegionMaintainTaskStatus.FAIL);
+ req.setRegionId(tRegionId).setMigrateResult(status);
req.setFailedNodeAndReason(failedNodeAndReason);
- try {
- reportRegionMigrateResultToConfigNode(req);
- } catch (Exception e) {
- LOGGER.error(
- "{}, Report region {} migrate error in reportFailed, result:{}",
- REGION_MIGRATE_PROCESS,
- tRegionId,
- req,
- e);
- }
+ taskResultMap.put(taskId, req);
}
- private static void reportRegionMigrateResultToConfigNode(TRegionMigrateResultReportReq req)
- throws TException, ClientManagerException {
- TSStatus status;
- try (ConfigNodeClient client =
- ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
- status = client.reportRegionMigrateResult(req);
- LOGGER.info(
- "{}, Report region {} migrate result {} to Config node succeed, result: {}",
- REGION_MIGRATE_PROCESS,
- req.getRegionId(),
- req,
- status);
+ public TRegionMigrateResult getRegionMaintainResult(Long taskId) {
+ TRegionMigrateResult result = new TRegionMigrateResult();
+ if (!taskResultMap.containsKey(taskId)) {
+ result.setTaskStatus(TRegionMaintainTaskStatus.TASK_NOT_EXIST);
+ } else if (taskResultMap.get(taskId) == unfinishedResult) {
+ result.setTaskStatus(TRegionMaintainTaskStatus.PROCESSING);
+ } else {
+ result = taskResultMap.get(taskId);
}
+ return result;
}
private static boolean isSucceed(TSStatus status) {
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/RegionTransitionState.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/DataNodeKillPoints.java
similarity index 75%
copy from iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/RegionTransitionState.java
copy to iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/DataNodeKillPoints.java
index a21a6af941b..9d08cc09b78 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/RegionTransitionState.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/DataNodeKillPoints.java
@@ -17,14 +17,12 @@
* under the License.
*/
-package org.apache.iotdb.confignode.procedure.state;
+package org.apache.iotdb.commons.utils;
-public enum RegionTransitionState {
- REGION_MIGRATE_PREPARE,
- CREATE_NEW_REGION_PEER,
- ADD_REGION_PEER,
- CHANGE_REGION_LEADER,
- REMOVE_REGION_PEER,
- DELETE_OLD_REGION_PEER,
- UPDATE_REGION_LOCATION_CACHE
+@TestOnly
+public enum DataNodeKillPoints {
+ OriginalRemovePeer,
+ OriginalDeleteOldRegionPeer,
+
+ CoordinatorRemovePeer,
}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/FileUtils.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/FileUtils.java
index e197366c28d..86558d4d14c 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/FileUtils.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/FileUtils.java
@@ -34,6 +34,7 @@ import java.nio.file.Files;
import java.nio.file.NoSuchFileException;
import java.util.Arrays;
import java.util.Objects;
+import java.util.concurrent.TimeUnit;
public class FileUtils {
private static final Logger LOGGER = LoggerFactory.getLogger(FileUtils.class);
@@ -257,4 +258,13 @@ public class FileUtils {
LOGGER.info("move file success, {}", fromTo);
return true;
}
+
+ public static void logBreakpoint(String logContent) {
+ LOGGER.info("breakpoint:{}", logContent);
+ try {
+ TimeUnit.SECONDS.sleep(1);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
}
diff --git a/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift b/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift
index ba5a2da6b09..8261ea6e748 100644
--- a/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift
+++ b/iotdb-protocol/thrift-commons/src/main/thrift/common.thrift
@@ -95,7 +95,22 @@ enum TRegionMigrateFailedType {
RemovePeerFailed,
RemoveConsensusGroupFailed,
DeleteRegionFailed,
- CreateRegionFailed
+ CreateRegionFailed,
+ Disconnect,
+}
+
+enum TRegionMaintainTaskStatus {
+ TASK_NOT_EXIST,
+ PROCESSING,
+ SUCCESS,
+ FAIL,
+}
+
+struct TRegionMigrateResult {
+ 1: optional TConsensusGroupId regionId
+ 2: optional TSStatus migrateResult
+ 3: optional map<TDataNodeLocation, TRegionMigrateFailedType> failedNodeAndReason
+ 4: required TRegionMaintainTaskStatus taskStatus
}
struct TFlushReq {
diff --git a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
index b42590483c0..b32e4acb0f9 100644
--- a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
+++ b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
@@ -141,12 +141,6 @@ struct TDataNodeRemoveResp {
2: optional map<common.TDataNodeLocation, common.TSStatus> nodeToStatus
}
-struct TRegionMigrateResultReportReq {
- 1: required common.TConsensusGroupId regionId
- 2: required common.TSStatus migrateResult
- 3: optional map<common.TDataNodeLocation, common.TRegionMigrateFailedType> failedNodeAndReason
-}
-
struct TDataNodeConfigurationResp {
1: required common.TSStatus status
// map<DataNodeId, DataNodeConfiguration>
@@ -987,9 +981,6 @@ service IConfigNodeRPCService {
*/
TDataNodeConfigurationResp getDataNodeConfiguration(i32 dataNodeId)
- /** Report region migration complete */
- common.TSStatus reportRegionMigrateResult(TRegionMigrateResultReportReq req)
-
// ======================================================
// Database
// ======================================================
diff --git a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
index ddc14327a16..632a536137e 100644
--- a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
+++ b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
@@ -51,6 +51,12 @@ struct TCreatePeerReq {
struct TMaintainPeerReq {
1: required common.TConsensusGroupId regionId
2: required common.TDataNodeLocation destNode
+ 3: required i64 taskId
+}
+
+struct TResetPeerListReq {
+ 1: required common.TConsensusGroupId regionId
+ 2: required list<common.TDataNodeLocation> correctLocations
}
struct TFragmentInstanceId {
@@ -673,6 +679,16 @@ service IDataNodeRPCService {
*/
common.TSStatus deleteOldRegionPeer(TMaintainPeerReq req)
+ /**
+ * Reset a consensus group's peer list
+ */
+ common.TSStatus resetPeerList(TResetPeerListReq req);
+
+ /**
+ * Get the result of a region maintainance task
+ */
+ common.TRegionMigrateResult getRegionMaintainResult(i64 taskId)
+
/**
* Config node will disable the Data node, the Data node will not accept read/write request when disabled
* @param data node location