diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java index 4aef9cd74f050..339c1bc25c05f 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java @@ -132,34 +132,6 @@ private DataNodeMiscMessages() {} public static final String SET_SERVER_CONTEXT_FACTORY = "Will set ServerContextFactory from {} "; - // --------------------------------------------------------------------------- - // protocol – ConfigNodeInfo - // --------------------------------------------------------------------------- - public static final String UPDATE_CONFIG_NODE_SUCCESSFULLY = - "Update ConfigNode Successfully: {}, which takes {} ms."; - public static final String UPDATE_CONFIG_NODE_FAILED = "Update ConfigNode failed."; - public static final String SYSTEM_PROPERTIES_NOT_EXIST = - "System properties file not exist, not necessary to store ConfigNode list"; - public static final String LOAD_CONFIG_NODE_SUCCESSFULLY = - "Load ConfigNode successfully: {}, which takes {} ms."; - public static final String CANNOT_PARSE_CONFIG_NODE_LIST = - "Cannot parse config node list in system.properties"; - - // --------------------------------------------------------------------------- - // protocol – ConfigNodeClient - // --------------------------------------------------------------------------- - public static final String NODE_LEADER_MAY_DOWN_TRY_NEXT = - "The current node leader may have been down {}, try next node"; - public static final String UNEXPECTED_INTERRUPTION_CONNECT_CONFIG_NODE = - "Unexpected interruption when waiting to try to connect to ConfigNode"; - public static final String NODE_MAY_DOWN_TRY_NEXT = - "The current node may have been down {},try next node"; - public static final String FAILED_CONNECT_CONFIG_NODE_NOT_LEADER = - "Failed to connect to ConfigNode {} from DataNode {}, because the current node is not " - + "leader or not ready yet, will try again later"; - public static final String UNEXPECTED_INTERRUPTION_CONNECT_CONFIG_NODE_BREAK = - "Unexpected interruption when waiting to try to connect to ConfigNode, may because current node has been down. Will break current execution process to avoid meaningless wait."; - // --------------------------------------------------------------------------- // protocol – DataNodeInternalClient // --------------------------------------------------------------------------- @@ -670,8 +642,6 @@ private DataNodeMiscMessages() {} public static final String STARTING_IOTDB = "Starting IoTDB {}"; public static final String CANNOT_CREATE_SCHEMA_DIR = "Can not create schema dir: {}"; public static final String SCHEMA_DIR_CREATED = " {} dir has been created."; - public static final String IOTDB_VERSION_TOO_OLD = "IoTDB version is too old"; - public static final String REPAIR_SYSTEM_PROPERTIES = "repair system.properties, lack {}"; public static final String UNEXPECTED_CONSENSUS_GROUP_TYPE = "Unexpected consensus group type"; public static final String ENCRYPT_MAGIC_STRING_NOT_MATCHED = @@ -1149,12 +1119,8 @@ private DataNodeMiscMessages() {} "Totally find {} tsFiles to be settled."; public static final String MISC_LOG_SYSTEM_PROPERTIES_FILE_HAS_BEEN_MOVED_SUCCESSFULLY_4445A448 = "system.properties file has been moved successfully: {} -> {}"; - public static final String MISC_LOG_SERIALIZE_MUTABLE_SYSTEM_PROPERTIES_SUCCESSFULLY_WHICH_TAKES_4656A206 = - "Serialize mutable system properties successfully, which takes {} ms."; public static final String MISC_LOG_SYSTEMPROPERTIES_NORMALIZE_FROM_TO_FOR_COMPATIBILITY_BE1C725F = "[SystemProperties] Normalize {} from {} to {} for compatibility."; - public static final String MISC_LOG_DO_NOT_UPGRADE_IOTDB_FROM_V0_9_OR_LOWER_VERSION_TO_V1_0_9878EC88 = - "DO NOT UPGRADE IoTDB from v0.9 or lower version to v1.0! Please upgrade to v0.10 first"; public static final String MISC_LOG_CANNOT_FIND_IOTDB_HOME_OR_IOTDB_CONF_ENVIRONMENT_VARIABLE_BE01B2FE = "Cannot find IOTDB_HOME or IOTDB_CONF environment variable when loading config file {}, use " + "default configuration"; @@ -1402,10 +1368,6 @@ private DataNodeMiscMessages() {} "Illegal defaultDatabaseLevel: %d, should >= 1"; public static final String MISC_EXCEPTION_LOADTSFILESPILTPARTITIONMAXSIZE_SHOULD_BE_GREATER_THAN_OR_95B4DB23 = "loadTsFileSpiltPartitionMaxSize should be greater than or equal to 0"; - public static final String MISC_EXCEPTION_REMOVING_IS_ONLY_ALLOWED_IN_AN_ENVIRONMENT_WHERE_THE_DATANODE_5A3E1FEA = - "Removing is only allowed in an environment where the datanode has been successfully " - + "started. Please check whether it is removed on the confignode, or if you have deleted the " - + "system.properties file by mistake."; public static final String MISC_EXCEPTION_STATEMENTID_SDOESN_T_EXIST_IN_THIS_SESSION_S_BD5B4733 = "StatementId: %sdoesn't exist in this session %s"; public static final String MISC_EXCEPTION_INTERNALCLIENTSESSION_SHOULD_NEVER_CALL_PREPARE_STATEMENT_CCAB3CDC = @@ -1428,8 +1390,6 @@ private DataNodeMiscMessages() {} "%s is not allowed, only support %s"; public static final String MISC_EXCEPTION_S_IS_NOT_ALLOWED_ONLY_SUPPORT_S_1B06E0B7 = " %s is not allowed, only support %s"; - public static final String PARAMETER_CANNOT_BE_MODIFIED_AFTER_FIRST_STARTUP_FMT = - "%s can't be modified after first startup"; public static final String UNSUPPORTED_INVOCATION_BY_DATANODE = "This method is not supported for invocation by DataNode"; public static final String UNSUPPORTED_INVOCATION_BY_DATANODE_USE_SUBMIT_LOAD_CONFIGURATION_TASK = @@ -1494,9 +1454,6 @@ private DataNodeMiscMessages() {} public static final String EXCEPTION_RUNTIMESTATE_D4D018BA = "runtimeState"; public static final String EXCEPTION_STATEMENTNAME_IS_NULL_C03BB8D4 = "statementName is null"; public static final String EXCEPTION_SQL_IS_NULL_BEDB2B7A = "sql is null"; - public static final String MESSAGE_CONFIGNODE_LEADER_ARG_IS_WARMING_UP_BEFORE_SERVING_DATANODE_ARG_WILL_WAIT_AND_RETRY_AAB5F962 = "ConfigNode leader {} is warming up before serving DataNode {}, will wait and retry."; - public static final String MESSAGE_CONFIGNODE_LEADER_ARG_IS_WARMING_UP_BEFORE_SERVING_DATANODE_ARG_WILL_WAIT_AND_RETRY_REASON_ARG_3A2A4163 = - "ConfigNode leader {} is warming up before serving DataNode {}, will wait and retry. Reason: {}"; public static final String EXCEPTION_TRUST_STORE_PATH_MUST_BE_SET_WHEN_THRIFT_SSL_CLIENT_AUTH_IS_TRUE_36016171 = "trust_store_path must be set when thrift_ssl_client_auth is true"; public static final String EXCEPTION_CONTINUOUS_QUERY_MIN_EVERY_INTERVAL_IN_MS_SHOULD_BE_GREATER_THAN_0_BUT_CURRENT_VALUE_IS_F9A1BEC4 = "continuous_query_min_every_interval_in_ms should be greater than 0, but current value is "; public static final String EXCEPTION_UNKNOWN_READ_CONSISTENCY_LEVEL_ARG_PLEASE_SET_TO_STRONG_OR_WEAK_8CF29949 = "Unknown read_consistency_level: %s, please set to \"strong\" or \"weak\""; diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java index a6656b5ace369..676713cbe8958 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java @@ -62,7 +62,6 @@ private StorageEngineMessages() {} // ======================== Resource Control - Disk ======================== - public static final String FAILED_TO_DEREGISTER_FILE_LOCK = "Failed to deregister file lock because {}"; public static final String ALL_FOLDERS_FULL_CHANGE_TO_READ_ONLY = "All folders are full, change system mode to read-only."; public static final String FAILED_TO_PROCESS_FOLDER = "Failed to process folder '"; public static final String FAIL_TO_GET_CANONICAL_PATH = "Fail to get canonical path of data dir {}"; @@ -1442,14 +1441,6 @@ private StorageEngineMessages() {} "number of read requests exceeded - wait %sms"; public static final String STORAGE_EXCEPTION_READ_SIZE_LIMIT_EXCEEDED_WAIT_SMS_E19598BA = "read size limit exceeded - wait %sms"; - public static final String STORAGE_EXCEPTION_UNABLE_TO_CREATE_DIRECTORY_S_BECAUSE_THERE_IS_FILE_UNDER_1C59ACFC = - "Unable to create directory %s because there is file under the path, please check " - + "configuration and restart."; - public static final String STORAGE_EXCEPTION_UNABLE_TO_CREATE_DIRECTORY_S_PLEASE_CHECK_CONFIGURATION_BA580B67 = - "Unable to create directory %s, please check configuration and restart."; - public static final String STORAGE_EXCEPTION_CONFLICT_IS_DETECTED_IN_DIRECTORY_S_WHICH_MAY_BE_BEING_USED_CB5C77FC = - "Conflict is detected in directory %s, which may be being used by another IoTDB " - + "(ProcessId=%s). Please check configuration and restart."; public static final String COMPACTION_INNER_SPACE = "inner"; public static final String COMPACTION_CROSS_SPACE = "cross"; public static final String DEVICE_DOES_NOT_EXIST_IN_RESOURCE_FILE_FMT = diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java index bf909f2774521..bf7962c8b0fb2 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java @@ -132,33 +132,6 @@ private DataNodeMiscMessages() {} public static final String SET_SERVER_CONTEXT_FACTORY = "将从 {} 设置 ServerContextFactory"; - // --------------------------------------------------------------------------- - // protocol – ConfigNodeInfo - // --------------------------------------------------------------------------- - public static final String UPDATE_CONFIG_NODE_SUCCESSFULLY = - "成功更新 ConfigNode：{}，耗时 {} 毫秒。"; - public static final String UPDATE_CONFIG_NODE_FAILED = "更新 ConfigNode 失败。"; - public static final String SYSTEM_PROPERTIES_NOT_EXIST = - "系统属性文件不存在，无需存储 ConfigNode 列表"; - public static final String LOAD_CONFIG_NODE_SUCCESSFULLY = - "成功加载 ConfigNode：{}，耗时 {} 毫秒。"; - public static final String CANNOT_PARSE_CONFIG_NODE_LIST = - "无法解析 system.properties 中的 ConfigNode 列表"; - - // --------------------------------------------------------------------------- - // protocol – ConfigNodeClient - // --------------------------------------------------------------------------- - public static final String NODE_LEADER_MAY_DOWN_TRY_NEXT = - "当前节点 Leader {} 可能已宕机，尝试下一个节点"; - public static final String UNEXPECTED_INTERRUPTION_CONNECT_CONFIG_NODE = - "等待连接 ConfigNode 时发生意外中断"; - public static final String NODE_MAY_DOWN_TRY_NEXT = - "当前节点 {} 可能已宕机，尝试下一个节点"; - public static final String FAILED_CONNECT_CONFIG_NODE_NOT_LEADER = - "从 DataNode {} 连接 ConfigNode {} 失败，因为当前节点不是 Leader 或尚未就绪，稍后将重试"; - public static final String UNEXPECTED_INTERRUPTION_CONNECT_CONFIG_NODE_BREAK = - "等待连接 ConfigNode 时发生意外中断，可能是当前节点已宕机，将中断当前执行流程以避免无意义等待。"; - // --------------------------------------------------------------------------- // protocol – DataNodeInternalClient // --------------------------------------------------------------------------- @@ -669,8 +642,6 @@ private DataNodeMiscMessages() {} public static final String STARTING_IOTDB = "正在启动 IoTDB {}"; public static final String CANNOT_CREATE_SCHEMA_DIR = "无法创建 Schema 目录：{}"; public static final String SCHEMA_DIR_CREATED = " {} 目录已创建。"; - public static final String IOTDB_VERSION_TOO_OLD = "IoTDB 版本过旧"; - public static final String REPAIR_SYSTEM_PROPERTIES = "修复 system.properties，缺少 {}"; public static final String UNEXPECTED_CONSENSUS_GROUP_TYPE = "未预期的共识组类型"; public static final String ENCRYPT_MAGIC_STRING_NOT_MATCHED = @@ -1143,12 +1114,8 @@ private DataNodeMiscMessages() {} "共找到 {} 个待 settle 的 TsFile。"; public static final String MISC_LOG_SYSTEM_PROPERTIES_FILE_HAS_BEEN_MOVED_SUCCESSFULLY_4445A448 = "system.properties 文件已成功移动：{} -> {}"; - public static final String MISC_LOG_SERIALIZE_MUTABLE_SYSTEM_PROPERTIES_SUCCESSFULLY_WHICH_TAKES_4656A206 = - "成功序列化 mutable system properties，耗时 {} ms。"; public static final String MISC_LOG_SYSTEMPROPERTIES_NORMALIZE_FROM_TO_FOR_COMPATIBILITY_BE1C725F = "[SystemProperties] 为兼容性将 {} 从 {} 规范化为 {}。"; - public static final String MISC_LOG_DO_NOT_UPGRADE_IOTDB_FROM_V0_9_OR_LOWER_VERSION_TO_V1_0_9878EC88 = - "请勿将 IoTDB 从 v0.9 或更低版本直接升级到 v1.0！请先升级到 v0.10"; public static final String MISC_LOG_CANNOT_FIND_IOTDB_HOME_OR_IOTDB_CONF_ENVIRONMENT_VARIABLE_BE01B2FE = "加载配置文件 {} 时找不到 IOTDB_HOME 或 IOTDB_CONF 环境变量，使用默认配置"; public static final String MISC_LOG_COULDN_T_LOAD_THE_CONFIGURATION_FROM_ANY_OF_THE_KNOWN_SOURCES_EE3ED103 = @@ -1372,8 +1339,6 @@ private DataNodeMiscMessages() {} "非法的 defaultDatabaseLevel：%d，应 >= 1"; public static final String MISC_EXCEPTION_LOADTSFILESPILTPARTITIONMAXSIZE_SHOULD_BE_GREATER_THAN_OR_95B4DB23 = "loadTsFileSpiltPartitionMaxSize 应大于或等于 0"; - public static final String MISC_EXCEPTION_REMOVING_IS_ONLY_ALLOWED_IN_AN_ENVIRONMENT_WHERE_THE_DATANODE_5A3E1FEA = - "只有在 DataNode 已成功启动的环境中才允许移除。请检查它是否已在 ConfigNode 上移除，或是否误删了 system.properties 文件。"; public static final String MISC_EXCEPTION_STATEMENTID_SDOESN_T_EXIST_IN_THIS_SESSION_S_BD5B4733 = "StatementId：%s 在会话 %s 中不存在"; public static final String MISC_EXCEPTION_INTERNALCLIENTSESSION_SHOULD_NEVER_CALL_PREPARE_STATEMENT_CCAB3CDC = @@ -1396,8 +1361,6 @@ private DataNodeMiscMessages() {} "%s 不允许，仅支持 %s"; public static final String MISC_EXCEPTION_S_IS_NOT_ALLOWED_ONLY_SUPPORT_S_1B06E0B7 = " %s 不允许，仅支持 %s"; - public static final String PARAMETER_CANNOT_BE_MODIFIED_AFTER_FIRST_STARTUP_FMT = - "%s 首次启动后不能修改"; public static final String UNSUPPORTED_INVOCATION_BY_DATANODE = "DataNode 不支持调用此方法"; public static final String UNSUPPORTED_INVOCATION_BY_DATANODE_USE_SUBMIT_LOAD_CONFIGURATION_TASK = @@ -1473,9 +1436,6 @@ private DataNodeMiscMessages() {} public static final String EXCEPTION_RUNTIMESTATE_D4D018BA = "runtimeState"; public static final String EXCEPTION_STATEMENTNAME_IS_NULL_C03BB8D4 = "statementName 不能为空"; public static final String EXCEPTION_SQL_IS_NULL_BEDB2B7A = "sql 不能为空"; - public static final String MESSAGE_CONFIGNODE_LEADER_ARG_IS_WARMING_UP_BEFORE_SERVING_DATANODE_ARG_WILL_WAIT_AND_RETRY_AAB5F962 = "ConfigNode leader {} 正在预热，暂未对 DataNode {} 提供服务，将等待并重试。"; - public static final String MESSAGE_CONFIGNODE_LEADER_ARG_IS_WARMING_UP_BEFORE_SERVING_DATANODE_ARG_WILL_WAIT_AND_RETRY_REASON_ARG_3A2A4163 = - "ConfigNode leader {} 正在预热，暂未对 DataNode {} 提供服务，将等待并重试。原因：{}"; public static final String EXCEPTION_TRUST_STORE_PATH_MUST_BE_SET_WHEN_THRIFT_SSL_CLIENT_AUTH_IS_TRUE_36016171 = "当 thrift_ssl_client_auth 为 true 时，必须设置 trust_store_path"; public static final String EXCEPTION_CONTINUOUS_QUERY_MIN_EVERY_INTERVAL_IN_MS_SHOULD_BE_GREATER_THAN_0_BUT_CURRENT_VALUE_IS_F9A1BEC4 = "continuous_query_min_every_interval_in_ms 必须大于 0，但当前值为 "; public static final String EXCEPTION_UNKNOWN_READ_CONSISTENCY_LEVEL_ARG_PLEASE_SET_TO_STRONG_OR_WEAK_8CF29949 = "未知的 read_consistency_level：%s，请设置为 \"strong\" 或 \"weak\""; diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java index 2786c2e5d206d..92810efb52e83 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java @@ -61,8 +61,7 @@ private StorageEngineMessages() {} public static final String FILE_NO_SUCH_TIME_SERIES = "文件中不包含该时间序列 {}。"; // ======================== Resource Control - Disk ======================== - - public static final String FAILED_TO_DEREGISTER_FILE_LOCK = "注销文件锁失败，原因: {}"; + public static final String ALL_FOLDERS_FULL_CHANGE_TO_READ_ONLY = "所有目录已满，切换系统为只读模式。"; public static final String FAILED_TO_PROCESS_FOLDER = "处理目录失败 '"; public static final String FAIL_TO_GET_CANONICAL_PATH = "获取数据目录 {} 的规范路径失败"; @@ -1371,12 +1370,6 @@ private StorageEngineMessages() {} "读取请求数量超限 - 等待 %sms"; public static final String STORAGE_EXCEPTION_READ_SIZE_LIMIT_EXCEEDED_WAIT_SMS_E19598BA = "读取大小超限 - 等待 %sms"; - public static final String STORAGE_EXCEPTION_UNABLE_TO_CREATE_DIRECTORY_S_BECAUSE_THERE_IS_FILE_UNDER_1C59ACFC = - "无法创建目录 %s，原因：该路径下存在文件，请检查配置并重启。"; - public static final String STORAGE_EXCEPTION_UNABLE_TO_CREATE_DIRECTORY_S_PLEASE_CHECK_CONFIGURATION_BA580B67 = - "无法创建目录 %s，请检查配置并重启。"; - public static final String STORAGE_EXCEPTION_CONFLICT_IS_DETECTED_IN_DIRECTORY_S_WHICH_MAY_BE_BEING_USED_CB5C77FC = - "目录 %s 检测到冲突，可能正被另一个 IoTDB 使用 (ProcessId=%s)。请检查配置并重启。"; public static final String COMPACTION_INNER_SPACE = "内部"; public static final String COMPACTION_CROSS_SPACE = "跨空间"; public static final String DEVICE_DOES_NOT_EXIST_IN_RESOURCE_FILE_FMT = diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBStartCheck.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBStartCheck.java index ffc22191679ec..ec36efc900ec8 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBStartCheck.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBStartCheck.java @@ -19,58 +19,40 @@ package org.apache.iotdb.db.conf; import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType; +import org.apache.iotdb.commons.conf.AbstractIoTDBStartCheck; import org.apache.iotdb.commons.conf.CommonConfig; import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.conf.IoTDBConstant; +import org.apache.iotdb.commons.disk.utils.DirectoryChecker; import org.apache.iotdb.commons.exception.ConfigurationException; import org.apache.iotdb.commons.file.SystemFileFactory; import org.apache.iotdb.commons.file.SystemPropertiesHandler; import org.apache.iotdb.consensus.ConsensusFactory; import org.apache.iotdb.db.i18n.DataNodeMiscMessages; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALMode; -import org.apache.iotdb.db.storageengine.rescon.disk.DirectoryChecker; import org.apache.tsfile.common.conf.TSFileConfig; import org.apache.tsfile.common.conf.TSFileDescriptor; import org.apache.tsfile.encrypt.EncryptUtils; import org.apache.tsfile.exception.encrypt.EncryptException; import org.apache.tsfile.external.commons.io.FileUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import java.io.File; import java.io.IOException; import java.util.HashMap; import java.util.Map; -import java.util.Map.Entry; import java.util.Objects; -import java.util.Properties; import java.util.function.Supplier; -public class IoTDBStartCheck { - - private static final Logger logger = LoggerFactory.getLogger(IoTDBStartCheck.class); +public class IoTDBStartCheck extends AbstractIoTDBStartCheck { private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig(); private static final CommonConfig commonConfig = CommonDescriptor.getInstance().getConfig(); - // this file is located in data/system/schema/system.properties - // If user delete folder "data", system.properties can reset. - public static final String PROPERTIES_FILE_NAME = "system.properties"; private static final String SCHEMA_DIR = config.getSchemaDir(); private boolean isFirstStart = false; - private Properties properties = new Properties(); - - private final Map> systemProperties = new HashMap<>(); - private final SystemPropertiesHandler systemPropertiesHandler; - - // region params need checking, determined when first start - private static final String SYSTEM_PROPERTIES_STRING = "System properties:"; - private static final String DATA_REGION_NUM = "data_region_num"; - - // endregion // region params don't need checking and can be updated private static final String INTERNAL_ADDRESS = "dn_internal_address"; private static final String INTERNAL_PORT = "dn_internal_port"; @@ -103,17 +85,28 @@ public class IoTDBStartCheck { // endregion // region params don't need checking, determined by the system - private static final String IOTDB_VERSION_STRING = "iotdb_version"; - private static final String COMMIT_ID_STRING = "commit_id"; private static final String DATA_NODE_ID = "data_node_id"; - private static final String CLUSTER_ID = "cluster_id"; private static final String SCHEMA_REGION_CONSENSUS_PROTOCOL = "schema_region_consensus_protocol"; private static final String DATA_REGION_CONSENSUS_PROTOCOL = "data_region_consensus_protocol"; + // endregion // region params of old versions private static final String VIRTUAL_STORAGE_GROUP_NUM = "virtual_storage_group_num"; - // endregion + private IoTDBStartCheck() { + logger.info(DataNodeMiscMessages.STARTING_IOTDB, IoTDBConstant.VERSION_WITH_BUILD); + + // check whether SCHEMA_DIR exists, create if not exists + File dir = SystemFileFactory.INSTANCE.getFile(SCHEMA_DIR); + if (!dir.exists()) { + if (!dir.mkdirs()) { + logger.error(DataNodeMiscMessages.CANNOT_CREATE_SCHEMA_DIR, SCHEMA_DIR); + System.exit(-1); + } else { + logger.info(DataNodeMiscMessages.SCHEMA_DIR_CREATED, SCHEMA_DIR); + } + } + } public static IoTDBStartCheck getInstance() { return IoTDBConfigCheckHolder.INSTANCE; @@ -128,34 +121,77 @@ private static class IoTDBConfigCheckHolder { private static IoTDBStartCheck INSTANCE = new IoTDBStartCheck(); } - private String getVal(String paramName) { - if (variableParamValueTable.containsKey(paramName)) { - return variableParamValueTable.get(paramName).get(); - } else { - return null; - } + @Override + protected SystemPropertiesHandler getSystemPropertiesHandler() { + return DataNodeSystemPropertiesHandler.getInstance(); } - private IoTDBStartCheck() { - logger.info(DataNodeMiscMessages.STARTING_IOTDB, IoTDBConstant.VERSION_WITH_BUILD); + @Override + protected Map> getVariableParamValueTable() { + return variableParamValueTable; + } - // check whether SCHEMA_DIR exists, create if not exists - File dir = SystemFileFactory.INSTANCE.getFile(SCHEMA_DIR); - if (!dir.exists()) { - if (!dir.mkdirs()) { - logger.error(DataNodeMiscMessages.CANNOT_CREATE_SCHEMA_DIR, SCHEMA_DIR); - System.exit(-1); - } else { - logger.info(DataNodeMiscMessages.SCHEMA_DIR_CREATED, SCHEMA_DIR); - } - } + @Override + protected String getNodeIdKey() { + return DATA_NODE_ID; + } - systemPropertiesHandler = DataNodeSystemPropertiesHandler.getInstance(); + @Override + protected void loadNodeId(int nodeId) { + config.setDataNodeId(nodeId); + } - systemProperties.put(IOTDB_VERSION_STRING, () -> IoTDBConstant.VERSION); - systemProperties.put(COMMIT_ID_STRING, () -> IoTDBConstant.BUILD_INFO); - for (String param : variableParamValueTable.keySet()) { - systemProperties.put(param, () -> getVal(param)); + @Override + protected void loadClusterId(String clusterId) { + config.setClusterId(clusterId); + } + + @Override + protected void validateFirstStart() throws ConfigurationException { + if ((config.getDataRegionConsensusProtocolClass().equals(ConsensusFactory.IOT_CONSENSUS) + || (config + .getDataRegionConsensusProtocolClass() + .equals(ConsensusFactory.IOT_CONSENSUS_V2) + && config + .getIotConsensusV2Mode() + .equals(ConsensusFactory.IOT_CONSENSUS_V2_STREAM_MODE))) + && config.getWalMode().equals(WALMode.DISABLE)) { + throw new ConfigurationException( + DataNodeMiscMessages + .MISC_EXCEPTION_CONFIGURING_THE_WALMODE_AS_DISABLE_IS_NOT_SUPPORTED_UNDER_49298819); + } + } + + @Override + protected void checkExtraImmutableProperties() throws IOException { + // Only the data region protocol could have been persisted as the old PipeConsensus name + // during a jar-only upgrade, so only that field needs compatibility normalization. + boolean needRewriteConsensusProtocol = false; + if (properties.containsKey(SCHEMA_REGION_CONSENSUS_PROTOCOL)) { + config.setSchemaRegionConsensusProtocolClass( + properties.getProperty(SCHEMA_REGION_CONSENSUS_PROTOCOL)); + } + if (properties.containsKey(DATA_REGION_CONSENSUS_PROTOCOL)) { + final String persistedDataRegionConsensusProtocolClass = + properties.getProperty(DATA_REGION_CONSENSUS_PROTOCOL); + final String dataRegionConsensusProtocolClass = + ConsensusFactory.normalizeConsensusProtocolClass( + persistedDataRegionConsensusProtocolClass); + if (!Objects.equals( + persistedDataRegionConsensusProtocolClass, dataRegionConsensusProtocolClass)) { + properties.setProperty(DATA_REGION_CONSENSUS_PROTOCOL, dataRegionConsensusProtocolClass); + needRewriteConsensusProtocol = true; + logger.warn( + DataNodeMiscMessages + .MISC_LOG_SYSTEMPROPERTIES_NORMALIZE_FROM_TO_FOR_COMPATIBILITY_BE1C725F, + DATA_REGION_CONSENSUS_PROTOCOL, + persistedDataRegionConsensusProtocolClass, + dataRegionConsensusProtocolClass); + } + config.setDataRegionConsensusProtocolClass(dataRegionConsensusProtocolClass); + } + if (needRewriteConsensusProtocol) { + systemPropertiesHandler.overwrite(properties); } } @@ -166,6 +202,7 @@ private IoTDBStartCheck() { * of permissions. (2) try to check if the directory is occupied, avoid multiple IoTDB processes * accessing same director. */ + @Override public void checkDirectory() throws ConfigurationException, IOException { for (String dataDir : config.getLocalDataDirs()) { DirectoryChecker.getInstance().registerDirectory(new File(dataDir)); @@ -199,8 +236,7 @@ public void checkDirectory() throws ConfigurationException, IOException { */ public static void checkOldSystemConfig() throws IOException { File oldPropertiesFile = - SystemFileFactory.INSTANCE.getFile( - IoTDBStartCheck.SCHEMA_DIR + File.separator + PROPERTIES_FILE_NAME); + SystemFileFactory.INSTANCE.getFile(SCHEMA_DIR + File.separator + PROPERTIES_FILE_NAME); if (oldPropertiesFile.exists()) { File correctPropertiesFile = SystemFileFactory.INSTANCE.getFile( @@ -214,125 +250,6 @@ public static void checkOldSystemConfig() throws IOException { } } - /** - * check configuration in system.properties when starting IoTDB - * - *

