This is an automated email from the ASF dual-hosted git repository.
Caideyipi 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 08380be95ae Fixed multiple configNode bugs (#17609)
08380be95ae is described below
commit 08380be95ae82bffc9e1aa8adf7cc1a92eb027ee
Author: Caideyipi <[email protected]>
AuthorDate: Tue Oct 20 14:11:18 2026 +0800
Fixed multiple configNode bugs (#17609)
* Fixed multiple config bugs
* fix
* Update ConfigRegionStateMachine.java
* Address ConfigNode review feedback
* Fix compile
* Compile fix
* Fix NodeManager Sonar issues
* Fix ConfigNode WAL recovery and registration review feedback
---
.../iotdb/confignode/i18n/ConfigNodeMessages.java | 37 +++
.../iotdb/confignode/i18n/ConfigNodeMessages.java | 45 ++-
.../statemachine/ConfigRegionStateMachine.java | 322 ++++++++++++++++-----
.../iotdb/confignode/manager/node/NodeManager.java | 246 ++++++++++++----
.../partition/DatabasePartitionTable.java | 2 +-
.../persistence/partition/PartitionInfo.java | 13 +-
.../impl/node/AddConfigNodeProcedure.java | 4 +-
.../statemachine/ConfigRegionStateMachineTest.java | 302 +++++++++++++++++++
.../manager/node/NodeManagerRegistrationTest.java | 260 +++++++++++++++++
.../iotdb/confignode/persistence/NodeInfoTest.java | 43 +++
.../confignode/persistence/PartitionInfoTest.java | 29 ++
11 files changed, 1162 insertions(+), 141 deletions(-)
diff --git
a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
index 60d672cc490..4aea615a183 100644
---
a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
+++
b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
@@ -20,6 +20,31 @@
package org.apache.iotdb.confignode.i18n;
public final class ConfigNodeMessages {
+ public static final String
EXCEPTION_CONFIGNODE_SIMPLECONSENSUS_WRITES_ARE_DISABLED_AFTER_A_WAL_CLEANUP_FAILURE_RESTART_IS_REQUIRED_D52F0107
=
+ "ConfigNode SimpleConsensus writes are disabled after a WAL cleanup
failure; restart is required.";
+ public static final String
EXCEPTION_FAILED_TO_CLOSE_THE_CONFIGNODE_SIMPLECONSENSUS_LOG_WRITER_BC0F417C =
+ "Failed to close the ConfigNode SimpleConsensus log writer.";
+ public static final String
EXCEPTION_FAILED_TO_APPLY_CONFIGNODE_ARG_THROUGH_CONSENSUS_LAYER_ARG_B4A985D7 =
+ "Failed to apply ConfigNode %s through consensus layer: %s";
+ public static final String
EXCEPTION_FAILED_TO_PERSIST_VERSION_INFO_FOR_CONFIGNODE_ARG_ARG_REGISTRATION_REMAINS_PENDING_AND_MUST_BE_RETRIED_F662C350
=
+ "Failed to persist version info for ConfigNode %d: %s. Registration
remains pending and must be retried.";
+ public static final String MESSAGE_UNKNOWN_ERROR_E9EEAFF8 =
+ "unknown error";
+ public static final String MESSAGE_STATUS_CODE_ARG_4B3BCD8B =
+ "status code %d";
+ public static final String
MESSAGE_FAILED_TO_PERSIST_VERSION_INFO_FOR_DATANODE_ARG_ARG_THE_REGISTRATION_HAS_BEEN_ROLLED_BACK_PLEASE_RETRY_THE_REGISTRATION_5DDB311F
=
+ "Failed to persist version info for DataNode %d: %s. The registration
has been rolled back. Please retry the registration.";
+ public static final String
MESSAGE_FAILED_TO_PERSIST_VERSION_INFO_FOR_DATANODE_ARG_ARG_THE_REGISTRATION_ROLLBACK_ALSO_FAILED_ARG_MANUAL_CLEANUP_MAY_BE_REQUIRED_BEFORE_RETRYING_THE_REGISTRATION_5463B5FF
=
+ "Failed to persist version info for DataNode %d: %s. The registration
rollback also failed: %s. Manual cleanup may be required before retrying the
registration.";
+ public static final String
LOG_FAILED_TO_ROLL_BACK_DATANODE_REGISTRATION_ARG_AFTER_VERSION_INFO_PERSISTENCE_FAILURE_VERSIONUPDATESTATUS_ARG_ROLLBACKSTATUS_ARG_437E6F09
=
+ "Failed to roll back DataNode registration {} after version info
persistence failure. versionUpdateStatus: {}, rollbackStatus: {}";
+ public static final String
MESSAGE_FAILED_TO_PERSIST_VERSION_INFO_FOR_AINODE_ARG_ARG_THE_REGISTRATION_HAS_BEEN_ROLLED_BACK_PLEASE_RETRY_THE_REGISTRATION_7571C626
=
+ "Failed to persist version info for AINode %d: %s. The registration has
been rolled back. Please retry the registration.";
+ public static final String
MESSAGE_FAILED_TO_PERSIST_VERSION_INFO_FOR_AINODE_ARG_ARG_THE_REGISTRATION_ROLLBACK_ALSO_FAILED_ARG_MANUAL_CLEANUP_MAY_BE_REQUIRED_BEFORE_RETRYING_THE_REGISTRATION_B9B1DCCD
=
+ "Failed to persist version info for AINode %d: %s. The registration
rollback also failed: %s. Manual cleanup may be required before retrying the
registration.";
+ public static final String
LOG_FAILED_TO_ROLL_BACK_AINODE_REGISTRATION_ARG_AFTER_VERSION_INFO_PERSISTENCE_FAILURE_VERSIONUPDATESTATUS_ARG_ROLLBACKSTATUS_ARG_BFF24BD4
=
+ "Failed to roll back AINode registration {} after version info
persistence failure. versionUpdateStatus: {}, rollbackStatus: {}";
+
public static final String
EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ILLEGAL_ARG_ARG_EXPECTED_A_LONG_VALUE_IN_MILLISECONDS_F79736B2
=
"Failed to create or alter topic, illegal %s=%s, expected a long
value in milliseconds";
@@ -61,6 +86,18 @@ public final class ConfigNodeMessages {
"[ConfigRegion Update] Can't force logWriter for ConfigNode
SimpleConsensus mode";
public static final String
CAN_T_SERIALIZE_CURRENT_CONFIGPHYSICALPLAN_FOR_CONFIGNODE_SIMPLECONSENSUS_MODE =
"[ConfigRegion Update] Can't serialize current ConfigPhysicalPlan for
ConfigNode SimpleConsensus mode";
+ public static final String PERSIST_CONFIGNODE_SIMPLECONSENSUS_LOG_FAILED =
+ "Persist ConfigNode SimpleConsensus log failed: %s";
+ public static final String
PERSIST_CURRENT_CONFIGPHYSICALPLAN_FOR_CONFIGNODE_SIMPLECONSENSUS_MODE_FAILED =
+ "Persist current ConfigPhysicalPlan for ConfigNode SimpleConsensus mode
failed";
+ public static final String
FAILED_TO_ROLLBACK_PERSISTED_CONFIGNODE_SIMPLECONSENSUS_LOG =
+ "Failed to rollback persisted ConfigNode SimpleConsensus log. The
persisted plan may be replayed after restart. Further SimpleConsensus writes
are disabled until restart.";
+ public static final String
ROLLBACK_FAILED_CONFIGPHYSICALPLAN_FOR_CONFIGNODE_SIMPLECONSENSUS_MODE_FAILED =
+ "Rollback failed ConfigPhysicalPlan for ConfigNode SimpleConsensus mode
failed, planType: {}, logFile: {}, truncateOffset: {}, endIndexBeforeWrite: {}.
Further SimpleConsensus writes are disabled until restart.";
+ public static final String
SEAL_RECOVERED_CONFIGNODE_SIMPLECONSENSUS_LOG_FAILED =
+ "Seal recovered ConfigNode SimpleConsensus log failed: {}";
+ public static final String SIMPLECONSENSUS_LOG_WRITER_IS_NOT_INITIALIZED =
+ "SimpleConsensus log writer is not initialized.";
public static final String CAN_T_START_CONFIGNODE_CONSENSUS_GROUP =
"Can't start ConfigNode consensus group!";
public static final String CHANGE_REGIONS_LEADER_ERROR_ON_DATE_NODE =
diff --git
a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
index 1b3926590cb..35436eee446 100644
---
a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
+++
b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
@@ -20,6 +20,31 @@
package org.apache.iotdb.confignode.i18n;
public final class ConfigNodeMessages {
+ public static final String
EXCEPTION_CONFIGNODE_SIMPLECONSENSUS_WRITES_ARE_DISABLED_AFTER_A_WAL_CLEANUP_FAILURE_RESTART_IS_REQUIRED_D52F0107
=
+ "ConfigNode SimpleConsensus 因 WAL 清理失败已禁用写入，需要重启。";
+ public static final String
EXCEPTION_FAILED_TO_CLOSE_THE_CONFIGNODE_SIMPLECONSENSUS_LOG_WRITER_BC0F417C =
+ "关闭 ConfigNode SimpleConsensus 日志写入器失败。";
+ public static final String
EXCEPTION_FAILED_TO_APPLY_CONFIGNODE_ARG_THROUGH_CONSENSUS_LAYER_ARG_B4A985D7 =
+ "通过共识层应用 ConfigNode %s 失败：%s";
+ public static final String
EXCEPTION_FAILED_TO_PERSIST_VERSION_INFO_FOR_CONFIGNODE_ARG_ARG_REGISTRATION_REMAINS_PENDING_AND_MUST_BE_RETRIED_F662C350
=
+ "持久化 ConfigNode %d 的版本信息失败：%s。注册尚未完成，需要重试。";
+ public static final String MESSAGE_UNKNOWN_ERROR_E9EEAFF8 =
+ "未知错误";
+ public static final String MESSAGE_STATUS_CODE_ARG_4B3BCD8B =
+ "状态码 %d";
+ public static final String
MESSAGE_FAILED_TO_PERSIST_VERSION_INFO_FOR_DATANODE_ARG_ARG_THE_REGISTRATION_HAS_BEEN_ROLLED_BACK_PLEASE_RETRY_THE_REGISTRATION_5DDB311F
=
+ "持久化 DataNode %d 的版本信息失败：%s。注册已回滚，请重试注册。";
+ public static final String
MESSAGE_FAILED_TO_PERSIST_VERSION_INFO_FOR_DATANODE_ARG_ARG_THE_REGISTRATION_ROLLBACK_ALSO_FAILED_ARG_MANUAL_CLEANUP_MAY_BE_REQUIRED_BEFORE_RETRYING_THE_REGISTRATION_5463B5FF
=
+ "持久化 DataNode %d 的版本信息失败：%s。注册回滚也失败：%s。重试注册前可能需要手动清理。";
+ public static final String
LOG_FAILED_TO_ROLL_BACK_DATANODE_REGISTRATION_ARG_AFTER_VERSION_INFO_PERSISTENCE_FAILURE_VERSIONUPDATESTATUS_ARG_ROLLBACKSTATUS_ARG_437E6F09
=
+ "版本信息持久化失败后，回滚 DataNode {} 的注册失败。versionUpdateStatus: {}，rollbackStatus:
{}";
+ public static final String
MESSAGE_FAILED_TO_PERSIST_VERSION_INFO_FOR_AINODE_ARG_ARG_THE_REGISTRATION_HAS_BEEN_ROLLED_BACK_PLEASE_RETRY_THE_REGISTRATION_7571C626
=
+ "持久化 AINode %d 的版本信息失败：%s。注册已回滚，请重试注册。";
+ public static final String
MESSAGE_FAILED_TO_PERSIST_VERSION_INFO_FOR_AINODE_ARG_ARG_THE_REGISTRATION_ROLLBACK_ALSO_FAILED_ARG_MANUAL_CLEANUP_MAY_BE_REQUIRED_BEFORE_RETRYING_THE_REGISTRATION_B9B1DCCD
=
+ "持久化 AINode %d 的版本信息失败：%s。注册回滚也失败：%s。重试注册前可能需要手动清理。";
+ public static final String
LOG_FAILED_TO_ROLL_BACK_AINODE_REGISTRATION_ARG_AFTER_VERSION_INFO_PERSISTENCE_FAILURE_VERSIONUPDATESTATUS_ARG_ROLLBACKSTATUS_ARG_BFF24BD4
=
+ "版本信息持久化失败后，回滚 AINode {} 的注册失败。versionUpdateStatus: {}，rollbackStatus:
{}";
+
public static final String
EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ILLEGAL_ARG_ARG_EXPECTED_A_LONG_VALUE_IN_MILLISECONDS_F79736B2
=
"创建或修改 topic 失败，%s=%s 无效，需指定以毫秒为单位的 long 值";
@@ -57,10 +82,22 @@ public final class ConfigNodeMessages {
"[ConfigRegion Update] 无法为 ConfigNode flushWALForSimpleConsensus 强制写入
logWriter";
public static final String
CAN_T_FORCE_LOGWRITER_FOR_CONFIGNODE_SIMPLECONSENSUS_MODE =
"[ConfigRegion Update] 无法为 ConfigNode SimpleConsensus 模式强制写入 logWriter";
- public static final String
-
CAN_T_SERIALIZE_CURRENT_CONFIGPHYSICALPLAN_FOR_CONFIGNODE_SIMPLECONSENSUS_MODE =
- "[ConfigRegion Update] 无法为 ConfigNode SimpleConsensus 模式序列化当前
ConfigPhysicalPlan";
- public static final String CAN_T_START_CONFIGNODE_CONSENSUS_GROUP = "无法启动
ConfigNode 共识组！";
+ public static final String
CAN_T_SERIALIZE_CURRENT_CONFIGPHYSICALPLAN_FOR_CONFIGNODE_SIMPLECONSENSUS_MODE =
+ "[ConfigRegion Update] 无法为 ConfigNode SimpleConsensus 模式序列化当前
ConfigPhysicalPlan";
+ public static final String PERSIST_CONFIGNODE_SIMPLECONSENSUS_LOG_FAILED =
+ "持久化 ConfigNode SimpleConsensus 日志失败：%s";
+ public static final String
PERSIST_CURRENT_CONFIGPHYSICALPLAN_FOR_CONFIGNODE_SIMPLECONSENSUS_MODE_FAILED =
+ "为 ConfigNode SimpleConsensus 模式持久化当前 ConfigPhysicalPlan 失败";
+ public static final String
FAILED_TO_ROLLBACK_PERSISTED_CONFIGNODE_SIMPLECONSENSUS_LOG =
+ "回滚已持久化的 ConfigNode SimpleConsensus 日志失败。该 plan 可能会在重启后被重放。重启前将禁止后续
SimpleConsensus 写入。";
+ public static final String
ROLLBACK_FAILED_CONFIGPHYSICALPLAN_FOR_CONFIGNODE_SIMPLECONSENSUS_MODE_FAILED =
+ "回滚 ConfigNode SimpleConsensus 模式下失败的 ConfigPhysicalPlan 失败，planType:
{}, logFile: {}, truncateOffset: {}, endIndexBeforeWrite: {}。重启前将禁止后续
SimpleConsensus 写入。";
+ public static final String
SEAL_RECOVERED_CONFIGNODE_SIMPLECONSENSUS_LOG_FAILED =
+ "封存恢复出的 ConfigNode SimpleConsensus 日志失败：{}";
+ public static final String SIMPLECONSENSUS_LOG_WRITER_IS_NOT_INITIALIZED =
+ "SimpleConsensus log writer 未初始化。";
+ public static final String CAN_T_START_CONFIGNODE_CONSENSUS_GROUP =
+ "无法启动 ConfigNode 共识组！";
public static final String CHANGE_REGIONS_LEADER_ERROR_ON_DATE_NODE =
"在 DataNode: {} 上切换 region leader 失败";
public static final String CHECK_BEFORE_DROPPING_TOPIC_TOPIC_EXISTS =
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachine.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachine.java
index 6b4da5445d3..53bc04d38c4 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachine.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachine.java
@@ -24,7 +24,6 @@ import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.auth.AuthException;
import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
import org.apache.iotdb.commons.concurrent.ThreadName;
-import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
import org.apache.iotdb.commons.file.SystemFileFactory;
@@ -57,14 +56,16 @@ import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
import java.nio.file.Files;
import java.nio.file.StandardCopyOption;
+import java.nio.file.StandardOpenOption;
import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
+import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
-import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
@@ -116,6 +117,7 @@ public class ConfigRegionStateMachine implements
IStateMachine, IStateMachine.Ev
private File simpleLogFile;
private int startIndex;
private int endIndex;
+ private boolean simpleConsensusWritesDisabled;
private static final String CURRENT_FILE_DIR =
ConsensusManager.getConfigRegionDir() + File.separator + "current";
@@ -157,16 +159,32 @@ public class ConfigRegionStateMachine implements
IStateMachine, IStateMachine.Ev
/** Transmit {@link ConfigPhysicalPlan} to {@link ConfigPlanExecutor} */
protected TSStatus write(ConfigPhysicalPlan plan) {
+ SimpleConsensusPersistResult persistResult = null;
+ if
(ConsensusFactory.SIMPLE_CONSENSUS.equals(CONF.getConfigNodeConsensusProtocolClass()))
{
+ persistResult = persistPlanForSimpleConsensus(plan);
+ final TSStatus persistStatus = persistResult.status;
+ if (persistStatus.getCode() !=
TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ return persistStatus;
+ }
+ }
+
TSStatus result;
try {
result = executor.executeNonQueryPlan(plan);
- } catch (UnknownPhysicalPlanTypeException e) {
+ } catch (UnknownPhysicalPlanTypeException | RuntimeException e) {
LOGGER.error(ConfigNodeMessages.EXECUTE_NON_QUERY_PLAN_FAILED, e);
result = new
TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode());
}
- if
(ConsensusFactory.SIMPLE_CONSENSUS.equals(CONF.getConfigNodeConsensusProtocolClass()))
{
- writeLogForSimpleConsensus(plan);
+ if (persistResult != null
+ && result.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()
+ && !rollbackFailedPlanForSimpleConsensus(plan, persistResult)) {
+ final String rollbackFailureMessage =
+
ConfigNodeMessages.FAILED_TO_ROLLBACK_PERSISTED_CONFIGNODE_SIMPLECONSENSUS_LOG;
+ result.setMessage(
+ Optional.ofNullable(result.getMessage())
+ .map(message -> message + " " + rollbackFailureMessage)
+ .orElse(rollbackFailureMessage));
}
if (result.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
@@ -233,7 +251,6 @@ public class ConfigRegionStateMachine implements
IStateMachine, IStateMachine.Ev
PipeConfigNodeAgent.runtime()
.listener()
.tryListenToSnapshots(ConfigNodeSnapshotParser.getSnapshots());
- return true;
} catch (IOException e) {
if (PipeConfigNodeAgent.runtime().listener().isOpened()) {
LOGGER.warn(
@@ -242,6 +259,7 @@ public class ConfigRegionStateMachine implements
IStateMachine, IStateMachine.Ev
e);
}
}
+ return true;
}
return false;
}
@@ -251,9 +269,12 @@ public class ConfigRegionStateMachine implements
IStateMachine, IStateMachine.Ev
// The boolean result must reflect whether the ConfigRegion state-machine
data was loaded, so
// callers (e.g. the AddPeer flow) can detect a real failure. The
pipe-listener recomputation
// below is best-effort post-processing: a failure there is logged but
must NOT be reported as a
- // snapshot-load failure, otherwise it would (e.g.) abort ConfigNode
(re)initialization on what
- // is actually a healthy data load.
+ // snapshot-load failure.
final boolean loadSucceeded = executor.loadSnapshot(latestSnapshotRootDir);
+ if (!loadSucceeded) {
+ return false;
+ }
+
try {
// We recompute the snapshot for pipe listener when loading snapshot
// to recover the newest snapshot in cache
@@ -267,7 +288,7 @@ public class ConfigRegionStateMachine implements
IStateMachine, IStateMachine.Ev
e);
}
}
- return loadSucceeded;
+ return true;
}
@Override
@@ -570,6 +591,9 @@ public class ConfigRegionStateMachine implements
IStateMachine, IStateMachine.Ev
@Override
public void stop() {
+ if
(ConsensusFactory.SIMPLE_CONSENSUS.equals(CONF.getConfigNodeConsensusProtocolClass()))
{
+ closeSimpleLogWriter();
+ }
// Shutdown leader related service for config pipe
PipeConfigNodeAgent.runtime().notifyLeaderUnavailable();
}
@@ -579,60 +603,126 @@ public class ConfigRegionStateMachine implements
IStateMachine, IStateMachine.Ev
return CommonDescriptor.getInstance().getConfig().isReadOnly();
}
- private void writeLogForSimpleConsensus(ConfigPhysicalPlan plan) {
- if (simpleLogFile.length() > LOG_FILE_MAX_SIZE) {
- try {
- simpleLogWriter.force();
- File completedFilePath = new File(FILE_PATH + startIndex + "_" +
endIndex);
- Files.move(
- simpleLogFile.toPath(), completedFilePath.toPath(),
StandardCopyOption.ATOMIC_MOVE);
- } catch (IOException e) {
- LOGGER.error(
-
ConfigNodeMessages.CAN_T_FORCE_LOGWRITER_FOR_CONFIGNODE_SIMPLECONSENSUS_MODE,
e);
+ private SimpleConsensusPersistResult
persistPlanForSimpleConsensus(ConfigPhysicalPlan plan) {
+ File persistedLogFile = null;
+ long logFileSizeBeforeWrite = 0;
+ int endIndexBeforeWrite = endIndex;
+ try {
+ if (simpleConsensusWritesDisabled) {
+ throw new IOException(
+ ConfigNodeMessages
+
.EXCEPTION_CONFIGNODE_SIMPLECONSENSUS_WRITES_ARE_DISABLED_AFTER_A_WAL_CLEANUP_FAILURE_RESTART_IS_REQUIRED_D52F0107);
}
- for (int retry = 0; retry < 5; retry++) {
- try {
- simpleLogWriter.close();
- } catch (IOException e) {
- LOGGER.warn(
- ConfigNodeMessages
-
.LOG_CONFIGREGION_UPDATE_FAILED_TO_CLOSE_THE_CONFIGNODE_SIMPLECONSENSUS_LOG_FILEPATH_ARG_RETRY_ARG_188A702E,
- simpleLogFile.getAbsolutePath(),
- retry);
- try {
- // Sleep 1s and retry
- TimeUnit.SECONDS.sleep(1);
- } catch (InterruptedException e2) {
- Thread.currentThread().interrupt();
- LOGGER.warn(
-
ConfigNodeMessages.UNEXPECTED_INTERRUPTION_DURING_THE_CLOSE_METHOD_OF_LOGWRITER);
- }
- continue;
- }
- break;
+ if (simpleLogWriter == null || simpleLogFile == null) {
+ throw new
IOException(ConfigNodeMessages.SIMPLECONSENSUS_LOG_WRITER_IS_NOT_INITIALIZED);
}
- startIndex = endIndex + 1;
- createLogFile(startIndex);
- }
- try {
+ if (simpleLogFile.length() > LOG_FILE_MAX_SIZE) {
+ rollSimpleConsensusLogFile();
+ }
+
+ persistedLogFile = simpleLogFile;
+ logFileSizeBeforeWrite = persistedLogFile.length();
+ endIndexBeforeWrite = endIndex;
+
ByteBuffer buffer = plan.serializeToByteBuffer();
buffer.position(buffer.limit());
simpleLogWriter.write(buffer);
+ simpleLogWriter.force();
endIndex = endIndex + 1;
} catch (Exception e) {
LOGGER.error(
ConfigNodeMessages
-
.CAN_T_SERIALIZE_CURRENT_CONFIGPHYSICALPLAN_FOR_CONFIGNODE_SIMPLECONSENSUS_MODE,
+
.PERSIST_CURRENT_CONFIGPHYSICALPLAN_FOR_CONFIGNODE_SIMPLECONSENSUS_MODE_FAILED,
+ e);
+ TSStatus status =
+ new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode())
+ .setMessage(
+ String.format(
+
ConfigNodeMessages.PERSIST_CONFIGNODE_SIMPLECONSENSUS_LOG_FAILED,
+ e.getMessage()));
+ // A failed append or force may still leave a complete or partial record
in the WAL.
+ if (persistedLogFile != null
+ && !rollbackFailedPlanForSimpleConsensus(
+ plan,
+ SimpleConsensusPersistResult.success(
+ persistedLogFile, logFileSizeBeforeWrite,
endIndexBeforeWrite))) {
+ status.setMessage(
+ status.getMessage()
+ + " "
+ +
ConfigNodeMessages.FAILED_TO_ROLLBACK_PERSISTED_CONFIGNODE_SIMPLECONSENSUS_LOG);
+ }
+ return SimpleConsensusPersistResult.failure(status);
+ }
+ return SimpleConsensusPersistResult.success(
+ persistedLogFile, logFileSizeBeforeWrite, endIndexBeforeWrite);
+ }
+
+ private boolean rollbackFailedPlanForSimpleConsensus(
+ ConfigPhysicalPlan plan, SimpleConsensusPersistResult persistResult) {
+ try {
+ if (!closeSimpleLogWriter()) {
+ throw new IOException(
+ ConfigNodeMessages
+
.EXCEPTION_FAILED_TO_CLOSE_THE_CONFIGNODE_SIMPLECONSENSUS_LOG_WRITER_BC0F417C);
+ }
+ try (FileChannel channel =
+ FileChannel.open(persistResult.logFile.toPath(),
StandardOpenOption.WRITE)) {
+ channel.truncate(persistResult.logFileSizeBeforeWrite);
+ channel.force(true);
+ }
+ simpleLogFile = persistResult.logFile;
+ simpleLogWriter = new LogWriter(simpleLogFile, false);
+ endIndex = persistResult.endIndexBeforeWrite;
+ return true;
+ } catch (IOException e) {
+ // Never append past an entry whose removal could not be made durable.
+ simpleConsensusWritesDisabled = true;
+ LOGGER.error(
+ ConfigNodeMessages
+
.ROLLBACK_FAILED_CONFIGPHYSICALPLAN_FOR_CONFIGNODE_SIMPLECONSENSUS_MODE_FAILED,
+ plan.getType(),
+ persistResult.logFile,
+ persistResult.logFileSizeBeforeWrite,
+ persistResult.endIndexBeforeWrite,
e);
+ return false;
}
}
+ private void rollSimpleConsensusLogFile() throws IOException {
+ simpleLogWriter.force();
+ if (!closeSimpleLogWriter()) {
+ simpleConsensusWritesDisabled = true;
+ throw new IOException(
+ ConfigNodeMessages
+
.EXCEPTION_FAILED_TO_CLOSE_THE_CONFIGNODE_SIMPLECONSENSUS_LOG_WRITER_BC0F417C);
+ }
+ try {
+ Files.move(
+ simpleLogFile.toPath(),
+ simpleLogFile.toPath().resolveSibling(LOG_FILE_PREFIX + startIndex +
"_" + endIndex),
+ StandardCopyOption.ATOMIC_MOVE);
+ } catch (IOException e) {
+ // The source file and indexes are unchanged when renaming fails. Reopen
it for retries.
+ try {
+ simpleLogWriter = new LogWriter(simpleLogFile, false);
+ } catch (IOException reopenException) {
+ simpleConsensusWritesDisabled = true;
+ e.addSuppressed(reopenException);
+ }
+ throw e;
+ }
+ startIndex = endIndex + 1;
+ createLogFile(startIndex);
+ }
+
private void initStandAloneConfigNode() {
File dir = new File(CURRENT_FILE_DIR);
dir.mkdirs();
String[] list = new File(CURRENT_FILE_DIR).list();
+ endIndex = 0;
if (list != null && list.length != 0) {
Arrays.sort(list,
Comparator.comparingLong(ConfigRegionStateMachine::parseEndIndex));
for (String logFileName : list) {
@@ -650,7 +740,7 @@ public class ConfigRegionStateMachine implements
IStateMachine, IStateMachine.Ev
continue;
}
- startIndex = endIndex;
+ final int recoveredStartIndex = parseStartIndex(logFileName);
while (logReader.hasNext()) {
endIndex++;
// Read and re-serialize the PhysicalPlan
@@ -668,34 +758,13 @@ public class ConfigRegionStateMachine implements
IStateMachine, IStateMachine.Ev
}
}
logReader.close();
- }
- } else {
- startIndex = 0;
- endIndex = 0;
- }
- startIndex = startIndex + 1;
- createLogFile(endIndex);
-
- ScheduledExecutorService simpleConsensusThread =
- IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor(
- ThreadName.CONFIG_NODE_SIMPLE_CONSENSUS_WAL_FLUSH.getName());
- ScheduledExecutorUtil.safelyScheduleWithFixedDelay(
- simpleConsensusThread,
- this::flushWALForSimpleConsensus,
- 0,
- CONF.getForceWalPeriodForConfigNodeSimpleInMs(),
- TimeUnit.MILLISECONDS);
- }
-
- private void flushWALForSimpleConsensus() {
- if (simpleLogWriter != null) {
- try {
- simpleLogWriter.force();
- } catch (IOException e) {
- LOGGER.error(
-
ConfigNodeMessages.CAN_T_FORCE_LOGWRITER_FOR_CONFIGNODE_FLUSHWALFORSIMPLECONSENSUS,
e);
+ if (isInProgressLogFile(logFileName)) {
+ sealRecoveredInProgressLogFile(logFile, recoveredStartIndex,
endIndex);
+ }
}
}
+ startIndex = endIndex + 1;
+ createLogFile(startIndex);
}
private void createLogFile(int startIndex) {
@@ -718,7 +787,56 @@ public class ConfigRegionStateMachine implements
IStateMachine, IStateMachine.Ev
}
}
- private static long parseEndIndex(String filename) {
+ private void sealRecoveredInProgressLogFile(
+ File logFile, int recoveredStartIndex, int recoveredEndIndex) {
+ try {
+ if (recoveredStartIndex > recoveredEndIndex) {
+ Files.deleteIfExists(logFile.toPath());
+ return;
+ }
+ Files.move(
+ logFile.toPath(),
+ new File(FILE_PATH + recoveredStartIndex + "_" +
recoveredEndIndex).toPath(),
+ StandardCopyOption.ATOMIC_MOVE);
+ } catch (IOException e) {
+ LOGGER.warn(
+
ConfigNodeMessages.SEAL_RECOVERED_CONFIGNODE_SIMPLECONSENSUS_LOG_FAILED,
logFile, e);
+ }
+ }
+
+ private boolean isInProgressLogFile(String filename) {
+ return filename.startsWith(LOG_INPROGRESS_FILE_PREFIX);
+ }
+
+ private boolean closeSimpleLogWriter() {
+ if (simpleLogWriter == null) {
+ return true;
+ }
+ for (int retry = 0; retry < 5; retry++) {
+ try {
+ simpleLogWriter.close();
+ simpleLogWriter = null;
+ return true;
+ } catch (IOException e) {
+ LOGGER.warn(
+ ConfigNodeMessages
+
.LOG_CONFIGREGION_UPDATE_FAILED_TO_CLOSE_THE_CONFIGNODE_SIMPLECONSENSUS_LOG_FILEPATH_ARG_RETRY_ARG_188A702E,
+ simpleLogFile == null ? null : simpleLogFile.getAbsolutePath(),
+ retry);
+ try {
+ TimeUnit.SECONDS.sleep(1);
+ } catch (InterruptedException e2) {
+ Thread.currentThread().interrupt();
+ LOGGER.warn(
+
ConfigNodeMessages.UNEXPECTED_INTERRUPTION_DURING_THE_CLOSE_METHOD_OF_LOGWRITER);
+ break;
+ }
+ }
+ }
+ return false;
+ }
+
+ static long parseEndIndex(String filename) {
final String endIndexString;
if (filename.startsWith(LOG_INPROGRESS_FILE_PREFIX)) {
endIndexString = filename.substring(LOG_INPROGRESS_FILE_PREFIX.length());
@@ -732,15 +850,36 @@ public class ConfigRegionStateMachine implements
IStateMachine, IStateMachine.Ev
return 0;
}
- if (endIndexString.isEmpty()) {
+ return isDigits(endIndexString) ? Long.parseLong(endIndexString) : 0;
+ }
+
+ static int parseStartIndex(String filename) {
+ final String startIndexString;
+ if (filename.startsWith(LOG_INPROGRESS_FILE_PREFIX)) {
+ startIndexString =
filename.substring(LOG_INPROGRESS_FILE_PREFIX.length());
+ } else if (filename.startsWith(LOG_FILE_PREFIX)) {
+ final int lastSeparatorIndex = filename.lastIndexOf('_');
+ if (lastSeparatorIndex <= LOG_FILE_PREFIX.length()) {
+ return 0;
+ }
+ startIndexString = filename.substring(LOG_FILE_PREFIX.length(),
lastSeparatorIndex);
+ } else {
return 0;
}
- for (int i = 0; i < endIndexString.length(); i++) {
- if (!Character.isDigit(endIndexString.charAt(i))) {
- return 0;
+
+ return isDigits(startIndexString) ? Integer.parseInt(startIndexString) : 0;
+ }
+
+ private static boolean isDigits(String value) {
+ if (value.isEmpty()) {
+ return false;
+ }
+ for (int i = 0; i < value.length(); i++) {
+ if (!Character.isDigit(value.charAt(i))) {
+ return false;
}
}
- return Long.parseLong(endIndexString);
+ return true;
}
/**
@@ -765,4 +904,33 @@ public class ConfigRegionStateMachine implements
IStateMachine, IStateMachine.Ev
startup.run();
}
}
+
+ private static class SimpleConsensusPersistResult {
+
+ private final TSStatus status;
+ private final File logFile;
+ private final long logFileSizeBeforeWrite;
+ private final int endIndexBeforeWrite;
+
+ private SimpleConsensusPersistResult(
+ TSStatus status, File logFile, long logFileSizeBeforeWrite, int
endIndexBeforeWrite) {
+ this.status = status;
+ this.logFile = logFile;
+ this.logFileSizeBeforeWrite = logFileSizeBeforeWrite;
+ this.endIndexBeforeWrite = endIndexBeforeWrite;
+ }
+
+ private static SimpleConsensusPersistResult success(
+ File logFile, long logFileSizeBeforeWrite, int endIndexBeforeWrite) {
+ return new SimpleConsensusPersistResult(
+ new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()),
+ logFile,
+ logFileSizeBeforeWrite,
+ endIndexBeforeWrite);
+ }
+
+ private static SimpleConsensusPersistResult failure(TSStatus status) {
+ return new SimpleConsensusPersistResult(status, null, 0, 0);
+ }
+ }
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
index c6790da284a..1b344229172 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.confignode.manager.node;
import org.apache.iotdb.common.rpc.thrift.TAINodeConfiguration;
+import org.apache.iotdb.common.rpc.thrift.TAINodeLocation;
import org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TDataNodeConfiguration;
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
@@ -46,6 +47,7 @@ import
org.apache.iotdb.confignode.client.sync.SyncConfigNodeClientPool;
import org.apache.iotdb.confignode.client.sync.SyncDataNodeClientPool;
import org.apache.iotdb.confignode.conf.ConfigNodeConfig;
import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
+import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan;
import
org.apache.iotdb.confignode.consensus.request.read.ainode.GetAINodeConfigurationPlan;
import
org.apache.iotdb.confignode.consensus.request.read.datanode.GetDataNodeConfigurationPlan;
import
org.apache.iotdb.confignode.consensus.request.write.ainode.RegisterAINodePlan;
@@ -337,30 +339,32 @@ public class NodeManager {
DataNodeRegisterResp resp = new DataNodeRegisterResp();
resp.setConfigNodeList(getRegisteredConfigNodes());
- // Create a new DataNodeHeartbeatCache and force update NodeStatus
int dataNodeId = nodeInfo.generateNextNodeId();
-
getLoadManager().getLoadCache().createNodeHeartbeatCache(NodeType.DataNode,
dataNodeId);
- // TODO: invoke a force heartbeat to update new DataNode's status
immediately
RegisterDataNodePlan registerDataNodePlan =
new RegisterDataNodePlan(req.getDataNodeConfiguration());
// Register new DataNode
registerDataNodePlan.getDataNodeConfiguration().getLocation().setDataNodeId(dataNodeId);
- try {
- getConsensusManager().write(registerDataNodePlan);
- } catch (ConsensusException e) {
- LOGGER.warn(CONSENSUS_WRITE_ERROR, e);
+ TSStatus registerStatus = writeConfigPhysicalPlan(registerDataNodePlan);
+ if (!isConsensusWriteSuccessful(registerStatus)) {
+ resp.setStatus(registerStatus);
+ return resp;
}
// update datanode's versionInfo
UpdateVersionInfoPlan updateVersionInfoPlan =
new UpdateVersionInfoPlan(req.getVersionInfo(), dataNodeId);
- try {
- getConsensusManager().write(updateVersionInfoPlan);
- } catch (ConsensusException e) {
- LOGGER.warn(CONSENSUS_WRITE_ERROR, e);
+ TSStatus updateVersionStatus =
writeConfigPhysicalPlan(updateVersionInfoPlan);
+ if (!isConsensusWriteSuccessful(updateVersionStatus)) {
+ resp.setStatus(
+ rollbackDataNodeRegistration(
+ registerDataNodePlan.getDataNodeConfiguration().getLocation(),
updateVersionStatus));
+ return resp;
}
+ // Create a new DataNodeHeartbeatCache. The heartbeat service will refresh
NodeStatus shortly.
+
getLoadManager().getLoadCache().createNodeHeartbeatCache(NodeType.DataNode,
dataNodeId);
+
// Bind DataNode metrics
PartitionMetrics.bindDataNodePartitionMetricsWhenUpdate(
MetricService.getInstance(), configManager, dataNodeId);
@@ -368,7 +372,10 @@ public class NodeManager {
// Adjust the maximum RegionGroup number of each Database
getClusterSchemaManager().adjustMaxRegionGroupNum();
- resp.setStatus(ClusterNodeStartUtils.ACCEPT_NODE_REGISTRATION);
+ resp.setStatus(
+ buildSuccessStatus(
+ ClusterNodeStartUtils.ACCEPT_NODE_REGISTRATION.getMessage(),
+ registerStatus.getMessage()));
resp.setDataNodeId(
registerDataNodePlan.getDataNodeConfiguration().getLocation().getDataNodeId());
resp.setRuntimeConfiguration(getRuntimeConfiguration(dataNodeId));
@@ -396,10 +403,10 @@ public class NodeManager {
// Update DataNodeConfiguration when modified during restart
UpdateDataNodePlan updateDataNodePlan =
new UpdateDataNodePlan(req.getDataNodeConfiguration());
- try {
- getConsensusManager().write(updateDataNodePlan);
- } catch (ConsensusException e) {
- LOGGER.warn(CONSENSUS_WRITE_ERROR, e);
+ TSStatus updateStatus = writeConfigPhysicalPlan(updateDataNodePlan);
+ if (!isConsensusWriteSuccessful(updateStatus)) {
+ resp.setStatus(updateStatus);
+ return resp;
}
}
TNodeVersionInfo versionInfo = nodeInfo.getVersionInfo(nodeId);
@@ -407,14 +414,14 @@ public class NodeManager {
// Update versionInfo when modified during restart
UpdateVersionInfoPlan updateVersionInfoPlan =
new UpdateVersionInfoPlan(req.getVersionInfo(), nodeId);
- try {
- getConsensusManager().write(updateVersionInfoPlan);
- } catch (ConsensusException e) {
- LOGGER.warn(CONSENSUS_WRITE_ERROR, e);
+ TSStatus updateStatus = writeConfigPhysicalPlan(updateVersionInfoPlan);
+ if (!isConsensusWriteSuccessful(updateStatus)) {
+ resp.setStatus(updateStatus);
+ return resp;
}
}
- resp.setStatus(ClusterNodeStartUtils.ACCEPT_NODE_RESTART);
+
resp.setStatus(buildSuccessStatus(ClusterNodeStartUtils.ACCEPT_NODE_RESTART.getMessage()));
resp.setRuntimeConfiguration(getRuntimeConfiguration(nodeId));
resp.setCorrectConsensusGroups(getPartitionManager().getAllReplicaSets(nodeId));
@@ -492,13 +499,12 @@ public class NodeManager {
// Update versionInfo when modified during restart
UpdateVersionInfoPlan updateConfigNodePlan =
new UpdateVersionInfoPlan(versionInfo, configNodeId);
- try {
- return getConsensusManager().write(updateConfigNodePlan);
- } catch (ConsensusException e) {
- return new
TSStatus(TSStatusCode.CONSENSUS_NOT_INITIALIZED.getStatusCode());
+ TSStatus updateStatus = writeConfigPhysicalPlan(updateConfigNodePlan);
+ if (!isConsensusWriteSuccessful(updateStatus)) {
+ return updateStatus;
}
}
- return ClusterNodeStartUtils.ACCEPT_NODE_RESTART;
+ return
buildSuccessStatus(ClusterNodeStartUtils.ACCEPT_NODE_RESTART.getMessage());
}
public List<TAINodeInfo> getRegisteredAINodeInfoList() {
@@ -544,27 +550,37 @@ public class NodeManager {
}
int aiNodeId = nodeInfo.generateNextNodeId();
- getLoadManager().getLoadCache().createNodeHeartbeatCache(NodeType.AINode,
aiNodeId);
RegisterAINodePlan registerAINodePlan = new
RegisterAINodePlan(req.getAiNodeConfiguration());
// Register new DataNode
registerAINodePlan.getAINodeConfiguration().getLocation().setAiNodeId(aiNodeId);
- try {
- getConsensusManager().write(registerAINodePlan);
- } catch (ConsensusException e) {
- LOGGER.warn(CONSENSUS_WRITE_ERROR, e);
+ TSStatus registerStatus = writeConfigPhysicalPlan(registerAINodePlan);
+ if (!isConsensusWriteSuccessful(registerStatus)) {
+ AINodeRegisterResp resp = new AINodeRegisterResp();
+ resp.setConfigNodeList(getRegisteredConfigNodes());
+ resp.setStatus(registerStatus);
+ return resp;
}
// update datanode's versionInfo
UpdateVersionInfoPlan updateVersionInfoPlan =
new UpdateVersionInfoPlan(req.getVersionInfo(), aiNodeId);
- try {
- getConsensusManager().write(updateVersionInfoPlan);
- } catch (ConsensusException e) {
- LOGGER.warn(CONSENSUS_WRITE_ERROR, e);
+ TSStatus updateVersionStatus =
writeConfigPhysicalPlan(updateVersionInfoPlan);
+ if (!isConsensusWriteSuccessful(updateVersionStatus)) {
+ AINodeRegisterResp resp = new AINodeRegisterResp();
+ resp.setConfigNodeList(getRegisteredConfigNodes());
+ resp.setStatus(
+ rollbackAiNodeRegistration(
+ registerAINodePlan.getAINodeConfiguration().getLocation(),
updateVersionStatus));
+ return resp;
}
+ getLoadManager().getLoadCache().createNodeHeartbeatCache(NodeType.AINode,
aiNodeId);
+
AINodeRegisterResp resp = new AINodeRegisterResp();
- resp.setStatus(ClusterNodeStartUtils.ACCEPT_NODE_REGISTRATION);
+ resp.setStatus(
+ buildSuccessStatus(
+ ClusterNodeStartUtils.ACCEPT_NODE_REGISTRATION.getMessage(),
+ registerStatus.getMessage()));
resp.setConfigNodeList(getRegisteredConfigNodes());
resp.setAINodeId(registerAINodePlan.getAINodeConfiguration().getLocation().getAiNodeId());
return resp;
@@ -602,10 +618,12 @@ public class NodeManager {
if (!req.getAiNodeConfiguration().equals(aiNodeConfiguration)) {
// Update AINodeConfiguration when modified during restart
UpdateAINodePlan updateAINodePlan = new
UpdateAINodePlan(req.getAiNodeConfiguration());
- try {
- getConsensusManager().write(updateAINodePlan);
- } catch (ConsensusException e) {
- LOGGER.warn(CONSENSUS_WRITE_ERROR, e);
+ TSStatus updateStatus = writeConfigPhysicalPlan(updateAINodePlan);
+ if (!isConsensusWriteSuccessful(updateStatus)) {
+ TAINodeRestartResp resp = new TAINodeRestartResp();
+ resp.setConfigNodeList(getRegisteredConfigNodes());
+ resp.setStatus(updateStatus);
+ return resp;
}
}
TNodeVersionInfo versionInfo = nodeInfo.getVersionInfo(nodeId);
@@ -613,15 +631,17 @@ public class NodeManager {
// Update versionInfo when modified during restart
UpdateVersionInfoPlan updateVersionInfoPlan =
new UpdateVersionInfoPlan(req.getVersionInfo(), nodeId);
- try {
- getConsensusManager().write(updateVersionInfoPlan);
- } catch (ConsensusException e) {
- LOGGER.warn(CONSENSUS_WRITE_ERROR, e);
+ TSStatus updateStatus = writeConfigPhysicalPlan(updateVersionInfoPlan);
+ if (!isConsensusWriteSuccessful(updateStatus)) {
+ TAINodeRestartResp resp = new TAINodeRestartResp();
+ resp.setConfigNodeList(getRegisteredConfigNodes());
+ resp.setStatus(updateStatus);
+ return resp;
}
}
TAINodeRestartResp resp = new TAINodeRestartResp();
- resp.setStatus(ClusterNodeStartUtils.ACCEPT_NODE_RESTART);
+
resp.setStatus(buildSuccessStatus(ClusterNodeStartUtils.ACCEPT_NODE_RESTART.getMessage()));
resp.setConfigNodeList(getRegisteredConfigNodes());
return resp;
}
@@ -916,21 +936,32 @@ public class NodeManager {
*
* @param configNodeLocation The new ConfigNode.
* @param versionInfo The new ConfigNode's versionInfo.
+ * @throws IllegalStateException if the ConfigNode information or version
cannot be persisted
*/
public void applyConfigNode(
TConfigNodeLocation configNodeLocation, TNodeVersionInfo versionInfo) {
ApplyConfigNodePlan applyConfigNodePlan = new
ApplyConfigNodePlan(configNodeLocation);
- try {
- getConsensusManager().write(applyConfigNodePlan);
- } catch (ConsensusException e) {
- LOGGER.warn(CONSENSUS_WRITE_ERROR, e);
+ final TSStatus applyStatus = writeConfigPhysicalPlan(applyConfigNodePlan);
+ if (!isConsensusWriteSuccessful(applyStatus)) {
+ throw new IllegalStateException(
+ String.format(
+ ConfigNodeMessages
+
.EXCEPTION_FAILED_TO_APPLY_CONFIGNODE_ARG_THROUGH_CONSENSUS_LAYER_ARG_B4A985D7,
+ configNodeLocation,
+ describeStatus(applyStatus)));
}
UpdateVersionInfoPlan updateVersionInfoPlan =
new UpdateVersionInfoPlan(versionInfo,
configNodeLocation.getConfigNodeId());
- try {
- getConsensusManager().write(updateVersionInfoPlan);
- } catch (ConsensusException e) {
- LOGGER.warn(CONSENSUS_WRITE_ERROR, e);
+ final TSStatus updateStatus =
writeConfigPhysicalPlan(updateVersionInfoPlan);
+ if (!isConsensusWriteSuccessful(updateStatus)) {
+ // ADD_PEER has already joined this node to Raft. Removing it here would
also mark the
+ // joining node as Removing. Keep the membership and let
REGISTER_SUCCESS retry both plans.
+ throw new IllegalStateException(
+ String.format(
+ ConfigNodeMessages
+
.EXCEPTION_FAILED_TO_PERSIST_VERSION_INFO_FOR_CONFIGNODE_ARG_ARG_REGISTRATION_REMAINS_PENDING_AND_MUST_BE_RETRIED_F662C350,
+ configNodeLocation.getConfigNodeId(),
+ describeStatus(updateStatus)));
}
}
@@ -1003,6 +1034,7 @@ public class NodeManager {
.setMessage(ManagerMessages.SUCCESSFULLY_REMOVE_CONFIGNODE);
}
+ @SuppressWarnings("checkstyle:LineLength")
private TSStatus transferLeader(
RemoveConfigNodePlan removeConfigNodePlan, ConsensusGroupId groupId) {
final TConfigNodeLocation removedConfigNode =
removeConfigNodePlan.getConfigNodeLocation();
@@ -1010,7 +1042,7 @@ public class NodeManager {
filterConfigNodeThroughStatus(NodeStatus.Running).stream()
.filter(configNode -> !configNode.equals(removedConfigNode))
.sorted(Comparator.comparingInt(TConfigNodeLocation::getConfigNodeId))
- .collect(Collectors.toList());
+ .toList();
if (newLeaderCandidates.isEmpty()) {
return new TSStatus(TSStatusCode.TRANSFER_LEADER_ERROR.getStatusCode())
.setMessage(
@@ -1385,6 +1417,112 @@ public class NodeManager {
return getRegisteredDataNode(dataNodeId).getLocation();
}
+ private TSStatus writeConfigPhysicalPlan(ConfigPhysicalPlan plan) {
+ try {
+ return getConsensusManager().write(plan);
+ } catch (ConsensusException e) {
+ LOGGER.warn(CONSENSUS_WRITE_ERROR, e);
+ return new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode())
+ .setMessage(e.getMessage());
+ }
+ }
+
+ private boolean isConsensusWriteSuccessful(TSStatus status) {
+ return status != null && status.getCode() ==
TSStatusCode.SUCCESS_STATUS.getStatusCode();
+ }
+
+ private TSStatus rollbackDataNodeRegistration(
+ TDataNodeLocation dataNodeLocation, TSStatus versionUpdateStatus) {
+ final TSStatus rollbackStatus =
+ writeConfigPhysicalPlan(
+ new
RemoveDataNodePlan(Collections.singletonList(dataNodeLocation)));
+ if (isConsensusWriteSuccessful(rollbackStatus)) {
+ return new TSStatus(versionUpdateStatus.getCode())
+ .setMessage(
+ String.format(
+ ConfigNodeMessages
+
.MESSAGE_FAILED_TO_PERSIST_VERSION_INFO_FOR_DATANODE_ARG_ARG_THE_REGISTRATION_HAS_BEEN_ROLLED_BACK_PLEASE_RETRY_THE_REGISTRATION_5DDB311F,
+ dataNodeLocation.getDataNodeId(),
+ describeStatus(versionUpdateStatus)));
+ }
+
+ LOGGER.error(
+ ConfigNodeMessages
+
.LOG_FAILED_TO_ROLL_BACK_DATANODE_REGISTRATION_ARG_AFTER_VERSION_INFO_PERSISTENCE_FAILURE_VERSIONUPDATESTATUS_ARG_ROLLBACKSTATUS_ARG_437E6F09,
+ dataNodeLocation,
+ versionUpdateStatus,
+ rollbackStatus);
+ return new TSStatus(rollbackStatus.getCode())
+ .setMessage(
+ String.format(
+ ConfigNodeMessages
+
.MESSAGE_FAILED_TO_PERSIST_VERSION_INFO_FOR_DATANODE_ARG_ARG_THE_REGISTRATION_ROLLBACK_ALSO_FAILED_ARG_MANUAL_CLEANUP_MAY_BE_REQUIRED_BEFORE_RETRYING_THE_REGISTRATION_5463B5FF,
+ dataNodeLocation.getDataNodeId(),
+ describeStatus(versionUpdateStatus),
+ describeStatus(rollbackStatus)));
+ }
+
+ private TSStatus rollbackAiNodeRegistration(
+ TAINodeLocation aiNodeLocation, TSStatus versionUpdateStatus) {
+ final TSStatus rollbackStatus = writeConfigPhysicalPlan(new
RemoveAINodePlan(aiNodeLocation));
+ if (isConsensusWriteSuccessful(rollbackStatus)) {
+ return new TSStatus(versionUpdateStatus.getCode())
+ .setMessage(
+ String.format(
+ ConfigNodeMessages
+
.MESSAGE_FAILED_TO_PERSIST_VERSION_INFO_FOR_AINODE_ARG_ARG_THE_REGISTRATION_HAS_BEEN_ROLLED_BACK_PLEASE_RETRY_THE_REGISTRATION_7571C626,
+ aiNodeLocation.getAiNodeId(),
+ describeStatus(versionUpdateStatus)));
+ }
+
+ LOGGER.error(
+ ConfigNodeMessages
+
.LOG_FAILED_TO_ROLL_BACK_AINODE_REGISTRATION_ARG_AFTER_VERSION_INFO_PERSISTENCE_FAILURE_VERSIONUPDATESTATUS_ARG_ROLLBACKSTATUS_ARG_BFF24BD4,
+ aiNodeLocation,
+ versionUpdateStatus,
+ rollbackStatus);
+ return new TSStatus(rollbackStatus.getCode())
+ .setMessage(
+ String.format(
+ ConfigNodeMessages
+
.MESSAGE_FAILED_TO_PERSIST_VERSION_INFO_FOR_AINODE_ARG_ARG_THE_REGISTRATION_ROLLBACK_ALSO_FAILED_ARG_MANUAL_CLEANUP_MAY_BE_REQUIRED_BEFORE_RETRYING_THE_REGISTRATION_B9B1DCCD,
+ aiNodeLocation.getAiNodeId(),
+ describeStatus(versionUpdateStatus),
+ describeStatus(rollbackStatus)));
+ }
+
+ private TSStatus buildStatus(int statusCode, String... messages) {
+ final TSStatus status = new TSStatus(statusCode);
+ final StringBuilder builder = new StringBuilder();
+ for (String message : messages) {
+ if (message == null || message.isEmpty()) {
+ continue;
+ }
+ if (!builder.isEmpty()) {
+ builder.append(' ');
+ }
+ builder.append(message);
+ }
+ if (!builder.isEmpty()) {
+ status.setMessage(builder.toString());
+ }
+ return status;
+ }
+
+ private TSStatus buildSuccessStatus(String... messages) {
+ return buildStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode(), messages);
+ }
+
+ private String describeStatus(TSStatus status) {
+ if (status == null) {
+ return ConfigNodeMessages.MESSAGE_UNKNOWN_ERROR_E9EEAFF8;
+ }
+ if (status.getMessage() != null && !status.getMessage().isEmpty()) {
+ return status.getMessage();
+ }
+ return String.format(ConfigNodeMessages.MESSAGE_STATUS_CODE_ARG_4B3BCD8B,
status.getCode());
+ }
+
private ConsensusManager getConsensusManager() {
return configManager.getConsensusManager();
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/DatabasePartitionTable.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/DatabasePartitionTable.java
index d3528c40ef8..3e3170af19f 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/DatabasePartitionTable.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/DatabasePartitionTable.java
@@ -245,7 +245,7 @@ public class DatabasePartitionTable {
result.getAndIncrement();
}
});
- return result.getAndIncrement();
+ return result.get();
}
/**
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/PartitionInfo.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/PartitionInfo.java
index 3c990a89578..900ab7ab4ae 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/PartitionInfo.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/PartitionInfo.java
@@ -244,7 +244,9 @@ public class PartitionInfo implements SnapshotProcessor {
*/
public TSStatus pollRegionMaintainTask() {
synchronized (regionMaintainTaskList) {
- regionMaintainTaskList.remove(0);
+ if (!regionMaintainTaskList.isEmpty()) {
+ regionMaintainTaskList.remove(0);
+ }
return RpcUtils.SUCCESS_STATUS;
}
}
@@ -1020,9 +1022,14 @@ public class PartitionInfo implements SnapshotProcessor {
databasePartitionTableEntry.getValue().serialize(bufferedOutputStream,
protocol);
}
+ final List<RegionMaintainTask> copiedRegionMaintainTaskList;
+ synchronized (regionMaintainTaskList) {
+ copiedRegionMaintainTaskList = new ArrayList<>(regionMaintainTaskList);
+ }
+
// serialize regionCleanList
- ReadWriteIOUtils.write(regionMaintainTaskList.size(),
bufferedOutputStream);
- for (RegionMaintainTask task : regionMaintainTaskList) {
+ ReadWriteIOUtils.write(copiedRegionMaintainTaskList.size(),
bufferedOutputStream);
+ for (RegionMaintainTask task : copiedRegionMaintainTaskList) {
task.serialize(bufferedOutputStream, protocol);
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/AddConfigNodeProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/AddConfigNodeProcedure.java
index a3a5ffb41de..5895f004717 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/AddConfigNodeProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/AddConfigNodeProcedure.java
@@ -81,9 +81,9 @@ public class AddConfigNodeProcedure extends
AbstractNodeProcedure<AddConfigNodeS
LOG.info(ProcedureMessages.SUCCESSFULLY_ADD_PEER,
tConfigNodeLocation);
break;
case REGISTER_SUCCESS:
- env.notifyRegisterSuccess(tConfigNodeLocation);
-
env.createConfigNodeHeartbeatCache(tConfigNodeLocation.getConfigNodeId());
env.applyConfigNode(tConfigNodeLocation, versionInfo);
+
env.createConfigNodeHeartbeatCache(tConfigNodeLocation.getConfigNodeId());
+ env.notifyRegisterSuccess(tConfigNodeLocation);
LOG.info(
ProcedureMessages.THE_CONFIGNODE_IS_SUCCESSFULLY_ADDED_TO_THE_CLUSTER,
tConfigNodeLocation);
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachineTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachineTest.java
new file mode 100644
index 00000000000..1df8d658531
--- /dev/null
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachineTest.java
@@ -0,0 +1,302 @@
+/*
+ * 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.consensus.statemachine;
+
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.confignode.conf.ConfigNodeConfig;
+import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
+import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan;
+import org.apache.iotdb.confignode.consensus.request.TestOnlyPlan;
+import
org.apache.iotdb.confignode.consensus.request.write.confignode.UpdateVersionInfoPlan;
+import org.apache.iotdb.confignode.i18n.ConfigNodeMessages;
+import org.apache.iotdb.confignode.persistence.executor.ConfigPlanExecutor;
+import org.apache.iotdb.confignode.rpc.thrift.TNodeVersionInfo;
+import org.apache.iotdb.confignode.writelog.io.SingleFileLogReader;
+import org.apache.iotdb.consensus.ConsensusFactory;
+import org.apache.iotdb.db.utils.writelog.LogWriter;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.List;
+
+public class ConfigRegionStateMachineTest {
+
+ @Test
+ public void testParseStartIndex() {
+ Assert.assertEquals(1,
ConfigRegionStateMachine.parseStartIndex("log_1_10"));
+ Assert.assertEquals(11,
ConfigRegionStateMachine.parseStartIndex("log_11_20"));
+ Assert.assertEquals(21,
ConfigRegionStateMachine.parseStartIndex("log_inprogress_21"));
+ Assert.assertEquals(0,
ConfigRegionStateMachine.parseStartIndex("invalid"));
+ }
+
+ @Test
+ public void testParseEndIndex() {
+ Assert.assertEquals(10,
ConfigRegionStateMachine.parseEndIndex("log_1_10"));
+ Assert.assertEquals(20,
ConfigRegionStateMachine.parseEndIndex("log_11_20"));
+ Assert.assertEquals(21,
ConfigRegionStateMachine.parseEndIndex("log_inprogress_21"));
+ Assert.assertEquals(0, ConfigRegionStateMachine.parseEndIndex("invalid"));
+ }
+
+ @Test
+ public void testLogFilesSortByEndIndex() {
+ List<String> filenames =
+ new ArrayList<>(Arrays.asList("log_inprogress_21", "log_11_20",
"log_1_10"));
+
+
filenames.sort(Comparator.comparingLong(ConfigRegionStateMachine::parseEndIndex));
+
+ Assert.assertEquals(Arrays.asList("log_1_10", "log_11_20",
"log_inprogress_21"), filenames);
+ }
+
+ @Test
+ public void testFailedSimpleConsensusWriteRollsBackPersistedPlan() throws
Exception {
+ ConfigNodeConfig conf = ConfigNodeDescriptor.getInstance().getConf();
+ String originalConsensusProtocol =
conf.getConfigNodeConsensusProtocolClass();
+ Path tempLogFile = Files.createTempFile("confignode-simple-consensus",
".wal");
+ ConfigPlanExecutor executor = Mockito.mock(ConfigPlanExecutor.class);
+
Mockito.when(executor.executeNonQueryPlan(Mockito.any(ConfigPhysicalPlan.class)))
+ .thenReturn(new
TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode()));
+ ConfigRegionStateMachine stateMachine = new ConfigRegionStateMachine(null,
executor);
+
+ try {
+
conf.setConfigNodeConsensusProtocolClass(ConsensusFactory.SIMPLE_CONSENSUS);
+ setField(stateMachine, "simpleLogFile", tempLogFile.toFile());
+ setField(stateMachine, "simpleLogWriter", new
LogWriter(tempLogFile.toFile(), false));
+ setField(stateMachine, "endIndex", 0);
+
+ TSStatus status = stateMachine.write(new TestOnlyPlan());
+
+
Assert.assertEquals(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode(),
status.getCode());
+ Assert.assertEquals(0, Files.size(tempLogFile));
+ Assert.assertEquals(0, getField(stateMachine, "endIndex"));
+ closeSimpleLogWriter(stateMachine);
+ assertNoReplayablePlans(tempLogFile);
+ } finally {
+ closeSimpleLogWriter(stateMachine);
+ Files.deleteIfExists(tempLogFile);
+ conf.setConfigNodeConsensusProtocolClass(originalConsensusProtocol);
+ }
+ }
+
+ private static void setField(Object target, String fieldName, Object value)
throws Exception {
+ Field field = target.getClass().getDeclaredField(fieldName);
+ field.setAccessible(true);
+ field.set(target, value);
+ }
+
+ @Test
+ public void testForceFailureRollsBackAppendedPlan() throws Exception {
+ assertPersistenceFailureRollsBack(true);
+ }
+
+ @Test
+ public void testPartialAppendFailureRollsBackPlan() throws Exception {
+ assertPersistenceFailureRollsBack(false);
+ }
+
+ private void assertPersistenceFailureRollsBack(boolean failOnForce) throws
Exception {
+ ConfigNodeConfig conf = ConfigNodeDescriptor.getInstance().getConf();
+ String originalConsensusProtocol =
conf.getConfigNodeConsensusProtocolClass();
+ Path logFile = Files.createTempFile("confignode-persist-failure", ".wal");
+ ConfigPlanExecutor executor = Mockito.mock(ConfigPlanExecutor.class);
+
Mockito.when(executor.executeNonQueryPlan(Mockito.any(ConfigPhysicalPlan.class)))
+ .thenReturn(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+ ConfigRegionStateMachine stateMachine = new ConfigRegionStateMachine(null,
executor);
+ LogWriter writer = Mockito.spy(new LogWriter(logFile.toFile(), false));
+ ConfigPhysicalPlan firstPlan = versionPlan(1);
+ ConfigPhysicalPlan failedPlan = versionPlan(2);
+ ConfigPhysicalPlan retryPlan = versionPlan(3);
+ try {
+
conf.setConfigNodeConsensusProtocolClass(ConsensusFactory.SIMPLE_CONSENSUS);
+ setField(stateMachine, "simpleLogFile", logFile.toFile());
+ setField(stateMachine, "simpleLogWriter", writer);
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
stateMachine.write(firstPlan).getCode());
+ long sizeBeforeWrite = Files.size(logFile);
+ if (failOnForce) {
+ Mockito.doThrow(new IOException("injected force
failure")).when(writer).force();
+ } else {
+ Mockito.doAnswer(
+ invocation -> {
+ Files.write(logFile, new byte[] {0, 0, 0},
StandardOpenOption.APPEND);
+ throw new IOException("injected partial append failure");
+ })
+ .when(writer)
+ .write(Mockito.any(ByteBuffer.class));
+ }
+
+ TSStatus status = stateMachine.write(failedPlan);
+
+
Assert.assertEquals(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode(),
status.getCode());
+ Mockito.verify(executor,
Mockito.never()).executeNonQueryPlan(failedPlan);
+ Assert.assertEquals(sizeBeforeWrite, Files.size(logFile));
+ Assert.assertEquals(1, getField(stateMachine, "endIndex"));
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
stateMachine.write(retryPlan).getCode());
+ Assert.assertEquals(2, getField(stateMachine, "endIndex"));
+ closeSimpleLogWriter(stateMachine);
+ assertReplayablePlans(logFile, firstPlan, retryPlan);
+ } finally {
+ closeSimpleLogWriter(stateMachine);
+ Files.deleteIfExists(logFile);
+ conf.setConfigNodeConsensusProtocolClass(originalConsensusProtocol);
+ }
+ }
+
+ @Test
+ public void testCleanupFailureDisablesFurtherWrites() throws Exception {
+ ConfigNodeConfig conf = ConfigNodeDescriptor.getInstance().getConf();
+ String originalConsensusProtocol =
conf.getConfigNodeConsensusProtocolClass();
+ Path logFile = Files.createTempFile("confignode-cleanup-failure", ".wal");
+ ConfigPlanExecutor executor = Mockito.mock(ConfigPlanExecutor.class);
+ ConfigRegionStateMachine stateMachine = new ConfigRegionStateMachine(null,
executor);
+ LogWriter writer = Mockito.spy(new LogWriter(logFile.toFile(), false));
+ try {
+
conf.setConfigNodeConsensusProtocolClass(ConsensusFactory.SIMPLE_CONSENSUS);
+ setField(stateMachine, "simpleLogFile", logFile.toFile());
+ setField(stateMachine, "simpleLogWriter", writer);
+ Mockito.doThrow(new IOException("injected force
failure")).when(writer).force();
+ Mockito.doThrow(new IOException("injected close
failure")).when(writer).close();
+
+ TSStatus status = stateMachine.write(versionPlan(1));
+
+
Assert.assertEquals(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode(),
status.getCode());
+ Assert.assertTrue(
+ status
+ .getMessage()
+ .contains(
+
ConfigNodeMessages.FAILED_TO_ROLLBACK_PERSISTED_CONFIGNODE_SIMPLECONSENSUS_LOG));
+ long failedLogSize = Files.size(logFile);
+ Assert.assertTrue(failedLogSize > 0);
+ Assert.assertEquals(
+ TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode(),
+ stateMachine.write(versionPlan(2)).getCode());
+ Assert.assertEquals(failedLogSize, Files.size(logFile));
+ Mockito.verify(writer,
Mockito.times(1)).write(Mockito.any(ByteBuffer.class));
+ Mockito.verify(executor, Mockito.never())
+ .executeNonQueryPlan(Mockito.any(ConfigPhysicalPlan.class));
+ } finally {
+ Mockito.doCallRealMethod().when(writer).close();
+ closeSimpleLogWriter(stateMachine);
+ Files.deleteIfExists(logFile);
+ conf.setConfigNodeConsensusProtocolClass(originalConsensusProtocol);
+ }
+ }
+
+ @Test
+ public void testRotationRenameFailureReopensOriginalLog() throws Exception {
+ ConfigNodeConfig conf = ConfigNodeDescriptor.getInstance().getConf();
+ String originalConsensusProtocol =
conf.getConfigNodeConsensusProtocolClass();
+ Path directory = Files.createTempDirectory("confignode-rotation-failure");
+ Path logFile = directory.resolve("log_inprogress_1");
+ Path conflictingTarget =
Files.createDirectory(directory.resolve("log_1_1"));
+ Path blocker = Files.createFile(conflictingTarget.resolve("blocker"));
+ ConfigPlanExecutor executor = Mockito.mock(ConfigPlanExecutor.class);
+
Mockito.when(executor.executeNonQueryPlan(Mockito.any(ConfigPhysicalPlan.class)))
+ .thenReturn(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+ ConfigRegionStateMachine stateMachine = new ConfigRegionStateMachine(null,
executor);
+ ConfigPhysicalPlan firstPlan = versionPlan(1);
+ ConfigPhysicalPlan retryPlan = versionPlan(2);
+ try {
+
conf.setConfigNodeConsensusProtocolClass(ConsensusFactory.SIMPLE_CONSENSUS);
+ setField(stateMachine, "simpleLogFile", logFile.toFile());
+ setField(stateMachine, "simpleLogWriter", new
LogWriter(logFile.toFile(), false));
+ setField(stateMachine, "startIndex", 1);
+ stateMachine.write(firstPlan);
+ Method rotate =
+
ConfigRegionStateMachine.class.getDeclaredMethod("rollSimpleConsensusLogFile");
+ rotate.setAccessible(true);
+ try {
+ rotate.invoke(stateMachine);
+ Assert.fail("Expected the conflicting target to prevent renaming");
+ } catch (InvocationTargetException e) {
+ Assert.assertTrue(e.getCause() instanceof IOException);
+ }
+
+ Assert.assertNotNull(getField(stateMachine, "simpleLogWriter"));
+ Assert.assertEquals(1, getField(stateMachine, "startIndex"));
+ Assert.assertEquals(1, getField(stateMachine, "endIndex"));
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
stateMachine.write(retryPlan).getCode());
+ closeSimpleLogWriter(stateMachine);
+ assertReplayablePlans(logFile, firstPlan, retryPlan);
+ } finally {
+ closeSimpleLogWriter(stateMachine);
+ Files.deleteIfExists(logFile);
+ Files.deleteIfExists(blocker);
+ Files.deleteIfExists(conflictingTarget);
+ Files.deleteIfExists(directory);
+ conf.setConfigNodeConsensusProtocolClass(originalConsensusProtocol);
+ }
+ }
+
+ private static ConfigPhysicalPlan versionPlan(int nodeId) {
+ return new UpdateVersionInfoPlan(new TNodeVersionInfo("version", "build"),
nodeId);
+ }
+
+ private static void assertReplayablePlans(Path logFile,
ConfigPhysicalPlan... plans)
+ throws Exception {
+ SingleFileLogReader reader = new SingleFileLogReader(logFile.toFile());
+ try {
+ for (ConfigPhysicalPlan plan : plans) {
+ Assert.assertTrue(reader.hasNext());
+ Assert.assertEquals(plan, reader.next());
+ }
+ Assert.assertFalse(reader.hasNext());
+ } finally {
+ reader.close();
+ }
+ }
+
+ private static Object getField(Object target, String fieldName) throws
Exception {
+ Field field = target.getClass().getDeclaredField(fieldName);
+ field.setAccessible(true);
+ return field.get(target);
+ }
+
+ private static void assertNoReplayablePlans(Path tempLogFile) throws
Exception {
+ SingleFileLogReader logReader = new
SingleFileLogReader(tempLogFile.toFile());
+ try {
+ Assert.assertFalse(logReader.hasNext());
+ } finally {
+ logReader.close();
+ }
+ }
+
+ private static void closeSimpleLogWriter(ConfigRegionStateMachine
stateMachine) throws Exception {
+ LogWriter logWriter = (LogWriter) getField(stateMachine,
"simpleLogWriter");
+ if (logWriter != null) {
+ logWriter.close();
+ }
+ }
+}
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/node/NodeManagerRegistrationTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/node/NodeManagerRegistrationTest.java
new file mode 100644
index 00000000000..5356be0494f
--- /dev/null
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/node/NodeManagerRegistrationTest.java
@@ -0,0 +1,260 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.confignode.manager.node;
+
+import org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TDataNodeConfiguration;
+import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.cluster.NodeStatus;
+import org.apache.iotdb.commons.cluster.NodeType;
+import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.confignode.conf.ConfigNodeConfig;
+import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
+import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan;
+import
org.apache.iotdb.confignode.consensus.request.write.confignode.ApplyConfigNodePlan;
+import
org.apache.iotdb.confignode.consensus.request.write.confignode.RemoveConfigNodePlan;
+import
org.apache.iotdb.confignode.consensus.request.write.confignode.UpdateVersionInfoPlan;
+import
org.apache.iotdb.confignode.consensus.request.write.datanode.RegisterDataNodePlan;
+import
org.apache.iotdb.confignode.consensus.request.write.datanode.RemoveDataNodePlan;
+import
org.apache.iotdb.confignode.consensus.response.datanode.DataNodeRegisterResp;
+import org.apache.iotdb.confignode.i18n.ConfigNodeMessages;
+import org.apache.iotdb.confignode.manager.ClusterManager;
+import org.apache.iotdb.confignode.manager.IManager;
+import org.apache.iotdb.confignode.manager.consensus.ConsensusManager;
+import org.apache.iotdb.confignode.manager.load.LoadManager;
+import org.apache.iotdb.confignode.manager.load.cache.LoadCache;
+import org.apache.iotdb.confignode.manager.schema.ClusterSchemaManager;
+import org.apache.iotdb.confignode.persistence.node.NodeInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRegisterReq;
+import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRestartReq;
+import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRestartResp;
+import org.apache.iotdb.confignode.rpc.thrift.TNodeVersionInfo;
+import org.apache.iotdb.consensus.exception.ConsensusException;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+
+import java.util.Collections;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class NodeManagerRegistrationTest {
+
+ private IManager configManager;
+ private ConsensusManager consensusManager;
+ private LoadManager loadManager;
+ private LoadCache loadCache;
+ private ClusterSchemaManager clusterSchemaManager;
+ private ClusterManager clusterManager;
+ private NodeInfo nodeInfo;
+ private NodeManager nodeManager;
+
+ @Before
+ public void setUp() {
+ configManager = Mockito.mock(IManager.class);
+ consensusManager = Mockito.mock(ConsensusManager.class);
+ loadManager = Mockito.mock(LoadManager.class);
+ loadCache = Mockito.mock(LoadCache.class);
+ clusterSchemaManager = Mockito.mock(ClusterSchemaManager.class);
+ clusterManager = Mockito.mock(ClusterManager.class);
+ nodeInfo = new NodeInfo();
+ nodeManager = new NodeManager(configManager, nodeInfo);
+
+ when(configManager.getConsensusManager()).thenReturn(consensusManager);
+ when(configManager.getLoadManager()).thenReturn(loadManager);
+ when(loadManager.getLoadCache()).thenReturn(loadCache);
+
when(configManager.getClusterSchemaManager()).thenReturn(clusterSchemaManager);
+ when(configManager.getClusterManager()).thenReturn(clusterManager);
+ }
+
+ @Test
+ public void testRegisterDataNodeStopsWhenRegisterWriteFails() throws
ConsensusException {
+ TSStatus failureStatus =
+ new
TSStatus(TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()).setMessage("redirect");
+ when(consensusManager.write(any())).thenReturn(failureStatus);
+
+ DataNodeRegisterResp resp =
+ (DataNodeRegisterResp)
nodeManager.registerDataNode(generateDataNodeRegisterReq(1));
+
+ Assert.assertEquals(failureStatus, resp.getStatus());
+ verify(loadCache, never()).createNodeHeartbeatCache(eq(NodeType.DataNode),
anyInt());
+ verify(clusterSchemaManager, never()).adjustMaxRegionGroupNum();
+ }
+
+ @Test
+ public void testRegisterDataNodeRollsBackWhenVersionWriteFails() throws
ConsensusException {
+ TSStatus successStatus = new
TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ TSStatus failureStatus =
+ new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode())
+ .setMessage("update failed");
+ when(consensusManager.write(any())).thenReturn(successStatus,
failureStatus, successStatus);
+
+ DataNodeRegisterResp resp =
+ (DataNodeRegisterResp)
nodeManager.registerDataNode(generateDataNodeRegisterReq(1));
+
+ Assert.assertEquals(
+ TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode(),
resp.getStatus().getCode());
+ Assert.assertEquals(
+ String.format(
+ ConfigNodeMessages
+
.MESSAGE_FAILED_TO_PERSIST_VERSION_INFO_FOR_DATANODE_ARG_ARG_THE_REGISTRATION_HAS_BEEN_ROLLED_BACK_PLEASE_RETRY_THE_REGISTRATION_5DDB311F,
+ 0,
+ failureStatus.getMessage()),
+ resp.getStatus().getMessage());
+ verify(loadCache, never()).createNodeHeartbeatCache(eq(NodeType.DataNode),
anyInt());
+ verify(clusterSchemaManager, never()).adjustMaxRegionGroupNum();
+
+ ArgumentCaptor<ConfigPhysicalPlan> planCaptor =
+ ArgumentCaptor.forClass(ConfigPhysicalPlan.class);
+ verify(consensusManager, Mockito.times(3)).write(planCaptor.capture());
+ Assert.assertTrue(planCaptor.getAllValues().get(0) instanceof
RegisterDataNodePlan);
+ Assert.assertTrue(planCaptor.getAllValues().get(1) instanceof
UpdateVersionInfoPlan);
+ Assert.assertTrue(planCaptor.getAllValues().get(2) instanceof
RemoveDataNodePlan);
+ }
+
+ @Test
+ public void testRestartDataNodeReturnsFailureWhenUpdateWriteFails() throws
ConsensusException {
+ final TDataNodeConfiguration registeredConfig =
generateDataNodeConfiguration(1, "127.0.0.1");
+ nodeInfo.registerDataNode(new RegisterDataNodePlan(registeredConfig));
+
when(clusterManager.getClusterIdWithRetry(anyLong())).thenReturn("cluster");
+
+ TSStatus failureStatus =
+ new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode())
+ .setMessage("update failed");
+ when(consensusManager.write(any())).thenReturn(failureStatus);
+
+ final TDataNodeRestartReq req = new TDataNodeRestartReq();
+ req.setDataNodeConfiguration(generateDataNodeConfiguration(1,
"127.0.0.2"));
+ req.setVersionInfo(new TNodeVersionInfo("version", "build"));
+
+ final TDataNodeRestartResp resp =
nodeManager.updateDataNodeIfNecessary(req);
+
+ Assert.assertEquals(failureStatus, resp.getStatus());
+ }
+
+ @Test
+ public void testApplyConfigNodeRetainsMembershipAndRetriesVersionWrite()
+ throws ConsensusException {
+ TSStatus failureStatus =
+ new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode())
+ .setMessage("version update failed");
+ AtomicBoolean failVersionUpdate = new AtomicBoolean(true);
+ when(consensusManager.write(any()))
+ .thenAnswer(
+ invocation -> {
+ ConfigPhysicalPlan plan = invocation.getArgument(0);
+ if (plan instanceof ApplyConfigNodePlan) {
+ return nodeInfo.applyConfigNode((ApplyConfigNodePlan) plan);
+ }
+ if (plan instanceof RemoveConfigNodePlan) {
+ return nodeInfo.removeConfigNode((RemoveConfigNodePlan) plan);
+ }
+ if (failVersionUpdate.getAndSet(false)) {
+ return failureStatus;
+ }
+ return nodeInfo.updateVersionInfo((UpdateVersionInfoPlan) plan);
+ });
+ TConfigNodeLocation location =
+ new TConfigNodeLocation(
+ 1, new TEndPoint("127.0.0.1", 10710), new TEndPoint("127.0.0.1",
10720));
+ TNodeVersionInfo version = new TNodeVersionInfo("version", "build");
+ TNodeVersionInfo previousVersion =
nodeInfo.getVersionInfo(location.getConfigNodeId());
+ ConfigNodeConfig conf = ConfigNodeDescriptor.getInstance().getConf();
+ int originalNodeId = conf.getConfigNodeId();
+ NodeStatus originalNodeStatus =
CommonDescriptor.getInstance().getConfig().getNodeStatus();
+ String originalStatusReason =
CommonDescriptor.getInstance().getConfig().getStatusReason();
+ try {
+ // Model the joining Raft member applying its own registration and
rollback plans.
+ conf.setConfigNodeId(location.getConfigNodeId());
+
CommonDescriptor.getInstance().getConfig().setNodeStatus(NodeStatus.Running);
+ try {
+ nodeManager.applyConfigNode(location, version);
+ Assert.fail("Expected the failed version write to keep registration
pending");
+ } catch (IllegalStateException e) {
+ Assert.assertEquals(
+ String.format(
+ ConfigNodeMessages
+
.EXCEPTION_FAILED_TO_PERSIST_VERSION_INFO_FOR_CONFIGNODE_ARG_ARG_REGISTRATION_REMAINS_PENDING_AND_MUST_BE_RETRIED_F662C350,
+ location.getConfigNodeId(),
+ failureStatus.getMessage()),
+ e.getMessage());
+ }
+ Assert.assertEquals(
+ NodeStatus.Running,
CommonDescriptor.getInstance().getConfig().getNodeStatus());
+ Assert.assertEquals(Collections.singletonList(location),
nodeInfo.getRegisteredConfigNodes());
+ Assert.assertEquals(previousVersion,
nodeInfo.getVersionInfo(location.getConfigNodeId()));
+
+ nodeManager.applyConfigNode(location, version);
+
+ Assert.assertEquals(
+ NodeStatus.Running,
CommonDescriptor.getInstance().getConfig().getNodeStatus());
+ Assert.assertEquals(Collections.singletonList(location),
nodeInfo.getRegisteredConfigNodes());
+ Assert.assertEquals(version,
nodeInfo.getVersionInfo(location.getConfigNodeId()));
+ } finally {
+ conf.setConfigNodeId(originalNodeId);
+ CommonDescriptor.getInstance()
+ .getConfig()
+ .setNodeStatusWithReason(originalNodeStatus, originalStatusReason);
+ }
+
+ ArgumentCaptor<ConfigPhysicalPlan> planCaptor =
+ ArgumentCaptor.forClass(ConfigPhysicalPlan.class);
+ verify(consensusManager, Mockito.times(4)).write(planCaptor.capture());
+ Assert.assertTrue(planCaptor.getAllValues().get(0) instanceof
ApplyConfigNodePlan);
+ Assert.assertTrue(planCaptor.getAllValues().get(1) instanceof
UpdateVersionInfoPlan);
+ Assert.assertTrue(planCaptor.getAllValues().get(2) instanceof
ApplyConfigNodePlan);
+ Assert.assertTrue(planCaptor.getAllValues().get(3) instanceof
UpdateVersionInfoPlan);
+ verify(consensusManager, never()).write(any(RemoveConfigNodePlan.class));
+ }
+
+ private TDataNodeRegisterReq generateDataNodeRegisterReq(int dataNodeId) {
+ final TDataNodeRegisterReq req = new TDataNodeRegisterReq();
+ req.setDataNodeConfiguration(generateDataNodeConfiguration(dataNodeId,
"127.0.0.1"));
+ req.setVersionInfo(new TNodeVersionInfo("version", "build"));
+ return req;
+ }
+
+ private TDataNodeConfiguration generateDataNodeConfiguration(int dataNodeId,
String ip) {
+ final TDataNodeLocation dataNodeLocation = new TDataNodeLocation();
+ dataNodeLocation.setDataNodeId(dataNodeId);
+ dataNodeLocation.setClientRpcEndPoint(new TEndPoint(ip, 6667 +
dataNodeId));
+ dataNodeLocation.setInternalEndPoint(new TEndPoint(ip, 10730 +
dataNodeId));
+ dataNodeLocation.setMPPDataExchangeEndPoint(new TEndPoint(ip, 10740 +
dataNodeId));
+ dataNodeLocation.setDataRegionConsensusEndPoint(new TEndPoint(ip, 10760 +
dataNodeId));
+ dataNodeLocation.setSchemaRegionConsensusEndPoint(new TEndPoint(ip, 10750
+ dataNodeId));
+
+ final TDataNodeConfiguration dataNodeConfiguration = new
TDataNodeConfiguration();
+ dataNodeConfiguration.setLocation(dataNodeLocation);
+ return dataNodeConfiguration;
+ }
+}
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/NodeInfoTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/NodeInfoTest.java
index 667cbba357e..0426a21e3cb 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/NodeInfoTest.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/NodeInfoTest.java
@@ -284,6 +284,49 @@ public class NodeInfoTest {
new TAINodeLocation(10006, new TEndPoint("127.0.0.1", 10810)), new
TNodeResource(8, 1024L));
}
+ @Test
+ public void testRegistrationPlansAreIdempotentForWalReplay() {
+ NodeInfo replayNodeInfo = new NodeInfo();
+
+ TDataNodeConfiguration dataNodeConfiguration =
generateTDataNodeConfiguration(100);
+ RegisterDataNodePlan registerDataNodePlan = new
RegisterDataNodePlan(dataNodeConfiguration);
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ replayNodeInfo.registerDataNode(registerDataNodePlan).getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ replayNodeInfo.registerDataNode(registerDataNodePlan).getCode());
+ Assert.assertEquals(1, replayNodeInfo.getRegisteredDataNodeCount());
+ Assert.assertEquals(
+ dataNodeConfiguration,
+
replayNodeInfo.getRegisteredDataNode(dataNodeConfiguration.getLocation().getDataNodeId()));
+
+ TConfigNodeLocation configNodeLocation =
+ new TConfigNodeLocation(
+ 20000, new TEndPoint("127.0.0.1", 22200), new
TEndPoint("127.0.0.1", 22300));
+ ApplyConfigNodePlan applyConfigNodePlan = new
ApplyConfigNodePlan(configNodeLocation);
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ replayNodeInfo.applyConfigNode(applyConfigNodePlan).getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ replayNodeInfo.applyConfigNode(applyConfigNodePlan).getCode());
+ Assert.assertEquals(1, replayNodeInfo.getRegisteredConfigNodes().size());
+ Assert.assertEquals(configNodeLocation,
replayNodeInfo.getRegisteredConfigNodes().get(0));
+
+ TNodeVersionInfo versionInfo = new TNodeVersionInfo("version", "build");
+ UpdateVersionInfoPlan updateVersionInfoPlan =
+ new UpdateVersionInfoPlan(versionInfo,
configNodeLocation.getConfigNodeId());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ replayNodeInfo.updateVersionInfo(updateVersionInfoPlan).getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ replayNodeInfo.updateVersionInfo(updateVersionInfoPlan).getCode());
+ Assert.assertEquals(
+ versionInfo,
replayNodeInfo.getVersionInfo(configNodeLocation.getConfigNodeId()));
+ }
+
private void registerConfigNodes() {
for (int i = 0; i < 3; i++) {
ApplyConfigNodePlan applyConfigNodePlan =
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/PartitionInfoTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/PartitionInfoTest.java
index 21744963066..b34096a9621 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/PartitionInfoTest.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/PartitionInfoTest.java
@@ -38,6 +38,7 @@ import
org.apache.iotdb.confignode.consensus.request.write.partition.CreateSchem
import
org.apache.iotdb.confignode.consensus.request.write.region.CreateRegionGroupsPlan;
import
org.apache.iotdb.confignode.consensus.request.write.region.OfferRegionMaintainTasksPlan;
import
org.apache.iotdb.confignode.consensus.response.partition.RegionInfoListResp;
+import org.apache.iotdb.confignode.exception.DatabaseNotExistsException;
import org.apache.iotdb.confignode.persistence.partition.PartitionInfo;
import
org.apache.iotdb.confignode.persistence.partition.maintainer.RegionCreateTask;
import
org.apache.iotdb.confignode.persistence.partition.maintainer.RegionDeleteTask;
@@ -332,6 +333,34 @@ public class PartitionInfoTest {
});
}
+ @Test
+ public void testRegionGroupCount() throws DatabaseNotExistsException {
+ partitionInfo.createDatabase(
+ new DatabaseSchemaPlan(
+ ConfigPhysicalPlanType.CreateDatabase, new
TDatabaseSchema("root.region_count")));
+
+ CreateRegionGroupsPlan createRegionGroupsPlan = new
CreateRegionGroupsPlan();
+ createRegionGroupsPlan.addRegionGroup(
+ "root.region_count",
+ generateTRegionReplicaSet(
+ testFlag.SchemaPartition.getFlag(),
+ generateTConsensusGroupId(
+ testFlag.SchemaPartition.getFlag(),
TConsensusGroupType.SchemaRegion)));
+ createRegionGroupsPlan.addRegionGroup(
+ "root.region_count",
+ generateTRegionReplicaSet(
+ testFlag.DataPartition.getFlag(),
+ generateTConsensusGroupId(
+ testFlag.DataPartition.getFlag(),
TConsensusGroupType.DataRegion)));
+ partitionInfo.createRegionGroups(createRegionGroupsPlan);
+
+ Assert.assertEquals(
+ 1,
+ partitionInfo.getRegionGroupCount("root.region_count",
TConsensusGroupType.SchemaRegion));
+ Assert.assertEquals(
+ 1, partitionInfo.getRegionGroupCount("root.region_count",
TConsensusGroupType.DataRegion));
+ }
+
private TRegionReplicaSet generateTRegionReplicaSet(
int startFlag, TConsensusGroupId tConsensusGroupId) {
TRegionReplicaSet tRegionReplicaSet = new TRegionReplicaSet();
