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 f52d752bf84 Region migration related work (#12293)
f52d752bf84 is described below
commit f52d752bf84e1a05a75db04e84125ecb49072a73
Author: Li Yu Heng <[email protected]>
AuthorDate: Fri Apr 12 19:09:52 2024 +0800
Region migration related work (#12293)
---
.../iotdb/it/env/cluster/env/AbstractEnv.java | 29 +-
.../it/env/cluster/node/AbstractNodeWrapper.java | 5 +
.../apache/iotdb/itbase/env/BaseNodeWrapper.java | 2 +
...IoTDBRegionMigrateDataNodeCrashITFramework.java | 15 +-
.../IoTDBRegionMigrateReliabilityITFramework.java | 412 +++++++++++++++------
.../it/regionmigration/KillPointContext.java | 22 +-
.../pass/IoTDBRegionMigrateClusterCrashIT.java | 68 ++++
.../pass/IoTDBRegionMigrateConfigNodeCrashIT.java | 92 ++++-
.../pass/IoTDBRegionMigrateNormalIT.java | 8 +-
.../pass/IoTDBRegionMigrateOtherIT.java | 9 +-
...ateCoordinatorCrashWhenRemoveRemotePeerIT.java} | 24 +-
.../IoTDBRegionMigrateDataNodeCrashIT.java | 59 ++-
...MigrateOriginalCrashWhenDeleteLocalPeerIT.java} | 19 +-
...igrateOriginalCrashWhenRemoveRemotePeerIT.java} | 19 +-
.../apache/iotdb/rpc/TElasticFramedTransport.java | 3 +-
.../client/async/AsyncDataNodeClientPool.java | 7 +
.../client/async/handlers/AsyncClientHandler.java | 1 +
.../iotdb/confignode/manager/ProcedureManager.java | 9 +-
.../iotdb/confignode/procedure/Procedure.java | 15 -
.../procedure/env/RegionMaintainHandler.java | 85 +++--
.../procedure/impl/StateMachineProcedure.java | 18 +
.../impl/region/AddRegionPeerProcedure.java | 91 +++--
.../impl/region/RemoveRegionPeerProcedure.java | 23 +-
.../testonly/CreateManyDatabasesProcedure.java | 10 +-
iotdb-core/consensus/pom.xml | 4 +
.../org/apache/iotdb/consensus/IConsensus.java | 19 +
.../consensus/exception/ConsensusException.java | 4 +
.../apache/iotdb/consensus/iot/IoTConsensus.java | 91 +++--
.../consensus/iot/IoTConsensusServerImpl.java | 130 +++++--
.../service/IoTConsensusRPCServiceProcessor.java | 8 +-
.../iotdb/consensus/ratis/RatisConsensus.java | 37 +-
.../iotdb/consensus/simple/SimpleConsensus.java | 11 +
.../apache/iotdb/consensus/iot/ReplicateTest.java | 21 ++
.../apache/iotdb/consensus/iot/StabilityTest.java | 12 +-
.../iotdb/consensus/ratis/RatisConsensusTest.java | 19 +
.../dataregion/DataRegionStateMachine.java | 15 +-
.../java/org/apache/iotdb/db/service/DataNode.java | 69 +---
.../iotdb/db/service/RegionMigrateService.java | 7 +-
.../dataregion/snapshot/SnapshotLoader.java | 6 -
... => IoTConsensusDeleteLocalPeerKillPoints.java} | 8 +-
...a => IoTConsensusInactivatePeerKillPoints.java} | 8 +-
...TConsensusRemovePeerCoordinatorKillPoints.java} | 2 +-
...nsusRemovePeerKillPoints.java => KillNode.java} | 11 +-
.../src/main/thrift/iotconsensus.thrift | 1 +
44 files changed, 1114 insertions(+), 414 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 68be2f40c51..5b64cb87234 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
@@ -468,7 +468,8 @@ public abstract class AbstractEnv implements BaseEnv {
dataNode = this.dataNodeWrapperList.get(0);
}
- return getWriteConnectionWithSpecifiedDataNode(dataNode, version,
username, password);
+ return getWriteConnectionFromDataNodeList(
+ this.dataNodeWrapperList, version, username, password);
}
protected NodeConnection getWriteConnectionWithSpecifiedDataNode(
@@ -487,6 +488,26 @@ public abstract class AbstractEnv implements BaseEnv {
writeConnection);
}
+ protected NodeConnection getWriteConnectionFromDataNodeList(
+ List<DataNodeWrapper> dataNodeList,
+ Constant.Version version,
+ String username,
+ String password)
+ throws SQLException {
+ List<DataNodeWrapper> dataNodeWrapperListCopy = new
ArrayList<>(dataNodeList);
+ Collections.shuffle(dataNodeWrapperListCopy);
+ SQLException lastException = null;
+ for (DataNodeWrapper dataNode : dataNodeWrapperListCopy) {
+ try {
+ return getWriteConnectionWithSpecifiedDataNode(dataNode, version,
username, password);
+ } catch (SQLException e) {
+ lastException = e;
+ }
+ }
+ logger.error("Failed to get connection from any DataNode, last exception
is ", lastException);
+ throw lastException;
+ }
+
protected List<NodeConnection> getReadConnections(
Constant.Version version, String username, String password) throws
SQLException {
List<String> endpoints = new ArrayList<>();
@@ -600,6 +621,12 @@ public abstract class AbstractEnv implements BaseEnv {
return dataNodeWrapperList;
}
+ public List<AbstractNodeWrapper> getNodeWrapperList() {
+ List<AbstractNodeWrapper> result = new ArrayList<>(configNodeWrapperList);
+ result.addAll(dataNodeWrapperList);
+ return result;
+ }
+
/**
* Get connection to ConfigNode-Leader in ClusterIT environment
*
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 a517fc95829..45a1188a4ca 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
@@ -502,6 +502,11 @@ public abstract class AbstractNodeWrapper implements
BaseNodeWrapper {
}
}
+ @Override
+ public boolean isAlive() {
+ return this.instance != null && this.instance.isAlive();
+ }
+
@Override
public final String getIp() {
return this.nodeAddress;
diff --git
a/integration-test/src/main/java/org/apache/iotdb/itbase/env/BaseNodeWrapper.java
b/integration-test/src/main/java/org/apache/iotdb/itbase/env/BaseNodeWrapper.java
index 73cd372b634..cd0eb86bde1 100644
---
a/integration-test/src/main/java/org/apache/iotdb/itbase/env/BaseNodeWrapper.java
+++
b/integration-test/src/main/java/org/apache/iotdb/itbase/env/BaseNodeWrapper.java
@@ -33,6 +33,8 @@ public interface BaseNodeWrapper {
void stopForcibly();
+ boolean isAlive();
+
String getIp();
int getPort();
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/exception/ConsensusException.java
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/IoTDBRegionMigrateDataNodeCrashITFramework.java
similarity index 65%
copy from
iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/exception/ConsensusException.java
copy to
integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/IoTDBRegionMigrateDataNodeCrashITFramework.java
index 385ecaa89fc..4a363d196e7 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/exception/ConsensusException.java
+++
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/IoTDBRegionMigrateDataNodeCrashITFramework.java
@@ -17,15 +17,14 @@
* under the License.
*/
-package org.apache.iotdb.consensus.exception;
+package org.apache.iotdb.confignode.it.regionmigration;
-public class ConsensusException extends Exception {
+import org.apache.iotdb.commons.utils.KillPoint.KillNode;
- public ConsensusException(String message) {
- super(message);
- }
-
- public ConsensusException(String message, Throwable cause) {
- super(message, cause);
+public class IoTDBRegionMigrateDataNodeCrashITFramework
+ extends IoTDBRegionMigrateReliabilityITFramework {
+ @SafeVarargs
+ public final <T extends Enum<T>> void success(T... dataNodeKillPoints)
throws Exception {
+ successTest(1, 1, 1, 2, noKillPoints(), buildSet(dataNodeKillPoints),
KillNode.ALL_NODES);
}
}
diff --git
a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/IoTDBRegionMigrateReliabilityITFramework.java
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/IoTDBRegionMigrateReliabilityITFramework.java
index 3f33c857fcd..0994ddfb389 100644
---
a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/IoTDBRegionMigrateReliabilityITFramework.java
+++
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/IoTDBRegionMigrateReliabilityITFramework.java
@@ -20,18 +20,26 @@
package org.apache.iotdb.confignode.it.regionmigration;
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.KillPoint.KillNode;
import org.apache.iotdb.commons.utils.KillPoint.KillPoint;
+import org.apache.iotdb.confignode.rpc.thrift.TRegionInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TShowRegionReq;
+import org.apache.iotdb.confignode.rpc.thrift.TShowRegionResp;
import org.apache.iotdb.consensus.ConsensusFactory;
import org.apache.iotdb.consensus.iot.IoTConsensusServerImpl;
import org.apache.iotdb.db.queryengine.common.header.ColumnHeaderConstant;
import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.env.cluster.env.AbstractEnv;
import org.apache.iotdb.it.env.cluster.node.AbstractNodeWrapper;
import org.apache.iotdb.it.env.cluster.node.ConfigNodeWrapper;
+import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
import org.apache.iotdb.itbase.exception.InconsistentDataException;
import org.apache.iotdb.metrics.utils.SystemType;
+import org.apache.thrift.TException;
import org.awaitility.Awaitility;
import org.awaitility.core.ConditionTimeoutException;
import org.junit.After;
@@ -50,8 +58,10 @@ import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
@@ -61,18 +71,48 @@ 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.function.Consumer;
import java.util.stream.Collectors;
public class IoTDBRegionMigrateReliabilityITFramework {
private static final Logger LOGGER =
LoggerFactory.getLogger(IoTDBRegionMigrateReliabilityITFramework.class);
- private static final String INSERTION =
- "INSERT INTO root.sg.d1(timestamp,speed,temperature) values(100, 10.1,
20.7)";
+ private static final String INSERTION1 =
+ "INSERT INTO root.sg.d1(timestamp,speed,temperature) values(100, 1, 2)";
+ private static final String INSERTION2 =
+ "INSERT INTO root.sg.d1(timestamp,speed,temperature) values(101, 3, 4)";
+ private static final String FLUSH_COMMAND = "flush";
private static final String SHOW_REGIONS = "show regions";
private static final String SHOW_DATANODES = "show datanodes";
+ private static final String COUNT_TIMESERIES = "select count(*) from
root.sg.**";
private static final String REGION_MIGRATE_COMMAND_FORMAT = "migrate region
%d from %d to %d";
+ private static final String CONFIGURATION_FILE_NAME = "configuration.dat";
ExecutorService executorService =
IoTDBThreadPoolFactory.newCachedThreadPool("regionMigrateIT");
+ public static Consumer<KillPointContext> actionOfKillNode =
+ context -> {
+ context.getNodeWrapper().stopForcibly();
+ LOGGER.info("Node {} stopped.", context.getNodeWrapper().getId());
+ Assert.assertFalse(context.getNodeWrapper().isAlive());
+ if (context.getNodeWrapper() instanceof ConfigNodeWrapper) {
+ context.getNodeWrapper().start();
+ LOGGER.info("Node {} restarted.", context.getNodeWrapper().getId());
+ Assert.assertTrue(context.getNodeWrapper().isAlive());
+ }
+ };
+
+ public static Consumer<KillPointContext> actionOfRestartCluster =
+ context -> {
+ context
+ .getEnv()
+ .getNodeWrapperList()
+ .parallelStream()
+ .forEach(AbstractNodeWrapper::stopForcibly);
+ LOGGER.info("Cluster has been stopped");
+
context.getEnv().getNodeWrapperList().parallelStream().forEach(AbstractNodeWrapper::start);
+ LOGGER.info("Cluster has been restarted");
+ };
+
@Before
public void setUp() throws Exception {
EnvFactory.getEnv()
@@ -94,7 +134,8 @@ public class IoTDBRegionMigrateReliabilityITFramework {
final int configNodeNum,
final int dataNodeNum,
KeySetView<String, Boolean> killConfigNodeKeywords,
- KeySetView<String, Boolean> killDataNodeKeywords)
+ KeySetView<String, Boolean> killDataNodeKeywords,
+ KillNode killNode)
throws Exception {
generalTestWithAllOptions(
dataReplicateFactor,
@@ -103,10 +144,9 @@ public class IoTDBRegionMigrateReliabilityITFramework {
dataNodeNum,
killConfigNodeKeywords,
killDataNodeKeywords,
+ actionOfKillNode,
true,
- true,
- 0,
- true);
+ killNode);
}
public void failTest(
@@ -115,7 +155,8 @@ public class IoTDBRegionMigrateReliabilityITFramework {
final int configNodeNum,
final int dataNodeNum,
KeySetView<String, Boolean> killConfigNodeKeywords,
- KeySetView<String, Boolean> killDataNodeKeywords)
+ KeySetView<String, Boolean> killDataNodeKeywords,
+ KillNode killNode)
throws Exception {
generalTestWithAllOptions(
dataReplicateFactor,
@@ -124,23 +165,38 @@ public class IoTDBRegionMigrateReliabilityITFramework {
dataNodeNum,
killConfigNodeKeywords,
killDataNodeKeywords,
- true,
- true,
- 60,
- false);
+ actionOfKillNode,
+ false,
+ killNode);
+ }
+
+ public void killClusterTest(
+ KeySetView<String, Boolean> configNodeKeywords, boolean
expectMigrateSuccess)
+ throws Exception {
+ generalTestWithAllOptions(
+ 2,
+ 3,
+ 3,
+ 3,
+ configNodeKeywords,
+ noKillPoints(),
+ actionOfRestartCluster,
+ expectMigrateSuccess,
+ KillNode.ALL_NODES);
}
+ // region general test
+
public void generalTestWithAllOptions(
final int dataReplicateFactor,
final int schemaReplicationFactor,
final int configNodeNum,
final int dataNodeNum,
- KeySetView<String, Boolean> killConfigNodeKeywords,
- KeySetView<String, Boolean> killDataNodeKeywords,
- final boolean checkOriginalRegionDirDeleted,
- final boolean checkConfigurationFileDeleted,
- final int restartTime,
- final boolean isMigrateSuccess)
+ KeySetView<String, Boolean> configNodeKeywords,
+ KeySetView<String, Boolean> dataNodeKeywords,
+ Consumer<KillPointContext> actionWhenDetectKeyWords,
+ final boolean expectMigrateSuccess,
+ KillNode killNode)
throws Exception {
// prepare env
EnvFactory.getEnv()
@@ -148,14 +204,16 @@ public class IoTDBRegionMigrateReliabilityITFramework {
.getCommonConfig()
.setDataReplicationFactor(dataReplicateFactor)
.setSchemaReplicationFactor(schemaReplicationFactor);
- EnvFactory.getEnv().registerConfigNodeKillPoints(new
ArrayList<>(killConfigNodeKeywords));
- EnvFactory.getEnv().registerDataNodeKillPoints(new
ArrayList<>(killDataNodeKeywords));
+ EnvFactory.getEnv().registerConfigNodeKillPoints(new
ArrayList<>(configNodeKeywords));
+ EnvFactory.getEnv().registerDataNodeKillPoints(new
ArrayList<>(dataNodeKeywords));
EnvFactory.getEnv().initClusterEnvironment(configNodeNum, dataNodeNum);
try (final Connection connection = EnvFactory.getEnv().getConnection();
- final Statement statement = connection.createStatement()) {
+ final Statement statement = connection.createStatement();
+ SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient)
EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
- statement.execute(INSERTION);
+ statement.execute(INSERTION1);
ResultSet result = statement.executeQuery(SHOW_REGIONS);
Map<Integer, Set<Integer>> regionMap = getRegionMap(result);
@@ -173,29 +231,35 @@ public class IoTDBRegionMigrateReliabilityITFramework {
checkRegionFileExist(originalDataNode);
checkPeersExist(regionMap.get(selectedRegion), originalDataNode,
selectedRegion);
- // set kill points
- setConfigNodeKillPoints(killConfigNodeKeywords, restartTime);
- setDataNodeKillPoints(killDataNodeKeywords, restartTime);
-
- // region migration start
- statement.execute(regionMigrateCommand(selectedRegion, originalDataNode,
destDataNode));
-
- boolean success = false;
try {
- awaitUntilSuccess(statement, selectedRegion, originalDataNode,
destDataNode);
- success = true;
+ awaitUntilFlush(statement, originalDataNode);
} catch (ConditionTimeoutException e) {
- LOGGER.error("Region migrate failed", e);
+ LOGGER.error("Flush timeout:", e);
+ Assert.fail();
}
- // Assert.assertTrue(isMigrateSuccess == success);
-
- // make sure all kill points have been triggered
- checkKillPointsAllTriggered(killConfigNodeKeywords);
- checkKillPointsAllTriggered(killDataNodeKeywords);
- if (!success) {
- restartAllDataNodes();
+ // set kill points
+ if (killNode == KillNode.ORIGINAL_DATANODE) {
+ setDataNodeKillPoints(
+ Collections.singletonList(
+
EnvFactory.getEnv().dataNodeIdToWrapper(originalDataNode).get()),
+ dataNodeKeywords,
+ actionWhenDetectKeyWords);
+ } else if (killNode == KillNode.DESTINATION_DATANODE) {
+ setDataNodeKillPoints(
+
Collections.singletonList(EnvFactory.getEnv().dataNodeIdToWrapper(destDataNode).get()),
+ dataNodeKeywords,
+ actionWhenDetectKeyWords);
+ } else {
+ setConfigNodeKillPoints(configNodeKeywords, actionWhenDetectKeyWords);
+ setDataNodeKillPoints(
+ EnvFactory.getEnv().getDataNodeWrapperList(),
+ dataNodeKeywords,
+ actionWhenDetectKeyWords);
}
+
+ LOGGER.info("DataNode set before migration: {}",
regionMap.get(selectedRegion));
+
System.out.println(
"originalDataNode: "
+
EnvFactory.getEnv().dataNodeIdToWrapper(originalDataNode).get().getNodePath());
@@ -203,75 +267,103 @@ public class IoTDBRegionMigrateReliabilityITFramework {
"destDataNode: "
+
EnvFactory.getEnv().dataNodeIdToWrapper(destDataNode).get().getNodePath());
- // check if there is anything remain
- if (checkOriginalRegionDirDeleted) {
- if (success) {
- checkRegionFileClear(originalDataNode);
- checkRegionFileExist(destDataNode);
- } else {
- checkRegionFileClear(destDataNode);
- checkRegionFileExist(originalDataNode);
+ // region migration start
+ statement.execute(buildRegionMigrateCommand(selectedRegion,
originalDataNode, destDataNode));
+
+ boolean success = false;
+ try {
+ awaitUntilSuccess(client, selectedRegion, originalDataNode,
destDataNode);
+ success = true;
+ } catch (ConditionTimeoutException e) {
+ if (expectMigrateSuccess) {
+ LOGGER.error("Region migrate failed", e);
+ Assert.fail();
}
}
- if (checkConfigurationFileDeleted) {
- if (success) {
- checkPeersClear(allDataNodeId, originalDataNode, selectedRegion);
- } else {
- checkPeersClear(allDataNodeId, destDataNode, selectedRegion);
- }
+ if (!expectMigrateSuccess && success) {
+ LOGGER.error("Region migrate succeeded unexpectedly");
+ Assert.fail();
}
+ // make sure all kill points have been triggered
+ checkKillPointsAllTriggered(configNodeKeywords);
+ checkKillPointsAllTriggered(dataNodeKeywords);
+
+ // check the remaining file
+ if (success) {
+ checkRegionFileClearIfNodeAlive(originalDataNode);
+ checkRegionFileExistIfNodeAlive(destDataNode);
+ checkPeersClearIfNodeAlive(allDataNodeId, originalDataNode,
selectedRegion);
+ checkClusterStillWritable();
+ } else {
+ checkRegionFileClearIfNodeAlive(destDataNode);
+ checkRegionFileExistIfNodeAlive(originalDataNode);
+ checkPeersClearIfNodeAlive(allDataNodeId, destDataNode,
selectedRegion);
+ }
} catch (InconsistentDataException ignore) {
}
LOGGER.info("test pass");
}
- private void restartAllDataNodes() {
- EnvFactory.getEnv()
- .getDataNodeWrapperList()
+ private void restartDataNodes(List<DataNodeWrapper> dataNodeWrappers) {
+ dataNodeWrappers
.parallelStream()
.forEach(
nodeWrapper -> {
- nodeWrapper.stopForcibly();
+ nodeWrapper.stop();
+ Awaitility.await()
+ .atMost(1, TimeUnit.MINUTES)
+ .pollDelay(2, TimeUnit.SECONDS)
+ .until(() -> !nodeWrapper.isAlive());
+ LOGGER.info("Node {} stopped.", nodeWrapper.getId());
nodeWrapper.start();
+ Awaitility.await()
+ .atMost(1, TimeUnit.MINUTES)
+ .pollDelay(2, TimeUnit.SECONDS)
+ .until(nodeWrapper::isAlive);
+ try {
+ TimeUnit.SECONDS.sleep(10);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ LOGGER.info("Node {} restarted.", nodeWrapper.getId());
});
}
private void setConfigNodeKillPoints(
- KeySetView<String, Boolean> killConfigNodeKeywords, int nodeRestartTime)
{
+ KeySetView<String, Boolean> killConfigNodeKeywords,
Consumer<KillPointContext> action) {
EnvFactory.getEnv()
.getConfigNodeWrapperList()
.forEach(
configNodeWrapper ->
executorService.submit(
() ->
- nodeLogKillPoint(
- configNodeWrapper, killConfigNodeKeywords,
nodeRestartTime)));
+ doActionWhenDetectKeywords(
+ configNodeWrapper, killConfigNodeKeywords,
action)));
}
private void setDataNodeKillPoints(
- KeySetView<String, Boolean> killDataNodeKeywords, int nodeRestartTime) {
- EnvFactory.getEnv()
- .getDataNodeWrapperList()
- .forEach(
- dataNodeWrapper ->
- executorService.submit(
- () ->
- nodeLogKillPoint(dataNodeWrapper,
killDataNodeKeywords, nodeRestartTime)));
+ List<DataNodeWrapper> dataNodeWrappers,
+ KeySetView<String, Boolean> killDataNodeKeywords,
+ Consumer<KillPointContext> action) {
+ dataNodeWrappers.forEach(
+ dataNodeWrapper ->
+ executorService.submit(
+ () -> doActionWhenDetectKeywords(dataNodeWrapper,
killDataNodeKeywords, action)));
}
/**
* Monitor the node's log and kill it when detect specific log.
*
* @param nodeWrapper Easy to understand
- * @param killNodeKeywords When detect these keywords in node's log, stop
the node forcibly
+ * @param keywords When detect these keywords in node's log, stop the node
forcibly
*/
- private static void nodeLogKillPoint(
+ private static void doActionWhenDetectKeywords(
AbstractNodeWrapper nodeWrapper,
- KeySetView<String, Boolean> killNodeKeywords,
- int restartTime) {
- if (killNodeKeywords.isEmpty()) {
+ KeySetView<String, Boolean> keywords,
+ Consumer<KillPointContext> action) {
+ if (keywords.isEmpty()) {
return;
}
final String logFileName;
@@ -314,32 +406,20 @@ public class IoTDBRegionMigrateReliabilityITFramework {
// if trigger more than one keyword at a same time, test code may
have mistakes
Assert.assertTrue(
line,
- killNodeKeywords.stream()
- .map(KillPoint::addKillPointPrefix)
- .filter(line::contains)
- .count()
+
keywords.stream().map(KillPoint::addKillPointPrefix).filter(line::contains).count()
<= 1);
String finalLine = line;
Optional<String> detectedKeyword =
- killNodeKeywords.stream()
+ keywords.stream()
.filter(keyword ->
finalLine.contains(KillPoint.addKillPointPrefix(keyword)))
.findAny();
if (detectedKeyword.isPresent()) {
// each keyword only trigger once
- killNodeKeywords.remove(detectedKeyword.get());
- LOGGER.info("Kill point is triggered: {}", detectedKeyword);
- // reboot the node
- nodeWrapper.stopForcibly();
- if (restartTime > 0) {
- try {
- TimeUnit.SECONDS.sleep(restartTime);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- }
- }
- nodeWrapper.start();
+ keywords.remove(detectedKeyword.get());
+ action.accept(new KillPointContext(nodeWrapper, (AbstractEnv)
EnvFactory.getEnv()));
+ LOGGER.info("Kill point triggered: {}", detectedKeyword.get());
}
- if (killNodeKeywords.isEmpty()) {
+ if (keywords.isEmpty()) {
break;
}
}
@@ -359,8 +439,10 @@ public class IoTDBRegionMigrateReliabilityITFramework {
}
}
- private static String regionMigrateCommand(int who, int from, int to) {
- return String.format(REGION_MIGRATE_COMMAND_FORMAT, who, from, to);
+ private static String buildRegionMigrateCommand(int who, int from, int to) {
+ String result = String.format(REGION_MIGRATE_COMMAND_FORMAT, who, from,
to);
+ LOGGER.info(result);
+ return result;
}
private static Map<Integer, Set<Integer>> getRegionMap(ResultSet
showRegionsResult)
@@ -369,15 +451,26 @@ public class IoTDBRegionMigrateReliabilityITFramework {
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);
+ int regionId =
showRegionsResult.getInt(ColumnHeaderConstant.REGION_ID);
+ int dataNodeId =
showRegionsResult.getInt(ColumnHeaderConstant.DATA_NODE_ID);
+ regionMap.computeIfAbsent(regionId, id -> new
HashSet<>()).add(dataNodeId);
}
}
return regionMap;
}
+ private static Map<Integer, Set<Integer>> getRegionMap(List<TRegionInfo>
regionInfoList) {
+ Map<Integer, Set<Integer>> regionMap = new HashMap<>();
+ regionInfoList.forEach(
+ regionInfo -> {
+ int regionId = regionInfo.getConsensusGroupId().getId();
+ regionMap
+ .computeIfAbsent(regionId, regionId1 -> new HashSet<>())
+ .add(regionInfo.getDataNodeId());
+ });
+ return regionMap;
+ }
+
private static int selectRegion(Map<Integer, Set<Integer>> regionMap) {
return regionMap.keySet().stream().findAny().orElseThrow(() -> new
RuntimeException("gg"));
}
@@ -397,21 +490,55 @@ public class IoTDBRegionMigrateReliabilityITFramework {
.orElseThrow(() -> new RuntimeException("cannot find dest DataNode"));
}
+ private static void awaitUntilFlush(Statement statement, int
originalDataNode) {
+ long startTime = System.currentTimeMillis();
+ File sequence = new File(buildDataPath(originalDataNode, true));
+ File unsequence = new File(buildDataPath(originalDataNode, false));
+ Awaitility.await()
+ .atMost(1, TimeUnit.MINUTES)
+ .pollDelay(2, TimeUnit.SECONDS)
+ .until(
+ () -> {
+ statement.execute(FLUSH_COMMAND);
+ int fileNum = 0;
+ if (sequence.exists() && sequence.listFiles() != null) {
+ fileNum += Objects.requireNonNull(sequence.listFiles()).length;
+ }
+ if (unsequence.exists() && unsequence.listFiles() != null) {
+ fileNum +=
Objects.requireNonNull(unsequence.listFiles()).length;
+ }
+ return fileNum > 0;
+ });
+ LOGGER.info("DataNode {} has been flushed", originalDataNode);
+ LOGGER.info("Flush cost time: {}ms", System.currentTimeMillis() -
startTime);
+ }
+
private static void awaitUntilSuccess(
- Statement statement, int selectedRegion, int originalDataNode, int
destDataNode) {
+ SyncConfigNodeIServiceClient client,
+ int selectedRegion,
+ int originalDataNode,
+ int destDataNode) {
AtomicReference<Set<Integer>> lastTimeDataNodes = new AtomicReference<>();
AtomicReference<Exception> lastException = new AtomicReference<>();
+ AtomicReference<SyncConfigNodeIServiceClient> clientRef = new
AtomicReference<>(client);
try {
Awaitility.await()
- .atMost(2, TimeUnit.MINUTES)
+ .atMost(1, TimeUnit.MINUTES)
+ .pollDelay(2, TimeUnit.SECONDS)
.until(
() -> {
try {
- Map<Integer, Set<Integer>> newRegionMap =
- getRegionMap(statement.executeQuery(SHOW_REGIONS));
+ TShowRegionResp resp = clientRef.get().showRegion(new
TShowRegionReq());
+ Map<Integer, Set<Integer>> newRegionMap =
getRegionMap(resp.getRegionInfoList());
Set<Integer> dataNodes = newRegionMap.get(selectedRegion);
lastTimeDataNodes.set(dataNodes);
return !dataNodes.contains(originalDataNode) &&
dataNodes.contains(destDataNode);
+ } catch (TException e) {
+ clientRef.set(
+ (SyncConfigNodeIServiceClient)
+ EnvFactory.getEnv().getLeaderConfigNodeConnection());
+ lastException.set(e);
+ return false;
} catch (Exception e) {
// Any exception can be ignored
lastException.set(e);
@@ -429,9 +556,21 @@ public class IoTDBRegionMigrateReliabilityITFramework {
lastTimeDataNodes.get().remove(originalDataNode);
lastTimeDataNodes.get().add(destDataNode);
String expectSetStr = lastTimeDataNodes.toString();
- LOGGER.info("DataNode Set {} is unexpected, expect {}", actualSetStr,
expectSetStr);
+ LOGGER.error("DataNode Set {} is unexpected, expect {}", actualSetStr,
expectSetStr);
+ if (lastException.get() == null) {
+ LOGGER.info("No exception during awaiting");
+ } else {
+ LOGGER.error("Last exception during awaiting:", lastException.get());
+ }
throw e;
}
+ LOGGER.info("DataNode set has been successfully changed to {}",
lastTimeDataNodes.get());
+ }
+
+ private static void checkRegionFileExistIfNodeAlive(int dataNode) {
+ if (EnvFactory.getEnv().dataNodeIdToWrapper(dataNode).get().isAlive()) {
+ checkRegionFileExist(dataNode);
+ }
}
private static void checkRegionFileExist(int dataNode) {
@@ -440,18 +579,37 @@ public class IoTDBRegionMigrateReliabilityITFramework {
Assert.assertNotEquals(0,
Objects.requireNonNull(originalRegionDir.listFiles()).length);
}
+ private static void checkRegionFileClearIfNodeAlive(int dataNode) {
+ if (EnvFactory.getEnv().dataNodeIdToWrapper(dataNode).get().isAlive()) {
+ checkRegionFileClear(dataNode);
+ }
+ }
+
/** Check whether the original DataNode's region file has been deleted. */
private static void checkRegionFileClear(int dataNode) {
File originalRegionDir = new File(buildRegionDirPath(dataNode));
Assert.assertTrue(originalRegionDir.isDirectory());
Assert.assertEquals(0,
Objects.requireNonNull(originalRegionDir.listFiles()).length);
- LOGGER.info("Original region clear");
+ LOGGER.info("Original DataNode {} region file clear", dataNode);
+ }
+
+ private static void checkPeersExistIfNodeAlive(
+ Set<Integer> dataNodes, int originalDataNode, int regionId) {
+ dataNodes.forEach(
+ targetDataNode -> checkPeerExistIfNodeAlive(targetDataNode,
originalDataNode, regionId));
}
private static void checkPeersExist(Set<Integer> dataNodes, int
originalDataNode, int regionId) {
dataNodes.forEach(targetDataNode -> checkPeerExist(targetDataNode,
originalDataNode, regionId));
}
+ private static void checkPeerExistIfNodeAlive(
+ int checkTargetDataNode, int originalDataNode, int regionId) {
+ if
(EnvFactory.getEnv().dataNodeIdToWrapper(checkTargetDataNode).get().isAlive()) {
+ checkPeerExist(checkTargetDataNode, originalDataNode, regionId);
+ }
+ }
+
private static void checkPeerExist(int checkTargetDataNode, int
originalDataNode, int regionId) {
File expectExistedFile =
new File(buildConfigurationDataFilePath(checkTargetDataNode,
originalDataNode, regionId));
@@ -460,6 +618,15 @@ public class IoTDBRegionMigrateReliabilityITFramework {
expectExistedFile.exists());
}
+ private static void checkPeersClearIfNodeAlive(
+ Set<Integer> dataNodes, int originalDataNode, int regionId) {
+ dataNodes.stream()
+ .filter(dataNode -> dataNode != originalDataNode)
+ .forEach(
+ targetDataNode ->
+ checkPeerClearIfNodeAlive(targetDataNode, originalDataNode,
regionId));
+ }
+
private static void checkPeersClear(Set<Integer> dataNodes, int
originalDataNode, int regionId) {
dataNodes.stream()
.filter(dataNode -> dataNode != originalDataNode)
@@ -467,12 +634,35 @@ public class IoTDBRegionMigrateReliabilityITFramework {
LOGGER.info("Peer clear");
}
+ private static void checkPeerClearIfNodeAlive(
+ int checkTargetDataNode, int originalDataNode, int regionId) {
+ if
(EnvFactory.getEnv().dataNodeIdToWrapper(checkTargetDataNode).get().isAlive()) {
+ checkPeerClear(checkTargetDataNode, originalDataNode, regionId);
+ }
+ }
+
private static void checkPeerClear(int checkTargetDataNode, int
originalDataNode, int regionId) {
File expectDeletedFile =
new File(buildConfigurationDataFilePath(checkTargetDataNode,
originalDataNode, regionId));
Assert.assertFalse(
"configuration file should be deleted, but it didn't: " +
expectDeletedFile.getPath(),
expectDeletedFile.exists());
+ LOGGER.info("configuration file has been deleted: {}",
expectDeletedFile.getPath());
+ }
+
+ private void checkClusterStillWritable() {
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute(INSERTION2);
+ ResultSet resultSet = statement.executeQuery(COUNT_TIMESERIES);
+ resultSet.next();
+ Assert.assertEquals(2, resultSet.getLong(1));
+ Assert.assertEquals(2, resultSet.getLong(2));
+ LOGGER.info("Region group is still writable");
+ } catch (SQLException e) {
+ LOGGER.error("Something wrong", e);
+ Assert.fail("Something wrong");
+ }
}
private static String buildRegionDirPath(int dataNode) {
@@ -488,12 +678,26 @@ public class IoTDBRegionMigrateReliabilityITFramework {
+ IoTDBConstant.DATA_REGION_FOLDER_NAME;
}
+ private static String buildDataPath(int dataNode, boolean isSequence) {
+ String nodePath =
EnvFactory.getEnv().dataNodeIdToWrapper(dataNode).get().getNodePath();
+ return nodePath
+ + File.separator
+ + IoTDBConstant.DATA_FOLDER_NAME
+ + File.separator
+ + "datanode"
+ + File.separator
+ + IoTDBConstant.DATA_FOLDER_NAME
+ + File.separator
+ + (isSequence ? IoTDBConstant.SEQUENCE_FOLDER_NAME :
IoTDBConstant.UNSEQUENCE_FOLDER_NAME);
+ }
+
private static String buildConfigurationDataFilePath(
int localDataNodeId, int remoteDataNodeId, int regionId) {
String configurationDatDirName =
buildRegionDirPath(localDataNodeId) + File.separator + "1_" + regionId;
String expectDeletedFileName =
-
IoTConsensusServerImpl.generateConfigurationDatFileName(remoteDataNodeId);
+ IoTConsensusServerImpl.generateConfigurationDatFileName(
+ remoteDataNodeId, CONFIGURATION_FILE_NAME);
return configurationDatDirName + File.separator + expectDeletedFileName;
}
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/exception/ConsensusException.java
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/KillPointContext.java
similarity index 61%
copy from
iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/exception/ConsensusException.java
copy to
integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/KillPointContext.java
index 385ecaa89fc..d22b6dc1528 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/exception/ConsensusException.java
+++
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/KillPointContext.java
@@ -17,15 +17,25 @@
* under the License.
*/
-package org.apache.iotdb.consensus.exception;
+package org.apache.iotdb.confignode.it.regionmigration;
-public class ConsensusException extends Exception {
+import org.apache.iotdb.it.env.cluster.env.AbstractEnv;
+import org.apache.iotdb.it.env.cluster.node.AbstractNodeWrapper;
- public ConsensusException(String message) {
- super(message);
+public class KillPointContext {
+ AbstractNodeWrapper nodeWrapper;
+ AbstractEnv env;
+
+ public KillPointContext(AbstractNodeWrapper nodeWrapper, AbstractEnv env) {
+ this.nodeWrapper = nodeWrapper;
+ this.env = env;
+ }
+
+ public AbstractNodeWrapper getNodeWrapper() {
+ return nodeWrapper;
}
- public ConsensusException(String message, Throwable cause) {
- super(message, cause);
+ public AbstractEnv getEnv() {
+ return env;
}
}
diff --git
a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/IoTDBRegionMigrateClusterCrashIT.java
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/IoTDBRegionMigrateClusterCrashIT.java
new file mode 100644
index 00000000000..db8ad76e491
--- /dev/null
+++
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/IoTDBRegionMigrateClusterCrashIT.java
@@ -0,0 +1,68 @@
+/*
+ * 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.regionmigration.pass;
+
+import
org.apache.iotdb.confignode.it.regionmigration.IoTDBRegionMigrateReliabilityITFramework;
+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.it.framework.IoTDBTestRunner;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+@RunWith(IoTDBTestRunner.class)
+public class IoTDBRegionMigrateClusterCrashIT extends
IoTDBRegionMigrateReliabilityITFramework {
+
+ @Test
+ public void clusterCrash1() throws Exception {
+ killClusterTest(buildSet(AddRegionPeerState.CREATE_NEW_REGION_PEER), true);
+ }
+
+ @Test
+ public void clusterCrash2() throws Exception {
+ killClusterTest(buildSet(AddRegionPeerState.DO_ADD_REGION_PEER), false);
+ }
+
+ @Test
+ public void clusterCrash3() throws Exception {
+ killClusterTest(buildSet(AddRegionPeerState.UPDATE_REGION_LOCATION_CACHE),
true);
+ }
+
+ @Test
+ public void clusterCrash4() throws Exception {
+ killClusterTest(buildSet(RegionTransitionState.CHANGE_REGION_LEADER),
true);
+ }
+
+ @Test
+ public void clusterCrash6() throws Exception {
+ killClusterTest(buildSet(RemoveRegionPeerState.REMOVE_REGION_PEER), true);
+ }
+
+ @Test
+ public void clusterCrash7() throws Exception {
+ killClusterTest(buildSet(RemoveRegionPeerState.DELETE_OLD_REGION_PEER),
true);
+ }
+
+ @Test
+ public void clusterCrash8() throws Exception {
+
killClusterTest(buildSet(RemoveRegionPeerState.REMOVE_REGION_LOCATION_CACHE),
true);
+ }
+}
diff --git
a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/IoTDBRegionMigrateConfigNodeCrashIT.java
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/IoTDBRegionMigrateConfigNodeCrashIT.java
index 5b21cc81bb1..d9bb2acf2d8 100644
---
a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/IoTDBRegionMigrateConfigNodeCrashIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/IoTDBRegionMigrateConfigNodeCrashIT.java
@@ -19,61 +19,119 @@
package org.apache.iotdb.confignode.it.regionmigration.pass;
+import org.apache.iotdb.commons.utils.KillPoint.KillNode;
import org.apache.iotdb.commons.utils.KillPoint.KillPoint;
import
org.apache.iotdb.confignode.it.regionmigration.IoTDBRegionMigrateReliabilityITFramework;
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.it.framework.IoTDBTestRunner;
import org.junit.Ignore;
import org.junit.Test;
+import org.junit.runner.RunWith;
import java.util.Arrays;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
+@RunWith(IoTDBTestRunner.class)
public class IoTDBRegionMigrateConfigNodeCrashIT extends
IoTDBRegionMigrateReliabilityITFramework {
@Test
@Ignore
- public void cnCrashDuringPreCheck() throws Exception {
- successTest(1, 1, 1, 2,
buildSet(RegionTransitionState.REGION_MIGRATE_PREPARE), noKillPoints());
+ public void cnCrashDuringPreCheckTest() throws Exception {
+ successTest(
+ 1,
+ 1,
+ 1,
+ 2,
+ buildSet(RegionTransitionState.REGION_MIGRATE_PREPARE),
+ noKillPoints(),
+ KillNode.CONFIG_NODE);
}
@Test
- public void cnCrashDuringCreatePeer() throws Exception {
- successTest(1, 1, 1, 2,
buildSet(AddRegionPeerState.CREATE_NEW_REGION_PEER), noKillPoints());
+ public void cnCrashDuringCreatePeerTest() throws Exception {
+ successTest(
+ 1,
+ 1,
+ 1,
+ 2,
+ buildSet(AddRegionPeerState.CREATE_NEW_REGION_PEER),
+ noKillPoints(),
+ KillNode.CONFIG_NODE);
}
@Test
- public void cnCrashDuringDoAddPeer() throws Exception {
- successTest(1, 1, 1, 2, buildSet(AddRegionPeerState.DO_ADD_REGION_PEER),
noKillPoints());
+ public void testCnCrashDuringDoAddPeer() throws Exception {
+ successTest(
+ 1,
+ 1,
+ 1,
+ 2,
+ buildSet(AddRegionPeerState.DO_ADD_REGION_PEER),
+ noKillPoints(),
+ KillNode.CONFIG_NODE);
}
@Test
- public void cnCrashDuringUpdateCache() throws Exception {
+ public void cnCrashDuringUpdateCacheTest() throws Exception {
successTest(
- 1, 1, 1, 2, buildSet(AddRegionPeerState.UPDATE_REGION_LOCATION_CACHE),
noKillPoints());
+ 1,
+ 1,
+ 1,
+ 2,
+ buildSet(AddRegionPeerState.UPDATE_REGION_LOCATION_CACHE),
+ noKillPoints(),
+ KillNode.CONFIG_NODE);
}
@Test
- public void cnCrashDuringChangeRegionLeader() throws Exception {
- successTest(1, 1, 1, 2,
buildSet(RegionTransitionState.CHANGE_REGION_LEADER), noKillPoints());
+ public void cnCrashDuringChangeRegionLeaderTest() throws Exception {
+ successTest(
+ 1,
+ 1,
+ 1,
+ 2,
+ buildSet(RegionTransitionState.CHANGE_REGION_LEADER),
+ noKillPoints(),
+ KillNode.CONFIG_NODE);
}
@Test
- public void cnCrashDuringRemoveRegionPeer() throws Exception {
- successTest(1, 1, 1, 2,
buildSet(RemoveRegionPeerState.REMOVE_REGION_PEER), noKillPoints());
+ public void cnCrashDuringRemoveRegionPeerTest() throws Exception {
+ successTest(
+ 1,
+ 1,
+ 1,
+ 2,
+ buildSet(RemoveRegionPeerState.REMOVE_REGION_PEER),
+ noKillPoints(),
+ KillNode.CONFIG_NODE);
}
@Test
- public void cnCrashDuringDeleteOldRegionPeer() throws Exception {
- successTest(1, 1, 1, 2,
buildSet(RemoveRegionPeerState.DELETE_OLD_REGION_PEER), noKillPoints());
+ public void cnCrashDuringDeleteOldRegionPeerTest() throws Exception {
+ successTest(
+ 1,
+ 1,
+ 1,
+ 2,
+ buildSet(RemoveRegionPeerState.DELETE_OLD_REGION_PEER),
+ noKillPoints(),
+ KillNode.CONFIG_NODE);
}
@Test
- public void cnCrashDuringRemoveRegionLocationCache() throws Exception {
+ public void cnCrashDuringRemoveRegionLocationCacheTest() throws Exception {
successTest(
- 1, 1, 1, 2,
buildSet(RemoveRegionPeerState.REMOVE_REGION_LOCATION_CACHE), noKillPoints());
+ 1,
+ 1,
+ 1,
+ 2,
+ buildSet(RemoveRegionPeerState.REMOVE_REGION_LOCATION_CACHE),
+ noKillPoints(),
+ KillNode.CONFIG_NODE);
}
@Test
@@ -87,6 +145,6 @@ public class IoTDBRegionMigrateConfigNodeCrashIT extends
IoTDBRegionMigrateRelia
Arrays.stream(RemoveRegionPeerState.values())
.map(KillPoint::enumToString)
.collect(Collectors.toList()));
- successTest(1, 1, 1, 2, killConfigNodeKeywords, noKillPoints());
+ successTest(1, 1, 1, 2, killConfigNodeKeywords, noKillPoints(),
KillNode.CONFIG_NODE);
}
}
diff --git
a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/IoTDBRegionMigrateNormalIT.java
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/IoTDBRegionMigrateNormalIT.java
index 072530da626..f6e31d08304 100644
---
a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/IoTDBRegionMigrateNormalIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/IoTDBRegionMigrateNormalIT.java
@@ -19,24 +19,22 @@
package org.apache.iotdb.confignode.it.regionmigration.pass;
+import org.apache.iotdb.commons.utils.KillPoint.KillNode;
import
org.apache.iotdb.confignode.it.regionmigration.IoTDBRegionMigrateReliabilityITFramework;
import org.apache.iotdb.it.framework.IoTDBTestRunner;
-import org.apache.iotdb.itbase.category.ClusterIT;
import org.junit.Test;
-import org.junit.experimental.categories.Category;
import org.junit.runner.RunWith;
@RunWith(IoTDBTestRunner.class)
-@Category({ClusterIT.class})
public class IoTDBRegionMigrateNormalIT extends
IoTDBRegionMigrateReliabilityITFramework {
@Test
public void normal1C2DTest() throws Exception {
- successTest(1, 1, 1, 2, noKillPoints(), noKillPoints());
+ successTest(1, 1, 1, 2, noKillPoints(), noKillPoints(),
KillNode.ALL_NODES);
}
@Test
public void normal3C3DTest() throws Exception {
- successTest(2, 3, 3, 3, noKillPoints(), noKillPoints());
+ successTest(2, 3, 3, 3, noKillPoints(), noKillPoints(),
KillNode.ALL_NODES);
}
}
diff --git
a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/IoTDBRegionMigrateOtherIT.java
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/IoTDBRegionMigrateOtherIT.java
index 6d9444c9e4f..f4ca461edd8 100644
---
a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/IoTDBRegionMigrateOtherIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/IoTDBRegionMigrateOtherIT.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.confignode.it.regionmigration.pass;
+import org.apache.iotdb.commons.utils.KillPoint.KillNode;
import org.apache.iotdb.commons.utils.KillPoint.NeverTriggeredKillPoint;
import
org.apache.iotdb.confignode.it.regionmigration.IoTDBRegionMigrateReliabilityITFramework;
import org.apache.iotdb.it.framework.IoTDBTestRunner;
@@ -36,7 +37,13 @@ public class IoTDBRegionMigrateOtherIT extends
IoTDBRegionMigrateReliabilityITFr
public void badKillPoint() throws Exception {
try {
successTest(
- 1, 1, 1, 2,
buildSet(NeverTriggeredKillPoint.NEVER_TRIGGERED_KILL_POINT), noKillPoints());
+ 1,
+ 1,
+ 1,
+ 2,
+ buildSet(NeverTriggeredKillPoint.NEVER_TRIGGERED_KILL_POINT),
+ noKillPoints(),
+ KillNode.ALL_NODES);
} catch (AssertionError e) {
return;
}
diff --git
a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/notpass/datanodecrash/CoordinatorRemoveRemotePeerCrashIT.java
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/datanodecrash/IoTDBRegionMigrateCoordinatorCrashWhenRemoveRemotePeerIT.java
similarity index 64%
rename from
integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/notpass/datanodecrash/CoordinatorRemoveRemotePeerCrashIT.java
rename to
integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/datanodecrash/IoTDBRegionMigrateCoordinatorCrashWhenRemoveRemotePeerIT.java
index 7533424909b..5a63e76f979 100644
---
a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/notpass/datanodecrash/CoordinatorRemoveRemotePeerCrashIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/datanodecrash/IoTDBRegionMigrateCoordinatorCrashWhenRemoveRemotePeerIT.java
@@ -17,35 +17,37 @@
* under the License.
*/
-package org.apache.iotdb.confignode.it.regionmigration.notpass.datanodecrash;
+package org.apache.iotdb.confignode.it.regionmigration.pass.datanodecrash;
-import
org.apache.iotdb.commons.utils.KillPoint.IoTConsensusRemovePeerKillPoints;
-import
org.apache.iotdb.confignode.it.regionmigration.IoTDBRegionMigrateReliabilityITFramework;
+import
org.apache.iotdb.commons.utils.KillPoint.IoTConsensusRemovePeerCoordinatorKillPoints;
+import
org.apache.iotdb.confignode.it.regionmigration.IoTDBRegionMigrateDataNodeCrashITFramework;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
import org.junit.Test;
+import org.junit.runner.RunWith;
-public class CoordinatorRemoveRemotePeerCrashIT extends
IoTDBRegionMigrateReliabilityITFramework {
- private <T extends Enum<T>> void base(T... dataNodeKillPoints) throws
Exception {
- successTest(1, 1, 1, 2, noKillPoints(), buildSet(dataNodeKillPoints));
- }
+@RunWith(IoTDBTestRunner.class)
+public class IoTDBRegionMigrateCoordinatorCrashWhenRemoveRemotePeerIT
+ extends IoTDBRegionMigrateDataNodeCrashITFramework {
@Test
public void initCrash() throws Exception {
- base(IoTConsensusRemovePeerKillPoints.INIT);
+ success(IoTConsensusRemovePeerCoordinatorKillPoints.INIT);
}
@Test
public void crashAfterNotifyPeersToRemoveSyncLogChannel() throws Exception {
-
base(IoTConsensusRemovePeerKillPoints.AFTER_NOTIFY_PEERS_TO_REMOVE_SYNC_LOG_CHANNEL);
+ success(
+
IoTConsensusRemovePeerCoordinatorKillPoints.AFTER_NOTIFY_PEERS_TO_REMOVE_SYNC_LOG_CHANNEL);
}
@Test
public void crashAfterInactivePeer() throws Exception {
- base(IoTConsensusRemovePeerKillPoints.AFTER_INACTIVE_PEER);
+ success(IoTConsensusRemovePeerCoordinatorKillPoints.AFTER_INACTIVE_PEER);
}
@Test
public void crashAfterFinish() throws Exception {
- base(IoTConsensusRemovePeerKillPoints.FINISH);
+ success(IoTConsensusRemovePeerCoordinatorKillPoints.FINISH);
}
}
diff --git
a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/notpass/datanodecrash/IoTDBRegionMigrateDataNodeCrashIT.java
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/datanodecrash/IoTDBRegionMigrateDataNodeCrashIT.java
similarity index 58%
rename from
integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/notpass/datanodecrash/IoTDBRegionMigrateDataNodeCrashIT.java
rename to
integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/datanodecrash/IoTDBRegionMigrateDataNodeCrashIT.java
index 6b023274aef..a5c619ec704 100644
---
a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/notpass/datanodecrash/IoTDBRegionMigrateDataNodeCrashIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/datanodecrash/IoTDBRegionMigrateDataNodeCrashIT.java
@@ -17,9 +17,10 @@
* under the License.
*/
-package org.apache.iotdb.confignode.it.regionmigration.notpass.datanodecrash;
+package org.apache.iotdb.confignode.it.regionmigration.pass.datanodecrash;
import org.apache.iotdb.commons.utils.KillPoint.DataNodeKillPoints;
+import org.apache.iotdb.commons.utils.KillPoint.KillNode;
import
org.apache.iotdb.confignode.it.regionmigration.IoTDBRegionMigrateReliabilityITFramework;
import org.junit.Test;
@@ -27,15 +28,33 @@ import org.junit.Test;
public class IoTDBRegionMigrateDataNodeCrashIT extends
IoTDBRegionMigrateReliabilityITFramework {
// region Coordinator DataNode crash tests
+ private final int dataReplicateFactor = 2;
+ private final int schemaReplicationFactor = 2;
+ private final int configNodeNum = 1;
+ private final int dataNodeNum = 3;
+
@Test
public void coordinatorCrashDuringAddPeerTransition() throws Exception {
failTest(
- 2, 2, 1, 3, noKillPoints(),
buildSet(DataNodeKillPoints.COORDINATOR_ADD_PEER_TRANSITION));
+ 2,
+ 2,
+ 1,
+ 3,
+ noKillPoints(),
+ buildSet(DataNodeKillPoints.COORDINATOR_ADD_PEER_TRANSITION),
+ KillNode.COORDINATOR_DATANODE);
}
@Test
public void coordinatorCrashDuringAddPeerDone() throws Exception {
- failTest(2, 2, 1, 3, noKillPoints(),
buildSet(DataNodeKillPoints.COORDINATOR_ADD_PEER_DONE));
+ failTest(
+ 2,
+ 2,
+ 1,
+ 3,
+ noKillPoints(),
+ buildSet(DataNodeKillPoints.COORDINATOR_ADD_PEER_DONE),
+ KillNode.COORDINATOR_DATANODE);
}
// endregion ----------------------------------------------
@@ -44,7 +63,14 @@ public class IoTDBRegionMigrateDataNodeCrashIT extends
IoTDBRegionMigrateReliabi
@Test
public void originalCrashDuringAddPeerDone() throws Exception {
- failTest(2, 2, 1, 3, noKillPoints(),
buildSet(DataNodeKillPoints.ORIGINAL_ADD_PEER_DONE));
+ failTest(
+ 2,
+ 2,
+ 1,
+ 3,
+ noKillPoints(),
+ buildSet(DataNodeKillPoints.ORIGINAL_ADD_PEER_DONE),
+ KillNode.ORIGINAL_DATANODE);
}
// endregion ----------------------------------------------
@@ -54,18 +80,37 @@ public class IoTDBRegionMigrateDataNodeCrashIT extends
IoTDBRegionMigrateReliabi
@Test
public void destinationCrashDuringCreateLocalPeer() throws Exception {
failTest(
- 2, 2, 1, 3, noKillPoints(),
buildSet(DataNodeKillPoints.DESTINATION_CREATE_LOCAL_PEER));
+ 2,
+ 2,
+ 1,
+ 3,
+ noKillPoints(),
+ buildSet(DataNodeKillPoints.DESTINATION_CREATE_LOCAL_PEER),
+ KillNode.DESTINATION_DATANODE);
}
@Test
public void destinationCrashDuringAddPeerTransition() throws Exception {
failTest(
- 2, 2, 1, 3, noKillPoints(),
buildSet(DataNodeKillPoints.DESTINATION_ADD_PEER_TRANSITION));
+ 2,
+ 2,
+ 1,
+ 3,
+ noKillPoints(),
+ buildSet(DataNodeKillPoints.DESTINATION_ADD_PEER_TRANSITION),
+ KillNode.DESTINATION_DATANODE);
}
@Test
public void destinationCrashDuringAddPeerDone() throws Exception {
- failTest(2, 2, 1, 3, noKillPoints(),
buildSet(DataNodeKillPoints.DESTINATION_ADD_PEER_DONE));
+ failTest(
+ 2,
+ 2,
+ 1,
+ 3,
+ noKillPoints(),
+ buildSet(DataNodeKillPoints.DESTINATION_ADD_PEER_DONE),
+ KillNode.DESTINATION_DATANODE);
}
// endregion ----------------------------------------------
diff --git
a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/IoTDBRegionMigrateNormalIT.java
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/datanodecrash/IoTDBRegionMigrateOriginalCrashWhenDeleteLocalPeerIT.java
similarity index 65%
copy from
integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/IoTDBRegionMigrateNormalIT.java
copy to
integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/datanodecrash/IoTDBRegionMigrateOriginalCrashWhenDeleteLocalPeerIT.java
index 072530da626..a21884b5da2 100644
---
a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/IoTDBRegionMigrateNormalIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/datanodecrash/IoTDBRegionMigrateOriginalCrashWhenDeleteLocalPeerIT.java
@@ -17,26 +17,25 @@
* under the License.
*/
-package org.apache.iotdb.confignode.it.regionmigration.pass;
+package org.apache.iotdb.confignode.it.regionmigration.pass.datanodecrash;
-import
org.apache.iotdb.confignode.it.regionmigration.IoTDBRegionMigrateReliabilityITFramework;
+import
org.apache.iotdb.commons.utils.KillPoint.IoTConsensusDeleteLocalPeerKillPoints;
+import
org.apache.iotdb.confignode.it.regionmigration.IoTDBRegionMigrateDataNodeCrashITFramework;
import org.apache.iotdb.it.framework.IoTDBTestRunner;
-import org.apache.iotdb.itbase.category.ClusterIT;
import org.junit.Test;
-import org.junit.experimental.categories.Category;
import org.junit.runner.RunWith;
@RunWith(IoTDBTestRunner.class)
-@Category({ClusterIT.class})
-public class IoTDBRegionMigrateNormalIT extends
IoTDBRegionMigrateReliabilityITFramework {
+public class IoTDBRegionMigrateOriginalCrashWhenDeleteLocalPeerIT
+ extends IoTDBRegionMigrateDataNodeCrashITFramework {
@Test
- public void normal1C2DTest() throws Exception {
- successTest(1, 1, 1, 2, noKillPoints(), noKillPoints());
+ public void crashBeforeDelete() throws Exception {
+ success(IoTConsensusDeleteLocalPeerKillPoints.BEFORE_DELETE);
}
@Test
- public void normal3C3DTest() throws Exception {
- successTest(2, 3, 3, 3, noKillPoints(), noKillPoints());
+ public void crashAfterDelete() throws Exception {
+ success(IoTConsensusDeleteLocalPeerKillPoints.AFTER_DELETE);
}
}
diff --git
a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/IoTDBRegionMigrateNormalIT.java
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/datanodecrash/IoTDBRegionMigrateOriginalCrashWhenRemoveRemotePeerIT.java
similarity index 64%
copy from
integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/IoTDBRegionMigrateNormalIT.java
copy to
integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/datanodecrash/IoTDBRegionMigrateOriginalCrashWhenRemoveRemotePeerIT.java
index 072530da626..cf6e0eb19e6 100644
---
a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/IoTDBRegionMigrateNormalIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/datanodecrash/IoTDBRegionMigrateOriginalCrashWhenRemoveRemotePeerIT.java
@@ -17,26 +17,25 @@
* under the License.
*/
-package org.apache.iotdb.confignode.it.regionmigration.pass;
+package org.apache.iotdb.confignode.it.regionmigration.pass.datanodecrash;
-import
org.apache.iotdb.confignode.it.regionmigration.IoTDBRegionMigrateReliabilityITFramework;
+import
org.apache.iotdb.commons.utils.KillPoint.IoTConsensusInactivatePeerKillPoints;
+import
org.apache.iotdb.confignode.it.regionmigration.IoTDBRegionMigrateDataNodeCrashITFramework;
import org.apache.iotdb.it.framework.IoTDBTestRunner;
-import org.apache.iotdb.itbase.category.ClusterIT;
import org.junit.Test;
-import org.junit.experimental.categories.Category;
import org.junit.runner.RunWith;
@RunWith(IoTDBTestRunner.class)
-@Category({ClusterIT.class})
-public class IoTDBRegionMigrateNormalIT extends
IoTDBRegionMigrateReliabilityITFramework {
+public class IoTDBRegionMigrateOriginalCrashWhenRemoveRemotePeerIT
+ extends IoTDBRegionMigrateDataNodeCrashITFramework {
@Test
- public void normal1C2DTest() throws Exception {
- successTest(1, 1, 1, 2, noKillPoints(), noKillPoints());
+ public void crashBeforeInactivate() throws Exception {
+ success(IoTConsensusInactivatePeerKillPoints.BEFORE_INACTIVATE);
}
@Test
- public void normal3C3DTest() throws Exception {
- successTest(2, 3, 3, 3, noKillPoints(), noKillPoints());
+ public void crashAfterInactivate() throws Exception {
+ success(IoTConsensusInactivatePeerKillPoints.AFTER_INACTIVATE);
}
}
diff --git
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TElasticFramedTransport.java
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TElasticFramedTransport.java
index 1075e1de865..b0c55c21bd0 100644
---
a/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TElasticFramedTransport.java
+++
b/iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TElasticFramedTransport.java
@@ -178,8 +178,7 @@ public class TElasticFramedTransport extends TTransport {
@Override
public int getBytesRemainingInBuffer() {
// return -1 can make the caller protocol to copy binary data from the
underlying transport.
- if (copyBinary) return -1;
- return readBuffer.getBytesRemainingInBuffer();
+ return copyBinary ? -1 : readBuffer.getBytesRemainingInBuffer();
}
@Override
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/AsyncDataNodeClientPool.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/AsyncDataNodeClientPool.java
index 35cd65fceeb..e52367276c8 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/AsyncDataNodeClientPool.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/AsyncDataNodeClientPool.java
@@ -74,6 +74,7 @@ import
org.apache.iotdb.mpp.rpc.thrift.TPushSingleTopicMetaReq;
import org.apache.iotdb.mpp.rpc.thrift.TPushTopicMetaReq;
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;
@@ -484,6 +485,12 @@ public class AsyncDataNodeClientPool {
(AsyncTSStatusRPCHandler)
clientHandler.createAsyncRPCHandler(requestId,
targetDataNode));
break;
+ case RESET_PEER_LIST:
+ client.resetPeerList(
+ (TResetPeerListReq) clientHandler.getRequest(requestId),
+ (AsyncTSStatusRPCHandler)
+ clientHandler.createAsyncRPCHandler(requestId,
targetDataNode));
+ break;
default:
LOGGER.error(
"Unexpected DataNode Request Type: {} when
sendAsyncRequestToDataNode",
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/AsyncClientHandler.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/AsyncClientHandler.java
index 4f05f892afc..a8c445665d0 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/AsyncClientHandler.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/AsyncClientHandler.java
@@ -266,6 +266,7 @@ public class AsyncClientHandler<Q, R> {
case UPDATE_TEMPLATE:
case CHANGE_REGION_LEADER:
case KILL_QUERY_INSTANCE:
+ case RESET_PEER_LIST:
default:
return new AsyncTSStatusRPCHandler(
requestType,
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 4e4ed126f68..f5898d4bc17 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
@@ -654,7 +654,14 @@ public class ProcedureManager {
// select coordinator for adding peer
RegionMaintainHandler handler = new RegionMaintainHandler(configManager);
final TDataNodeLocation coordinatorForAddPeer =
- handler.filterDataNodeWithOtherRegionReplica(regionGroupId,
destDataNode).orElse(null);
+ handler
+ .filterDataNodeWithOtherRegionReplica(
+ regionGroupId,
+ destDataNode,
+ NodeStatus.Running,
+ NodeStatus.Removing,
+ NodeStatus.ReadOnly)
+ .orElse(null);
// Select coordinator for removing peer
// For now, destDataNode temporarily acts as the coordinatorForRemovePeer
final TDataNodeLocation coordinatorForRemovePeer = destDataNode;
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java
index 1008fe62dfd..f0a6378aefb 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java
@@ -51,11 +51,6 @@ public abstract class Procedure<Env> implements
Comparable<Procedure<Env>> {
private static final Logger LOG = LoggerFactory.getLogger(Procedure.class);
public static final long NO_PROC_ID = -1;
public static final long NO_TIMEOUT = -1;
- /**
- * The isDeserialized of a newly created procedure is false. When a leader
switch or ConfigNode
- * restart occurs during the execution of the procedure, isDeserialized
becomes true.
- */
- private boolean isDeserialized = false;
private long parentProcId = NO_PROC_ID;
private long rootProcId = NO_PROC_ID;
@@ -192,8 +187,6 @@ public abstract class Procedure<Env> implements
Comparable<Procedure<Env>> {
public void deserialize(ByteBuffer byteBuffer) {
// procid
this.setProcId(byteBuffer.getLong());
- // isDeserialized
- this.setDeserialized(true);
// state
this.setState(ProcedureState.values()[byteBuffer.getInt()]);
// submit time
@@ -545,10 +538,6 @@ public abstract class Procedure<Env> implements
Comparable<Procedure<Env>> {
return procId;
}
- public boolean isDeserialized() {
- return isDeserialized;
- }
-
public boolean hasParent() {
return parentProcId != NO_PROC_ID;
}
@@ -574,10 +563,6 @@ public abstract class Procedure<Env> implements
Comparable<Procedure<Env>> {
this.procId = procId;
}
- private void setDeserialized(boolean isDeserialized) {
- this.isDeserialized = isDeserialized;
- }
-
public void setProcRunnable() {
this.submittedTime = System.currentTimeMillis();
setState(ProcedureState.RUNNABLE);
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java
index b9ccd7e3418..64282e0e3ff 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java
@@ -34,6 +34,8 @@ import org.apache.iotdb.commons.cluster.NodeStatus;
import org.apache.iotdb.commons.service.metric.MetricService;
import org.apache.iotdb.commons.utils.NodeUrlUtils;
import org.apache.iotdb.confignode.client.DataNodeRequestType;
+import org.apache.iotdb.confignode.client.async.AsyncDataNodeClientPool;
+import org.apache.iotdb.confignode.client.async.handlers.AsyncClientHandler;
import org.apache.iotdb.confignode.client.sync.SyncDataNodeClientPool;
import org.apache.iotdb.confignode.conf.ConfigNodeConfig;
import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
@@ -60,6 +62,7 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
@@ -249,7 +252,7 @@ public class RegionMaintainHandler {
* @param regionId region id
* @return TSStatus
*/
- public TSStatus addRegionPeer(
+ public TSStatus submitAddRegionPeerTask(
long procedureId,
TDataNodeLocation destDataNode,
TConsensusGroupId regionId,
@@ -266,11 +269,12 @@ public class RegionMaintainHandler {
maintainPeerReq,
DataNodeRequestType.ADD_REGION_PEER);
LOGGER.info(
- "{}, Send action addRegionPeer finished, regionId: {}, rpcDataNode:
{}, destDataNode: {}",
+ "{}, Send action addRegionPeer finished, regionId: {}, rpcDataNode:
{}, destDataNode: {}, status: {}",
REGION_MIGRATE_PROCESS,
regionId,
getIdWithRpcEndpoint(coordinator),
- getIdWithRpcEndpoint(destDataNode));
+ getIdWithRpcEndpoint(destDataNode),
+ status);
return status;
}
@@ -284,7 +288,7 @@ public class RegionMaintainHandler {
* @param regionId region id
* @return TSStatus
*/
- public TSStatus removeRegionPeer(
+ public TSStatus submitRemoveRegionPeerTask(
long procedureId,
TDataNodeLocation originalDataNode,
TConsensusGroupId regionId,
@@ -318,7 +322,7 @@ public class RegionMaintainHandler {
* @param regionId region id
* @return TSStatus
*/
- public TSStatus deleteOldRegionPeer(
+ public TSStatus submitDeleteOldRegionPeerTask(
long procedureId, TDataNodeLocation originalDataNode, TConsensusGroupId
regionId) {
TSStatus status;
@@ -348,27 +352,43 @@ public class RegionMaintainHandler {
}
public TSStatus resetPeerList(
- TConsensusGroupId regionId, List<TDataNodeLocation>
correctDataNodeLocations) {
- Optional<TDataNodeLocation> optional =
filterDataNodeWithOtherRegionReplica(regionId, null);
- TDataNodeLocation selectDataNode = optional.get();
+ TConsensusGroupId regionId,
+ List<TDataNodeLocation> correctDataNodeLocations,
+ TDataNodeLocation target) {
TSStatus status =
SyncDataNodeClientPool.getInstance()
.sendSyncRequestToDataNodeWithRetry(
- selectDataNode.getInternalEndPoint(),
+ target.getInternalEndPoint(),
new TResetPeerListReq(regionId, correctDataNodeLocations),
DataNodeRequestType.RESET_PEER_LIST);
return status;
}
+ public Map<Integer, TSStatus> resetPeerList(
+ TConsensusGroupId regionId,
+ List<TDataNodeLocation> correctDataNodeLocations,
+ Map<Integer, TDataNodeLocation> dataNodeLocationMap) {
+ AsyncClientHandler<TResetPeerListReq, TSStatus> clientHandler =
+ new AsyncClientHandler<>(
+ DataNodeRequestType.RESET_PEER_LIST,
+ new TResetPeerListReq(regionId, correctDataNodeLocations),
+ dataNodeLocationMap);
+
AsyncDataNodeClientPool.getInstance().sendAsyncRequestToDataNodeWithRetry(clientHandler);
+ return clientHandler.getResponseMap();
+ }
+
// 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) {
+ // In some cases the DataNode is still working, but its status is unknown.
+ // In order to make task continue under this circumstance, some
unconditional retries are
+ // performed here.
+ int unconditionallyRetry = 0;
+ while (unconditionallyRetry < 6
+ ||
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;
}
@@ -381,6 +401,7 @@ public class RegionMaintainHandler {
Thread.currentThread().interrupt();
}
}
+ unconditionallyRetry++;
}
LOGGER.warn(
"{} task {} cannot contact to DataNode {}",
@@ -395,10 +416,6 @@ public class RegionMaintainHandler {
}
public void addRegionLocation(TConsensusGroupId regionId, TDataNodeLocation
newLocation) {
- LOGGER.info(
- "AddRegionLocation started, add region {} to {}",
- regionId,
- getIdWithRpcEndpoint(newLocation));
AddRegionLocationPlan req = new AddRegionLocationPlan(regionId,
newLocation);
TSStatus status =
configManager.getPartitionManager().addRegionLocation(req);
LOGGER.info(
@@ -406,31 +423,28 @@ public class RegionMaintainHandler {
regionId,
getIdWithRpcEndpoint(newLocation),
status);
-
- // Remove the RegionGroupCache of the regionId
- configManager.getLoadManager().removeRegionGroupCache(regionId);
-
- // Broadcast the latest RegionRouteMap when Region migration finished
- configManager.getLoadManager().broadcastLatestRegionRouteMap();
+ updateRegionLocation(regionId);
}
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 {}",
+ "RemoveRegionLocation remove region {} from DataNode {}, result is {}",
regionId,
getIdWithRpcEndpoint(deprecatedLocation),
status);
+ updateRegionLocation(regionId);
+ }
+ private void updateRegionLocation(TConsensusGroupId regionId) {
// Remove the RegionGroupCache of the regionId
configManager.getLoadManager().removeRegionGroupCache(regionId);
-
+ // force balance region leader to skip waiting for leader election
+ configManager.getLoadManager().forceBalanceRegionLeader();
+ // Wait for leader election
+
configManager.getLoadManager().waitForLeaderElection(Collections.singletonList(regionId));
// Broadcast the latest RegionRouteMap when Region migration finished
configManager.getLoadManager().broadcastLatestRegionRouteMap();
}
@@ -699,6 +713,12 @@ public class RegionMaintainHandler {
*/
public Optional<TDataNodeLocation> filterDataNodeWithOtherRegionReplica(
TConsensusGroupId regionId, TDataNodeLocation filterLocation) {
+ return filterDataNodeWithOtherRegionReplica(
+ regionId, filterLocation, NodeStatus.Running, NodeStatus.ReadOnly);
+ }
+
+ public Optional<TDataNodeLocation> filterDataNodeWithOtherRegionReplica(
+ TConsensusGroupId regionId, TDataNodeLocation filterLocation,
NodeStatus... allowingStatus) {
List<TDataNodeLocation> regionLocations = findRegionLocations(regionId);
if (regionLocations.isEmpty()) {
LOGGER.warn("Cannot find DataNodes contain the given region: {}",
regionId);
@@ -708,15 +728,10 @@ public class RegionMaintainHandler {
// Choosing the RUNNING DataNodes to execute firstly
// If all DataNodes are not RUNNING, then choose the REMOVING DataNodes
secondly
List<TDataNodeLocation> aliveDataNodes =
-
configManager.getNodeManager().filterDataNodeThroughStatus(NodeStatus.Running).stream()
+
configManager.getNodeManager().filterDataNodeThroughStatus(allowingStatus).stream()
.map(TDataNodeConfiguration::getLocation)
.collect(Collectors.toList());
- aliveDataNodes.addAll(
-
configManager.getNodeManager().filterDataNodeThroughStatus(NodeStatus.Removing).stream()
- .map(TDataNodeConfiguration::getLocation)
- .collect(Collectors.toList()));
-
// TODO return the node which has lowest load.
for (TDataNodeLocation aliveDataNode : aliveDataNodes) {
if (regionLocations.contains(aliveDataNode) &&
!aliveDataNode.equals(filterLocation)) {
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 b29d5034915..87743003ff0 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
@@ -67,6 +67,8 @@ public abstract class StateMachineProcedure<Env, TState>
extends Procedure<Env>
/** Mark whether this procedure is called by a pipe forwarded request. */
protected boolean isGeneratedByPipe;
+ private boolean stateDeserialized = false;
+
protected StateMachineProcedure() {
this(false);
}
@@ -191,6 +193,7 @@ public abstract class StateMachineProcedure<Env, TState>
extends Procedure<Env>
LOG.trace("{}", this);
stateFlow = executeFromState(env, state);
+ setStateDeserialized(false);
if (!hasMoreState()) {
setNextState(EOF_STATE);
}
@@ -334,5 +337,20 @@ public abstract class StateMachineProcedure<Env, TState>
extends Procedure<Env>
} else {
states = null;
}
+ this.setStateDeserialized(true);
+ }
+
+ /**
+ * The isStateDeserialized indicates whether the current stage of this
procedure was generated by
+ * deserialization. If true, this means the procedure has undergone a leader
switch or a restart
+ * recovery at this stage. After the procedure is recovered, you may not
want to re-execute all
+ * the code in this stage, which is the purpose of this variable.
+ */
+ public boolean isStateDeserialized() {
+ return stateDeserialized;
+ }
+
+ private void setStateDeserialized(boolean isDeserialized) {
+ this.stateDeserialized = isDeserialized;
}
}
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
index 853262625b7..9cefb239489 100644
---
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
@@ -40,6 +40,11 @@ import org.slf4j.LoggerFactory;
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
import static org.apache.iotdb.commons.utils.KillPoint.KillPoint.setKillPoint;
import static
org.apache.iotdb.confignode.procedure.state.AddRegionPeerState.UPDATE_REGION_LOCATION_CACHE;
@@ -79,39 +84,35 @@ public class AddRegionPeerProcedure
outerSwitch:
switch (state) {
case CREATE_NEW_REGION_PEER:
- handler.createNewRegionPeer(consensusGroupId, destDataNode);
+ TSStatus status = handler.createNewRegionPeer(consensusGroupId,
destDataNode);
setKillPoint(state);
+ if (status.getCode() != SUCCESS_STATUS.getStatusCode()) {
+ rollback(env, handler);
+ }
setNextState(AddRegionPeerState.DO_ADD_REGION_PEER);
break;
case DO_ADD_REGION_PEER:
- TSStatus tsStatus =
- handler.addRegionPeer(this.getProcId(), destDataNode,
consensusGroupId, coordinator);
- setKillPoint(state);
- 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");
+ // We don't want to re-submit AddRegionPeerTask when leader change
or ConfigNode reboot
+ if (!this.isStateDeserialized()) {
+ TSStatus tsStatus =
+ handler.submitAddRegionPeerTask(
+ this.getProcId(), destDataNode, consensusGroupId,
coordinator);
+ setKillPoint(state);
+ if (tsStatus.getCode() != SUCCESS_STATUS.getStatusCode()) {
+ throw new ProcedureException("ADD_REGION_PEER executed failed in
DataNode");
+ }
}
+ TRegionMigrateResult result =
handler.waitTaskFinish(this.getProcId(), coordinator);
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 is {}, procedure failed. Will try to reset peer
list automatically...",
+ state,
result.getTaskStatus());
- // List<TDataNodeLocation> correctDataNodeLocations
=
- //
- //
env.getConfigManager().getPartitionManager().getAllReplicaSets().stream()
- // .filter(
- // tRegionReplicaSet ->
- //
- // tRegionReplicaSet.getRegionId().equals(consensusGroupId))
- // .findAny()
- // .get()
- // .getDataNodeLocations();
- // handler.resetPeerList(consensusGroupId,
correctDataNodeLocations);
+ rollback(env, handler);
return Flow.NO_MORE_STATE;
case PROCESSING:
// should never happen
@@ -145,6 +146,54 @@ public class AddRegionPeerProcedure
return Flow.HAS_MORE_STATE;
}
+ private void rollback(ConfigNodeProcedureEnv env, RegionMaintainHandler
handler) {
+ List<TDataNodeLocation> correctDataNodeLocations =
+
env.getConfigManager().getPartitionManager().getAllReplicaSets().stream()
+ .filter(tRegionReplicaSet ->
tRegionReplicaSet.getRegionId().equals(consensusGroupId))
+ .findAny()
+ .get()
+ .getDataNodeLocations();
+
+ String correctStr =
+ correctDataNodeLocations.stream()
+ .map(TDataNodeLocation::getDataNodeId)
+ .collect(Collectors.toList())
+ .toString();
+ List<TDataNodeLocation> relatedDataNodeLocations = new
ArrayList<>(correctDataNodeLocations);
+ relatedDataNodeLocations.add(destDataNode);
+ Map<Integer, TDataNodeLocation> relatedDataNodeLocationMap = new
HashMap<>();
+ relatedDataNodeLocations.forEach(
+ location -> relatedDataNodeLocationMap.put(location.dataNodeId,
location));
+ LOGGER.info(
+ "Will reset peer list of consensus group {} on DataNode {}",
+ consensusGroupId,
+ relatedDataNodeLocations.stream()
+ .map(TDataNodeLocation::getDataNodeId)
+ .collect(Collectors.toList()));
+
+ Map<Integer, TSStatus> resultMap =
+ handler.resetPeerList(
+ consensusGroupId, correctDataNodeLocations,
relatedDataNodeLocationMap);
+
+ resultMap.forEach(
+ (dataNodeId, resetResult) -> {
+ if (resetResult.getCode() == SUCCESS_STATUS.getStatusCode()) {
+ LOGGER.info(
+ "reset peer list: peer list of consensus group {} on DataNode
{} has been successfully to {}",
+ consensusGroupId,
+ dataNodeId,
+ correctStr);
+ } else {
+ // TODO: more precise
+ LOGGER.warn(
+ "reset peer list: peer list of consensus group {} on DataNode
{} failed to reset to {}, you may manually reset it",
+ consensusGroupId,
+ dataNodeId,
+ correctStr);
+ }
+ });
+ }
+
@Override
protected void rollbackState(
ConfigNodeProcedureEnv configNodeProcedureEnv, AddRegionPeerState
addRegionPeerState)
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
index 7055a274874..1e183a631e6 100644
---
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
@@ -80,30 +80,41 @@ public class RemoveRegionPeerProcedure
switch (state) {
case REMOVE_REGION_PEER:
tsStatus =
- handler.removeRegionPeer(
+ handler.submitRemoveRegionPeerTask(
this.getProcId(), targetDataNode, consensusGroupId,
coordinator);
setKillPoint(state);
if (tsStatus.getCode() != SUCCESS_STATUS.getStatusCode()) {
- throw new ProcedureException("REMOVE_REGION_PEER executed failed
in DataNode");
+ LOGGER.warn(
+ "{} task submitted failed, procedure will continue. You should
manually clear peer list.",
+ state);
+ setNextState(DELETE_OLD_REGION_PEER);
+ break;
}
TRegionMigrateResult removeRegionPeerResult =
handler.waitTaskFinish(this.getProcId(), coordinator);
if (removeRegionPeerResult.getTaskStatus() !=
TRegionMaintainTaskStatus.SUCCESS) {
- throw new ProcedureException("REMOVE_REGION_PEER executed failed
in DataNode");
+ LOGGER.warn(
+ "{} executed failed, procedure will continue. You should
manually clear peer list.",
+ state);
}
setNextState(DELETE_OLD_REGION_PEER);
break;
case DELETE_OLD_REGION_PEER:
tsStatus =
- handler.deleteOldRegionPeer(this.getProcId(), targetDataNode,
consensusGroupId);
+ handler.submitDeleteOldRegionPeerTask(
+ this.getProcId(), targetDataNode, consensusGroupId);
setKillPoint(state);
if (tsStatus.getCode() != SUCCESS_STATUS.getStatusCode()) {
- throw new ProcedureException("DELETE_OLD_REGION_PEER executed
failed in DataNode");
+ LOGGER.warn(
+ "DELETE_OLD_REGION_PEER task submitted failed, procedure will
continue. You should manually delete region file.");
+ setNextState(REMOVE_REGION_LOCATION_CACHE);
+ break;
}
TRegionMigrateResult deleteOldRegionPeerResult =
handler.waitTaskFinish(this.getProcId(), targetDataNode);
if (deleteOldRegionPeerResult.getTaskStatus() !=
TRegionMaintainTaskStatus.SUCCESS) {
- throw new ProcedureException("DELETE_OLD_REGION_PEER executed
failed in DataNode");
+ LOGGER.warn(
+ "DELETE_OLD_REGION_PEER executed failed, procedure will
continue. You should manually delete region file.");
}
setNextState(REMOVE_REGION_LOCATION_CACHE);
break;
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/testonly/CreateManyDatabasesProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/testonly/CreateManyDatabasesProcedure.java
index 7d1c0119c23..4693923bb30 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/testonly/CreateManyDatabasesProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/testonly/CreateManyDatabasesProcedure.java
@@ -32,6 +32,7 @@ import org.slf4j.LoggerFactory;
import java.io.DataOutputStream;
import java.io.IOException;
+import java.nio.ByteBuffer;
/**
* This procedure will create numerous databases (perhaps 100), during which
the confignode leader
@@ -47,12 +48,13 @@ public class CreateManyDatabasesProcedure
public static final String DATABASE_NAME_PREFIX = "root.test_";
public static final long SLEEP_FOREVER = Long.MAX_VALUE;
private boolean createFailedOnce = false;
+ private boolean isDeserialized = false;
@Override
protected Flow executeFromState(ConfigNodeProcedureEnv
configNodeProcedureEnv, Integer state)
throws InterruptedException {
if (state < MAX_STATE) {
- if (state == MAX_STATE - 1 && !isDeserialized()) {
+ if (state == MAX_STATE - 1 && !isDeserialized) {
Thread.sleep(SLEEP_FOREVER);
}
try {
@@ -106,4 +108,10 @@ public class CreateManyDatabasesProcedure
stream.writeShort(ProcedureType.CREATE_MANY_DATABASES_PROCEDURE.getTypeCode());
super.serialize(stream);
}
+
+ @Override
+ public void deserialize(ByteBuffer byteBuffer) {
+ super.deserialize(byteBuffer);
+ isDeserialized = true;
+ }
}
diff --git a/iotdb-core/consensus/pom.xml b/iotdb-core/consensus/pom.xml
index b3a160c0dfd..e149e62f6f6 100644
--- a/iotdb-core/consensus/pom.xml
+++ b/iotdb-core/consensus/pom.xml
@@ -128,6 +128,10 @@
<artifactId>junit</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ </dependency>
</dependencies>
<build>
<plugins>
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 32dae2c0ff5..27d1340965d 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
@@ -207,4 +207,23 @@ public interface IConsensus {
* @return consensusGroupId list
*/
List<ConsensusGroupId> getAllConsensusGroupIds();
+
+ /**
+ * Return all consensus group ids from disk.
+ *
+ * <p>We need to parse all the RegionGroupIds from the disk directory before
starting the
+ * consensus layer, and {@link #getAllConsensusGroupIds()} returns an empty
list, so we need to
+ * add a new interface.
+ *
+ * @return consensusGroupId list
+ */
+ List<ConsensusGroupId> getAllConsensusGroupIdsWithoutStarting();
+
+ /**
+ * Return the region directory of the corresponding consensus group.
+ *
+ * @param groupId the consensus group
+ * @return region directory
+ */
+ String getRegionDirFromConsensusGroupId(ConsensusGroupId groupId);
}
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/exception/ConsensusException.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/exception/ConsensusException.java
index 385ecaa89fc..a43a5b995a8 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/exception/ConsensusException.java
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/exception/ConsensusException.java
@@ -28,4 +28,8 @@ public class ConsensusException extends Exception {
public ConsensusException(String message, Throwable cause) {
super(message, cause);
}
+
+ public ConsensusException(Exception e) {
+ super(e);
+ }
}
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 e9bb1fb6eb5..4da51bd0ad1 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
@@ -30,7 +30,8 @@ import org.apache.iotdb.commons.exception.StartupException;
import org.apache.iotdb.commons.service.RegisterManager;
import org.apache.iotdb.commons.utils.FileUtils;
import org.apache.iotdb.commons.utils.KillPoint.DataNodeKillPoints;
-import
org.apache.iotdb.commons.utils.KillPoint.IoTConsensusRemovePeerKillPoints;
+import
org.apache.iotdb.commons.utils.KillPoint.IoTConsensusDeleteLocalPeerKillPoints;
+import
org.apache.iotdb.commons.utils.KillPoint.IoTConsensusRemovePeerCoordinatorKillPoints;
import org.apache.iotdb.commons.utils.KillPoint.KillPoint;
import org.apache.iotdb.commons.utils.StatusUtils;
import org.apache.iotdb.consensus.IConsensus;
@@ -264,6 +265,7 @@ public class IoTConsensus implements IConsensus {
@Override
public void deleteLocalPeer(ConsensusGroupId groupId) throws
ConsensusException {
+
KillPoint.setKillPoint(IoTConsensusDeleteLocalPeerKillPoints.BEFORE_DELETE);
AtomicBoolean exist = new AtomicBoolean(false);
stateMachineMap.computeIfPresent(
groupId,
@@ -273,6 +275,7 @@ public class IoTConsensus implements IConsensus {
FileUtils.deleteFileOrDirectory(new File(buildPeerDir(storageDir,
groupId)));
return null;
});
+ KillPoint.setKillPoint(IoTConsensusDeleteLocalPeerKillPoints.AFTER_DELETE);
if (!exist.get()) {
throw new ConsensusGroupNotExistException(groupId);
}
@@ -289,7 +292,7 @@ public class IoTConsensus implements IConsensus {
try {
// step 1: inactive new Peer to prepare for following steps
logger.info("[IoTConsensus] inactivate new peer: {}", peer);
- impl.inactivePeer(peer);
+ impl.inactivePeer(peer, false);
// step 2: take snapshot
logger.info("[IoTConsensus] start to take snapshot...");
@@ -329,7 +332,7 @@ public class IoTConsensus implements IConsensus {
logger.error(
"[IoTConsensus] failed to cleanup side effects after failed to add
remote peer", mpe);
}
- throw new ConsensusException(e.getMessage());
+ throw new ConsensusException(e);
}
}
@@ -351,7 +354,7 @@ public class IoTConsensus implements IConsensus {
throw new PeerNotInConsensusGroupException(groupId, peer.toString());
}
- KillPoint.setKillPoint(IoTConsensusRemovePeerKillPoints.INIT);
+ KillPoint.setKillPoint(IoTConsensusRemovePeerCoordinatorKillPoints.INIT);
try {
// let other peers remove the sync channel with target peer
@@ -360,18 +363,18 @@ public class IoTConsensus implements IConsensus {
throw new ConsensusException(e.getMessage());
}
KillPoint.setKillPoint(
-
IoTConsensusRemovePeerKillPoints.AFTER_NOTIFY_PEERS_TO_REMOVE_SYNC_LOG_CHANNEL);
+
IoTConsensusRemovePeerCoordinatorKillPoints.AFTER_NOTIFY_PEERS_TO_REMOVE_SYNC_LOG_CHANNEL);
try {
// let target peer reject new write
- impl.inactivePeer(peer);
-
KillPoint.setKillPoint(IoTConsensusRemovePeerKillPoints.AFTER_INACTIVE_PEER);
+ impl.inactivePeer(peer, true);
+
KillPoint.setKillPoint(IoTConsensusRemovePeerCoordinatorKillPoints.AFTER_INACTIVE_PEER);
// wait its SyncLog to complete
impl.waitTargetPeerUntilSyncLogCompleted(peer);
} catch (ConsensusGroupModifyPeerException e) {
throw new ConsensusException(e.getMessage());
}
- KillPoint.setKillPoint(IoTConsensusRemovePeerKillPoints.FINISH);
+ KillPoint.setKillPoint(IoTConsensusRemovePeerCoordinatorKillPoints.FINISH);
}
@Override
@@ -414,28 +417,74 @@ public class IoTConsensus implements IConsensus {
return new ArrayList<>(stateMachineMap.keySet());
}
- public void resetPeerList(ConsensusGroupId groupId, List<Peer> peers) throws
ConsensusException {
+ @Override
+ public List<ConsensusGroupId> getAllConsensusGroupIdsWithoutStarting() {
+ return getConsensusGroupIdsFromDir(storageDir, logger);
+ }
+
+ public static List<ConsensusGroupId> getConsensusGroupIdsFromDir(File
storageDir, Logger logger) {
+ List<ConsensusGroupId> consensusGroupIds = new ArrayList<>();
+ try (DirectoryStream<Path> stream =
Files.newDirectoryStream(storageDir.toPath())) {
+ for (Path path : stream) {
+ try {
+ String[] items = path.getFileName().toString().split("_");
+ ConsensusGroupId consensusGroupId =
+ ConsensusGroupId.Factory.create(
+ Integer.parseInt(items[0]), Integer.parseInt(items[1]));
+ consensusGroupIds.add(consensusGroupId);
+ } catch (Exception e) {
+ logger.info(
+ "The directory {} is not a group directory;" + " ignoring it. ",
+ path.getFileName().toString());
+ }
+ }
+ } catch (IOException e) {
+ logger.error("Failed to get all consensus group ids from disk", e);
+ }
+ return consensusGroupIds;
+ }
+
+ @Override
+ public String getRegionDirFromConsensusGroupId(ConsensusGroupId groupId) {
+ return buildPeerDir(storageDir, groupId);
+ }
+
+ public void resetPeerList(ConsensusGroupId groupId, List<Peer> correctPeers)
+ 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.");
+ Peer localPeer = new Peer(groupId, thisNodeId, thisNode);
+ if (!correctPeers.contains(localPeer)) {
+ logger.warn(
+ "[RESET PEER LIST] Local peer is not in the correct configuration,
delete local peer {}",
+ groupId);
+ deleteLocalPeer(groupId);
+ return;
}
-
+ String previousPeerListStr = impl.getConfiguration().toString();
for (Peer peer : impl.getConfiguration()) {
- if (!peers.contains(peer)) {
+ if (!correctPeers.contains(peer)) {
try {
- removeRemotePeer(groupId, peer);
- } catch (ConsensusException e) {
- logger.error("Failed to remove peer {} from group {}", peer,
groupId, e);
- throw e;
+ impl.removeSyncLogChannel(peer);
+ } catch (ConsensusGroupModifyPeerException e) {
+ logger.error(
+ "[RESET PEER LIST] Failed to remove peer {}'s sync log channel
from group {}",
+ peer,
+ groupId,
+ e);
}
}
}
- impl.resetConfiguration(peers);
+ logger.info(
+ "[RESET PEER LIST] Local peer list has been reset: {} -> {}",
+ previousPeerListStr,
+ impl.getConfiguration());
+ for (Peer peer : correctPeers) {
+ if (!impl.getConfiguration().contains(peer)) {
+ logger.warn("[RESET PEER LIST] \"Correct peer\" {} is not in local
peer list", peer);
+ }
+ }
}
public IoTConsensusServerImpl getImpl(ConsensusGroupId 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 cdb797da4e4..9a7153b158b 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
@@ -27,6 +27,8 @@ import
org.apache.iotdb.commons.consensus.index.ComparableConsensusRequest;
import org.apache.iotdb.commons.consensus.index.impl.IoTProgressIndex;
import org.apache.iotdb.commons.service.metric.MetricService;
import org.apache.iotdb.commons.service.metric.PerformanceOverviewMetrics;
+import org.apache.iotdb.commons.utils.KillPoint.DataNodeKillPoints;
+import org.apache.iotdb.commons.utils.KillPoint.KillPoint;
import org.apache.iotdb.consensus.IStateMachine;
import org.apache.iotdb.consensus.common.DataSet;
import org.apache.iotdb.consensus.common.Peer;
@@ -60,6 +62,7 @@ import
org.apache.iotdb.consensus.iot.thrift.TWaitSyncLogCompleteRes;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
+import com.google.common.collect.ImmutableList;
import org.apache.commons.io.FileUtils;
import org.apache.thrift.TException;
import org.slf4j.Logger;
@@ -70,10 +73,10 @@ import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
-import java.nio.file.StandardOpenOption;
import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedList;
@@ -87,6 +90,7 @@ import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.regex.Pattern;
+import java.util.stream.Collectors;
import java.util.stream.Stream;
public class IoTConsensusServerImpl {
@@ -134,8 +138,9 @@ public class IoTConsensusServerImpl {
this.configuration = configuration;
if (configuration.isEmpty()) {
recoverConfiguration();
+ } else {
+ persistConfiguration();
}
- persistConfiguration();
this.backgroundTaskService = backgroundTaskService;
this.config = config;
this.consensusGroupId = thisNode.getGroupId().toString();
@@ -317,11 +322,10 @@ public class IoTConsensusServerImpl {
if (!Files.exists(parentDir)) {
Files.createDirectories(parentDir);
}
- Files.write(
- Paths.get(targetFile.getAbsolutePath()),
- fileChunk.array(),
- StandardOpenOption.CREATE,
- StandardOpenOption.APPEND);
+ try (FileOutputStream fos = new
FileOutputStream(targetFile.getAbsolutePath(), true);
+ FileChannel channel = fos.getChannel()) {
+ channel.write(fileChunk.slice());
+ }
} catch (IOException e) {
throw new ConsensusGroupModifyPeerException(
String.format("error when receiving snapshot %s", snapshotId), e);
@@ -379,19 +383,30 @@ public class IoTConsensusServerImpl {
stateMachine.loadSnapshot(new File(storageDir, snapshotId));
}
- public void inactivePeer(Peer peer) throws ConsensusGroupModifyPeerException
{
+ @FunctionalInterface
+ public interface ThrowableFunction<T, R> {
+ R apply(T t) throws Exception;
+ }
+
+ public void inactivePeer(Peer peer, boolean forDeletionPurpose)
+ throws ConsensusGroupModifyPeerException {
try (SyncIoTConsensusServiceClient client =
syncClientManager.borrowClient(peer.getEndpoint())) {
- TInactivatePeerRes res =
- client.inactivatePeer(
- new
TInactivatePeerReq(peer.getGroupId().convertToTConsensusGroupId()));
- if (!isSuccess(res.status)) {
+ try {
+ TInactivatePeerRes res =
+ client.inactivatePeer(
+ new
TInactivatePeerReq(peer.getGroupId().convertToTConsensusGroupId())
+ .setForDeletionPurpose(forDeletionPurpose));
+ if (!isSuccess(res.status)) {
+ throw new ConsensusGroupModifyPeerException(
+ String.format("error when inactivating %s. %s", peer,
res.getStatus()));
+ }
+ } catch (Exception e) {
throw new ConsensusGroupModifyPeerException(
- String.format("error when inactivating %s. %s", peer,
res.getStatus()));
+ String.format("error when inactivating %s", peer), e);
}
- } catch (Exception e) {
- throw new ConsensusGroupModifyPeerException(
- String.format("error when inactivating %s", peer), e);
+ } catch (ClientManagerException e) {
+ throw new ConsensusGroupModifyPeerException(e);
}
}
@@ -478,7 +493,7 @@ public class IoTConsensusServerImpl {
throws ConsensusGroupModifyPeerException {
// The configuration will be modified during iterating because we will add
the targetPeer to
// configuration
- List<Peer> currentMembers = new ArrayList<>(this.configuration);
+ ImmutableList<Peer> currentMembers =
ImmutableList.copyOf(this.configuration);
for (Peer peer : currentMembers) {
if (peer.equals(targetPeer)) {
// if the targetPeer is the same as current peer, skip it because
removing itself is illegal
@@ -562,6 +577,7 @@ public class IoTConsensusServerImpl {
public void buildSyncLogChannel(Peer targetPeer, long initialSyncIndex)
throws ConsensusGroupModifyPeerException {
+ KillPoint.setKillPoint(DataNodeKillPoints.ORIGINAL_ADD_PEER_DONE);
// step 1, build sync channel in LogDispatcher
logger.info(
"[IoTConsensus] build sync log channel to {} with initialSyncIndex {}",
@@ -591,23 +607,12 @@ public class IoTConsensusServerImpl {
}
}
- // TODO: persist first and then delete old configuration file
public void persistConfiguration() {
try {
- try (Stream<Path> stream = Files.walk(Paths.get(storageDir))) {
- stream
- .filter(Files::isRegularFile)
- .filter(filePath ->
filePath.getFileName().toString().contains("configuration"))
- .forEach(
- filePath -> {
- try {
- Files.delete(filePath);
- } catch (IOException e) {
- logger.error("Unexpected error occurs when deleting old
configuration file", e);
- }
- });
- }
+ renameTmpConfigurationFileToRemoveSuffix();
serializeConfigurationAndFsyncToDisk();
+ deleteConfiguration();
+ renameTmpConfigurationFileToRemoveSuffix();
} catch (IOException e) {
// TODO: (xingtanzjr) need to handle the IOException because the
IoTConsensus won't
// work expectedly
@@ -642,6 +647,7 @@ public class IoTConsensusServerImpl {
configuration.add(peer);
}
}
+ persistConfiguration();
}
logger.info("Recover IoTConsensus server Impl, configuration: {}",
configuration);
} catch (IOException e) {
@@ -657,12 +663,12 @@ public class IoTConsensusServerImpl {
for (int i = 0; i < size; i++) {
configuration.add(Peer.deserialize(buffer));
}
- // TODO: delete old file before new file persisted is unsafe
+ persistConfiguration();
Files.delete(oldConfigurationPath);
}
- public static String generateConfigurationDatFileName(int nodeId) {
- return nodeId + "_" + CONFIGURATION_FILE_NAME;
+ public static String generateConfigurationDatFileName(int nodeId, String
suffix) {
+ return nodeId + "_" + suffix;
}
private List<Peer> getConfiguration(Path dirPath, String
configurationFileName)
@@ -681,11 +687,6 @@ public class IoTConsensusServerImpl {
return tmpConfiguration;
}
- public void resetConfiguration(List<Peer> newConfiguration) {
- configuration.clear();
- configuration.addAll(newConfiguration);
- }
-
public IndexedConsensusRequest buildIndexedConsensusRequestForLocalRequest(
IConsensusRequest request) {
if (request instanceof ComparableConsensusRequest) {
@@ -876,7 +877,8 @@ public class IoTConsensusServerImpl {
private void serializeConfigurationAndFsyncToDisk() throws IOException {
for (Peer peer : configuration) {
- String peerConfigurationFileName =
generateConfigurationDatFileName(peer.getNodeId());
+ String peerConfigurationFileName =
+ generateConfigurationDatFileName(peer.getNodeId(),
CONFIGURATION_TMP_FILE_NAME);
FileOutputStream fileOutputStream =
new FileOutputStream(new File(storageDir,
peerConfigurationFileName));
try (DataOutputStream outputStream = new
DataOutputStream(fileOutputStream)) {
@@ -886,12 +888,58 @@ public class IoTConsensusServerImpl {
fileOutputStream.flush();
fileOutputStream.getFD().sync();
} catch (IOException ignore) {
- // ignore
+ // ignore sync exception
}
}
}
}
+ private void renameTmpConfigurationFileToRemoveSuffix() throws IOException {
+ try (Stream<Path> stream = Files.walk(Paths.get(storageDir))) {
+ List<Path> paths =
+ stream
+ .filter(Files::isRegularFile)
+ .filter(
+ filePath ->
+
filePath.getFileName().toString().endsWith(CONFIGURATION_TMP_FILE_NAME))
+ .collect(Collectors.toList());
+ for (Path filePath : paths) {
+ String targetPath =
+ filePath.toString().replace(CONFIGURATION_TMP_FILE_NAME,
CONFIGURATION_FILE_NAME);
+ File targetFile = new File(targetPath);
+ if (targetFile.exists()) {
+ try {
+ Files.delete(targetFile.toPath());
+ } catch (IOException e) {
+ logger.error("Unexpected error occurs when delete file: {}",
targetPath);
+ }
+ }
+ if (!filePath.toFile().renameTo(targetFile)) {
+ logger.error("Unexpected error occurs when rename file: {} -> {}",
filePath, targetPath);
+ }
+ }
+ }
+ }
+
+ private void deleteConfiguration() throws IOException {
+ try (Stream<Path> stream = Files.walk(Paths.get(storageDir))) {
+ stream
+ .filter(Files::isRegularFile)
+ .filter(filePath ->
filePath.getFileName().toString().endsWith(CONFIGURATION_FILE_NAME))
+ .forEach(
+ filePath -> {
+ try {
+ Files.delete(filePath);
+ } catch (IOException e) {
+ logger.error(
+ "Unexpected error occurs when deleting old configuration
file {}",
+ filePath,
+ e);
+ }
+ });
+ }
+ }
+
/**
* This method is used for write of IoTConsensus SyncLog. By this method, we
can keep write order
* in follower the same as the leader. And besides order insurance, we can
make the
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCServiceProcessor.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCServiceProcessor.java
index 6c4b7cc6777..1c5f5354c82 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCServiceProcessor.java
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/service/IoTConsensusRPCServiceProcessor.java
@@ -22,6 +22,7 @@ package org.apache.iotdb.consensus.iot.service;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
import org.apache.iotdb.commons.utils.KillPoint.DataNodeKillPoints;
+import
org.apache.iotdb.commons.utils.KillPoint.IoTConsensusInactivatePeerKillPoints;
import org.apache.iotdb.commons.utils.KillPoint.KillPoint;
import org.apache.iotdb.consensus.common.Peer;
import org.apache.iotdb.consensus.common.request.BatchIndexedConsensusRequest;
@@ -139,6 +140,9 @@ public class IoTConsensusRPCServiceProcessor implements
IoTConsensusIService.Asy
public void inactivatePeer(
TInactivatePeerReq req, AsyncMethodCallback<TInactivatePeerRes>
resultHandler)
throws TException {
+ if (req.isForDeletionPurpose()) {
+
KillPoint.setKillPoint(IoTConsensusInactivatePeerKillPoints.BEFORE_INACTIVATE);
+ }
ConsensusGroupId groupId =
ConsensusGroupId.Factory.createFromTConsensusGroupId(req.getConsensusGroupId());
IoTConsensusServerImpl impl = consensus.getImpl(groupId);
@@ -154,6 +158,9 @@ public class IoTConsensusRPCServiceProcessor implements
IoTConsensusIService.Asy
impl.setActive(false);
resultHandler.onComplete(
new TInactivatePeerRes(new
TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())));
+ if (req.isForDeletionPurpose()) {
+
KillPoint.setKillPoint(IoTConsensusInactivatePeerKillPoints.AFTER_INACTIVATE);
+ }
}
@Override
@@ -201,7 +208,6 @@ public class IoTConsensusRPCServiceProcessor implements
IoTConsensusIService.Asy
responseStatus = new
TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode());
responseStatus.setMessage(e.getMessage());
}
- KillPoint.setKillPoint(DataNodeKillPoints.ORIGINAL_ADD_PEER_DONE);
resultHandler.onComplete(new TBuildSyncLogChannelRes(responseStatus));
}
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 3a6082ae2fe..d06c7c93fe2 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
@@ -90,11 +90,15 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
+import java.nio.file.DirectoryStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -110,6 +114,8 @@ class RatisConsensus implements IConsensus {
/** The unique net communication endpoint */
private final RaftPeer myself;
+ private final File storageDir;
+
private final RaftServer server;
private final RaftProperties properties = new RaftProperties();
@@ -143,9 +149,9 @@ class RatisConsensus implements IConsensus {
myself =
Utils.fromNodeInfoAndPriorityToRaftPeer(
config.getThisNodeId(), config.getThisNodeEndPoint(),
DEFAULT_PRIORITY);
+ this.storageDir = new File(config.getStorageDir());
- RaftServerConfigKeys.setStorageDir(
- properties, Collections.singletonList(new
File(config.getStorageDir())));
+ RaftServerConfigKeys.setStorageDir(properties,
Collections.singletonList(storageDir));
GrpcConfigKeys.Server.setPort(properties,
config.getThisNodeEndPoint().getPort());
Utils.initRatisConfig(properties, config.getRatisConfig());
@@ -688,6 +694,33 @@ class RatisConsensus implements IConsensus {
return ids;
}
+ @Override
+ public List<ConsensusGroupId> getAllConsensusGroupIdsWithoutStarting() {
+ List<ConsensusGroupId> consensusGroupIds = new ArrayList<>();
+ try (DirectoryStream<Path> stream =
Files.newDirectoryStream(storageDir.toPath())) {
+ for (Path path : stream) {
+ try {
+ RaftGroupId raftGroupId =
+
RaftGroupId.valueOf(UUID.fromString(path.getFileName().toString()));
+
consensusGroupIds.add(Utils.fromRaftGroupIdToConsensusGroupId(raftGroupId));
+ } catch (Exception e) {
+ logger.info(
+ "The directory {} is not a group directory;" + " ignoring it. ",
+ path.getFileName().toString());
+ }
+ }
+ } catch (IOException e) {
+ logger.error("Failed to get all consensus group ids from disk", e);
+ }
+ return consensusGroupIds;
+ }
+
+ @Override
+ public String getRegionDirFromConsensusGroupId(ConsensusGroupId
consensusGroupId) {
+ RaftGroupId raftGroupId =
Utils.fromConsensusGroupIdToRaftGroupId(consensusGroupId);
+ return storageDir + File.separator + raftGroupId.getUuid().toString();
+ }
+
@Override
public void triggerSnapshot(ConsensusGroupId groupId, boolean force) throws
ConsensusException {
final RaftGroupId raftGroupId =
Utils.fromConsensusGroupIdToRaftGroupId(groupId);
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 d23086b51a7..3f35944ced7 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
@@ -38,6 +38,7 @@ import
org.apache.iotdb.consensus.exception.ConsensusGroupAlreadyExistException;
import org.apache.iotdb.consensus.exception.ConsensusGroupNotExistException;
import org.apache.iotdb.consensus.exception.IllegalPeerEndpointException;
import org.apache.iotdb.consensus.exception.IllegalPeerNumException;
+import org.apache.iotdb.consensus.iot.IoTConsensus;
import org.apache.iotdb.rpc.TSStatusCode;
import org.slf4j.Logger;
@@ -243,6 +244,16 @@ class SimpleConsensus implements IConsensus {
return new ArrayList<>(stateMachineMap.keySet());
}
+ @Override
+ public List<ConsensusGroupId> getAllConsensusGroupIdsWithoutStarting() {
+ return IoTConsensus.getConsensusGroupIdsFromDir(storageDir, logger);
+ }
+
+ @Override
+ public String getRegionDirFromConsensusGroupId(ConsensusGroupId groupId) {
+ return buildPeerDir(groupId);
+ }
+
@Override
public void resetPeerList(ConsensusGroupId groupId, List<Peer> peers) throws
ConsensusException {
throw new ConsensusException("SimpleConsensus does not support reset peer
list");
diff --git
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/ReplicateTest.java
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/ReplicateTest.java
index 0e7eb9e1559..6a636542df3 100644
---
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/ReplicateTest.java
+++
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/ReplicateTest.java
@@ -307,6 +307,27 @@ public class ReplicateTest {
}
}
+ @Test
+ public void parsingAndConstructIDTest() throws Exception {
+ logger.info("Start ParsingAndConstructIDTest");
+ servers.get(0).createLocalPeer(group.getGroupId(), group.getPeers());
+ for (int i = 0; i < CHECK_POINT_GAP; i++) {
+ servers.get(0).write(gid, new TestEntry(i, peers.get(0)));
+ }
+ List<ConsensusGroupId> ids =
servers.get(0).getAllConsensusGroupIdsWithoutStarting();
+
+ Assert.assertEquals(1, ids.size());
+ Assert.assertEquals(gid, ids.get(0));
+
+ String regionDir = servers.get(0).getRegionDirFromConsensusGroupId(gid);
+ try {
+ File regionDirFile = new File(regionDir);
+ Assert.assertTrue(regionDirFile.exists());
+ } catch (Exception e) {
+ Assert.fail();
+ }
+ }
+
private void findPortAvailable(int i) {
long start = System.currentTimeMillis();
while (System.currentTimeMillis() - start < timeout) {
diff --git
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/StabilityTest.java
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/StabilityTest.java
index f94c489a2f7..7e176242dc7 100644
---
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/StabilityTest.java
+++
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/StabilityTest.java
@@ -102,7 +102,7 @@ public class StabilityTest {
try {
consensusImpl.createLocalPeer(
dataRegionId,
- Collections.singletonList(new Peer(dataRegionId, 1, new
TEndPoint("0.0.0.0", 6667))));
+ Collections.singletonList(new Peer(dataRegionId, 1, new
TEndPoint("0.0.0.0", basePort))));
} catch (ConsensusException e) {
Assert.fail();
}
@@ -110,7 +110,7 @@ public class StabilityTest {
try {
consensusImpl.createLocalPeer(
dataRegionId,
- Collections.singletonList(new Peer(dataRegionId, 1, new
TEndPoint("0.0.0.0", 6667))));
+ Collections.singletonList(new Peer(dataRegionId, 1, new
TEndPoint("0.0.0.0", basePort))));
Assert.fail();
} catch (ConsensusException e) {
assertTrue(e instanceof ConsensusGroupAlreadyExistException);
@@ -126,7 +126,7 @@ public class StabilityTest {
try {
consensusImpl.createLocalPeer(
dataRegionId,
- Collections.singletonList(new Peer(dataRegionId, 1, new
TEndPoint("0.0.0.1", 6667))));
+ Collections.singletonList(new Peer(dataRegionId, 1, new
TEndPoint("0.0.0.1", basePort))));
Assert.fail();
} catch (ConsensusException e) {
assertTrue(e instanceof IllegalPeerEndpointException);
@@ -150,7 +150,7 @@ public class StabilityTest {
try {
consensusImpl.createLocalPeer(
dataRegionId,
- Collections.singletonList(new Peer(dataRegionId, 1, new
TEndPoint("0.0.0.0", 6667))));
+ Collections.singletonList(new Peer(dataRegionId, 1, new
TEndPoint("0.0.0.0", basePort))));
consensusImpl.deleteLocalPeer(dataRegionId);
} catch (ConsensusException e) {
Assert.fail();
@@ -177,8 +177,8 @@ public class StabilityTest {
public void transferLeader() {
try {
consensusImpl.transferLeader(
- dataRegionId, new Peer(dataRegionId, 1, new TEndPoint("0.0.0.0",
6667)));
- Assert.fail("Can't transfer leader in SimpleConsensus.");
+ dataRegionId, new Peer(dataRegionId, 1, new TEndPoint("0.0.0.0",
basePort)));
+ Assert.fail("Can't transfer leader in IoTConsensus.");
} catch (ConsensusException e) {
// not handle
}
diff --git
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RatisConsensusTest.java
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RatisConsensusTest.java
index 7dcf98caceb..d164cac4316 100644
---
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RatisConsensusTest.java
+++
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RatisConsensusTest.java
@@ -38,6 +38,7 @@ import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.File;
import java.io.IOException;
import java.util.Collections;
import java.util.List;
@@ -274,6 +275,24 @@ public class RatisConsensusTest {
doConsensus(1, 10, 20);
}
+ @Test
+ public void parsingAndConstructIDs() throws Exception {
+ servers.get(0).createLocalPeer(gid, peers.subList(0, 1));
+ doConsensus(0, 10, 10);
+
+ List<ConsensusGroupId> ids =
servers.get(0).getAllConsensusGroupIdsWithoutStarting();
+ Assert.assertEquals(1, ids.size());
+ Assert.assertEquals(gid, ids.get(0));
+
+ String regionDir = servers.get(0).getRegionDirFromConsensusGroupId(gid);
+ try {
+ File regionDirFile = new File(regionDir);
+ Assert.assertTrue(regionDirFile.exists());
+ } catch (Exception e) {
+ Assert.fail();
+ }
+ }
+
private void doConsensus(int serverIndex, int count, int target) throws
Exception {
miniCluster.writeManyParallel(writeExecutor, serverIndex, count);
Assert.assertEquals(target, miniCluster.mustRead(serverIndex));
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java
index 38a973d554d..ae288312599 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java
@@ -287,15 +287,16 @@ public class DataRegionStateMachine extends
BaseStateMachine {
@Override
public File getSnapshotRoot() {
- String snapshotDir =
-
IoTDBDescriptor.getInstance().getConfig().getRatisDataRegionSnapshotDir()
- + File.separator
- + region.getDatabaseName()
- + "-"
- + region.getDataRegionId();
+ String snapshotDir = "";
try {
+ snapshotDir =
+
IoTDBDescriptor.getInstance().getConfig().getRatisDataRegionSnapshotDir()
+ + File.separator
+ + region.getDatabaseName()
+ + "-"
+ + region.getDataRegionId();
return new File(snapshotDir).getCanonicalFile();
- } catch (IOException e) {
+ } catch (IOException | NullPointerException e) {
logger.warn("{}: cannot get the canonical file of {} due to {}", this,
snapshotDir, e);
return null;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java
index 1cdd3406574..e9752f828c7 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java
@@ -57,7 +57,6 @@ import
org.apache.iotdb.confignode.rpc.thrift.TNodeVersionInfo;
import org.apache.iotdb.confignode.rpc.thrift.TRuntimeConfiguration;
import org.apache.iotdb.confignode.rpc.thrift.TSystemConfigurationResp;
import org.apache.iotdb.consensus.ConsensusFactory;
-import org.apache.iotdb.consensus.iot.IoTConsensus;
import org.apache.iotdb.db.conf.DataNodeStartupCheck;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
@@ -112,9 +111,6 @@ import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.Charset;
-import java.nio.file.DirectoryStream;
-import java.nio.file.Files;
-import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -457,55 +453,30 @@ public class DataNode implements DataNodeMBean {
}
}
- // TODO: Implement in IConsensus, not in DataNode
- private List<ConsensusGroupId> getConsensusGroupId() {
- List<ConsensusGroupId> consensusGroupIds = new ArrayList<>();
- String dataRegionConsensusDir = config.getDataRegionConsensusDir();
- if
(config.getDataRegionConsensusProtocolClass().equals(ConsensusFactory.RATIS_CONSENSUS))
{
- return consensusGroupIds;
- }
- try (DirectoryStream<Path> stream =
- Files.newDirectoryStream(new File(dataRegionConsensusDir).toPath())) {
- for (Path path : stream) {
- String[] items = path.getFileName().toString().split("_");
- ConsensusGroupId consensusGroupId =
- ConsensusGroupId.Factory.create(Integer.parseInt(items[0]),
Integer.parseInt(items[1]));
- consensusGroupIds.add(consensusGroupId);
- }
- } catch (IOException e) {
- logger.error("Cannot get consensus group id from {}",
dataRegionConsensusDir, e);
- }
- return consensusGroupIds;
- }
-
- // TODO: remove for current version, add todo for rename
- private void renameInvalidRegionDirs(List<ConsensusGroupId>
invalidConsensusGroupIds) {
- for (ConsensusGroupId consensusGroupId : invalidConsensusGroupIds) {
- File oldDir =
- new File(
- IoTConsensus.buildPeerDir(
- new File(config.getDataRegionConsensusDir()),
consensusGroupId));
- File newDir =
- new File(
- IoTConsensus.buildPeerDir(
- new File(config.getInvalidDataRegionConsensusDir()),
consensusGroupId));
- if (oldDir.exists() && !FileUtils.moveFileSafe(oldDir, newDir)) {
- logger.error("move {} to {} failed.", oldDir.getAbsolutePath(),
newDir.getAbsolutePath());
- try {
- FileUtils.recursivelyDeleteFolder(oldDir.getPath());
- } catch (IOException e) {
- logger.error("delete {} failed.", oldDir.getAbsolutePath());
- }
- }
- }
- }
-
private void removeInvalidRegions(List<ConsensusGroupId>
dataNodeConsensusGroupIds) {
List<ConsensusGroupId> invalidConsensusGroupIds =
- getConsensusGroupId().stream()
+
DataRegionConsensusImpl.getInstance().getAllConsensusGroupIdsWithoutStarting().stream()
.filter(consensusGroupId ->
!dataNodeConsensusGroupIds.contains(consensusGroupId))
.collect(Collectors.toList());
- renameInvalidRegionDirs(invalidConsensusGroupIds);
+ if (!invalidConsensusGroupIds.isEmpty()) {
+ logger.info("Remove invalid region directories... {}",
invalidConsensusGroupIds);
+ for (ConsensusGroupId consensusGroupId : invalidConsensusGroupIds) {
+ File oldDir =
+ new File(
+ DataRegionConsensusImpl.getInstance()
+ .getRegionDirFromConsensusGroupId(consensusGroupId));
+ if (oldDir.exists()) {
+ try {
+ FileUtils.recursivelyDeleteFolder(oldDir.getPath());
+ logger.info("delete {} succeed.", oldDir.getAbsolutePath());
+ } catch (IOException e) {
+ logger.error("delete {} failed.", oldDir.getAbsolutePath());
+ }
+ } else {
+ logger.info("delete {} failed, because it does not exist.",
oldDir.getAbsolutePath());
+ }
+ }
+ }
}
private void sendRestartRequestToConfigNode() throws StartupException {
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 df07d3d7487..c021920f27a 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
@@ -38,8 +38,7 @@ 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.schemaengine.SchemaEngine;
-import org.apache.iotdb.db.storageengine.StorageEngine;
+import org.apache.iotdb.db.protocol.thrift.impl.DataNodeRegionManager;
import org.apache.iotdb.db.storageengine.rescon.memory.AbstractPoolManager;
import org.apache.iotdb.mpp.rpc.thrift.TMaintainPeerReq;
import org.apache.iotdb.mpp.rpc.thrift.TRegionMigrateResult;
@@ -524,9 +523,9 @@ public class RegionMigrateService implements IService {
ConsensusGroupId regionId =
ConsensusGroupId.Factory.createFromTConsensusGroupId(tRegionId);
try {
if (regionId instanceof DataRegionId) {
- StorageEngine.getInstance().deleteDataRegion((DataRegionId)
regionId);
+ DataNodeRegionManager.getInstance().deleteDataRegion((DataRegionId)
regionId);
} else {
- SchemaEngine.getInstance().deleteSchemaRegion((SchemaRegionId)
regionId);
+
DataNodeRegionManager.getInstance().deleteSchemaRegion((SchemaRegionId)
regionId);
}
} catch (Exception e) {
taskLogger.error("{}, deleteRegion {} error", REGION_MIGRATE_PROCESS,
regionId, e);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/snapshot/SnapshotLoader.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/snapshot/SnapshotLoader.java
index 5e04f1f2f35..795bd960d7c 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/snapshot/SnapshotLoader.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/snapshot/SnapshotLoader.java
@@ -124,12 +124,6 @@ public class SnapshotLoader {
return null;
}
LOGGER.info("Moving snapshot file to data dirs");
- try {
- deleteAllFilesInDataDirs();
- LOGGER.info("Remove all data files in original data dir");
- } catch (IOException e) {
- return null;
- }
createLinksFromSnapshotDirToDataDirWithoutLog(new File(snapshotPath));
return loadSnapshot();
} catch (IOException | DiskSpaceInsufficientException e) {
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/KillPoint/IoTConsensusRemovePeerKillPoints.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/KillPoint/IoTConsensusDeleteLocalPeerKillPoints.java
similarity index 86%
copy from
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/KillPoint/IoTConsensusRemovePeerKillPoints.java
copy to
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/KillPoint/IoTConsensusDeleteLocalPeerKillPoints.java
index 4f80aa87212..ea8352c7634 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/KillPoint/IoTConsensusRemovePeerKillPoints.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/KillPoint/IoTConsensusDeleteLocalPeerKillPoints.java
@@ -19,9 +19,7 @@
package org.apache.iotdb.commons.utils.KillPoint;
-public enum IoTConsensusRemovePeerKillPoints {
- INIT,
- AFTER_NOTIFY_PEERS_TO_REMOVE_SYNC_LOG_CHANNEL,
- AFTER_INACTIVE_PEER,
- FINISH,
+public enum IoTConsensusDeleteLocalPeerKillPoints {
+ BEFORE_DELETE,
+ AFTER_DELETE,
}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/KillPoint/IoTConsensusRemovePeerKillPoints.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/KillPoint/IoTConsensusInactivatePeerKillPoints.java
similarity index 86%
copy from
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/KillPoint/IoTConsensusRemovePeerKillPoints.java
copy to
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/KillPoint/IoTConsensusInactivatePeerKillPoints.java
index 4f80aa87212..f21577db8ea 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/KillPoint/IoTConsensusRemovePeerKillPoints.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/KillPoint/IoTConsensusInactivatePeerKillPoints.java
@@ -19,9 +19,7 @@
package org.apache.iotdb.commons.utils.KillPoint;
-public enum IoTConsensusRemovePeerKillPoints {
- INIT,
- AFTER_NOTIFY_PEERS_TO_REMOVE_SYNC_LOG_CHANNEL,
- AFTER_INACTIVE_PEER,
- FINISH,
+public enum IoTConsensusInactivatePeerKillPoints {
+ BEFORE_INACTIVATE,
+ AFTER_INACTIVATE,
}
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/KillPoint/IoTConsensusRemovePeerKillPoints.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/KillPoint/IoTConsensusRemovePeerCoordinatorKillPoints.java
similarity index 94%
copy from
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/KillPoint/IoTConsensusRemovePeerKillPoints.java
copy to
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/KillPoint/IoTConsensusRemovePeerCoordinatorKillPoints.java
index 4f80aa87212..14a2ce9f439 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/KillPoint/IoTConsensusRemovePeerKillPoints.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/KillPoint/IoTConsensusRemovePeerCoordinatorKillPoints.java
@@ -19,7 +19,7 @@
package org.apache.iotdb.commons.utils.KillPoint;
-public enum IoTConsensusRemovePeerKillPoints {
+public enum IoTConsensusRemovePeerCoordinatorKillPoints {
INIT,
AFTER_NOTIFY_PEERS_TO_REMOVE_SYNC_LOG_CHANNEL,
AFTER_INACTIVE_PEER,
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/KillPoint/IoTConsensusRemovePeerKillPoints.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/KillPoint/KillNode.java
similarity index 86%
rename from
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/KillPoint/IoTConsensusRemovePeerKillPoints.java
rename to
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/KillPoint/KillNode.java
index 4f80aa87212..1ecba5f303b 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/KillPoint/IoTConsensusRemovePeerKillPoints.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/KillPoint/KillNode.java
@@ -19,9 +19,10 @@
package org.apache.iotdb.commons.utils.KillPoint;
-public enum IoTConsensusRemovePeerKillPoints {
- INIT,
- AFTER_NOTIFY_PEERS_TO_REMOVE_SYNC_LOG_CHANNEL,
- AFTER_INACTIVE_PEER,
- FINISH,
+public enum KillNode {
+ ALL_NODES,
+ CONFIG_NODE,
+ ORIGINAL_DATANODE,
+ DESTINATION_DATANODE,
+ COORDINATOR_DATANODE,
}
diff --git
a/iotdb-protocol/thrift-consensus/src/main/thrift/iotconsensus.thrift
b/iotdb-protocol/thrift-consensus/src/main/thrift/iotconsensus.thrift
index 5810ff97a96..c1957eaf114 100644
--- a/iotdb-protocol/thrift-consensus/src/main/thrift/iotconsensus.thrift
+++ b/iotdb-protocol/thrift-consensus/src/main/thrift/iotconsensus.thrift
@@ -41,6 +41,7 @@ struct TSyncLogEntriesRes {
struct TInactivatePeerReq {
1: required common.TConsensusGroupId consensusGroupId
+ 2: optional bool forDeletionPurpose
}
struct TInactivatePeerRes {