When init: create system.properties directly - * - *

When upgrading the system.properties: (1) create system.properties.tmp (2) delete - * system.properties (3) rename system.properties.tmp to system.properties - */ - public void checkSystemConfig() throws ConfigurationException, IOException { - // read properties from system.properties - properties = systemPropertiesHandler.read(); - - if (systemPropertiesHandler.isFirstStart()) { - if ((config.getDataRegionConsensusProtocolClass().equals(ConsensusFactory.IOT_CONSENSUS) - || (config - .getDataRegionConsensusProtocolClass() - .equals(ConsensusFactory.IOT_CONSENSUS_V2) - && config - .getIotConsensusV2Mode() - .equals(ConsensusFactory.IOT_CONSENSUS_V2_STREAM_MODE))) - && config.getWalMode().equals(WALMode.DISABLE)) { - throw new ConfigurationException( - DataNodeMiscMessages - .MISC_EXCEPTION_CONFIGURING_THE_WALMODE_AS_DISABLE_IS_NOT_SUPPORTED_UNDER_49298819); - } - } else { - // check whether upgrading from <=v0.9 - if (!properties.containsKey(IOTDB_VERSION_STRING)) { - logger.error( - DataNodeMiscMessages - .MISC_LOG_DO_NOT_UPGRADE_IOTDB_FROM_V0_9_OR_LOWER_VERSION_TO_V1_0_9878EC88); - System.exit(-1); - } - String versionString = properties.getProperty(IOTDB_VERSION_STRING); - if (versionString.startsWith("0.")) { - logger.error(DataNodeMiscMessages.IOTDB_VERSION_TOO_OLD); - System.exit(-1); - } - checkImmutableSystemProperties(); - } - } - - /** repair broken properties */ - private void upgradePropertiesFileFromBrokenFile() throws IOException { - systemProperties.forEach( - (k, v) -> { - if (!properties.containsKey(k)) { - properties.setProperty(k, v.get()); - } - }); - properties.setProperty(IOTDB_VERSION_STRING, IoTDBConstant.VERSION); - properties.setProperty(COMMIT_ID_STRING, IoTDBConstant.BUILD_INFO); - systemPropertiesHandler.overwrite(properties); - } - - /** Check all immutable properties */ - private void checkImmutableSystemProperties() throws IOException { - for (Entry> entry : systemProperties.entrySet()) { - if (!properties.containsKey(entry.getKey())) { - upgradePropertiesFileFromBrokenFile(); - logger.info(DataNodeMiscMessages.REPAIR_SYSTEM_PROPERTIES, entry.getKey()); - } - } - - // load configuration from system properties only when start as Data node - if (properties.containsKey(DATA_NODE_ID)) { - config.setDataNodeId(Integer.parseInt(properties.getProperty(DATA_NODE_ID))); - } - if (properties.containsKey(CLUSTER_ID)) { - config.setClusterId(properties.getProperty(CLUSTER_ID)); - } - // Only the data region protocol could have been persisted as the old PipeConsensus name - // during a jar-only upgrade, so only that field needs compatibility normalization. - boolean needRewriteConsensusProtocol = false; - if (properties.containsKey(SCHEMA_REGION_CONSENSUS_PROTOCOL)) { - config.setSchemaRegionConsensusProtocolClass( - properties.getProperty(SCHEMA_REGION_CONSENSUS_PROTOCOL)); - } - if (properties.containsKey(DATA_REGION_CONSENSUS_PROTOCOL)) { - final String persistedDataRegionConsensusProtocolClass = - properties.getProperty(DATA_REGION_CONSENSUS_PROTOCOL); - final String dataRegionConsensusProtocolClass = - ConsensusFactory.normalizeConsensusProtocolClass( - persistedDataRegionConsensusProtocolClass); - if (!Objects.equals( - persistedDataRegionConsensusProtocolClass, dataRegionConsensusProtocolClass)) { - properties.setProperty(DATA_REGION_CONSENSUS_PROTOCOL, dataRegionConsensusProtocolClass); - needRewriteConsensusProtocol = true; - logger.warn( - DataNodeMiscMessages - .MISC_LOG_SYSTEMPROPERTIES_NORMALIZE_FROM_TO_FOR_COMPATIBILITY_BE1C725F, - DATA_REGION_CONSENSUS_PROTOCOL, - persistedDataRegionConsensusProtocolClass, - dataRegionConsensusProtocolClass); - } - config.setDataRegionConsensusProtocolClass(dataRegionConsensusProtocolClass); - } - if (needRewriteConsensusProtocol) { - systemPropertiesHandler.overwrite(properties); - } - } - - private void throwException(String parameter, Object badValue) throws ConfigurationException { - throw new ConfigurationException( - parameter, - String.valueOf(badValue), - properties.getProperty(parameter), - String.format( - DataNodeMiscMessages.PARAMETER_CANNOT_BE_MODIFIED_AFTER_FIRST_STARTUP_FMT, parameter)); - } - - public void serializeDataNodeId(int dataNodeId) throws IOException { - systemPropertiesHandler.put(DATA_NODE_ID, String.valueOf(dataNodeId)); - } - - public void serializeClusterID(String clusterId) throws IOException { - systemPropertiesHandler.put(CLUSTER_ID, clusterId); - } - public void serializeEncryptMagicString() throws IOException { if (!Objects.equals(TSFileDescriptor.getInstance().getConfig().getEncryptType(), "UNENCRYPTED") && !Objects.equals( @@ -386,30 +303,6 @@ public boolean checkConsensusProtocolExists(TConsensusGroupType type) { return false; } - public void serializeMutableSystemPropertiesIfNecessary() throws IOException { - long startTime = System.currentTimeMillis(); - boolean needsSerialize = false; - for (String param : variableParamValueTable.keySet()) { - if (!properties.getProperty(param).equals(getVal(param))) { - needsSerialize = true; - } - } - - if (needsSerialize) { - generateOrOverwriteSystemPropertiesFile(); - } - long endTime = System.currentTimeMillis(); - logger.info( - DataNodeMiscMessages - .MISC_LOG_SERIALIZE_MUTABLE_SYSTEM_PROPERTIES_SUCCESSFULLY_WHICH_TAKES_4656A206, - (endTime - startTime)); - } - - public void generateOrOverwriteSystemPropertiesFile() throws IOException { - systemProperties.forEach((k, v) -> properties.setProperty(k, v.get())); - systemPropertiesHandler.overwrite(properties); - } - public void checkEncryptMagicString() throws IOException, ConfigurationException { if (!Objects.equals(TSFileDescriptor.getInstance().getConfig().getEncryptType(), "UNENCRYPTED") && !Objects.equals( @@ -448,8 +341,4 @@ public void checkEncryptMagicString() throws IOException, ConfigurationException } } } - - public Properties getProperties() { - return properties; - } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java index a9fe5a60cd4cb..8ad9c7e36e970 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java @@ -36,16 +36,15 @@ import org.apache.iotdb.common.rpc.thrift.TShowConfigurationResp; import org.apache.iotdb.common.rpc.thrift.TShowTTLReq; import org.apache.iotdb.common.rpc.thrift.TTestConnectionResp; +import org.apache.iotdb.commons.client.AbstractConfigNodeClient; +import org.apache.iotdb.commons.client.AbstractConfigNodeInfo; import org.apache.iotdb.commons.client.ClientManager; -import org.apache.iotdb.commons.client.ThriftClient; import org.apache.iotdb.commons.client.factory.ThriftClientFactory; import org.apache.iotdb.commons.client.property.ThriftClientProperty; import org.apache.iotdb.commons.client.request.TestConnectionUtils; import org.apache.iotdb.commons.client.sync.SyncThriftClientWithErrorHandler; -import org.apache.iotdb.commons.conf.CommonConfig; -import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.consensus.ConfigRegionId; -import org.apache.iotdb.confignode.rpc.thrift.IConfigNodeRPCService; +import org.apache.iotdb.commons.i18n.ConfigMessages; import org.apache.iotdb.confignode.rpc.thrift.TAINodeConfigurationResp; import org.apache.iotdb.confignode.rpc.thrift.TAINodeRegisterReq; import org.apache.iotdb.confignode.rpc.thrift.TAINodeRegisterResp; @@ -121,7 +120,6 @@ import org.apache.iotdb.confignode.rpc.thrift.TGetClusterIdResp; import org.apache.iotdb.confignode.rpc.thrift.TGetCommitProgressReq; import org.apache.iotdb.confignode.rpc.thrift.TGetCommitProgressResp; -import org.apache.iotdb.confignode.rpc.thrift.TGetDataNodeLocationsResp; import org.apache.iotdb.confignode.rpc.thrift.TGetDatabaseReq; import org.apache.iotdb.confignode.rpc.thrift.TGetJarInListReq; import org.apache.iotdb.confignode.rpc.thrift.TGetJarInListResp; @@ -185,7 +183,6 @@ import org.apache.iotdb.confignode.rpc.thrift.TStartPipeReq; import org.apache.iotdb.confignode.rpc.thrift.TStopPipeReq; import org.apache.iotdb.confignode.rpc.thrift.TSubscribeReq; -import org.apache.iotdb.confignode.rpc.thrift.TSystemConfigurationResp; import org.apache.iotdb.confignode.rpc.thrift.TTestOperation; import org.apache.iotdb.confignode.rpc.thrift.TThrottleQuotaResp; import org.apache.iotdb.confignode.rpc.thrift.TUnsetSchemaTemplateReq; @@ -193,7 +190,6 @@ import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.i18n.DataNodeMiscMessages; -import org.apache.iotdb.rpc.DeepCopyRpcTransportFactory; import org.apache.iotdb.rpc.TSStatusCode; import org.apache.commons.pool2.PooledObject; @@ -203,299 +199,55 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import javax.net.ssl.SSLHandshakeException; - -import java.io.IOException; import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.Optional; import java.util.Set; -import java.util.function.Predicate; - -public class ConfigNodeClient implements IConfigNodeRPCService.Iface, ThriftClient, AutoCloseable { +public class ConfigNodeClient extends AbstractConfigNodeClient { private static final Logger logger = LoggerFactory.getLogger(ConfigNodeClient.class); - private static final int RETRY_NUM = 15; + private static final String NODE_TYPE_NAME = "DataNode"; public static final String MSG_RECONNECTION_FAIL = - "Fail to connect to any config node. Please check status of ConfigNodes or logs of connected DataNode"; + String.format(ConfigMessages.MSG_RECONNECTION_FAIL, NODE_TYPE_NAME); - private static final String MSG_RECONNECTION_DATANODE_FAIL = - "Failed to connect to ConfigNode %s from DataNode %s when executing %s, Exception:"; - private static final long RETRY_INTERVAL_MS = 1000L; - private static final long WAIT_CN_LEADER_ELECTION_INTERVAL_MS = 2000L; private static final long REGISTER_LEADER_WARMING_UP_RETRY_TIMEOUT_MS = 60_000L; private static final String UNSUPPORTED_INVOCATION = DataNodeMiscMessages.UNSUPPORTED_INVOCATION_BY_DATANODE; - private final ThriftClientProperty property; - - private IConfigNodeRPCService.Iface client; - - private TTransport transport; - - private TEndPoint configLeader; - - private List configNodes; - - private TEndPoint configNode; - - private int cursor = 0; - - private boolean isFirstInitiated; - private final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig(); - private final CommonConfig commonConfig = CommonDescriptor.getInstance().getConfig(); - - ClientManager clientManager; - - ConfigRegionId configRegionId = ConfigNodeInfo.CONFIG_REGION_ID; - public ConfigNodeClient( List configNodes, ThriftClientProperty property, ClientManager clientManager) throws TException { - this.configNodes = configNodes; - this.property = property; - this.clientManager = clientManager; - // Set the first configNode as configLeader for a tentative connection - this.configLeader = this.configNodes.get(0); - this.isFirstInitiated = true; - - connectAndSync(); - } - - public void connect(TEndPoint endpoint, int timeoutMs) throws TException { - // Close existing transport before reassigning to prevent connection leaks. - if (transport != null) { - transport.close(); - } - transport = - commonConfig.isEnableInternalSSL() - ? DeepCopyRpcTransportFactory.INSTANCE.getTransport( - endpoint.getIp(), - endpoint.getPort(), - timeoutMs, - commonConfig.getTrustStorePath(), - commonConfig.getTrustStorePwd(), - commonConfig.getKeyStorePath(), - commonConfig.getKeyStorePwd()) - : DeepCopyRpcTransportFactory.INSTANCE.getTransport( - // As there is a try-catch already, we do not need to use TSocket.wrap - endpoint.getIp(), endpoint.getPort(), timeoutMs); - if (!transport.isOpen()) { - transport.open(); - } - configNode = endpoint; - - client = new IConfigNodeRPCService.Client(property.getProtocolFactory().getProtocol(transport)); - } - - private void connectAndSync() throws TException { - try { - tryToConnect(property.getConnectionTimeoutMs()); - } catch (TException e) { - // Can not connect to each config node - syncLatestConfigNodeList(); - tryToConnect(property.getConnectionTimeoutMs()); - } - } - - private void connectAndSync(int timeoutMs) throws TException { - try { - tryToConnect(timeoutMs); - } catch (TException e) { - // Can not connect to each config node - syncLatestConfigNodeList(); - tryToConnect(timeoutMs); - } - } - - private void tryToConnect(int timeoutMs) throws TException { - TException exception = null; - if (configLeader != null) { - try { - connect(configLeader, timeoutMs); - return; - } catch (TException e) { - logger.warn(DataNodeMiscMessages.NODE_LEADER_MAY_DOWN_TRY_NEXT, configLeader); - configLeader = null; - exception = e; - } - } else { - try { - // Wait to start the next try - Thread.sleep(RETRY_INTERVAL_MS); - } catch (InterruptedException ignore) { - Thread.currentThread().interrupt(); - logger.warn(DataNodeMiscMessages.UNEXPECTED_INTERRUPTION_CONNECT_CONFIG_NODE); - } - } - - for (int tryHostNum = 0; tryHostNum < configNodes.size(); tryHostNum++) { - cursor = (cursor + 1) % configNodes.size(); - TEndPoint tryEndpoint = configNodes.get(cursor); - - try { - connect(tryEndpoint, timeoutMs); - return; - } catch (TException e) { - logger.warn(DataNodeMiscMessages.NODE_MAY_DOWN_TRY_NEXT, tryEndpoint); - exception = e; - } - } - if (exception != null - && exception.getCause() != null - && exception.getCause().getCause() != null - && exception.getCause().getCause() instanceof IOException) { - throw new TException(exception.getCause().getCause()); - } - - throw new TException(MSG_RECONNECTION_FAIL); - } - - public TTransport getTransport() { - return transport; - } - - public void syncLatestConfigNodeList() { - configNodes = ConfigNodeInfo.getInstance().getLatestConfigNodes(); - cursor = 0; - } - - @Override - public void close() { - clientManager.returnClient(configRegionId, this); + super(configNodes, property, clientManager); } @Override - public void invalidate() { - Optional.ofNullable(transport).ifPresent(TTransport::close); + protected final String getNodeTypeName() { + return NODE_TYPE_NAME; } @Override - public void invalidateAll() { - clientManager.clear(ConfigNodeInfo.CONFIG_REGION_ID); + protected final TEndPoint getAddressAndPort() { + return config.getAddressAndPort(); } @Override - public boolean printLogWhenEncounterException() { - return property.isPrintLogWhenEncounterException(); - } - - private boolean updateConfigNodeLeader(TSStatus status) { - try { - if (status.getCode() == TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()) { - if (status.isSetRedirectNode()) { - configLeader = - new TEndPoint(status.getRedirectNode().getIp(), status.getRedirectNode().getPort()); - } else { - configLeader = null; - } - if (!isFirstInitiated) { - logger.info( - DataNodeMiscMessages.FAILED_CONNECT_CONFIG_NODE_NOT_LEADER, - configNode, - config.getAddressAndPort()); - } - return true; - } - if (status.getCode() == TSStatusCode.CONFIG_NODE_LEADER_WARMING_UP.getStatusCode()) { - if (!isFirstInitiated) { - logger.info( - DataNodeMiscMessages - .MESSAGE_CONFIGNODE_LEADER_ARG_IS_WARMING_UP_BEFORE_SERVING_DATANODE_ARG_WILL_WAIT_AND_RETRY_REASON_ARG_3A2A4163, - configNode, - config.getAddressAndPort(), - status.getMessage()); - } - try { - Thread.sleep(WAIT_CN_LEADER_ELECTION_INTERVAL_MS); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - logger.warn(DataNodeMiscMessages.UNEXPECTED_INTERRUPTION_CONNECT_CONFIG_NODE_BREAK); - } - return true; - } - return false; - } finally { - isFirstInitiated = false; - } + protected final AbstractConfigNodeInfo getConfigNodeInfo() { + return ConfigNodeInfo.getInstance(); } private boolean isConfigNodeLeaderWarmingUp(TSStatus status) { return status.getCode() == TSStatusCode.CONFIG_NODE_LEADER_WARMING_UP.getStatusCode(); } - /** - * The frame of execute RPC, include logic of retry and exception handling. - * - * @param call which rpc should call - * @param check check the rpc's result - * @return rpc's result - * @param the type of rpc result - * @throws TException if fails more than RETRY_NUM times, throw TException(MSG_RECONNECTION_FAIL) - */ - private T executeRemoteCallWithRetry(final Operation call, final Predicate check) - throws TException { - int detectedNodeNum = 0; - for (int i = 0; i < RETRY_NUM; i++) { - try { - final T result = call.execute(); - if (check.test(result)) { - return result; - } - detectedNodeNum++; - } catch (TException e) { - final String message = - String.format( - MSG_RECONNECTION_DATANODE_FAIL, - configNode, - config.getAddressAndPort(), - Thread.currentThread().getStackTrace()[2].getMethodName()); - logger.warn(message, e); - configLeader = null; - if (e.getCause() != null && e.getCause() instanceof SSLHandshakeException) { - throw e; - } - } - - // If we have detected all configNodes and still not return - if (detectedNodeNum >= configNodes.size()) { - // Clear count - detectedNodeNum = 0; - // Wait to start the next try - try { - Thread.sleep(WAIT_CN_LEADER_ELECTION_INTERVAL_MS); - } catch (InterruptedException ignore) { - Thread.currentThread().interrupt(); - logger.warn(DataNodeMiscMessages.UNEXPECTED_INTERRUPTION_CONNECT_CONFIG_NODE_BREAK); - break; - } - } - - connectAndSync(); - } - throw new TException(MSG_RECONNECTION_FAIL); - } - - @FunctionalInterface - private interface Operation { - T execute() throws TException; - } - - @Override - public TSystemConfigurationResp getSystemConfiguration() throws TException { - return executeRemoteCallWithRetry( - () -> client.getSystemConfiguration(), resp -> !updateConfigNodeLeader(resp.status)); - } - @Override public TGetClusterIdResp getClusterId() throws TException { return executeRemoteCallWithRetry( @@ -534,9 +286,10 @@ public TDataNodeRegisterResp registerDataNode(TDataNodeRegisterReq req) throws T } catch (TException e) { String message = String.format( - MSG_RECONNECTION_DATANODE_FAIL, + ConfigMessages.MSG_RECONNECTION_NODE_FAIL, configNode, - config.getAddressAndPort(), + getNodeTypeName(), + getAddressAndPort(), Thread.currentThread().getStackTrace()[1].getMethodName()); logger.warn(message, e); configLeader = null; @@ -665,6 +418,7 @@ public TShowTTLResp showTTL(TShowTTLReq req) throws TException { () -> client.showTTL(req), resp -> !updateConfigNodeLeader(resp.status)); } + @Override public TSStatus callSpecialProcedure(TTestOperation operation) throws TException { return executeRemoteCallWithRetry( () -> client.callSpecialProcedure(operation), status -> !updateConfigNodeLeader(status)); @@ -810,6 +564,7 @@ public TAuthizedPatternTreeResp fetchAuthizedPatternTree(TCheckUserPrivilegesReq () -> client.fetchAuthizedPatternTree(req), resp -> !updateConfigNodeLeader(resp.status)); } + @Override public TPermissionInfoResp checkRoleOfUser(TAuthorizerReq req) throws TException { return executeRemoteCallWithRetry( () -> client.checkRoleOfUser(req), resp -> !updateConfigNodeLeader(resp.status)); @@ -932,12 +687,6 @@ public TSStatus killQuery(String queryId, int dataNodeId, String allowedUsername status -> !updateConfigNodeLeader(status)); } - @Override - public TGetDataNodeLocationsResp getReadableDataNodeLocations() throws TException { - return executeRemoteCallWithRetry( - () -> client.getReadableDataNodeLocations(), resp -> !updateConfigNodeLeader(resp.status)); - } - @Override public TShowRegionResp showRegion(TShowRegionReq req) throws TException { return executeRemoteCallWithRetry( @@ -1061,6 +810,7 @@ public TGetLocationForTriggerResp getLocationOfStatefulTrigger(String triggerNam resp -> !updateConfigNodeLeader(resp.status)); } + @Override public TGetTriggerTableResp getTriggerTable() throws TException { return executeRemoteCallWithRetry( () -> client.getTriggerTable(), resp -> !updateConfigNodeLeader(resp.status)); @@ -1364,6 +1114,7 @@ public TGetTimeSlotListResp getTimeSlotList(TGetTimeSlotListReq req) throws TExc () -> client.getTimeSlotList(req), resp -> !updateConfigNodeLeader(resp.status)); } + @Override public TCountTimeSlotListResp countTimeSlotList(TCountTimeSlotListReq req) throws TException { return executeRemoteCallWithRetry( () -> client.countTimeSlotList(req), resp -> !updateConfigNodeLeader(resp.status)); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeInfo.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeInfo.java index 27f5143e88460..0188a44312250 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeInfo.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeInfo.java @@ -19,137 +19,19 @@ package org.apache.iotdb.db.protocol.client; -import org.apache.iotdb.common.rpc.thrift.TEndPoint; -import org.apache.iotdb.commons.consensus.ConfigRegionId; -import org.apache.iotdb.commons.exception.BadNodeUrlException; -import org.apache.iotdb.commons.exception.StartupException; -import org.apache.iotdb.commons.file.SystemPropertiesHandler; -import org.apache.iotdb.commons.utils.NodeUrlUtils; +import org.apache.iotdb.commons.client.AbstractConfigNodeInfo; import org.apache.iotdb.db.conf.DataNodeSystemPropertiesHandler; -import org.apache.iotdb.db.i18n.DataNodeMiscMessages; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.IOException; -import java.util.ArrayList; -import java.util.HashSet; -import java.util.List; -import java.util.Properties; -import java.util.Set; -import java.util.concurrent.locks.ReentrantReadWriteLock; - -public class ConfigNodeInfo { - private static final Logger logger = LoggerFactory.getLogger(ConfigNodeInfo.class); - - private static final String CONFIG_NODE_LIST = "config_node_list"; - - private final ReentrantReadWriteLock configNodeInfoReadWriteLock; - - /** latest config nodes. */ - private final Set onlineConfigNodes; - - public static final ConfigRegionId CONFIG_REGION_ID = new ConfigRegionId(0); - - SystemPropertiesHandler systemPropertiesHandler = DataNodeSystemPropertiesHandler.getInstance(); +public class ConfigNodeInfo extends AbstractConfigNodeInfo { + private static final String NODE_TYPE_NAME = "datanode"; private ConfigNodeInfo() { - this.configNodeInfoReadWriteLock = new ReentrantReadWriteLock(); - this.onlineConfigNodes = new HashSet<>(); - } - - public static void reinitializeStatics() { - ConfigNodeInfoHolder.INSTANCE = new ConfigNodeInfo(); - } - - /** Update ConfigNodeList both in memory and system.properties file */ - public boolean updateConfigNodeList(List latestConfigNodes) { - long startTime = System.currentTimeMillis(); - // Check whether the config nodes are latest or not - configNodeInfoReadWriteLock.readLock().lock(); - try { - if (onlineConfigNodes.size() == latestConfigNodes.size() - && onlineConfigNodes.containsAll(latestConfigNodes)) { - return true; - } - } finally { - configNodeInfoReadWriteLock.readLock().unlock(); - } - - // Update config nodes - configNodeInfoReadWriteLock.writeLock().lock(); - try { - onlineConfigNodes.clear(); - onlineConfigNodes.addAll(latestConfigNodes); - storeConfigNodeList(); - long endTime = System.currentTimeMillis(); - logger.info( - DataNodeMiscMessages.UPDATE_CONFIG_NODE_SUCCESSFULLY, - onlineConfigNodes, - (endTime - startTime)); - } catch (IOException e) { - logger.error(DataNodeMiscMessages.UPDATE_CONFIG_NODE_FAILED, e); - return false; - } finally { - configNodeInfoReadWriteLock.writeLock().unlock(); - } - return true; - } - - /** - * Call this method to store config node list. - * - * @throws IOException if properties deserialization or configNode list serialization failed. - */ - public void storeConfigNodeList() throws IOException { - if (!systemPropertiesHandler.fileExist()) { - logger.info(DataNodeMiscMessages.SYSTEM_PROPERTIES_NOT_EXIST); - return; - } - systemPropertiesHandler.put( - CONFIG_NODE_LIST, NodeUrlUtils.convertTEndPointUrls(new ArrayList<>(onlineConfigNodes))); + super(DataNodeSystemPropertiesHandler.getInstance()); } - public void loadConfigNodeList() throws StartupException { - long startTime = System.currentTimeMillis(); - // properties contain CONFIG_NODE_LIST only when start as Data node - configNodeInfoReadWriteLock.writeLock().lock(); - try { - Properties properties = systemPropertiesHandler.read(); - - if (properties.containsKey(CONFIG_NODE_LIST)) { - onlineConfigNodes.clear(); - onlineConfigNodes.addAll( - NodeUrlUtils.parseTEndPointUrls(properties.getProperty(CONFIG_NODE_LIST))); - } - if (onlineConfigNodes.isEmpty()) { - throw new StartupException( - DataNodeMiscMessages - .MISC_EXCEPTION_REMOVING_IS_ONLY_ALLOWED_IN_AN_ENVIRONMENT_WHERE_THE_DATANODE_5A3E1FEA); - } - long endTime = System.currentTimeMillis(); - logger.info( - DataNodeMiscMessages.LOAD_CONFIG_NODE_SUCCESSFULLY, - onlineConfigNodes, - (endTime - startTime)); - } catch (IOException e) { - throw new RuntimeException(e); - } catch (BadNodeUrlException e) { - logger.error(DataNodeMiscMessages.CANNOT_PARSE_CONFIG_NODE_LIST); - } finally { - configNodeInfoReadWriteLock.writeLock().unlock(); - } - } - - public List getLatestConfigNodes() { - List result; - configNodeInfoReadWriteLock.readLock().lock(); - try { - result = new ArrayList<>(onlineConfigNodes); - } finally { - configNodeInfoReadWriteLock.readLock().unlock(); - } - return result; + @Override + protected String getNodeTypeName() { + return NODE_TYPE_NAME; } private static class ConfigNodeInfoHolder { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java index 760c5747c1af5..5ce282db7a701 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java @@ -598,7 +598,7 @@ private void sendRegisterRequestToConfigNode(boolean isPreCheck) /* Store runtime configurations when register success */ int dataNodeID = dataNodeRegisterResp.getDataNodeId(); config.setDataNodeId(dataNodeID); - IoTDBStartCheck.getInstance().serializeDataNodeId(dataNodeID); + IoTDBStartCheck.getInstance().serializeNodeId(dataNodeID); storeRuntimeConfigurations( dataNodeRegisterResp.getConfigNodeList(), dataNodeRegisterResp.getRuntimeConfiguration()); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java index ca6048394ca76..6dd082c0efa4b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java @@ -28,6 +28,7 @@ import org.apache.iotdb.commons.cluster.NodeStatus; import org.apache.iotdb.commons.concurrent.ThreadName; import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.disk.utils.DirectoryChecker; import org.apache.iotdb.commons.pipe.config.PipeConfig; import org.apache.iotdb.consensus.ConsensusFactory; import org.apache.iotdb.consensus.exception.ConsensusException; @@ -45,7 +46,6 @@ import org.apache.iotdb.db.protocol.client.ConfigNodeInfo; import org.apache.iotdb.db.storageengine.StorageEngine; import org.apache.iotdb.db.storageengine.dataregion.wal.WALManager; -import org.apache.iotdb.db.storageengine.rescon.disk.DirectoryChecker; import org.apache.iotdb.db.utils.MemUtils; import org.apache.iotdb.rpc.TSStatusCode; diff --git a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java index 94194a086b47c..08ab3211d9d20 100644 --- a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java @@ -130,6 +130,16 @@ public final class CommonMessages { "unable to stop auto releaser after {} ms"; public static final String UNABLE_TO_STOP_EVICTOR = "unable to stop evictor after {} ms"; + // --- startcheck / system.properties --- + public static final String IOTDB_VERSION_TOO_OLD = "IoTDB version is too old"; + public static final String REPAIR_SYSTEM_PROPERTIES = "repair system.properties, lack {}"; + public static final String PARAMETER_CANNOT_BE_MODIFIED_AFTER_FIRST_STARTUP_FMT = + "%s can't be modified after first startup"; + public static final String MISC_LOG_SERIALIZE_MUTABLE_SYSTEM_PROPERTIES_SUCCESSFULLY_WHICH_TAKES_4656A206 = + "Serialize mutable system properties successfully, which takes {} ms."; + public static final String MISC_LOG_DO_NOT_UPGRADE_IOTDB_FROM_V0_9_OR_LOWER_VERSION_TO_V1_0_9878EC88 = + "DO NOT UPGRADE IoTDB from v0.9 or lower version to v1.0! Please upgrade to v0.10 first"; + private CommonMessages() {} public static final String COLLECTION_MUST_NOT_BE_NULL = "Collection must not be null."; diff --git a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/ConfigMessages.java b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/ConfigMessages.java index 8ce3e9047886f..e974dfeb52833 100644 --- a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/ConfigMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/ConfigMessages.java @@ -73,6 +73,45 @@ public final class ConfigMessages { + " There may have been an unexpected interruption in the last" + " configuration file update. Ignore temporary file {}"; + // --------------------------------------------------------------------------- + // protocol – ConfigNodeInfo + // --------------------------------------------------------------------------- + public static final String UPDATE_CONFIG_NODE_SUCCESSFULLY = + "Update ConfigNode Successfully: {}, which takes {} ms."; + public static final String UPDATE_CONFIG_NODE_FAILED = "Update ConfigNode failed."; + public static final String SYSTEM_PROPERTIES_NOT_EXIST = + "System properties file not exist, not necessary to store ConfigNode list"; + public static final String LOAD_CONFIG_NODE_SUCCESSFULLY = + "Load ConfigNode successfully: {}, which takes {} ms."; + public static final String CANNOT_PARSE_CONFIG_NODE_LIST = + "Cannot parse config node list in system.properties"; + public static final String MISC_EXCEPTION_REMOVING_IS_ONLY_ALLOWED_IN_AN_ENVIRONMENT_WHEN_NODE_STARTED_2ACA2BD0 = + "Removing is only allowed in an environment where the %s has been successfully " + + "started. Please check whether it is removed on the confignode, or if you have " + + "deleted the system.properties file by mistake."; + + // --------------------------------------------------------------------------- + // protocol – ConfigNodeClient + // --------------------------------------------------------------------------- + public static final String MSG_RECONNECTION_FAIL = + "Fail to connect to any config node. Please check status of ConfigNodes or logs of " + + "connected %s"; + public static final String MSG_RECONNECTION_NODE_FAIL = + "Failed to connect to ConfigNode %s from %s %s when executing %s, Exception:"; + public static final String NODE_LEADER_MAY_DOWN_TRY_NEXT = + "The current node leader may have been down {}, try next node"; + public static final String UNEXPECTED_INTERRUPTION_CONNECT_CONFIG_NODE = + "Unexpected interruption when waiting to try to connect to ConfigNode"; + public static final String NODE_MAY_DOWN_TRY_NEXT = + "The current node may have been down {},try next node"; + public static final String FAILED_CONNECT_CONFIG_NODE_NOT_LEADER = + "Failed to connect to ConfigNode {} from {} {}, because the current node is not leader " + + "or not ready yet, will try again later"; + public static final String UNEXPECTED_INTERRUPTION_CONNECT_CONFIG_NODE_BREAK = + "Unexpected interruption when waiting to try to connect to ConfigNode, may because current node has been down. Will break current execution process to avoid meaningless wait."; + public static final String MESSAGE_CONFIGNODE_LEADER_ARG_IS_WARMING_UP_BEFORE_SERVING_NODE_ARG_WILL_WAIT_AND_RETRY_REASON_ARG_2ACA2BD0 = + "ConfigNode leader {} is warming up before serving {} {}, will wait and retry. Reason: {}"; + private ConfigMessages() {} // --------------------------------------------------------------------------- // Additional auto-collected messages diff --git a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/StorageMessages.java b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/StorageMessages.java new file mode 100644 index 0000000000000..e0a2140e30427 --- /dev/null +++ b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/StorageMessages.java @@ -0,0 +1,38 @@ +/* + * 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.commons.i18n; + +public final class StorageMessages { + + private StorageMessages() {} + + // ======================== Directory ======================== + + public static final String FAILED_TO_DEREGISTER_FILE_LOCK = + "Failed to deregister file lock because {}"; + public static final String STORAGE_EXCEPTION_UNABLE_TO_CREATE_DIRECTORY_S_BECAUSE_THERE_IS_FILE_UNDER_1C59ACFC = + "Unable to create directory %s because there is file under the path, please check " + + "configuration and restart."; + public static final String STORAGE_EXCEPTION_UNABLE_TO_CREATE_DIRECTORY_S_PLEASE_CHECK_CONFIGURATION_BA580B67 = + "Unable to create directory %s, please check configuration and restart."; + public static final String STORAGE_EXCEPTION_CONFLICT_IS_DETECTED_IN_DIRECTORY_S_WHICH_MAY_BE_BEING_USED_CB5C77FC = + "Conflict is detected in directory %s, which may be being used by another IoTDB " + + "(ProcessId=%s). Please check configuration and restart."; +} \ No newline at end of file diff --git a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java index 07d886a44f3d4..2c7766d419989 100644 --- a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java @@ -126,6 +126,16 @@ public final class CommonMessages { "在 {} 毫秒后仍无法停止自动释放器"; public static final String UNABLE_TO_STOP_EVICTOR = "在 {} 毫秒后仍无法停止驱逐器"; + // --- startcheck / system.properties --- + public static final String IOTDB_VERSION_TOO_OLD = "IoTDB 版本过旧"; + public static final String REPAIR_SYSTEM_PROPERTIES = "修复 system.properties，缺少 {}"; + public static final String PARAMETER_CANNOT_BE_MODIFIED_AFTER_FIRST_STARTUP_FMT = + "%s 首次启动后不能修改"; + public static final String MISC_LOG_SERIALIZE_MUTABLE_SYSTEM_PROPERTIES_SUCCESSFULLY_WHICH_TAKES_4656A206 = + "成功序列化 mutable system properties，耗时 {} ms。"; + public static final String MISC_LOG_DO_NOT_UPGRADE_IOTDB_FROM_V0_9_OR_LOWER_VERSION_TO_V1_0_9878EC88 = + "请勿将 IoTDB 从 v0.9 或更低版本直接升级到 v1.0！请先升级到 v0.10"; + private CommonMessages() {} public static final String COLLECTION_MUST_NOT_BE_NULL = "集合不能为空。"; diff --git a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/ConfigMessages.java b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/ConfigMessages.java index f8b810fde13df..2ffaaff811e81 100644 --- a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/ConfigMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/ConfigMessages.java @@ -73,6 +73,39 @@ public final class ConfigMessages { + "上一次配置文件更新可能发生了意外中断。" + "忽略临时文件 {}"; + // --------------------------------------------------------------------------- + // protocol – ConfigNodeInfo + // --------------------------------------------------------------------------- + public static final String UPDATE_CONFIG_NODE_SUCCESSFULLY = + "成功更新 ConfigNode：{}，耗时 {} 毫秒。"; + public static final String UPDATE_CONFIG_NODE_FAILED = "更新 ConfigNode 失败。"; + public static final String SYSTEM_PROPERTIES_NOT_EXIST = + "系统属性文件不存在，无需存储 ConfigNode 列表"; + public static final String LOAD_CONFIG_NODE_SUCCESSFULLY = + "成功加载 ConfigNode：{}，耗时 {} 毫秒。"; + public static final String CANNOT_PARSE_CONFIG_NODE_LIST = + "无法解析 system.properties 中的 ConfigNode 列表"; + public static final String MISC_EXCEPTION_REMOVING_IS_ONLY_ALLOWED_IN_AN_ENVIRONMENT_WHEN_NODE_STARTED_2ACA2BD0 = + "只有在 %s 已成功启动的环境中才允许移除。请检查它是否已在 ConfigNode 上移除，或是否误删了 system.properties 文件。"; + + // --- ConfigNodeClient --- + public static final String MSG_RECONNECTION_FAIL = + "无法连接到任何 config node。请检查 ConfigNodes 的状态或已连接 %s 的日志"; + public static final String MSG_RECONNECTION_NODE_FAIL = + "连接 ConfigNode %s 失败：从 %s %s 执行 %s 时，异常:"; + public static final String NODE_LEADER_MAY_DOWN_TRY_NEXT = + "当前节点 leader 可能已宕机 {}，尝试下一个节点"; + public static final String UNEXPECTED_INTERRUPTION_CONNECT_CONFIG_NODE = + "等待尝试连接 ConfigNode 时发生意外中断"; + public static final String NODE_MAY_DOWN_TRY_NEXT = + "当前节点可能已宕机 {}，尝试下一个节点"; + public static final String FAILED_CONNECT_CONFIG_NODE_NOT_LEADER = + "连接 ConfigNode {} 失败：从 {} {} 发起，因为当前节点不是 leader 或尚未就绪，稍后将重试"; + public static final String UNEXPECTED_INTERRUPTION_CONNECT_CONFIG_NODE_BREAK = + "等待尝试连接 ConfigNode 时发生意外中断，可能因为当前节点已宕机。将中断当前执行流程以避免无意义的等待"; + public static final String MESSAGE_CONFIGNODE_LEADER_ARG_IS_WARMING_UP_BEFORE_SERVING_NODE_ARG_WILL_WAIT_AND_RETRY_REASON_ARG_2ACA2BD0 = + "ConfigNode leader {} 正在预热，暂未对 {} {} 提供服务，将等待并重试。原因：{}"; + private ConfigMessages() {} // --------------------------------------------------------------------------- // Additional auto-collected messages diff --git a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/StorageMessages.java b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/StorageMessages.java new file mode 100644 index 0000000000000..a5f0a5cd241f6 --- /dev/null +++ b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/StorageMessages.java @@ -0,0 +1,35 @@ +/* + * 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.commons.i18n; + +public final class StorageMessages { + + private StorageMessages() {} + + // ======================== Directory ======================== + + public static final String FAILED_TO_DEREGISTER_FILE_LOCK = "注销文件锁失败，原因: {}"; + public static final String STORAGE_EXCEPTION_UNABLE_TO_CREATE_DIRECTORY_S_BECAUSE_THERE_IS_FILE_UNDER_1C59ACFC = + "无法创建目录 %s，原因：该路径下存在文件，请检查配置并重启。"; + public static final String STORAGE_EXCEPTION_UNABLE_TO_CREATE_DIRECTORY_S_PLEASE_CHECK_CONFIGURATION_BA580B67 = + "无法创建目录 %s，请检查配置并重启。"; + public static final String STORAGE_EXCEPTION_CONFLICT_IS_DETECTED_IN_DIRECTORY_S_WHICH_MAY_BE_BEING_USED_CB5C77FC = + "目录 %s 检测到冲突，可能正被另一个 IoTDB 使用 (ProcessId=%s)。请检查配置并重启。"; +} \ No newline at end of file diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/AbstractConfigNodeClient.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/AbstractConfigNodeClient.java new file mode 100644 index 0000000000000..a95bf95488878 --- /dev/null +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/AbstractConfigNodeClient.java @@ -0,0 +1,1331 @@ +/* + * 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.commons.client; + +import org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation; +import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation; +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.common.rpc.thrift.TExternalServiceListResp; +import org.apache.iotdb.common.rpc.thrift.TFlushReq; +import org.apache.iotdb.common.rpc.thrift.TNodeLocations; +import org.apache.iotdb.common.rpc.thrift.TPipeHeartbeatResp; +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot; +import org.apache.iotdb.common.rpc.thrift.TSetConfigurationReq; +import org.apache.iotdb.common.rpc.thrift.TSetSpaceQuotaReq; +import org.apache.iotdb.common.rpc.thrift.TSetTTLReq; +import org.apache.iotdb.common.rpc.thrift.TSetThrottleQuotaReq; +import org.apache.iotdb.common.rpc.thrift.TShowAppliedConfigurationsResp; +import org.apache.iotdb.common.rpc.thrift.TShowConfigurationResp; +import org.apache.iotdb.common.rpc.thrift.TShowTTLReq; +import org.apache.iotdb.common.rpc.thrift.TTestConnectionResp; +import org.apache.iotdb.commons.client.property.ThriftClientProperty; +import org.apache.iotdb.commons.conf.CommonConfig; +import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.consensus.ConfigRegionId; +import org.apache.iotdb.commons.i18n.ConfigMessages; +import org.apache.iotdb.confignode.rpc.thrift.IConfigNodeRPCService; +import org.apache.iotdb.confignode.rpc.thrift.TAINodeConfigurationResp; +import org.apache.iotdb.confignode.rpc.thrift.TAINodeRegisterReq; +import org.apache.iotdb.confignode.rpc.thrift.TAINodeRegisterResp; +import org.apache.iotdb.confignode.rpc.thrift.TAINodeRemoveReq; +import org.apache.iotdb.confignode.rpc.thrift.TAINodeRestartReq; +import org.apache.iotdb.confignode.rpc.thrift.TAINodeRestartResp; +import org.apache.iotdb.confignode.rpc.thrift.TAddConsensusGroupReq; +import org.apache.iotdb.confignode.rpc.thrift.TAlterEncodingCompressorReq; +import org.apache.iotdb.confignode.rpc.thrift.TAlterLogicalViewReq; +import org.apache.iotdb.confignode.rpc.thrift.TAlterOrDropTableReq; +import org.apache.iotdb.confignode.rpc.thrift.TAlterPipeReq; +import org.apache.iotdb.confignode.rpc.thrift.TAlterSchemaTemplateReq; +import org.apache.iotdb.confignode.rpc.thrift.TAlterTimeSeriesReq; +import org.apache.iotdb.confignode.rpc.thrift.TAlterTopicReq; +import org.apache.iotdb.confignode.rpc.thrift.TAuthizedPatternTreeResp; +import org.apache.iotdb.confignode.rpc.thrift.TAuthorizerRelationalReq; +import org.apache.iotdb.confignode.rpc.thrift.TAuthorizerReq; +import org.apache.iotdb.confignode.rpc.thrift.TAuthorizerResp; +import org.apache.iotdb.confignode.rpc.thrift.TCheckUserPrivilegesReq; +import org.apache.iotdb.confignode.rpc.thrift.TCloseConsumerReq; +import org.apache.iotdb.confignode.rpc.thrift.TConfigNodeHeartbeatReq; +import org.apache.iotdb.confignode.rpc.thrift.TConfigNodeHeartbeatResp; +import org.apache.iotdb.confignode.rpc.thrift.TConfigNodeRegisterReq; +import org.apache.iotdb.confignode.rpc.thrift.TConfigNodeRegisterResp; +import org.apache.iotdb.confignode.rpc.thrift.TCountDatabaseResp; +import org.apache.iotdb.confignode.rpc.thrift.TCountTimeSlotListReq; +import org.apache.iotdb.confignode.rpc.thrift.TCountTimeSlotListResp; +import org.apache.iotdb.confignode.rpc.thrift.TCreateCQReq; +import org.apache.iotdb.confignode.rpc.thrift.TCreateConsumerReq; +import org.apache.iotdb.confignode.rpc.thrift.TCreateExternalServiceReq; +import org.apache.iotdb.confignode.rpc.thrift.TCreateFunctionReq; +import org.apache.iotdb.confignode.rpc.thrift.TCreatePipePluginReq; +import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq; +import org.apache.iotdb.confignode.rpc.thrift.TCreateSchemaTemplateReq; +import org.apache.iotdb.confignode.rpc.thrift.TCreateTableViewReq; +import org.apache.iotdb.confignode.rpc.thrift.TCreateTopicReq; +import org.apache.iotdb.confignode.rpc.thrift.TCreateTriggerReq; +import org.apache.iotdb.confignode.rpc.thrift.TDataNodeConfigurationResp; +import org.apache.iotdb.confignode.rpc.thrift.TDataNodeLeaseRecoveryResp; +import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRegisterReq; +import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRegisterResp; +import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRemoveReq; +import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRemoveResp; +import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRestartReq; +import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRestartResp; +import org.apache.iotdb.confignode.rpc.thrift.TDataPartitionReq; +import org.apache.iotdb.confignode.rpc.thrift.TDataPartitionTableResp; +import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema; +import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchemaResp; +import org.apache.iotdb.confignode.rpc.thrift.TDeactivateSchemaTemplateReq; +import org.apache.iotdb.confignode.rpc.thrift.TDeleteDatabaseReq; +import org.apache.iotdb.confignode.rpc.thrift.TDeleteDatabasesReq; +import org.apache.iotdb.confignode.rpc.thrift.TDeleteLogicalViewReq; +import org.apache.iotdb.confignode.rpc.thrift.TDeleteTableDeviceReq; +import org.apache.iotdb.confignode.rpc.thrift.TDeleteTableDeviceResp; +import org.apache.iotdb.confignode.rpc.thrift.TDeleteTimeSeriesReq; +import org.apache.iotdb.confignode.rpc.thrift.TDescTable4InformationSchemaResp; +import org.apache.iotdb.confignode.rpc.thrift.TDescTableResp; +import org.apache.iotdb.confignode.rpc.thrift.TDropCQReq; +import org.apache.iotdb.confignode.rpc.thrift.TDropFunctionReq; +import org.apache.iotdb.confignode.rpc.thrift.TDropPipePluginReq; +import org.apache.iotdb.confignode.rpc.thrift.TDropPipeReq; +import org.apache.iotdb.confignode.rpc.thrift.TDropSubscriptionReq; +import org.apache.iotdb.confignode.rpc.thrift.TDropTopicReq; +import org.apache.iotdb.confignode.rpc.thrift.TDropTriggerReq; +import org.apache.iotdb.confignode.rpc.thrift.TExtendRegionReq; +import org.apache.iotdb.confignode.rpc.thrift.TFetchTableResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetAINodeLocationResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetAllPipeInfoResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetAllSubscriptionInfoResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetAllTemplatesResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetAllTopicInfoResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetClusterIdResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetCommitProgressReq; +import org.apache.iotdb.confignode.rpc.thrift.TGetCommitProgressResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetDataNodeLocationsResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetDatabaseReq; +import org.apache.iotdb.confignode.rpc.thrift.TGetJarInListReq; +import org.apache.iotdb.confignode.rpc.thrift.TGetJarInListResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetLocationForTriggerResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetPathsSetTemplatesReq; +import org.apache.iotdb.confignode.rpc.thrift.TGetPathsSetTemplatesResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetPipePluginTableResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetRegionGroupsByTimeReq; +import org.apache.iotdb.confignode.rpc.thrift.TGetRegionGroupsByTimeResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetRegionIdReq; +import org.apache.iotdb.confignode.rpc.thrift.TGetRegionIdResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetSeriesSlotListReq; +import org.apache.iotdb.confignode.rpc.thrift.TGetSeriesSlotListResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetTemplateResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetTimeSlotListReq; +import org.apache.iotdb.confignode.rpc.thrift.TGetTimeSlotListResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetTriggerTableResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetUDFTableResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetUdfTableReq; +import org.apache.iotdb.confignode.rpc.thrift.TLoginReq; +import org.apache.iotdb.confignode.rpc.thrift.TMigrateRegionReq; +import org.apache.iotdb.confignode.rpc.thrift.TPermissionInfoResp; +import org.apache.iotdb.confignode.rpc.thrift.TPipeConfigTransferReq; +import org.apache.iotdb.confignode.rpc.thrift.TPipeConfigTransferResp; +import org.apache.iotdb.confignode.rpc.thrift.TReconstructRegionReq; +import org.apache.iotdb.confignode.rpc.thrift.TRegionRouteMapResp; +import org.apache.iotdb.confignode.rpc.thrift.TRemoveRegionReq; +import org.apache.iotdb.confignode.rpc.thrift.TSchemaNodeManagementReq; +import org.apache.iotdb.confignode.rpc.thrift.TSchemaNodeManagementResp; +import org.apache.iotdb.confignode.rpc.thrift.TSchemaPartitionReq; +import org.apache.iotdb.confignode.rpc.thrift.TSchemaPartitionTableResp; +import org.apache.iotdb.confignode.rpc.thrift.TSetDataNodeStatusReq; +import org.apache.iotdb.confignode.rpc.thrift.TSetDataReplicationFactorReq; +import org.apache.iotdb.confignode.rpc.thrift.TSetSchemaReplicationFactorReq; +import org.apache.iotdb.confignode.rpc.thrift.TSetSchemaTemplateReq; +import org.apache.iotdb.confignode.rpc.thrift.TSetTimePartitionIntervalReq; +import org.apache.iotdb.confignode.rpc.thrift.TShowAINodesResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowCQResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowClusterResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowConfigNodes4InformationSchemaResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowConfigNodesResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowDataNodes4InformationSchemaResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowDataNodesResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowDatabaseResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowPipePluginReq; +import org.apache.iotdb.confignode.rpc.thrift.TShowPipeReq; +import org.apache.iotdb.confignode.rpc.thrift.TShowPipeResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowRegionReq; +import org.apache.iotdb.confignode.rpc.thrift.TShowRegionResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowRepairDataPartitionTableProgressResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowSubscriptionReq; +import org.apache.iotdb.confignode.rpc.thrift.TShowSubscriptionResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowTTLResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowTable4InformationSchemaResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowTableResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowThrottleReq; +import org.apache.iotdb.confignode.rpc.thrift.TShowTopicReq; +import org.apache.iotdb.confignode.rpc.thrift.TShowTopicResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowVariablesResp; +import org.apache.iotdb.confignode.rpc.thrift.TSpaceQuotaResp; +import org.apache.iotdb.confignode.rpc.thrift.TStartPipeReq; +import org.apache.iotdb.confignode.rpc.thrift.TStopPipeReq; +import org.apache.iotdb.confignode.rpc.thrift.TSubscribeReq; +import org.apache.iotdb.confignode.rpc.thrift.TSystemConfigurationResp; +import org.apache.iotdb.confignode.rpc.thrift.TTestOperation; +import org.apache.iotdb.confignode.rpc.thrift.TThrottleQuotaResp; +import org.apache.iotdb.confignode.rpc.thrift.TUnsetSchemaTemplateReq; +import org.apache.iotdb.confignode.rpc.thrift.TUnsubscribeReq; +import org.apache.iotdb.rpc.DeepCopyRpcTransportFactory; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.apache.thrift.TException; +import org.apache.thrift.transport.TTransport; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.net.ssl.SSLHandshakeException; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.function.Predicate; + +public abstract class AbstractConfigNodeClient> + implements IConfigNodeRPCService.Iface, ThriftClient, AutoCloseable { + + private static final Logger LOGGER = LoggerFactory.getLogger(AbstractConfigNodeClient.class); + + protected static final int RETRY_NUM = 15; + protected static final long RETRY_INTERVAL_MS = 1000L; + protected static final long WAIT_CN_LEADER_ELECTION_INTERVAL_MS = 2000L; + + protected final ThriftClientProperty property; + + protected IConfigNodeRPCService.Iface client; + + protected TTransport transport; + + protected TEndPoint configLeader; + + protected List configNodes; + + protected TEndPoint configNode; + + protected int cursor = 0; + + protected boolean isFirstInitiated; + + protected final CommonConfig commonConfig = CommonDescriptor.getInstance().getConfig(); + + protected final ClientManager clientManager; + + protected final ConfigRegionId configRegionId = AbstractConfigNodeInfo.CONFIG_REGION_ID; + + protected AbstractConfigNodeClient( + List configNodes, + ThriftClientProperty property, + ClientManager clientManager) + throws TException { + this.configNodes = configNodes; + this.property = property; + this.clientManager = clientManager; + // Set the first configNode as configLeader for a tentative connection + this.configLeader = this.configNodes.get(0); + this.isFirstInitiated = true; + + connectAndSync(); + } + + public TTransport getTransport() { + return transport; + } + + public void connect(TEndPoint endpoint, int timeoutMs) throws TException { + // Close existing transport before reassigning to prevent connection leaks. + if (transport != null) { + transport.close(); + } + transport = + commonConfig.isEnableInternalSSL() + ? DeepCopyRpcTransportFactory.INSTANCE.getTransport( + endpoint.getIp(), + endpoint.getPort(), + timeoutMs, + commonConfig.getTrustStorePath(), + commonConfig.getTrustStorePwd(), + commonConfig.getKeyStorePath(), + commonConfig.getKeyStorePwd()) + : DeepCopyRpcTransportFactory.INSTANCE.getTransport( + // As there is a try-catch already, we do not need to use TSocket.wrap + endpoint.getIp(), endpoint.getPort(), timeoutMs); + if (!transport.isOpen()) { + transport.open(); + } + configNode = endpoint; + + client = new IConfigNodeRPCService.Client(property.getProtocolFactory().getProtocol(transport)); + } + + protected final void connectAndSync() throws TException { + try { + tryToConnect(property.getConnectionTimeoutMs()); + } catch (TException e) { + // Can not connect to each config node + syncLatestConfigNodeList(); + tryToConnect(property.getConnectionTimeoutMs()); + } + } + + protected final void connectAndSync(int timeoutMs) throws TException { + try { + tryToConnect(timeoutMs); + } catch (TException e) { + // Can not connect to each config node + syncLatestConfigNodeList(); + tryToConnect(timeoutMs); + } + } + + private void tryToConnect(int timeoutMs) throws TException { + TException exception = null; + if (configLeader != null) { + try { + connect(configLeader, timeoutMs); + return; + } catch (TException e) { + LOGGER.warn(ConfigMessages.NODE_LEADER_MAY_DOWN_TRY_NEXT, configLeader); + configLeader = null; + exception = e; + } + } else { + try { + // Wait to start the next try + Thread.sleep(RETRY_INTERVAL_MS); + } catch (InterruptedException ignore) { + Thread.currentThread().interrupt(); + LOGGER.warn(ConfigMessages.UNEXPECTED_INTERRUPTION_CONNECT_CONFIG_NODE); + } + } + + for (int tryHostNum = 0; tryHostNum < configNodes.size(); tryHostNum++) { + cursor = (cursor + 1) % configNodes.size(); + TEndPoint tryEndpoint = configNodes.get(cursor); + + try { + connect(tryEndpoint, timeoutMs); + return; + } catch (TException e) { + LOGGER.warn(ConfigMessages.NODE_MAY_DOWN_TRY_NEXT, tryEndpoint); + exception = e; + } + } + if (exception != null + && exception.getCause() != null + && exception.getCause().getCause() != null + && exception.getCause().getCause() instanceof IOException) { + throw new TException(exception.getCause().getCause()); + } + + throw new TException(String.format(ConfigMessages.MSG_RECONNECTION_FAIL, getNodeTypeName())); + } + + public void syncLatestConfigNodeList() { + configNodes = getConfigNodeInfo().getLatestConfigNodes(); + cursor = 0; + } + + @Override + public void close() { + clientManager.returnClient(configRegionId, (C) this); + } + + @Override + public void invalidate() { + Optional.ofNullable(transport).ifPresent(TTransport::close); + } + + @Override + public void invalidateAll() { + clientManager.clear(AbstractConfigNodeInfo.CONFIG_REGION_ID); + } + + @Override + public boolean printLogWhenEncounterException() { + return property.isPrintLogWhenEncounterException(); + } + + protected final boolean updateConfigNodeLeader(TSStatus status) { + try { + if (status.getCode() == TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()) { + if (status.isSetRedirectNode()) { + configLeader = + new TEndPoint(status.getRedirectNode().getIp(), status.getRedirectNode().getPort()); + } else { + configLeader = null; + } + if (!isFirstInitiated) { + LOGGER.info( + ConfigMessages.FAILED_CONNECT_CONFIG_NODE_NOT_LEADER, + configNode, + getNodeTypeName(), + getAddressAndPort()); + } + return true; + } + if (status.getCode() == TSStatusCode.CONFIG_NODE_LEADER_WARMING_UP.getStatusCode()) { + if (!isFirstInitiated) { + LOGGER.info( + ConfigMessages + .MESSAGE_CONFIGNODE_LEADER_ARG_IS_WARMING_UP_BEFORE_SERVING_NODE_ARG_WILL_WAIT_AND_RETRY_REASON_ARG_2ACA2BD0, + configNode, + getNodeTypeName(), + getAddressAndPort(), + status.getMessage()); + } + try { + Thread.sleep(WAIT_CN_LEADER_ELECTION_INTERVAL_MS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + LOGGER.warn(ConfigMessages.UNEXPECTED_INTERRUPTION_CONNECT_CONFIG_NODE_BREAK); + } + return true; + } + return false; + } finally { + isFirstInitiated = false; + } + } + + /** + * The frame of execute RPC, include logic of retry and exception handling. + * + * @param call which rpc should call + * @param check check the rpc's result + * @return rpc's result + * @param the type of rpc result + * @throws TException if fails more than RETRY_NUM times, throw TException(MSG_RECONNECTION_FAIL) + */ + protected final R executeRemoteCallWithRetry( + final Operation call, final Predicate check) throws TException { + int detectedNodeNum = 0; + for (int i = 0; i < RETRY_NUM; i++) { + try { + final R result = call.execute(); + if (check.test(result)) { + return result; + } + detectedNodeNum++; + } catch (TException e) { + final String message = + String.format( + ConfigMessages.MSG_RECONNECTION_NODE_FAIL, + configNode, + getNodeTypeName(), + getAddressAndPort(), + Thread.currentThread().getStackTrace()[2].getMethodName()); + LOGGER.warn(message, e); + configLeader = null; + if (e.getCause() != null && e.getCause() instanceof SSLHandshakeException) { + throw e; + } + } + + // If we have detected all configNodes and still not return + if (detectedNodeNum >= configNodes.size()) { + // Clear count + detectedNodeNum = 0; + // Wait to start the next try + try { + Thread.sleep(WAIT_CN_LEADER_ELECTION_INTERVAL_MS); + } catch (InterruptedException ignore) { + Thread.currentThread().interrupt(); + LOGGER.warn(ConfigMessages.UNEXPECTED_INTERRUPTION_CONNECT_CONFIG_NODE_BREAK); + break; + } + } + + connectAndSync(); + } + throw new TException(String.format(ConfigMessages.MSG_RECONNECTION_FAIL, getNodeTypeName())); + } + + @Override + public TSystemConfigurationResp getSystemConfiguration() throws TException { + return executeRemoteCallWithRetry( + () -> client.getSystemConfiguration(), resp -> !updateConfigNodeLeader(resp.status)); + } + + @Override + public TGetDataNodeLocationsResp getReadableDataNodeLocations() throws TException { + return executeRemoteCallWithRetry( + () -> client.getReadableDataNodeLocations(), resp -> !updateConfigNodeLeader(resp.status)); + } + + @FunctionalInterface + protected interface Operation { + R execute() throws TException; + } + + /** The node type name used in log messages, e.g. DataNode, StreamNode. */ + protected abstract String getNodeTypeName(); + + /** The address and port of the current node, used in log messages. */ + protected abstract TEndPoint getAddressAndPort(); + + /** The ConfigNodeInfo singleton of the current node. */ + protected abstract AbstractConfigNodeInfo getConfigNodeInfo(); + + @Override + public TGetClusterIdResp getClusterId() throws TException { + return null; + } + + @Override + public TDataNodeRegisterResp registerDataNode(TDataNodeRegisterReq req) throws TException { + return null; + } + + @Override + public TDataNodeRestartResp restartDataNode(TDataNodeRestartReq req) throws TException { + return null; + } + + @Override + public TDataNodeLeaseRecoveryResp reloadCacheAfterLeaseRecovery() throws TException { + return null; + } + + @Override + public TAINodeRegisterResp registerAINode(TAINodeRegisterReq req) throws TException { + return null; + } + + @Override + public TAINodeRestartResp restartAINode(TAINodeRestartReq req) throws TException { + return null; + } + + @Override + public TSStatus removeAINode(TAINodeRemoveReq req) throws TException { + return null; + } + + @Override + public TShowAINodesResp showAINodes() throws TException { + return null; + } + + @Override + public TAINodeConfigurationResp getAINodeConfiguration(int aiNodeId) throws TException { + return null; + } + + @Override + public TGetAINodeLocationResp getAINodeLocation() throws TException { + return null; + } + + @Override + public TDataNodeRemoveResp removeDataNode(TDataNodeRemoveReq req) throws TException { + return null; + } + + @Override + public TSStatus reportDataNodeShutdown(TDataNodeLocation dataNodeLocation) throws TException { + return null; + } + + @Override + public TDataNodeConfigurationResp getDataNodeConfiguration(int dataNodeId) throws TException { + return null; + } + + @Override + public TSStatus setDatabase(TDatabaseSchema databaseSchema) throws TException { + return null; + } + + @Override + public TSStatus alterDatabase(TDatabaseSchema databaseSchema) throws TException { + return null; + } + + @Override + public TSStatus deleteDatabase(TDeleteDatabaseReq req) throws TException { + return null; + } + + @Override + public TSStatus deleteDatabases(TDeleteDatabasesReq req) throws TException { + return null; + } + + @Override + public TSStatus setSchemaReplicationFactor(TSetSchemaReplicationFactorReq req) throws TException { + return null; + } + + @Override + public TSStatus setDataReplicationFactor(TSetDataReplicationFactorReq req) throws TException { + return null; + } + + @Override + public TSStatus setTimePartitionInterval(TSetTimePartitionIntervalReq req) throws TException { + return null; + } + + @Override + public TCountDatabaseResp countMatchedDatabases(TGetDatabaseReq req) throws TException { + return null; + } + + @Override + public TDatabaseSchemaResp getMatchedDatabaseSchemas(TGetDatabaseReq req) throws TException { + return null; + } + + @Override + public TSStatus callSpecialProcedure(TTestOperation operation) throws TException { + return null; + } + + @Override + public TSchemaPartitionTableResp getSchemaPartitionTable(TSchemaPartitionReq req) + throws TException { + return null; + } + + @Override + public TSchemaPartitionTableResp getSchemaPartitionTableWithSlots( + java.util.Map> dbSlotMap) throws TException { + return null; + } + + @Override + public TSchemaPartitionTableResp getOrCreateSchemaPartitionTable(TSchemaPartitionReq req) + throws TException { + return null; + } + + @Override + public TSchemaPartitionTableResp getOrCreateSchemaPartitionTableWithSlots( + java.util.Map> dbSlotMap) throws TException { + return null; + } + + @Override + public TSchemaNodeManagementResp getSchemaNodeManagementPartition(TSchemaNodeManagementReq req) + throws TException { + return null; + } + + @Override + public TDataPartitionTableResp getDataPartitionTable(TDataPartitionReq req) throws TException { + return null; + } + + @Override + public TDataPartitionTableResp getOrCreateDataPartitionTable(TDataPartitionReq req) + throws TException { + return null; + } + + @Override + public TSStatus dataPartitionTableIntegrityCheck() throws TException { + return null; + } + + @Override + public TShowRepairDataPartitionTableProgressResp showRepairDataPartitionTableProgress() + throws TException { + return null; + } + + @Override + public TSStatus operatePermission(TAuthorizerReq req) throws TException { + return null; + } + + @Override + public TSStatus operateRPermission(TAuthorizerRelationalReq req) throws TException { + return null; + } + + @Override + public TAuthorizerResp queryPermission(TAuthorizerReq req) throws TException { + return null; + } + + @Override + public TAuthorizerResp queryRPermission(TAuthorizerRelationalReq req) throws TException { + return null; + } + + @Override + public TPermissionInfoResp login(TLoginReq req) throws TException { + return null; + } + + @Override + public TPermissionInfoResp checkUserPrivileges(TCheckUserPrivilegesReq req) throws TException { + return null; + } + + @Override + public TAuthizedPatternTreeResp fetchAuthizedPatternTree(TCheckUserPrivilegesReq req) + throws TException { + return null; + } + + @Override + public TPermissionInfoResp checkRoleOfUser(TAuthorizerReq req) throws TException { + return null; + } + + @Override + public TPermissionInfoResp getUser(String userName) throws TException { + return null; + } + + @Override + public TConfigNodeRegisterResp registerConfigNode(TConfigNodeRegisterReq req) throws TException { + return null; + } + + @Override + public TSStatus addConsensusGroup(TAddConsensusGroupReq req) throws TException { + return null; + } + + @Override + public TSStatus notifyRegisterSuccess() throws TException { + return null; + } + + @Override + public TSStatus removeConfigNode(TConfigNodeLocation configNodeLocation) throws TException { + return null; + } + + @Override + public TSStatus deleteConfigNodePeer(TConfigNodeLocation configNodeLocation) throws TException { + return null; + } + + @Override + public TSStatus reportConfigNodeShutdown(TConfigNodeLocation configNodeLocation) + throws TException { + return null; + } + + @Override + public TSStatus stopAndClearConfigNode(TConfigNodeLocation configNodeLocation) throws TException { + return null; + } + + @Override + public TConfigNodeHeartbeatResp getConfigNodeHeartBeat(TConfigNodeHeartbeatReq req) + throws TException { + return null; + } + + @Override + public TSStatus createFunction(TCreateFunctionReq req) throws TException { + return null; + } + + @Override + public TSStatus dropFunction(TDropFunctionReq req) throws TException { + return null; + } + + @Override + public TGetUDFTableResp getUDFTable(TGetUdfTableReq req) throws TException { + return null; + } + + @Override + public TGetJarInListResp getUDFJar(TGetJarInListReq req) throws TException { + return null; + } + + @Override + public TSStatus createTrigger(TCreateTriggerReq req) throws TException { + return null; + } + + @Override + public TSStatus dropTrigger(TDropTriggerReq req) throws TException { + return null; + } + + @Override + public TGetLocationForTriggerResp getLocationOfStatefulTrigger(String triggerName) + throws TException { + return null; + } + + @Override + public TGetTriggerTableResp getTriggerTable() throws TException { + return null; + } + + @Override + public TGetTriggerTableResp getStatefulTriggerTable() throws TException { + return null; + } + + @Override + public TGetJarInListResp getTriggerJar(TGetJarInListReq req) throws TException { + return null; + } + + @Override + public TSStatus createPipePlugin(TCreatePipePluginReq req) throws TException { + return null; + } + + @Override + public TSStatus dropPipePlugin(TDropPipePluginReq req) throws TException { + return null; + } + + @Override + public TGetPipePluginTableResp getPipePluginTable() throws TException { + return null; + } + + @Override + public TGetPipePluginTableResp getPipePluginTableExtended(TShowPipePluginReq req) + throws TException { + return null; + } + + @Override + public TGetJarInListResp getPipePluginJar(TGetJarInListReq req) throws TException { + return null; + } + + @Override + public TShowTTLResp showTTL(TShowTTLReq req) throws TException { + return null; + } + + @Override + public TSStatus setTTL(TSetTTLReq req) throws TException { + return null; + } + + @Override + public TSStatus merge() throws TException { + return null; + } + + @Override + public TSStatus flush(TFlushReq req) throws TException { + return null; + } + + @Override + public TSStatus clearCache(java.util.Set cacheClearOptions) throws TException { + return null; + } + + @Override + public TSStatus setConfiguration(TSetConfigurationReq req) throws TException { + return null; + } + + @Override + public TShowConfigurationResp showConfiguration(int nodeId) throws TException { + return null; + } + + @Override + public TShowAppliedConfigurationsResp showAppliedConfigurations(int nodeId) throws TException { + return null; + } + + @Override + public TSStatus startRepairData() throws TException { + return null; + } + + @Override + public TSStatus stopRepairData() throws TException { + return null; + } + + @Override + public TSStatus submitLoadConfigurationTask() throws TException { + return null; + } + + @Override + public TSStatus loadConfiguration() throws TException { + return null; + } + + @Override + public TSStatus setSystemStatus(String status) throws TException { + return null; + } + + @Override + public TSStatus setDataNodeStatus(TSetDataNodeStatusReq req) throws TException { + return null; + } + + @Override + public TSStatus migrateRegion(TMigrateRegionReq req) throws TException { + return null; + } + + @Override + public TSStatus reconstructRegion(TReconstructRegionReq req) throws TException { + return null; + } + + @Override + public TSStatus extendRegion(TExtendRegionReq req) throws TException { + return null; + } + + @Override + public TSStatus removeRegion(TRemoveRegionReq req) throws TException { + return null; + } + + @Override + public TSStatus killQuery(String queryId, int dataNodeId, String allowedUsername) + throws TException { + return null; + } + + @Override + public TShowClusterResp showCluster() throws TException { + return null; + } + + @Override + public TShowVariablesResp showVariables() throws TException { + return null; + } + + @Override + public TShowDataNodesResp showDataNodes() throws TException { + return null; + } + + @Override + public TShowDataNodes4InformationSchemaResp showDataNodes4InformationSchema() throws TException { + return null; + } + + @Override + public TShowConfigNodesResp showConfigNodes() throws TException { + return null; + } + + @Override + public TShowConfigNodes4InformationSchemaResp showConfigNodes4InformationSchema() + throws TException { + return null; + } + + @Override + public TShowDatabaseResp showDatabase(TGetDatabaseReq req) throws TException { + return null; + } + + @Override + public TTestConnectionResp submitTestConnectionTask(TNodeLocations nodeLocations) + throws TException { + return null; + } + + @Override + public TTestConnectionResp submitTestConnectionTaskToLeader() throws TException { + return null; + } + + @Override + public TSStatus testConnectionEmptyRPC() throws TException { + return null; + } + + @Override + public TShowRegionResp showRegion(TShowRegionReq req) throws TException { + return null; + } + + @Override + public TRegionRouteMapResp getLatestRegionRouteMap() throws TException { + return null; + } + + @Override + public TSStatus createSchemaTemplate(TCreateSchemaTemplateReq req) throws TException { + return null; + } + + @Override + public TGetAllTemplatesResp getAllTemplates() throws TException { + return null; + } + + @Override + public TGetTemplateResp getTemplate(String req) throws TException { + return null; + } + + @Override + public TSStatus setSchemaTemplate(TSetSchemaTemplateReq req) throws TException { + return null; + } + + @Override + public TGetPathsSetTemplatesResp getPathsSetTemplate(TGetPathsSetTemplatesReq req) + throws TException { + return null; + } + + @Override + public TSStatus deactivateSchemaTemplate(TDeactivateSchemaTemplateReq req) throws TException { + return null; + } + + @Override + public TSStatus unsetSchemaTemplate(TUnsetSchemaTemplateReq req) throws TException { + return null; + } + + @Override + public TSStatus dropSchemaTemplate(String req) throws TException { + return null; + } + + @Override + public TSStatus alterSchemaTemplate(TAlterSchemaTemplateReq req) throws TException { + return null; + } + + @Override + public TSStatus alterEncodingCompressor(TAlterEncodingCompressorReq req) throws TException { + return null; + } + + @Override + public TSStatus deleteTimeSeries(TDeleteTimeSeriesReq req) throws TException { + return null; + } + + @Override + public TSStatus alterTimeSeriesDataType(TAlterTimeSeriesReq req) throws TException { + return null; + } + + @Override + public TSStatus deleteLogicalView(TDeleteLogicalViewReq req) throws TException { + return null; + } + + @Override + public TSStatus alterLogicalView(TAlterLogicalViewReq req) throws TException { + return null; + } + + @Override + public TSStatus createPipe(TCreatePipeReq req) throws TException { + return null; + } + + @Override + public TSStatus alterPipe(TAlterPipeReq req) throws TException { + return null; + } + + @Override + public TSStatus startPipe(String pipeName) throws TException { + return null; + } + + @Override + public TSStatus startPipeExtended(TStartPipeReq req) throws TException { + return null; + } + + @Override + public TSStatus stopPipe(String pipeName) throws TException { + return null; + } + + @Override + public TSStatus stopPipeExtended(TStopPipeReq req) throws TException { + return null; + } + + @Override + public TSStatus dropPipe(String pipeName) throws TException { + return null; + } + + @Override + public TSStatus dropPipeExtended(TDropPipeReq req) throws TException { + return null; + } + + @Override + public TShowPipeResp showPipe(TShowPipeReq req) throws TException { + return null; + } + + @Override + public TGetAllPipeInfoResp getAllPipeInfo() throws TException { + return null; + } + + @Override + public TPipeConfigTransferResp handleTransferConfigPlan(TPipeConfigTransferReq req) + throws TException { + return null; + } + + @Override + public TSStatus handlePipeConfigClientExit(String clientId) throws TException { + return null; + } + + @Override + public TSStatus createTopic(TCreateTopicReq req) throws TException { + return null; + } + + @Override + public TSStatus alterTopic(TAlterTopicReq req) throws TException { + return null; + } + + @Override + public TSStatus dropTopic(String topicName) throws TException { + return null; + } + + @Override + public TSStatus dropTopicExtended(TDropTopicReq req) throws TException { + return null; + } + + @Override + public TShowTopicResp showTopic(TShowTopicReq req) throws TException { + return null; + } + + @Override + public TGetAllTopicInfoResp getAllTopicInfo() throws TException { + return null; + } + + @Override + public TSStatus createConsumer(TCreateConsumerReq req) throws TException { + return null; + } + + @Override + public TSStatus closeConsumer(TCloseConsumerReq req) throws TException { + return null; + } + + @Override + public TSStatus createSubscription(TSubscribeReq req) throws TException { + return null; + } + + @Override + public TSStatus dropSubscription(TUnsubscribeReq req) throws TException { + return null; + } + + @Override + public TSStatus dropSubscriptionById(TDropSubscriptionReq req) throws TException { + return null; + } + + @Override + public TShowSubscriptionResp showSubscription(TShowSubscriptionReq req) throws TException { + return null; + } + + @Override + public TGetAllSubscriptionInfoResp getAllSubscriptionInfo() throws TException { + return null; + } + + @Override + public TGetCommitProgressResp getCommitProgress(TGetCommitProgressReq req) throws TException { + return null; + } + + @Override + public TGetRegionIdResp getRegionId(TGetRegionIdReq req) throws TException { + return null; + } + + @Override + public TGetTimeSlotListResp getTimeSlotList(TGetTimeSlotListReq req) throws TException { + return null; + } + + @Override + public TCountTimeSlotListResp countTimeSlotList(TCountTimeSlotListReq req) throws TException { + return null; + } + + @Override + public TGetSeriesSlotListResp getSeriesSlotList(TGetSeriesSlotListReq req) throws TException { + return null; + } + + @Override + public TGetRegionGroupsByTimeResp getRegionGroupsByTime(TGetRegionGroupsByTimeReq req) + throws TException { + return null; + } + + @Override + public TSStatus createCQ(TCreateCQReq req) throws TException { + return null; + } + + @Override + public TSStatus dropCQ(TDropCQReq req) throws TException { + return null; + } + + @Override + public TShowCQResp showCQ() throws TException { + return null; + } + + @Override + public TSStatus createExternalService(TCreateExternalServiceReq req) throws TException { + return null; + } + + @Override + public TSStatus startExternalService(int dataNodeId, String serviceName) throws TException { + return null; + } + + @Override + public TSStatus stopExternalService(int dataNodeId, String serviceName) throws TException { + return null; + } + + @Override + public TSStatus dropExternalService(int dataNodeId, String serviceName) throws TException { + return null; + } + + @Override + public TExternalServiceListResp showExternalService(int dataNodeId) throws TException { + return null; + } + + @Override + public TSStatus setSpaceQuota(TSetSpaceQuotaReq req) throws TException { + return null; + } + + @Override + public TSpaceQuotaResp showSpaceQuota(java.util.List databases) throws TException { + return null; + } + + @Override + public TSpaceQuotaResp getSpaceQuota() throws TException { + return null; + } + + @Override + public TSStatus setThrottleQuota(TSetThrottleQuotaReq req) throws TException { + return null; + } + + @Override + public TThrottleQuotaResp showThrottleQuota(TShowThrottleReq req) throws TException { + return null; + } + + @Override + public TThrottleQuotaResp getThrottleQuota() throws TException { + return null; + } + + @Override + public TSStatus pushHeartbeat(int dataNodeId, TPipeHeartbeatResp resp) throws TException { + return null; + } + + @Override + public TSStatus createTable(ByteBuffer tableInfo) throws TException { + return null; + } + + @Override + public TSStatus alterOrDropTable(TAlterOrDropTableReq req) throws TException { + return null; + } + + @Override + public TShowTableResp showTables(String database, boolean isDetails) throws TException { + return null; + } + + @Override + public TShowTable4InformationSchemaResp showTables4InformationSchema() throws TException { + return null; + } + + @Override + public TDescTableResp describeTable(String database, String tableName, boolean isDetails) + throws TException { + return null; + } + + @Override + public TDescTable4InformationSchemaResp descTables4InformationSchema() throws TException { + return null; + } + + @Override + public TFetchTableResp fetchTables(Map> fetchTableMap, byte tableNodeStatus) + throws TException { + return null; + } + + @Override + public TDeleteTableDeviceResp deleteDevice(TDeleteTableDeviceReq req) throws TException { + return null; + } + + @Override + public TSStatus createTableView(TCreateTableViewReq req) throws TException { + return null; + } +} diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/AbstractConfigNodeInfo.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/AbstractConfigNodeInfo.java new file mode 100644 index 0000000000000..5bb9b284d8a89 --- /dev/null +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/AbstractConfigNodeInfo.java @@ -0,0 +1,151 @@ +/* + * 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.commons.client; + +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.commons.consensus.ConfigRegionId; +import org.apache.iotdb.commons.exception.BadNodeUrlException; +import org.apache.iotdb.commons.exception.StartupException; +import org.apache.iotdb.commons.file.SystemPropertiesHandler; +import org.apache.iotdb.commons.i18n.ConfigMessages; +import org.apache.iotdb.commons.utils.NodeUrlUtils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Properties; +import java.util.Set; +import java.util.concurrent.locks.ReentrantReadWriteLock; + +public abstract class AbstractConfigNodeInfo { + private static final Logger logger = LoggerFactory.getLogger(AbstractConfigNodeInfo.class); + + private static final String CONFIG_NODE_LIST = "config_node_list"; + + public static final ConfigRegionId CONFIG_REGION_ID = new ConfigRegionId(0); + + protected final SystemPropertiesHandler systemPropertiesHandler; + + private final ReentrantReadWriteLock configNodeInfoReadWriteLock; + + /** latest config nodes. */ + private final Set onlineConfigNodes; + + protected AbstractConfigNodeInfo(SystemPropertiesHandler systemPropertiesHandler) { + this.configNodeInfoReadWriteLock = new ReentrantReadWriteLock(); + this.onlineConfigNodes = new HashSet<>(); + this.systemPropertiesHandler = systemPropertiesHandler; + } + + /** Update ConfigNodeList both in memory and system.properties file */ + public boolean updateConfigNodeList(List latestConfigNodes) { + long startTime = System.currentTimeMillis(); + // Check whether the config nodes are latest or not + configNodeInfoReadWriteLock.readLock().lock(); + try { + if (onlineConfigNodes.size() == latestConfigNodes.size() + && onlineConfigNodes.containsAll(latestConfigNodes)) { + return true; + } + } finally { + configNodeInfoReadWriteLock.readLock().unlock(); + } + + // Update config nodes + configNodeInfoReadWriteLock.writeLock().lock(); + try { + onlineConfigNodes.clear(); + onlineConfigNodes.addAll(latestConfigNodes); + storeConfigNodeList(); + long endTime = System.currentTimeMillis(); + logger.info( + ConfigMessages.UPDATE_CONFIG_NODE_SUCCESSFULLY, onlineConfigNodes, (endTime - startTime)); + } catch (IOException e) { + logger.error(ConfigMessages.UPDATE_CONFIG_NODE_FAILED, e); + return false; + } finally { + configNodeInfoReadWriteLock.writeLock().unlock(); + } + return true; + } + + /** + * Call this method to store config node list. + * + * @throws IOException if properties deserialization or configNode list serialization failed. + */ + public void storeConfigNodeList() throws IOException { + if (!systemPropertiesHandler.fileExist()) { + logger.info(ConfigMessages.SYSTEM_PROPERTIES_NOT_EXIST); + return; + } + systemPropertiesHandler.put( + CONFIG_NODE_LIST, NodeUrlUtils.convertTEndPointUrls(new ArrayList<>(onlineConfigNodes))); + } + + public void loadConfigNodeList() throws StartupException { + long startTime = System.currentTimeMillis(); + // properties contain CONFIG_NODE_LIST only when start as node + configNodeInfoReadWriteLock.writeLock().lock(); + try { + Properties properties = systemPropertiesHandler.read(); + + if (properties.containsKey(CONFIG_NODE_LIST)) { + onlineConfigNodes.clear(); + onlineConfigNodes.addAll( + NodeUrlUtils.parseTEndPointUrls(properties.getProperty(CONFIG_NODE_LIST))); + } + if (onlineConfigNodes.isEmpty()) { + throw new StartupException( + String.format( + ConfigMessages + .MISC_EXCEPTION_REMOVING_IS_ONLY_ALLOWED_IN_AN_ENVIRONMENT_WHEN_NODE_STARTED_2ACA2BD0, + getNodeTypeName())); + } + long endTime = System.currentTimeMillis(); + logger.info( + ConfigMessages.LOAD_CONFIG_NODE_SUCCESSFULLY, onlineConfigNodes, (endTime - startTime)); + } catch (IOException e) { + throw new RuntimeException(e); + } catch (BadNodeUrlException e) { + logger.error(ConfigMessages.CANNOT_PARSE_CONFIG_NODE_LIST); + } finally { + configNodeInfoReadWriteLock.writeLock().unlock(); + } + } + + public List getLatestConfigNodes() { + List result; + configNodeInfoReadWriteLock.readLock().lock(); + try { + result = new ArrayList<>(onlineConfigNodes); + } finally { + configNodeInfoReadWriteLock.readLock().unlock(); + } + return result; + } + + /** The node type name used in the exception message, e.g. datanode, streamnode. */ + protected abstract String getNodeTypeName(); +} diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/AbstractIoTDBStartCheck.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/AbstractIoTDBStartCheck.java new file mode 100644 index 0000000000000..310b88bd74287 --- /dev/null +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/AbstractIoTDBStartCheck.java @@ -0,0 +1,195 @@ +/* + * 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.commons.conf; + +import org.apache.iotdb.commons.exception.ConfigurationException; +import org.apache.iotdb.commons.file.SystemPropertiesHandler; +import org.apache.iotdb.commons.i18n.CommonMessages; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; +import java.util.Map.Entry; +import java.util.Properties; +import java.util.function.Supplier; + +public abstract class AbstractIoTDBStartCheck { + + protected static final Logger logger = LoggerFactory.getLogger(AbstractIoTDBStartCheck.class); + + // this file is located in data/system/system.properties + // If user delete folder "data", system.properties can reset. + public static final String PROPERTIES_FILE_NAME = "system.properties"; + + protected Properties properties = new Properties(); + + protected final Map> systemProperties = new HashMap<>(); + protected final SystemPropertiesHandler systemPropertiesHandler; + + protected static final String IOTDB_VERSION_STRING = "iotdb_version"; + protected static final String COMMIT_ID_STRING = "commit_id"; + protected static final String CLUSTER_ID = "cluster_id"; + + protected AbstractIoTDBStartCheck() { + systemPropertiesHandler = getSystemPropertiesHandler(); + + systemProperties.put(IOTDB_VERSION_STRING, () -> IoTDBConstant.VERSION); + systemProperties.put(COMMIT_ID_STRING, () -> IoTDBConstant.BUILD_INFO); + for (String param : getVariableParamValueTable().keySet()) { + systemProperties.put(param, () -> getVal(param)); + } + } + + protected abstract SystemPropertiesHandler getSystemPropertiesHandler(); + + protected abstract Map> getVariableParamValueTable(); + + protected abstract String getNodeIdKey(); + + protected abstract void loadNodeId(int nodeId); + + protected abstract void loadClusterId(String clusterId); + + /** check and create directory before start the node. */ + public abstract void checkDirectory() throws ConfigurationException, IOException; + + /** + * check configuration in system.properties when starting the node + * + *

When init: create system.properties directly + * + *

When upgrading the system.properties: (1) create system.properties.tmp (2) delete + * system.properties (3) rename system.properties.tmp to system.properties + */ + public void checkSystemConfig() throws ConfigurationException, IOException { + // read properties from system.properties + properties = systemPropertiesHandler.read(); + + if (systemPropertiesHandler.isFirstStart()) { + validateFirstStart(); + } else { + // check whether upgrading from <=v0.9 + if (!properties.containsKey(IOTDB_VERSION_STRING)) { + logger.error( + CommonMessages + .MISC_LOG_DO_NOT_UPGRADE_IOTDB_FROM_V0_9_OR_LOWER_VERSION_TO_V1_0_9878EC88); + System.exit(-1); + } + String versionString = properties.getProperty(IOTDB_VERSION_STRING); + if (versionString.startsWith("0.")) { + logger.error(CommonMessages.IOTDB_VERSION_TOO_OLD); + System.exit(-1); + } + checkImmutableSystemProperties(); + } + } + + /** Check all immutable properties */ + protected void checkImmutableSystemProperties() throws IOException { + for (Entry> entry : systemProperties.entrySet()) { + if (!properties.containsKey(entry.getKey())) { + upgradePropertiesFileFromBrokenFile(); + logger.info(CommonMessages.REPAIR_SYSTEM_PROPERTIES, entry.getKey()); + } + } + + if (properties.containsKey(getNodeIdKey())) { + loadNodeId(Integer.parseInt(properties.getProperty(getNodeIdKey()))); + } + if (properties.containsKey(CLUSTER_ID)) { + loadClusterId(properties.getProperty(CLUSTER_ID)); + } + checkExtraImmutableProperties(); + } + + /** Node-specific validation on first start, no-op by default. */ + protected void validateFirstStart() throws ConfigurationException {} + + /** Node-specific immutable properties, no-op by default. */ + protected void checkExtraImmutableProperties() throws IOException {} + + /** repair broken properties */ + protected void upgradePropertiesFileFromBrokenFile() throws IOException { + systemProperties.forEach( + (k, v) -> { + if (!properties.containsKey(k)) { + properties.setProperty(k, v.get()); + } + }); + properties.setProperty(IOTDB_VERSION_STRING, IoTDBConstant.VERSION); + properties.setProperty(COMMIT_ID_STRING, IoTDBConstant.BUILD_INFO); + systemPropertiesHandler.overwrite(properties); + } + + protected void throwException(String parameter, Object badValue) throws ConfigurationException { + throw new ConfigurationException( + parameter, + String.valueOf(badValue), + properties.getProperty(parameter), + String.format( + CommonMessages.PARAMETER_CANNOT_BE_MODIFIED_AFTER_FIRST_STARTUP_FMT, parameter)); + } + + protected String getVal(String paramName) { + if (getVariableParamValueTable().containsKey(paramName)) { + return getVariableParamValueTable().get(paramName).get(); + } else { + return null; + } + } + + public void serializeNodeId(int nodeId) throws IOException { + systemPropertiesHandler.put(getNodeIdKey(), String.valueOf(nodeId)); + } + + public void serializeClusterID(String clusterId) throws IOException { + systemPropertiesHandler.put(CLUSTER_ID, clusterId); + } + + public void serializeMutableSystemPropertiesIfNecessary() throws IOException { + long startTime = System.currentTimeMillis(); + boolean needsSerialize = false; + for (String param : getVariableParamValueTable().keySet()) { + if (!properties.getProperty(param).equals(getVal(param))) { + needsSerialize = true; + } + } + + if (needsSerialize) { + generateOrOverwriteSystemPropertiesFile(); + } + long endTime = System.currentTimeMillis(); + logger.info( + CommonMessages + .MISC_LOG_SERIALIZE_MUTABLE_SYSTEM_PROPERTIES_SUCCESSFULLY_WHICH_TAKES_4656A206, + (endTime - startTime)); + } + + public void generateOrOverwriteSystemPropertiesFile() throws IOException { + systemProperties.forEach((k, v) -> properties.setProperty(k, v.get())); + systemPropertiesHandler.overwrite(properties); + } + + public Properties getProperties() { + return properties; + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/disk/DirectoryChecker.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/disk/utils/DirectoryChecker.java similarity index 94% rename from iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/disk/DirectoryChecker.java rename to iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/disk/utils/DirectoryChecker.java index 77bf6f4572912..1d6e9590248db 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/disk/DirectoryChecker.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/disk/utils/DirectoryChecker.java @@ -16,11 +16,11 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.iotdb.db.storageengine.rescon.disk; +package org.apache.iotdb.commons.disk.utils; import org.apache.iotdb.commons.exception.ConfigurationException; +import org.apache.iotdb.commons.i18n.StorageMessages; import org.apache.iotdb.commons.utils.ProcessIdUtils; -import org.apache.iotdb.db.i18n.StorageEngineMessages; import org.apache.tsfile.external.commons.io.FileUtils; import org.apache.tsfile.utils.FSUtils; @@ -58,14 +58,14 @@ public void registerDirectory(File dir) throws ConfigurationException, IOExcepti if (dir.exists() && !dir.isDirectory()) { throw new ConfigurationException( String.format( - StorageEngineMessages + StorageMessages .STORAGE_EXCEPTION_UNABLE_TO_CREATE_DIRECTORY_S_BECAUSE_THERE_IS_FILE_UNDER_1C59ACFC, dir.getAbsolutePath())); } else if (!dir.exists()) { if (!dir.mkdirs()) { throw new ConfigurationException( String.format( - StorageEngineMessages + StorageMessages .STORAGE_EXCEPTION_UNABLE_TO_CREATE_DIRECTORY_S_PLEASE_CHECK_CONFIGURATION_BA580B67, dir.getAbsolutePath())); } @@ -91,7 +91,7 @@ public void registerDirectory(File dir) throws ConfigurationException, IOExcepti channel.close(); throw new ConfigurationException( String.format( - StorageEngineMessages + StorageMessages .STORAGE_EXCEPTION_CONFLICT_IS_DETECTED_IN_DIRECTORY_S_WHICH_MAY_BE_BEING_USED_CB5C77FC, dir.getAbsolutePath(), lockOwner)); @@ -147,7 +147,7 @@ public void deregisterAll() { FileUtils.delete(file); } } catch (IOException e) { - logger.warn(StorageEngineMessages.FAILED_TO_DEREGISTER_FILE_LOCK, e.getMessage(), e); + logger.warn(StorageMessages.FAILED_TO_DEREGISTER_FILE_LOCK, e.getMessage(), e); } }