diff --git a/iotdb-core/confignode/pom.xml b/iotdb-core/confignode/pom.xml index a7cd4b6298b5..5d4798f6f2df 100644 --- a/iotdb-core/confignode/pom.xml +++ b/iotdb-core/confignode/pom.xml @@ -32,6 +32,7 @@ ${confignode.test.skip} false ${confignode.test.skip} + en @@ -168,6 +169,37 @@ + + com.diffplug.spotless + spotless-maven-plugin + + + + src/main/java/**/*.java + src/test/java/**/*.java + src/main/i18n/**/*.java + + + + + + org.codehaus.mojo + build-helper-maven-plugin + + + add-i18n-source + generate-sources + + add-source + + + + src/main/i18n/${i18n.locale} + + + + + org.apache.maven.plugins maven-assembly-plugin @@ -241,6 +273,12 @@ + + with-zh-locale + + zh + + skipConfigNodeTests 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 new file mode 100644 index 000000000000..322b57ed59dc --- /dev/null +++ b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java @@ -0,0 +1,79 @@ +/* + * 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.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_APPLY_CONFIGNODE_ARG_THROUGH_CONSENSUS_LAYER_ARG_B4A985D7 = + "Failed to apply ConfigNode %s through consensus layer: %s"; + 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_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 EXECUTE_NON_QUERY_PLAN_FAILED = + "[ConfigRegion Update] Execute non-query plan 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 + LOG_CONFIGREGION_UPDATE_FAILED_TO_CLOSE_THE_CONFIGNODE_SIMPLECONSENSUS_LOG_FILEPATH_ARG_RETRY_ARG_188A702E = + "[ConfigRegion Update] Failed to close the ConfigNode SimpleConsensus log, filePath: {}, retry: {}."; + 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 + 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 + 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 MESSAGE_STATUS_CODE_ARG_4B3BCD8B = "status code %d"; + public static final String MESSAGE_UNKNOWN_ERROR_E9EEAFF8 = "unknown error"; + 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 + 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 UNEXPECTED_INTERRUPTION_DURING_THE_CLOSE_METHOD_OF_LOGWRITER = + "[ConfigRegion Update] Unexpected interruption during the close method of logWriter"; + + private ConfigNodeMessages() {} +} 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 new file mode 100644 index 000000000000..aaaff13b4785 --- /dev/null +++ b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java @@ -0,0 +1,78 @@ +/* + * 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.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_APPLY_CONFIGNODE_ARG_THROUGH_CONSENSUS_LAYER_ARG_B4A985D7 = + "通过共识层应用 ConfigNode %s 失败：%s"; + public static final String + EXCEPTION_FAILED_TO_CLOSE_THE_CONFIGNODE_SIMPLECONSENSUS_LOG_WRITER_BC0F417C = + "关闭 ConfigNode SimpleConsensus 日志写入器失败。"; + 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 EXECUTE_NON_QUERY_PLAN_FAILED = "[ConfigRegion Update] 执行非查询计划失败"; + public static final String FAILED_TO_ROLLBACK_PERSISTED_CONFIGNODE_SIMPLECONSENSUS_LOG = + "回滚已持久化的 ConfigNode SimpleConsensus 日志失败。该 plan 可能会在重启后被重放。重启前将禁止后续 SimpleConsensus 写入。"; + public static final String + LOG_CONFIGREGION_UPDATE_FAILED_TO_CLOSE_THE_CONFIGNODE_SIMPLECONSENSUS_LOG_FILEPATH_ARG_RETRY_ARG_188A702E = + "[ConfigRegion Update] 关闭 ConfigNode SimpleConsensus 日志失败，filePath: {}，重试: {}。"; + 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 + 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 + 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 MESSAGE_STATUS_CODE_ARG_4B3BCD8B = "状态码 %d"; + public static final String MESSAGE_UNKNOWN_ERROR_E9EEAFF8 = "未知错误"; + 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 + 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 UNEXPECTED_INTERRUPTION_DURING_THE_CLOSE_METHOD_OF_LOGWRITER = + "[ConfigRegion Update] 关闭 logWriter 过程中发生意外中断"; + + private ConfigNodeMessages() {} +} 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 bf18b7637186..e96c7e56e684 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.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; @@ -33,6 +32,7 @@ import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan; import org.apache.iotdb.confignode.consensus.request.read.ConfigPhysicalReadPlan; import org.apache.iotdb.confignode.exception.physical.UnknownPhysicalPlanTypeException; +import org.apache.iotdb.confignode.i18n.ConfigNodeMessages; import org.apache.iotdb.confignode.manager.ConfigManager; import org.apache.iotdb.confignode.manager.consensus.ConsensusManager; import org.apache.iotdb.confignode.manager.pipe.agent.PipeConfigNodeAgent; @@ -55,16 +55,15 @@ 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.Optional; import java.util.concurrent.ExecutorService; -import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; -import java.util.regex.Matcher; -import java.util.regex.Pattern; /** {@link IStateMachine} for ConfigRegion. */ public class ConfigRegionStateMachine implements IStateMachine, IStateMachine.EventApi { @@ -83,17 +82,18 @@ 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"; private static final String PROGRESS_FILE_PATH = CURRENT_FILE_DIR + File.separator + "log_inprogress_"; private static final String FILE_PATH = CURRENT_FILE_DIR + File.separator + "log_"; + private static final String LOG_INPROGRESS_FILE_PREFIX = "log_inprogress_"; + private static final String LOG_FILE_PREFIX = "log_"; private static final long LOG_FILE_MAX_SIZE = CONF.getConfigNodeSimpleConsensusLogSegmentSizeMax(); private final TEndPoint currentNodeTEndPoint; - private static Pattern LOG_INPROGRESS_PATTERN = Pattern.compile("\\d+"); - private static Pattern LOG_PATTERN = Pattern.compile("(?<=_)(\\d+)$"); public ConfigRegionStateMachine(ConfigManager configManager, ConfigPlanExecutor executor) { this.executor = executor; @@ -121,16 +121,32 @@ public TSStatus write(IConsensusRequest request) { /** 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) { - LOGGER.error("Execute non-query plan failed", 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()) { @@ -197,7 +213,6 @@ public boolean takeSnapshot(File snapshotDir) { PipeConfigNodeAgent.runtime() .listener() .tryListenToSnapshots(ConfignodeSnapshotParser.getSnapshots()); - return true; } catch (IOException e) { if (PipeConfigNodeAgent.runtime().listener().isOpened()) { LOGGER.warn( @@ -205,6 +220,7 @@ public boolean takeSnapshot(File snapshotDir) { e); } } + return true; } return false; } @@ -343,6 +359,9 @@ public void start() { @Override public void stop() { + if (ConsensusFactory.SIMPLE_CONSENSUS.equals(CONF.getConfigNodeConsensusProtocolClass())) { + closeSimpleLogWriter(); + } // Shutdown leader related service for config pipe PipeConfigNodeAgent.runtime().notifyLeaderUnavailable(); } @@ -352,56 +371,126 @@ public boolean isReadOnly() { 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("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( - "Can't close StandAloneLog for ConfigNode SimpleConsensus mode, " - + "filePath: {}, retry: {}", - simpleLogFile.getAbsolutePath(), - retry); - try { - // Sleep 1s and retry - TimeUnit.SECONDS.sleep(1); - } catch (InterruptedException e2) { - Thread.currentThread().interrupt(); - LOGGER.warn("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( - "Can't serialize current ConfigPhysicalPlan for ConfigNode SimpleConsensus mode", e); + ConfigNodeMessages + .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, new FileComparator()); for (String logFileName : list) { @@ -418,7 +507,7 @@ private void initStandAloneConfigNode() { continue; } - startIndex = endIndex; + final int recoveredStartIndex = parseStartIndex(logFileName); while (logReader.hasNext()) { endIndex++; // Read and re-serialize the PhysicalPlan @@ -436,33 +525,13 @@ private void initStandAloneConfigNode() { } } 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("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) { @@ -493,18 +562,127 @@ public int compare(String filename1, String filename2) { } } + 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) { - if (filename.startsWith("log_inprogress_")) { - Matcher matcher = LOG_INPROGRESS_PATTERN.matcher(filename); - if (matcher.find()) { - return Long.parseLong(matcher.group()); + final String endIndexString; + if (filename.startsWith(LOG_INPROGRESS_FILE_PREFIX)) { + endIndexString = 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; } + endIndexString = filename.substring(lastSeparatorIndex + 1); } else { - Matcher matcher = LOG_PATTERN.matcher(filename); - if (matcher.find()) { - return Long.parseLong(matcher.group()); + return 0; + } + + 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; + } + + 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 true; + } + + 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); } - return 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 81cad2f3befd..d8d6f239a06a 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; @@ -44,6 +45,7 @@ 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; @@ -61,6 +63,7 @@ import org.apache.iotdb.confignode.consensus.response.datanode.DataNodeConfigurationResp; import org.apache.iotdb.confignode.consensus.response.datanode.DataNodeRegisterResp; import org.apache.iotdb.confignode.consensus.response.datanode.DataNodeToStatusResp; +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.TTLManager; @@ -281,30 +284,32 @@ public DataSet registerDataNode(TDataNodeRegisterReq req) { 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); @@ -312,7 +317,10 @@ public DataSet registerDataNode(TDataNodeRegisterReq req) { // 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()); @@ -340,10 +348,10 @@ public TDataNodeRestartResp updateDataNodeIfNecessary(TDataNodeRestartReq req) { // 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); @@ -351,14 +359,14 @@ public TDataNodeRestartResp updateDataNodeIfNecessary(TDataNodeRestartReq req) { // 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()); resp.setCorrectConsensusGroups(getPartitionManager().getAllReplicaSets(nodeId)); @@ -436,13 +444,12 @@ public TSStatus updateConfigNodeIfNecessary(int configNodeId, TNodeVersionInfo v // 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 getRegisteredAINodeInfoList() { @@ -488,27 +495,37 @@ public synchronized DataSet registerAINode(TAINodeRegisterReq req) { } 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; @@ -550,10 +567,12 @@ public TAINodeRestartResp updateAINodeIfNecessary(TAINodeRestartReq req) { 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); @@ -561,15 +580,17 @@ public TAINodeRestartResp updateAINodeIfNecessary(TAINodeRestartReq req) { // 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; } @@ -764,21 +785,32 @@ public List getRegisteredConfigNodeInfoList() { * * @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))); } } @@ -1131,6 +1163,112 @@ public TDataNodeLocation getLowestLoadDataNode(Set nodes) { 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 b301b13b56ef..7d8e1ff97fcc 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 @@ -243,7 +243,7 @@ public int getRegionGroupCount(TConsensusGroupType type) { 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 8c91c99859ed..6ea8abdf97ba 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 @@ -231,7 +231,9 @@ public TSStatus offerRegionMaintainTasks( */ public TSStatus pollRegionMaintainTask() { synchronized (regionMaintainTaskList) { - regionMaintainTaskList.remove(0); + if (!regionMaintainTaskList.isEmpty()) { + regionMaintainTaskList.remove(0); + } return RpcUtils.SUCCESS_STATUS; } } @@ -986,9 +988,14 @@ public boolean processTakeSnapshot(File snapshotDir) throws TException, IOExcept databasePartitionTableEntry.getValue().serialize(bufferedOutputStream, protocol); } + final List 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 b38696a10ed2..c0d45d00d33c 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 @@ -80,9 +80,9 @@ protected Flow executeFromState(ConfigNodeProcedureEnv env, AddConfigNodeState s LOG.info("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("The ConfigNode: {} is successfully added to the cluster", tConfigNodeLocation); return Flow.NO_MORE_STATE; } 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 000000000000..1df8d6585311 --- /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 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 000000000000..518de349b970 --- /dev/null +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/node/NodeManagerRegistrationTest.java @@ -0,0 +1,259 @@ +/* + * 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 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().setNodeStatus(originalNodeStatus); + CommonDescriptor.getInstance().getConfig().setStatusReason(originalStatusReason); + } + + ArgumentCaptor 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 5ce94932036d..1774ec1812ca 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 @@ -26,8 +26,11 @@ import org.apache.iotdb.common.rpc.thrift.TNodeResource; import org.apache.iotdb.commons.exception.StartupException; import org.apache.iotdb.confignode.consensus.request.write.confignode.ApplyConfigNodePlan; +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.persistence.node.NodeInfo; +import org.apache.iotdb.confignode.rpc.thrift.TNodeVersionInfo; +import org.apache.iotdb.rpc.TSStatusCode; import org.apache.commons.io.FileUtils; import org.apache.thrift.TException; @@ -73,6 +76,49 @@ public void testSnapshot() throws TException, IOException { Assert.assertEquals(nodeInfo, nodeInfo1); } + @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 f6f209c93564..9d8d4648b8f0 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.region.OfferRegionMaintainTasksPlan; import org.apache.iotdb.confignode.consensus.request.write.region.PollSpecificRegionMaintainTaskPlan; 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; @@ -286,6 +287,34 @@ public void testShowRegion() { }); } + @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();