diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDeviceTemplateIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDeviceTemplateIT.java new file mode 100644 index 000000000000..b93113bc080f --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDeviceTemplateIT.java @@ -0,0 +1,291 @@ +/* + * 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.relational.it.schema; + +import org.apache.iotdb.commons.cluster.NodeStatus; +import org.apache.iotdb.consensus.ConsensusFactory; +import org.apache.iotdb.isession.SessionConfig; +import org.apache.iotdb.it.env.EnvFactory; +import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper; +import org.apache.iotdb.it.framework.IoTDBTestRunner; +import org.apache.iotdb.itbase.category.ClusterIT; +import org.apache.iotdb.itbase.env.BaseEnv; + +import org.junit.Assert; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.Statement; +import java.util.Collections; +import java.util.concurrent.Callable; + +import static org.junit.Assert.assertTrue; + +@RunWith(IoTDBTestRunner.class) +@Category({ClusterIT.class}) +public class IoTDBDeviceTemplateIT { + + private static final Logger LOGGER = LoggerFactory.getLogger(IoTDBDeviceTemplateIT.class); + + private static final String DATABASE = "root.device_template_ha"; + private static final String TEMPLATE1 = "temp1"; + private static final String TEMPLATE2 = "temp2"; + private static final String DEVICE1 = DATABASE + ".dev01"; + private static final String DEVICE2 = DATABASE + ".dev02"; + + private static void initCluster() { + EnvFactory.getEnv() + .getConfig() + .getCommonConfig() + .setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS) + .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS) + .setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS) + .setSchemaReplicationFactor(3) + .setDataReplicationFactor(2); + EnvFactory.getEnv().getConfig().getConfigNodeConfig().setMetadataLeaseFenceMs(20000); + EnvFactory.getEnv().initClusterEnvironment(1, 3); + } + + private static void cleanCluster() { + EnvFactory.getEnv().cleanClusterEnvironment(); + } + + private void preSetup(final Statement statement) throws Exception { + statement.execute("CREATE DATABASE " + DATABASE); + statement.execute("CREATE DEVICE TEMPLATE " + TEMPLATE1 + " (s1 INT32, s2 INT64)"); + statement.execute("SET DEVICE TEMPLATE " + TEMPLATE1 + " TO " + DATABASE); + statement.execute("CREATE TIMESERIES OF DEVICE TEMPLATE ON " + DEVICE1); + statement.execute("INSERT INTO " + DEVICE1 + "(time, s1, s2) VALUES(1, 1, 1)"); + } + + @Test + public void testHAWithOneDataNodeIsDown() throws Exception { + initCluster(); + try { + final DataNodeWrapper liveDataNode = EnvFactory.getEnv().getDataNodeWrapper(0); + final DataNodeWrapper victimDataNode = EnvFactory.getEnv().getDataNodeWrapper(2); + try (final Connection connection = + EnvFactory.getEnv() + .getConnection( + liveDataNode, + SessionConfig.DEFAULT_USER, + SessionConfig.DEFAULT_PASSWORD, + BaseEnv.TREE_SQL_DIALECT); + final Statement statement = connection.createStatement()) { + + preSetup(statement); + + victimDataNode.stop(); + Assert.assertFalse("victim DataNode should be stopped", victimDataNode.isAlive()); + + executeDeviceTemplateHATests(statement); + } + } finally { + cleanCluster(); + } + } + + @Test + public void testHAWithOneDataNodeIsReadOnly() throws Exception { + initCluster(); + try { + final DataNodeWrapper liveDataNode = EnvFactory.getEnv().getDataNodeWrapper(0); + final DataNodeWrapper victimDataNode = EnvFactory.getEnv().getDataNodeWrapper(2); + + try (final Connection connection = + EnvFactory.getEnv() + .getConnection( + liveDataNode, + SessionConfig.DEFAULT_USER, + SessionConfig.DEFAULT_PASSWORD, + BaseEnv.TREE_SQL_DIALECT); + final Statement statement = connection.createStatement()) { + + preSetup(statement); + + try (final Connection victimConn = + EnvFactory.getEnv() + .getConnection( + victimDataNode, + SessionConfig.DEFAULT_USER, + SessionConfig.DEFAULT_PASSWORD, + BaseEnv.TABLE_SQL_DIALECT); + final Statement victimStmt = victimConn.createStatement()) { + + victimStmt.execute("SET SYSTEM TO READONLY ON LOCAL"); + + EnvFactory.getEnv() + .ensureNodeStatus( + Collections.singletonList(victimDataNode), + Collections.singletonList(NodeStatus.ReadOnly)); + } + + executeDeviceTemplateHATests(statement); + } + } finally { + cleanCluster(); + } + } + + private void executeDeviceTemplateHATests(final Statement statement) throws Exception { + // 1. SHOW — verify all SHOW operations work + LOGGER.info("1. start to test high availability of show device template operations"); + assertTrue( + "SHOW DEVICE TEMPLATES must include " + TEMPLATE1, templateExists(statement, TEMPLATE1)); + assertTrue( + "SHOW NODES IN DEVICE TEMPLATE must include s1", + templateHasMeasurement(statement, TEMPLATE1, "s1")); + assertTrue( + "SHOW PATHS SET DEVICE TEMPLATE must include " + DATABASE, + pathSetTemplate(statement, TEMPLATE1, DATABASE)); + assertTrue( + "SHOW PATHS USING DEVICE TEMPLATE must include " + DEVICE1, + pathUsingTemplate(statement, TEMPLATE1, DEVICE1)); + + // 2. ALTER — add a measurement to t1 + LOGGER.info("2. start to test high availability of alter device template procedure"); + assertStatementEffect( + statement, + "ALTER DEVICE TEMPLATE " + TEMPLATE1 + " ADD (s3 FLOAT)", + () -> templateHasMeasurement(statement, TEMPLATE1, "s3"), + "ALTER DEVICE TEMPLATE must succeed with one DataNode down"); + + // 3. DEACTIVATE — deactivate t1 from dev01 + LOGGER.info("3. start to test high availability of deactivate device template procedure"); + assertStatementEffect( + statement, + "DEACTIVATE DEVICE TEMPLATE " + TEMPLATE1 + " FROM " + DEVICE1, + () -> !pathUsingTemplate(statement, TEMPLATE1, DEVICE1), + "DEACTIVATE DEVICE TEMPLATE must succeed with one DataNode down"); + + // 4. UNSET — unset t1 from database + LOGGER.info("4. start to test high availability of unset device template procedure"); + assertStatementEffect( + statement, + "UNSET DEVICE TEMPLATE " + TEMPLATE1 + " FROM " + DATABASE, + () -> !pathSetTemplate(statement, TEMPLATE1, DATABASE), + "UNSET DEVICE TEMPLATE must succeed with one DataNode down"); + + // 5. DROP — drop t1 + LOGGER.info("5. start to test high availability of drop device template procedure"); + assertStatementEffect( + statement, + "DROP DEVICE TEMPLATE " + TEMPLATE1, + () -> !templateExists(statement, TEMPLATE1), + "DROP DEVICE TEMPLATE must succeed with one DataNode down"); + + // 6. CREATE — create a new aligned template t2 + LOGGER.info("6. start to test high availability of create device template procedure"); + assertStatementEffect( + statement, + "CREATE DEVICE TEMPLATE " + TEMPLATE2 + " ALIGNED (lat FLOAT, lon FLOAT)", + () -> templateExists(statement, TEMPLATE2), + "CREATE DEVICE TEMPLATE must succeed with one DataNode down"); + assertTrue( + "SHOW NODES IN DEVICE TEMPLATE must include lat", + templateHasMeasurement(statement, TEMPLATE2, "lat")); + assertTrue( + "SHOW NODES IN DEVICE TEMPLATE must include lon", + templateHasMeasurement(statement, TEMPLATE2, "lon")); + + // 7. SET — set t2 to database + LOGGER.info("7. start to test high availability of set device template procedure"); + assertStatementEffect( + statement, + "SET DEVICE TEMPLATE " + TEMPLATE2 + " TO " + DATABASE, + () -> pathSetTemplate(statement, TEMPLATE2, DATABASE), + "SET DEVICE TEMPLATE must succeed with one DataNode down"); + + // 8. ACTIVATE — activate t2 on dev02 + LOGGER.info("8. start to test high availability of activate device template procedure"); + assertStatementEffect( + statement, + "CREATE TIMESERIES OF DEVICE TEMPLATE ON " + DEVICE2, + () -> pathUsingTemplate(statement, TEMPLATE2, DEVICE2), + "CREATE TIMESERIES USING DEVICE TEMPLATE must succeed with one DataNode down"); + } + + private void assertStatementEffect( + final Statement statement, + final String sql, + final Callable effect, + final String message) + throws Exception { + statement.execute(sql); + assertTrue(message, effect.call()); + } + + private boolean templateExists(final Statement statement, final String templateName) + throws Exception { + try (final ResultSet resultSet = statement.executeQuery("SHOW DEVICE TEMPLATES")) { + while (resultSet.next()) { + if (templateName.equalsIgnoreCase(resultSet.getString(1))) { + return true; + } + } + } + return false; + } + + private boolean templateHasMeasurement( + final Statement statement, final String templateName, final String measurement) + throws Exception { + try (final ResultSet resultSet = + statement.executeQuery("SHOW NODES IN DEVICE TEMPLATE " + templateName)) { + while (resultSet.next()) { + if (measurement.equalsIgnoreCase(resultSet.getString(1))) { + return true; + } + } + } + return false; + } + + private boolean pathSetTemplate( + final Statement statement, final String templateName, final String path) throws Exception { + try (final ResultSet resultSet = + statement.executeQuery("SHOW PATHS SET DEVICE TEMPLATE " + templateName)) { + while (resultSet.next()) { + if (path.equalsIgnoreCase(resultSet.getString(1))) { + return true; + } + } + } + return false; + } + + private boolean pathUsingTemplate( + final Statement statement, final String templateName, final String path) throws Exception { + try (final ResultSet resultSet = + statement.executeQuery("SHOW PATHS USING DEVICE TEMPLATE " + templateName)) { + while (resultSet.next()) { + if (path.equalsIgnoreCase(resultSet.getString(1))) { + return true; + } + } + } + return false; + } +} diff --git a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java index 494201b1e9fe..35d9a7b9bbb9 100644 --- a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java +++ b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java @@ -583,6 +583,7 @@ private ConfigNodeMessages() {} public static final String LOG_REDIRECTION_RECOMMENDED_REMOVECONFIGNODE_BUT_NO_LEADER_ENDPOINT_PROVIDED_ABORT_RETRY_520A4C64 = "Redirection recommended for removeConfigNode but no leader endpoint provided, abort retry."; public static final String LOG_FAILED_WRITE_AUDIT_LOG_DATANODE_ARG_RESPONSE_ARG_691ABC90 = "Failed to write audit log to DataNode {}, response: {}"; public static final String LOG_FAILED_WRITE_AUDIT_LOG_DATANODE_ARG_90F15E13 = "Failed to write audit log to DataNode {}"; + public static final String LOG_DATANODE_FAILED_TO_EXECUTE_METADATA_CACHE_PROPAGATION = "DataNode {} failed to execute metadata cache propagation, status: {}"; public static final String EXCEPTION_UNKNOWN_PHYSICALPLANTYPE_ARG_7F21B699 = "Unknown PhysicalPlanType: %d"; public static final String LOG_CANNOT_READ_MORE_PHYSICALPLANS_ARG_SUCCESSFULLY_READ_INDEX_ARG_REASON_2EC90E78 = "Cannot read more PhysicalPlans from {}, successfully read index is {}. The reason is"; public static final String EXCEPTION_FILE_11296840 = "file: "; diff --git a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java index 20316edc3971..4f1a2d423907 100644 --- a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java +++ b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java @@ -197,8 +197,8 @@ public final class ManagerMessages { "Failed to sync consumer group meta. Result status: {}."; public static final String FAILED_TO_SYNC_PIPE_META_RESULT_STATUS = "Failed to sync pipe meta. Result status: {}."; - public static final String FAILED_TO_SYNC_TEMPLATE_EXTENSION_INFO_TO_DATANODE = - "Failed to sync template {} extension info to DataNode {}"; + public static final String EXIST_DATANODE_FAILED_TO_EXECUTE_AND_COULD_NOT_BE_FENCED = + "Exist DataNode failed to execute template %s extension and could not be fenced; template state of DataNodes may be inconsistent"; public static final String FAILED_TO_SYNC_TOPIC_META_RESULT_STATUS = "Failed to sync topic meta. Result status: {}."; public static final String FAILED_TO_UNBIND_FROM_PIPE_CONFIG_REGION_CONNECTOR_METRICS_CONNECTOR = diff --git a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java index c771bb1ad98a..7e8ce6546695 100644 --- a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java +++ b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java @@ -489,7 +489,7 @@ public final class ProcedureMessages { "Failed to rollback pre-release {} for table {}.{} info to DataNode, failure results: {}"; public static final String FAILED_TO_PROVE_AN_UNREACHABLE_DN_IS_FENCED = "Failed to prove an unreachable DataNode is fenced"; public static final String FAILED_TO_ROLLBACK_PRE_RELEASE_TEMPLATE_INFO_OF_TEMPLATE_SET = - "Failed to rollback pre release template info of template {} set on path {} on DataNode {}"; + "Failed to rollback pre release template info of template {} set on path {} because {}"; public static final String FAILED_TO_ROLLBACK_PRE_SET_TEMPLATE_ON_PATH_DUE_TO = "Failed to rollback pre set template {} on path {} due to {}"; public static final String FAILED_TO_ROLLBACK_PRE_UNSET_TEMPLATE_OPERATION_OF_TEMPLATE_SET = @@ -510,7 +510,7 @@ public final class ProcedureMessages { "Failed to serialize finalDataPartitionTables"; public static final String FAILED_TO_SERIALIZE_SKIPDATANODE = "Failed to serialize skipDataNode"; public static final String FAILED_TO_SET_SCHEMAENGINE_TEMPLATE_ON_PATH_BECAUSE_THERE_S = - "Failed to set schemaengine template %s on path %s because there's failure on DataNode %s"; + "Failed to set schemaengine template %s on path %s because %s"; public static final String FAILED_TO_START_PIPE_DETAILS_METADATA_WILL_BE_SYNCHRONIZED_LATER = "Failed to start pipe {}, details: {}, metadata will be synchronized later."; public static final String FAILED_TO_STOP_AINODE_BECAUSE_BUT_THE_REMOVE_PROCESS_WILL = diff --git a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java index 0193e5a04854..ca10b04bb159 100644 --- a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java +++ b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java @@ -581,6 +581,7 @@ private ConfigNodeMessages() {} "向 DataNode {} 写入审计日志失败，响应：{}"; public static final String LOG_FAILED_WRITE_AUDIT_LOG_DATANODE_ARG_90F15E13 = "向 DataNode {} 写入审计日志失败"; + public static final String LOG_DATANODE_FAILED_TO_EXECUTE_METADATA_CACHE_PROPAGATION = "DataNode {} 执行元数据缓存传播失败，状态：{}"; public static final String EXCEPTION_UNKNOWN_PHYSICALPLANTYPE_ARG_7F21B699 = "未知的 PhysicalPlanType: %d"; public static final String diff --git a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java index 8dc9afd1e8bf..6d1b42163320 100644 --- a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java +++ b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java @@ -195,8 +195,8 @@ public final class ManagerMessages { "同步 consumer group 元数据失败。结果状态：{}。"; public static final String FAILED_TO_SYNC_PIPE_META_RESULT_STATUS = "同步 pipe 元数据失败。结果状态：{}。"; - public static final String FAILED_TO_SYNC_TEMPLATE_EXTENSION_INFO_TO_DATANODE = - "将模板 {} 的扩展信息同步到 DataNode {} 失败"; + public static final String EXIST_DATANODE_FAILED_TO_EXECUTE_AND_COULD_NOT_BE_FENCED = + "存在 DataNode 执行模板 %s 扩展失败且未能被隔离，DataNode之间模板状态可能不一致"; public static final String FAILED_TO_SYNC_TOPIC_META_RESULT_STATUS = "同步 topic 元数据失败。结果状态：{}。"; public static final String FAILED_TO_UNBIND_FROM_PIPE_CONFIG_REGION_CONNECTOR_METRICS_CONNECTOR = diff --git a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java index 5d20a0b7413e..6f27d7abc261 100644 --- a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java +++ b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java @@ -476,7 +476,7 @@ public final class ProcedureMessages { public static final String FAILED_TO_PROVE_AN_UNREACHABLE_DN_IS_FENCED = "不能认定一个不可达的datanode处于隔离状态"; public static final String FAILED_TO_ROLLBACK_PRE_RELEASE_TEMPLATE_INFO_OF_TEMPLATE_SET = - "回滚 DataNode {} 上路径 {} 处模板 {} 的预释放模板信息失败"; + "回滚模板 {} 在路径 {} 处的预释放模板信息失败，原因：{}"; public static final String FAILED_TO_ROLLBACK_PRE_SET_TEMPLATE_ON_PATH_DUE_TO = "回滚在路径 {} 上预设置的模板 {} 失败，原因：{}"; public static final String FAILED_TO_ROLLBACK_PRE_UNSET_TEMPLATE_OPERATION_OF_TEMPLATE_SET = @@ -494,7 +494,7 @@ public final class ProcedureMessages { "序列化 finalDataPartitionTables 失败"; public static final String FAILED_TO_SERIALIZE_SKIPDATANODE = "序列化 skipDataNode 失败"; public static final String FAILED_TO_SET_SCHEMAENGINE_TEMPLATE_ON_PATH_BECAUSE_THERE_S = - "在路径 %s 上设置 schemaengine 模板 %s 失败，原因：DataNode %s 上存在失败"; + "设置 schemaengine 模板 %s 到路径 %s 失败，原因：%s"; public static final String FAILED_TO_START_PIPE_DETAILS_METADATA_WILL_BE_SYNCHRONIZED_LATER = "启动 pipe {} 失败，详情：{}，元数据将稍后同步。"; public static final String FAILED_TO_STOP_AINODE_BECAUSE_BUT_THE_REMOVE_PROCESS_WILL = diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java index fee78f5568a0..53874873f62f 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java @@ -3308,7 +3308,8 @@ public TDataNodeLeaseRecoveryResp reloadCacheAfterLeaseRecovery() { } return new TDataNodeLeaseRecoveryResp() .setStatus(RpcUtils.SUCCESS_STATUS) - .setTableInfo(clusterSchemaManager.getAllTableInfoForDataNodeActivation()); + .setTableInfo(clusterSchemaManager.getAllTableInfoForDataNodeActivation()) + .setTemplateInfo(clusterSchemaManager.getAllTemplateSetInfo()); } @Override diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/lease/ClusterCachePropagator.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/lease/ClusterCachePropagator.java index d0cae2aeedf9..67880cf46f07 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/lease/ClusterCachePropagator.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/lease/ClusterCachePropagator.java @@ -22,10 +22,14 @@ import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation; import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor; +import org.apache.iotdb.confignode.i18n.ConfigNodeMessages; import org.apache.iotdb.confignode.manager.lease.MetadataBroadcastVerdict.DataNodeState; import org.apache.iotdb.confignode.manager.lease.MetadataBroadcastVerdict.Verdict; import org.apache.iotdb.rpc.TSStatusCode; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import java.util.ArrayList; import java.util.List; import java.util.Map; @@ -39,6 +43,8 @@ */ public class ClusterCachePropagator { + private static final Logger LOGGER = LoggerFactory.getLogger(ClusterCachePropagator.class); + /** * {@code T_proceed = T_fence + margin}. The margin covers heartbeat-recording granularity and * scheduling jitter. @@ -122,6 +128,11 @@ public Verdict propagateOnce(final CacheBroadcast broadcast, final boolean waitB break; default: // There is a DN executes procedure with internal failure + // all the procedures here are not expected to trigger internal failure + LOGGER.error( + ConfigNodeMessages.LOG_DATANODE_FAILED_TO_EXECUTE_METADATA_CACHE_PROPAGATION, + targets.get(nodeId), + status); return Verdict.FAIL; } } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java index 357c14eca4aa..07a18ae1c3bf 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java @@ -41,9 +41,6 @@ import org.apache.iotdb.commons.service.metric.MetricService; import org.apache.iotdb.commons.utils.PathUtils; import org.apache.iotdb.commons.utils.StatusUtils; -import org.apache.iotdb.confignode.client.async.CnToDnAsyncRequestType; -import org.apache.iotdb.confignode.client.async.CnToDnInternalServiceAsyncRequestManager; -import org.apache.iotdb.confignode.client.async.handlers.DataNodeAsyncRequestContext; import org.apache.iotdb.confignode.conf.ConfigNodeConfig; import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor; import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan; @@ -93,6 +90,7 @@ import org.apache.iotdb.confignode.i18n.ProcedureMessages; import org.apache.iotdb.confignode.manager.IManager; import org.apache.iotdb.confignode.manager.consensus.ConsensusManager; +import org.apache.iotdb.confignode.manager.lease.ClusterCachePropagator; import org.apache.iotdb.confignode.manager.node.NodeManager; import org.apache.iotdb.confignode.manager.partition.PartitionManager; import org.apache.iotdb.confignode.manager.partition.PartitionMetrics; @@ -137,6 +135,8 @@ import java.util.concurrent.locks.ReentrantLock; import java.util.stream.Collectors; +import static org.apache.iotdb.confignode.procedure.impl.schema.SchemaUtils.broadcastTemplateUpdate; + /** The ClusterSchemaManager Manages cluster schemaengine read and write requests. */ public class ClusterSchemaManager { @@ -1314,23 +1314,17 @@ public synchronized TSStatus extendSchemaTemplate( Map dataNodeLocationMap = configManager.getNodeManager().getRegisteredDataNodeLocations(); - DataNodeAsyncRequestContext clientHandler = - new DataNodeAsyncRequestContext<>( - CnToDnAsyncRequestType.UPDATE_TEMPLATE, updateTemplateReq, dataNodeLocationMap); - CnToDnInternalServiceAsyncRequestManager.getInstance().sendAsyncRequestWithRetry(clientHandler); - Map statusMap = clientHandler.getResponseMap(); - for (Map.Entry entry : statusMap.entrySet()) { - if (entry.getValue().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { - LOGGER.warn( - ManagerMessages.FAILED_TO_SYNC_TEMPLATE_EXTENSION_INFO_TO_DATANODE, - template.getName(), - dataNodeLocationMap.get(entry.getKey())); - return RpcUtils.getStatus( - TSStatusCode.EXECUTE_STATEMENT_ERROR, - String.format( - "Failed to sync template %s extension info to DataNode %s", - template.getName(), dataNodeLocationMap.get(entry.getKey()))); - } + // The template extension is already committed and cannot be rolled back. + // Unexpected DataNode internal failures cannot be handled here. + final boolean proceed = + new ClusterCachePropagator(dataNodeLocationMap) + .propagate(targets -> broadcastTemplateUpdate(updateTemplateReq, targets)); + if (!proceed) { + return RpcUtils.getStatus( + TSStatusCode.EXECUTE_STATEMENT_ERROR, + String.format( + ManagerMessages.EXIST_DATANODE_FAILED_TO_EXECUTE_AND_COULD_NOT_BE_FENCED, + template.getName())); } if (intersectionMeasurements.isEmpty()) { diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeactivateTemplateProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeactivateTemplateProcedure.java index b2875b7fc41b..09d955b53e0f 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeactivateTemplateProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeactivateTemplateProcedure.java @@ -30,8 +30,6 @@ import org.apache.iotdb.commons.path.PathPatternTree; import org.apache.iotdb.commons.schema.template.Template; import org.apache.iotdb.confignode.client.async.CnToDnAsyncRequestType; -import org.apache.iotdb.confignode.client.async.CnToDnInternalServiceAsyncRequestManager; -import org.apache.iotdb.confignode.client.async.handlers.DataNodeAsyncRequestContext; import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeDeactivateTemplatePlan; import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeEnrichedPlan; import org.apache.iotdb.confignode.i18n.ProcedureMessages; @@ -44,7 +42,6 @@ import org.apache.iotdb.mpp.rpc.thrift.TConstructSchemaBlackListWithTemplateReq; import org.apache.iotdb.mpp.rpc.thrift.TDeactivateTemplateReq; import org.apache.iotdb.mpp.rpc.thrift.TDeleteDataForDeleteSchemaReq; -import org.apache.iotdb.mpp.rpc.thrift.TInvalidateMatchedSchemaCacheReq; import org.apache.iotdb.mpp.rpc.thrift.TRollbackSchemaBlackListWithTemplateReq; import org.apache.iotdb.pipe.api.exception.PipeException; import org.apache.iotdb.rpc.TSStatusCode; @@ -180,30 +177,16 @@ protected List processResponseOfOneDataNode( } private void invalidateCache(final ConfigNodeProcedureEnv env) { - // if no target timeseries, return directly - if (!timeSeriesPatternTree.isEmpty()) { - Map dataNodeLocationMap = - env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations(); - DataNodeAsyncRequestContext clientHandler = - new DataNodeAsyncRequestContext<>( - CnToDnAsyncRequestType.INVALIDATE_MATCHED_SCHEMA_CACHE, - new TInvalidateMatchedSchemaCacheReq(timeSeriesPatternTreeBytes), - dataNodeLocationMap); - CnToDnInternalServiceAsyncRequestManager.getInstance() - .sendAsyncRequestWithRetry(clientHandler); - Map statusMap = clientHandler.getResponseMap(); - for (TSStatus status : statusMap.values()) { - // all dataNodes must clear the related schema cache - if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { - LOGGER.error( - ProcedureMessages.FAILED_TO_INVALIDATE_SCHEMA_CACHE_OF_TEMPLATE_TIMESERIES, - requestMessage); - setFailure( - new ProcedureException( - new MetadataException(ProcedureMessages.INVALIDATE_SCHEMA_CACHE_FAILED))); - return; - } - } + if (!timeSeriesPatternTree.isEmpty() + && !SchemaUtils.invalidateMatchedSchemaCache( + env.getConfigManager(), timeSeriesPatternTreeBytes, true)) { + LOGGER.error( + ProcedureMessages.FAILED_TO_INVALIDATE_SCHEMA_CACHE_OF_TEMPLATE_TIMESERIES, + requestMessage); + setFailure( + new ProcedureException( + new MetadataException(ProcedureMessages.INVALIDATE_SCHEMA_CACHE_FAILED))); + return; } setNextState(DeactivateTemplateState.DELETE_DATA); diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SchemaUtils.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SchemaUtils.java index 69a3cec44005..f476e9306401 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SchemaUtils.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SchemaUtils.java @@ -48,6 +48,7 @@ import org.apache.iotdb.mpp.rpc.thrift.TCountPathsUsingTemplateResp; import org.apache.iotdb.mpp.rpc.thrift.TInvalidateMatchedSchemaCacheReq; import org.apache.iotdb.mpp.rpc.thrift.TUpdateTableReq; +import org.apache.iotdb.mpp.rpc.thrift.TUpdateTemplateReq; import org.apache.iotdb.rpc.RpcUtils; import org.apache.iotdb.rpc.TSStatusCode; @@ -272,10 +273,16 @@ public static Map broadcastTableUpdate( return clientHandler.getResponseMap(); } - private static Map failedOnly(final Map responses) { - return responses.entrySet().stream() - .filter(entry -> entry.getValue().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) - .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); + public static Map broadcastTemplateUpdate( + final TUpdateTemplateReq req, final Map targets) { + final DataNodeAsyncRequestContext clientHandler = + new DataNodeAsyncRequestContext<>(CnToDnAsyncRequestType.UPDATE_TEMPLATE, req, targets); + CnToDnInternalServiceAsyncRequestManager.getInstance() + .sendAsyncRequest( + clientHandler, + ClusterCachePropagator.BROADCAST_RPC_RETRY, + ClusterCachePropagator.BROADCAST_RPC_TIMEOUT_MS); + return clientHandler.getResponseMap(); } public static Map filterFencedDataNode( diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SetTemplateProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SetTemplateProcedure.java index 52406f2519c3..25a6a6c38fe5 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SetTemplateProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SetTemplateProcedure.java @@ -30,8 +30,6 @@ import org.apache.iotdb.commons.path.PathPatternTree; import org.apache.iotdb.commons.schema.template.Template; import org.apache.iotdb.confignode.client.async.CnToDnAsyncRequestType; -import org.apache.iotdb.confignode.client.async.CnToDnInternalServiceAsyncRequestManager; -import org.apache.iotdb.confignode.client.async.handlers.DataNodeAsyncRequestContext; import org.apache.iotdb.confignode.consensus.request.read.template.CheckTemplateSettablePlan; import org.apache.iotdb.confignode.consensus.request.read.template.GetSchemaTemplatePlan; import org.apache.iotdb.confignode.consensus.request.write.pipe.payload.PipeEnrichedPlan; @@ -40,6 +38,7 @@ import org.apache.iotdb.confignode.consensus.response.template.TemplateInfoResp; import org.apache.iotdb.confignode.i18n.ConfigNodeMessages; import org.apache.iotdb.confignode.i18n.ProcedureMessages; +import org.apache.iotdb.confignode.manager.lease.ClusterCachePropagator; import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv; import org.apache.iotdb.confignode.procedure.exception.ProcedureException; import org.apache.iotdb.confignode.procedure.impl.StateMachineProcedure; @@ -219,26 +218,18 @@ private void preReleaseTemplate(final ConfigNodeProcedureEnv env) { req.setType(TemplateInternalRPCUpdateType.ADD_TEMPLATE_PRE_SET_INFO.toByte()); req.setTemplateInfo( TemplateInternalRPCUtil.generateAddTemplateSetInfoBytes(template, templateSetPath)); - - final Map dataNodeLocationMap = - env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations(); - final DataNodeAsyncRequestContext clientHandler = - new DataNodeAsyncRequestContext<>( - CnToDnAsyncRequestType.UPDATE_TEMPLATE, req, dataNodeLocationMap); - CnToDnInternalServiceAsyncRequestManager.getInstance().sendAsyncRequestWithRetry(clientHandler); - final Map statusMap = clientHandler.getResponseMap(); - for (final Map.Entry entry : statusMap.entrySet()) { - if (entry.getValue().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { - LOGGER.warn( - ProcedureMessages.FAILED_TO_SYNC_TEMPLATE_PRE_SET_INFO_ON_PATH_TO, - templateName, - templateSetPath, - dataNodeLocationMap.get(entry.getKey())); - setFailure( - new ProcedureException( - new MetadataException(ProcedureMessages.PRE_SET_TEMPLATE_FAILED))); - return; - } + boolean proceed = + new ClusterCachePropagator(SchemaUtils.filterFencedDataNode(env.getConfigManager())) + .propagate(targets -> SchemaUtils.broadcastTemplateUpdate(req, targets)); + if (!proceed) { + LOGGER.warn( + ProcedureMessages.FAILED_TO_SYNC_TEMPLATE_PRE_SET_INFO_ON_PATH_TO, + templateName, + templateSetPath, + ProcedureMessages.FAILED_TO_PROVE_AN_UNREACHABLE_DN_IS_FENCED); + setFailure( + new ProcedureException(new MetadataException(ProcedureMessages.PRE_SET_TEMPLATE_FAILED))); + return; } setNextState(SetTemplateState.VALIDATE_TIMESERIES_EXISTENCE); } @@ -400,37 +391,28 @@ private void commitReleaseTemplate(final ConfigNodeProcedureEnv env) { // already setFailure return; } - final TUpdateTemplateReq req = new TUpdateTemplateReq(); req.setType(TemplateInternalRPCUpdateType.COMMIT_TEMPLATE_SET_INFO.toByte()); req.setTemplateInfo( TemplateInternalRPCUtil.generateAddTemplateSetInfoBytes(template, templateSetPath)); - final Map dataNodeLocationMap = - env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations(); - final DataNodeAsyncRequestContext clientHandler = - new DataNodeAsyncRequestContext<>( - CnToDnAsyncRequestType.UPDATE_TEMPLATE, req, dataNodeLocationMap); - CnToDnInternalServiceAsyncRequestManager.getInstance().sendAsyncRequestWithRetry(clientHandler); - final Map statusMap = clientHandler.getResponseMap(); - for (final Map.Entry entry : statusMap.entrySet()) { - if (entry.getValue().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { - LOGGER.warn( - ProcedureMessages.FAILED_TO_SYNC_TEMPLATE_COMMIT_SET_INFO_ON_PATH_TO, - templateName, - templateSetPath, - dataNodeLocationMap.get(entry.getKey())); - setFailure( - new ProcedureException( - new MetadataException( - String.format( - ProcedureMessages - .FAILED_TO_SET_SCHEMAENGINE_TEMPLATE_ON_PATH_BECAUSE_THERE_S, - templateName, - templateSetPath, - dataNodeLocationMap.get(entry.getKey()))))); - return; - } + boolean proceed = + new ClusterCachePropagator(SchemaUtils.filterFencedDataNode(env.getConfigManager())) + .propagate(targets -> SchemaUtils.broadcastTemplateUpdate(req, targets)); + if (!proceed) { + LOGGER.warn( + ProcedureMessages.FAILED_TO_SYNC_TEMPLATE_COMMIT_SET_INFO_ON_PATH_TO, + templateName, + templateSetPath, + ProcedureMessages.FAILED_TO_PROVE_AN_UNREACHABLE_DN_IS_FENCED); + setFailure( + new ProcedureException( + new MetadataException( + String.format( + ProcedureMessages.FAILED_TO_SET_SCHEMAENGINE_TEMPLATE_ON_PATH_BECAUSE_THERE_S, + templateName, + templateSetPath, + ProcedureMessages.FAILED_TO_PROVE_AN_UNREACHABLE_DN_IS_FENCED)))); } } @@ -508,35 +490,25 @@ private void rollbackPreRelease(final ConfigNodeProcedureEnv env) { return; } - final Map dataNodeLocationMap = - env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations(); - - final TUpdateTemplateReq invalidateTemplateSetInfoReq = new TUpdateTemplateReq(); - invalidateTemplateSetInfoReq.setType( - TemplateInternalRPCUpdateType.INVALIDATE_TEMPLATE_SET_INFO.toByte()); - invalidateTemplateSetInfoReq.setTemplateInfo( + final TUpdateTemplateReq req = new TUpdateTemplateReq(); + req.setType(TemplateInternalRPCUpdateType.INVALIDATE_TEMPLATE_SET_INFO.toByte()); + req.setTemplateInfo( TemplateInternalRPCUtil.generateInvalidateTemplateSetInfoBytes( template.getId(), templateSetPath)); - final DataNodeAsyncRequestContext clientHandler = - new DataNodeAsyncRequestContext<>( - CnToDnAsyncRequestType.UPDATE_TEMPLATE, - invalidateTemplateSetInfoReq, - dataNodeLocationMap); - CnToDnInternalServiceAsyncRequestManager.getInstance().sendAsyncRequestWithRetry(clientHandler); - final Map statusMap = clientHandler.getResponseMap(); - for (final Map.Entry entry : statusMap.entrySet()) { - // all dataNodes must clear the related template cache - if (entry.getValue().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { - LOGGER.error( - ProcedureMessages.FAILED_TO_ROLLBACK_PRE_RELEASE_TEMPLATE_INFO_OF_TEMPLATE_SET, - template.getName(), - templateSetPath, - dataNodeLocationMap.get(entry.getKey())); - setFailure( - new ProcedureException( - new MetadataException(ProcedureMessages.ROLLBACK_PRE_RELEASE_TEMPLATE_FAILED))); - } + boolean proceed = + new ClusterCachePropagator(SchemaUtils.filterFencedDataNode(env.getConfigManager())) + .propagate(targets -> SchemaUtils.broadcastTemplateUpdate(req, targets)); + + if (!proceed) { + LOGGER.error( + ProcedureMessages.FAILED_TO_ROLLBACK_PRE_RELEASE_TEMPLATE_INFO_OF_TEMPLATE_SET, + template.getName(), + templateSetPath, + ProcedureMessages.FAILED_TO_PROVE_AN_UNREACHABLE_DN_IS_FENCED); + setFailure( + new ProcedureException( + new MetadataException(ProcedureMessages.ROLLBACK_PRE_RELEASE_TEMPLATE_FAILED))); } } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/UnsetTemplateProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/UnsetTemplateProcedure.java index 1fd7aefb3306..9966c9613af7 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/UnsetTemplateProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/UnsetTemplateProcedure.java @@ -31,6 +31,7 @@ import org.apache.iotdb.confignode.client.async.CnToDnInternalServiceAsyncRequestManager; import org.apache.iotdb.confignode.client.async.handlers.DataNodeAsyncRequestContext; import org.apache.iotdb.confignode.i18n.ProcedureMessages; +import org.apache.iotdb.confignode.manager.lease.ClusterCachePropagator; import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv; import org.apache.iotdb.confignode.procedure.exception.ProcedureException; import org.apache.iotdb.confignode.procedure.impl.StateMachineProcedure; @@ -154,30 +155,21 @@ private void invalidateCache(final ConfigNodeProcedureEnv env) { } } - private void executeInvalidateCache(final ConfigNodeProcedureEnv env) throws ProcedureException { - final Map dataNodeLocationMap = - env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations(); - final TUpdateTemplateReq invalidateTemplateSetInfoReq = new TUpdateTemplateReq(); - invalidateTemplateSetInfoReq.setType( - TemplateInternalRPCUpdateType.INVALIDATE_TEMPLATE_SET_INFO.toByte()); - invalidateTemplateSetInfoReq.setTemplateInfo(getInvalidateTemplateSetInfo()); - final DataNodeAsyncRequestContext clientHandler = - new DataNodeAsyncRequestContext<>( - CnToDnAsyncRequestType.UPDATE_TEMPLATE, - invalidateTemplateSetInfoReq, - dataNodeLocationMap); - CnToDnInternalServiceAsyncRequestManager.getInstance().sendAsyncRequestWithRetry(clientHandler); - final Map statusMap = clientHandler.getResponseMap(); - for (final TSStatus status : statusMap.values()) { - // all dataNodes must clear the related template cache - if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { - LOGGER.error( - ProcedureMessages.FAILED_TO_INVALIDATE_TEMPLATE_CACHE_OF_TEMPLATE_SET_ON, - template.getName(), - path); - throw new ProcedureException( - new MetadataException(ProcedureMessages.INVALIDATE_TEMPLATE_CACHE_FAILED)); - } + void executeInvalidateCache(final ConfigNodeProcedureEnv env) throws ProcedureException { + final TUpdateTemplateReq req = new TUpdateTemplateReq(); + req.setType(TemplateInternalRPCUpdateType.INVALIDATE_TEMPLATE_SET_INFO.toByte()); + req.setTemplateInfo(getInvalidateTemplateSetInfo()); + + final boolean proceed = + new ClusterCachePropagator(SchemaUtils.filterFencedDataNode(env.getConfigManager())) + .propagate(targets -> SchemaUtils.broadcastTemplateUpdate(req, targets)); + if (!proceed) { + LOGGER.error( + ProcedureMessages.FAILED_TO_INVALIDATE_TEMPLATE_CACHE_OF_TEMPLATE_SET_ON, + template.getName(), + path); + throw new ProcedureException( + new MetadataException(ProcedureMessages.INVALIDATE_TEMPLATE_CACHE_FAILED)); } } @@ -255,8 +247,7 @@ protected void rollbackState( } } - private void executeRollbackInvalidateCache(ConfigNodeProcedureEnv env) - throws ProcedureException { + void executeRollbackInvalidateCache(ConfigNodeProcedureEnv env) throws ProcedureException { Map dataNodeLocationMap = env.getConfigManager().getNodeManager().getRegisteredDataNodeLocations(); TUpdateTemplateReq rollbackTemplateSetInfoReq = new TUpdateTemplateReq(); diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/LeaseRecoverySnapshotRegressionTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/LeaseRecoverySnapshotRegressionTest.java new file mode 100644 index 000000000000..a92dd8c7e4cd --- /dev/null +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/LeaseRecoverySnapshotRegressionTest.java @@ -0,0 +1,75 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.confignode.manager; + +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.confignode.consensus.request.read.template.GetAllTemplateSetInfoPlan; +import org.apache.iotdb.confignode.manager.consensus.ConsensusManager; +import org.apache.iotdb.confignode.manager.schema.ClusterSchemaManager; +import org.apache.iotdb.confignode.rpc.thrift.TDataNodeLeaseRecoveryResp; +import org.apache.iotdb.consensus.exception.ConsensusException; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.junit.Assert; +import org.junit.Test; +import org.mockito.Mockito; + +import java.lang.reflect.Field; + +public class LeaseRecoverySnapshotRegressionTest { + + @Test + public void consensusFailureMustNotBecomeAnEmptyTemplateSnapshot() throws ConsensusException { + final IManager manager = Mockito.mock(IManager.class); + final ConsensusManager consensusManager = Mockito.mock(ConsensusManager.class); + Mockito.when(manager.getConsensusManager()).thenReturn(consensusManager); + Mockito.when(consensusManager.read(Mockito.any(GetAllTemplateSetInfoPlan.class))) + .thenThrow(new ConsensusException("injected ConfigRegion read failure")); + + final ClusterSchemaManager schemaManager = new ClusterSchemaManager(manager, null, null); + + Assert.assertNotEquals( + "A failed consensus read must not be represented as a valid empty snapshot", + 0, + schemaManager.getAllTemplateSetInfo().length); + } + + @Test + public void emptyTemplateSnapshotMustNotReturnSuccessfulLeaseRecovery() throws Exception { + final ConfigManager configManager = Mockito.mock(ConfigManager.class); + final ClusterSchemaManager schemaManager = Mockito.mock(ClusterSchemaManager.class); + Mockito.when(configManager.confirmLeader()) + .thenReturn(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())); + Mockito.when(schemaManager.getAllTableInfoForDataNodeActivation()).thenReturn(new byte[] {1}); + Mockito.when(schemaManager.getAllTemplateSetInfo()).thenReturn(new byte[0]); + Mockito.when(configManager.reloadCacheAfterLeaseRecovery()).thenCallRealMethod(); + + final Field schemaManagerField = ConfigManager.class.getDeclaredField("clusterSchemaManager"); + schemaManagerField.setAccessible(true); + schemaManagerField.set(configManager, schemaManager); + + final TDataNodeLeaseRecoveryResp response = configManager.reloadCacheAfterLeaseRecovery(); + + Assert.assertNotEquals( + "An incomplete cache snapshot must keep the DataNode fenced", + TSStatusCode.SUCCESS_STATUS.getStatusCode(), + response.getStatus().getCode()); + } +} diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/lease/ClusterCachePropagatorTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/lease/ClusterCachePropagatorTest.java index 00ccd9cd8f43..8f74b118e466 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/lease/ClusterCachePropagatorTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/lease/ClusterCachePropagatorTest.java @@ -21,14 +21,18 @@ import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation; import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.confignode.client.async.CnToDnAsyncRequestType; +import org.apache.iotdb.confignode.client.async.handlers.rpc.DataNodeTSStatusRPCHandler; import org.apache.iotdb.confignode.manager.lease.MetadataBroadcastVerdict.Verdict; import org.apache.iotdb.rpc.TSStatusCode; +import org.apache.thrift.TApplicationException; import org.junit.Assert; import org.junit.Test; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import java.util.function.IntToLongFunction; @@ -141,6 +145,30 @@ public void internalFailureFailsImmediately() { Assert.assertEquals(Verdict.FAIL, v); } + @Test + public void locallyFencedButReachableDataNodeMustBeRetried() { + final ClusterCachePropagator p = propagator(id -> 0L); + final Map targets = twoDataNodes(); + final Map responses = new HashMap<>(); + responses.put(1, success()); + + new DataNodeTSStatusRPCHandler( + CnToDnAsyncRequestType.UPDATE_TEMPLATE, + 2, + targets.get(2), + targets, + responses, + new CountDownLatch(1)) + .onError(new TApplicationException("Metadata lease is still fenced")); + + Assert.assertEquals( + TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode(), responses.get(2).getCode()); + + final Verdict v = p.propagateOnce(ignored -> responses, false); + Assert.assertEquals( + "The async recovery window must be retried instead of failing DDL", Verdict.WAIT, v); + } + @Test public void loopReturnsTrueWhenItEventuallyProceeds() { final AtomicInteger calls = new AtomicInteger(); diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/UnsetTemplateRollbackRegressionTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/UnsetTemplateRollbackRegressionTest.java new file mode 100644 index 000000000000..1e77942d4c33 --- /dev/null +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/UnsetTemplateRollbackRegressionTest.java @@ -0,0 +1,78 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.confignode.procedure.impl.schema; + +import org.apache.iotdb.commons.path.PartialPath; +import org.apache.iotdb.commons.schema.template.Template; +import org.apache.iotdb.confignode.manager.ConfigManager; +import org.apache.iotdb.confignode.manager.schema.ClusterSchemaManager; +import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv; +import org.apache.iotdb.confignode.procedure.exception.ProcedureException; +import org.apache.iotdb.confignode.procedure.state.schema.UnsetTemplateState; + +import org.junit.Test; +import org.mockito.Mockito; + +import java.io.IOException; + +public class UnsetTemplateRollbackRegressionTest { + + @Test + public void rollbackMustRestoreConfigNodeStateWhenDataNodeCacheRepairFails() throws Exception { + final Template template = new Template(); + template.setId(1); + template.setName("t1"); + final PartialPath path = new PartialPath("root.sg"); + final ConfigNodeProcedureEnv env = Mockito.mock(ConfigNodeProcedureEnv.class); + final ConfigManager configManager = Mockito.mock(ConfigManager.class); + final ClusterSchemaManager schemaManager = Mockito.mock(ClusterSchemaManager.class); + Mockito.when(env.getConfigManager()).thenReturn(configManager); + Mockito.when(configManager.getClusterSchemaManager()).thenReturn(schemaManager); + + final FailingDataNodeRollbackProcedure procedure = + new FailingDataNodeRollbackProcedure(template, path); + procedure.rollbackAfterActivationCheck(env); + + Mockito.verify(schemaManager).rollbackPreUnsetSchemaTemplate(template.getId(), path); + } + + private static class FailingDataNodeRollbackProcedure extends UnsetTemplateProcedure { + + private FailingDataNodeRollbackProcedure(final Template template, final PartialPath path) { + super("test", template, path, false); + } + + @Override + void executeRollbackInvalidateCache(final ConfigNodeProcedureEnv env) + throws ProcedureException { + throw new ProcedureException("injected offline DataNode"); + } + + @Override + void executeInvalidateCache(final ConfigNodeProcedureEnv env) { + // No-op: this test isolates ordering in the rollback path. + } + + private void rollbackAfterActivationCheck(final ConfigNodeProcedureEnv env) + throws IOException, InterruptedException, ProcedureException { + rollbackState(env, UnsetTemplateState.CHECK_DATANODE_TEMPLATE_ACTIVATION); + } + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/lease/MetadataLeaseManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/lease/MetadataLeaseManager.java index 9f00901da568..c3cbd0d48b5f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/lease/MetadataLeaseManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/lease/MetadataLeaseManager.java @@ -19,22 +19,30 @@ package org.apache.iotdb.db.schemaengine.lease; +import org.apache.iotdb.commons.client.exception.ClientManagerException; import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory; import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil; +import org.apache.iotdb.commons.exception.IoTDBRuntimeException; import org.apache.iotdb.commons.exception.MetadataLeaseFencedException; import org.apache.iotdb.commons.utils.TestOnly; +import org.apache.iotdb.confignode.rpc.thrift.TDataNodeLeaseRecoveryResp; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.i18n.DataNodeSchemaMessages; +import org.apache.iotdb.db.protocol.client.ConfigNodeClient; +import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager; +import org.apache.iotdb.db.protocol.client.ConfigNodeInfo; import org.apache.iotdb.db.queryengine.plan.analyze.ClusterPartitionFetcher; import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TreeDeviceSchemaCacheManager; import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache; +import org.apache.iotdb.db.schemaengine.template.ClusterTemplateManager; +import org.apache.iotdb.rpc.TSStatusCode; +import org.apache.thrift.TException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.ArrayList; import java.util.Arrays; -import java.util.Collections; import java.util.List; import java.util.concurrent.ExecutorService; import java.util.concurrent.ScheduledExecutorService; @@ -45,6 +53,7 @@ import static org.apache.iotdb.commons.concurrent.ThreadName.CHECK_DN_LEASE_STATUS; import static org.apache.iotdb.commons.concurrent.ThreadName.RELOAD_TABLE_METADATA_CACHE; +import static org.apache.iotdb.db.i18n.DataNodeSchemaMessages.FAILED_TO_REFRESH_CACHE_FROM_CN; /** * Tracks the DataNode's "metadata lease" with the ConfigNode. The ConfigNode periodically sends @@ -70,7 +79,6 @@ interface MetadataAction { private final Logger LOGGER = LoggerFactory.getLogger(MetadataLeaseManager.class); private final List clearCacheList; - private final List pullMetaList; private final LongSupplier nanoClock; private volatile long fenceThresholdMs = 20000; @@ -95,7 +103,6 @@ private MetadataLeaseManager() { this( System::nanoTime, defaultClearCacheList(), - defaultPullMetaList(), IoTDBThreadPoolFactory.newCachedThreadPool(RELOAD_TABLE_METADATA_CACHE.getName()), IoTDBDescriptor.getInstance().getConfig().getCheckDnLeaseStatusIntervalMs(), IoTDBThreadPoolFactory.newScheduledThreadPool(1, CHECK_DN_LEASE_STATUS.getName())); @@ -105,24 +112,18 @@ private static List defaultClearCacheList() { return Arrays.asList( () -> ClusterPartitionFetcher.getInstance().invalidAllCache(), () -> DataNodeTableCache.getInstance().invalidateAll(), - () -> TreeDeviceSchemaCacheManager.getInstance().cleanUp()); - } - - private static List defaultPullMetaList() { - return Collections.singletonList( - () -> DataNodeTableCache.getInstance().reloadTableCacheAfterLeaseRecovery()); + () -> TreeDeviceSchemaCacheManager.getInstance().cleanUp(), + () -> ClusterTemplateManager.getInstance().clear()); } MetadataLeaseManager( final LongSupplier nanoClock, final List clearCacheList, - final List pullMetaList, final ExecutorService pullExecutorService, final long checkDnLeaseStatusIntervalMs, final ScheduledExecutorService checkLeaseStatusExecutor) { this.nanoClock = nanoClock; this.clearCacheList = new ArrayList<>(clearCacheList); - this.pullMetaList = new ArrayList<>(pullMetaList); // Startup registration performs a full re-sync, so treat construction time as a fresh contact. this.lastConfigNodeHeartbeatNanos = nanoClock.getAsLong(); @@ -222,20 +223,38 @@ private void pullMetaDataAndInit() { LOGGER.error(DataNodeSchemaMessages.FAILED_TO_MARK_METADATA_STATE_AS_PULLING, metadataState); return; } - - for (final MetadataAction action : pullMetaList) { - try { - action.execute(); - } catch (final Throwable t) { - metadataStateRef.set(MetadataState.PULL_OR_INIT_FAILED, metadataStateRef.getStamp() + 1); - LOGGER.error(DataNodeSchemaMessages.FAILED_TO_PULL_OR_INIT_METADATA, t); - rethrowUnchecked(t); - } + try { + reloadRelatedCache(); + } catch (final Throwable t) { + metadataStateRef.set(MetadataState.PULL_OR_INIT_FAILED, metadataStateRef.getStamp() + 1); + LOGGER.error(DataNodeSchemaMessages.FAILED_TO_PULL_OR_INIT_METADATA, t); + rethrowUnchecked(t); } + this.lastConfigNodeHeartbeatNanos = nanoClock.getAsLong(); metadataStateRef.set(MetadataState.NORMAL, metadataStateRef.getStamp() + 1); } + void reloadRelatedCache() { + try (ConfigNodeClient configNodeClient = + ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) { + + final TDataNodeLeaseRecoveryResp resp = configNodeClient.reloadCacheAfterLeaseRecovery(); + if (resp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + throw new IoTDBRuntimeException(resp.getStatus().getMessage(), resp.getStatus().getCode()); + } + if (!resp.isSetTableInfo() || !resp.isSetTemplateInfo()) { + throw new RuntimeException(FAILED_TO_REFRESH_CACHE_FROM_CN); + } + DataNodeTableCache.getInstance().reloadTableCacheAfterLeaseRecovery(resp.getTableInfo()); + ClusterTemplateManager.getInstance() + .reloadTemplateCacheAfterLeaseRecovery(resp.getTemplateInfo()); + + } catch (final ClientManagerException | TException e) { + throw new RuntimeException(FAILED_TO_REFRESH_CACHE_FROM_CN, e); + } + } + private static void rethrowUnchecked(final Throwable t) { if (t instanceof Error) { throw (Error) t; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java index 5b90104de838..eb48a3aef02c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java @@ -21,9 +21,7 @@ import org.apache.iotdb.calc.plan.relational.metadata.CommonMetadataUtils; import org.apache.iotdb.commons.client.IClientManager; -import org.apache.iotdb.commons.client.exception.ClientManagerException; import org.apache.iotdb.commons.consensus.ConfigRegionId; -import org.apache.iotdb.commons.exception.IoTDBRuntimeException; import org.apache.iotdb.commons.exception.SemanticException; import org.apache.iotdb.commons.schema.table.NonCommittableTsTable; import org.apache.iotdb.commons.schema.table.PreDeleteTsTable; @@ -32,18 +30,15 @@ import org.apache.iotdb.commons.schema.table.TsTableInternalRPCUtil; import org.apache.iotdb.commons.schema.table.column.TsTableColumnSchema; import org.apache.iotdb.commons.utils.PathUtils; -import org.apache.iotdb.confignode.rpc.thrift.TDataNodeLeaseRecoveryResp; import org.apache.iotdb.confignode.rpc.thrift.TFetchTableResp; import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.i18n.DataNodeSchemaMessages; import org.apache.iotdb.db.protocol.client.ConfigNodeClient; import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager; -import org.apache.iotdb.db.protocol.client.ConfigNodeInfo; import org.apache.iotdb.db.queryengine.plan.execution.config.executor.ClusterConfigTaskExecutor; import org.apache.iotdb.db.schemaengine.lease.MetadataLeaseManager; import org.apache.iotdb.rpc.TSStatusCode; -import org.apache.thrift.TException; import org.apache.tsfile.utils.Pair; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -66,8 +61,6 @@ import java.util.function.Function; import java.util.stream.Collectors; -import static org.apache.iotdb.db.i18n.DataNodeSchemaMessages.FAILED_TO_REFRESH_CACHE_FROM_CN; - /** It contains all tables' latest column schema */ public class DataNodeTableCache implements ITableCache { @@ -145,24 +138,6 @@ public void init(final byte[] tableInitializationBytes) { } } - // No need to acquire a lock here; reloadTableCacheAfterLeaseRecovery is within a critical section - // protected by metadataLeaseManager - @Override - public void reloadTableCacheAfterLeaseRecovery() { - try (ConfigNodeClient configNodeClient = - CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) { - final TDataNodeLeaseRecoveryResp resp = configNodeClient.reloadCacheAfterLeaseRecovery(); - if (resp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { - throw new IoTDBRuntimeException(resp.getStatus().getMessage(), resp.getStatus().getCode()); - } - if (resp.isSetTableInfo()) { - init(resp.getTableInfo()); - } - } catch (final ClientManagerException | TException e) { - throw new RuntimeException(FAILED_TO_REFRESH_CACHE_FROM_CN, e); - } - } - /** * The case that pre update Table and pre delete Table procedures targeting the same table are * executed serially by CN. @@ -817,4 +792,15 @@ public String tryGetInternColumnName( return null; } } + + @Override + public void reloadTableCacheAfterLeaseRecovery(byte[] tableInfo) { + readWriteLock.writeLock().lock(); + try { + invalidateAll(); + init(tableInfo); + } finally { + readWriteLock.writeLock().unlock(); + } + } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/ITableCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/ITableCache.java index 26f67fb11295..090d1774956d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/ITableCache.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/ITableCache.java @@ -62,5 +62,5 @@ String tryGetInternColumnName( boolean isDatabaseExist(final String database); - void reloadTableCacheAfterLeaseRecovery(); + void reloadTableCacheAfterLeaseRecovery(byte[] tableInfo); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/template/ClusterTemplateManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/template/ClusterTemplateManager.java index 825ea757a777..ab5a777aa6cd 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/template/ClusterTemplateManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/template/ClusterTemplateManager.java @@ -44,6 +44,7 @@ import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager; import org.apache.iotdb.db.protocol.client.ConfigNodeInfo; import org.apache.iotdb.db.queryengine.plan.statement.metadata.template.CreateSchemaTemplateStatement; +import org.apache.iotdb.db.schemaengine.lease.MetadataLeaseManager; import org.apache.iotdb.db.utils.SchemaUtils; import org.apache.iotdb.rpc.RpcUtils; import org.apache.iotdb.rpc.TSStatusCode; @@ -273,6 +274,7 @@ public List getPathsSetTemplate(String name, PathPatternTree scope) public Template getTemplate(int id) { readWriteLock.readLock().lock(); try { + failIfMetadataLeaseFenced(); return templateIdMap.get(id); } finally { readWriteLock.readLock().unlock(); @@ -283,6 +285,7 @@ public Template getTemplate(int id) { public Pair checkTemplateSetInfo(PartialPath devicePath) { readWriteLock.readLock().lock(); try { + failIfMetadataLeaseFenced(); for (PartialPath templateSetPath : pathSetTemplateMap.keySet()) { if (devicePath.startsWithOrPrefixOf(templateSetPath.getNodes())) { return new Pair<>( @@ -300,6 +303,7 @@ public Pair checkTemplateSetAndPreSetInfo( PartialPath timeSeriesPath, String alias) { readWriteLock.readLock().lock(); try { + failIfMetadataLeaseFenced(); for (PartialPath templateSetPath : pathSetTemplateMap.keySet()) { if (timeSeriesPath.startsWithOrPrefixOf(templateSetPath.getNodes())) { return new Pair<>( @@ -343,6 +347,7 @@ public Pair checkTemplateSetAndPreSetInfo( public Pair> getAllPathsSetTemplate(String templateName) { readWriteLock.readLock().lock(); try { + failIfMetadataLeaseFenced(); if (!templateNameMap.containsKey(templateName)) { return null; } @@ -357,6 +362,7 @@ public Pair> getAllPathsSetTemplate(String templateN public Map checkAllRelatedTemplate(PartialPath pathPattern) { readWriteLock.readLock().lock(); try { + failIfMetadataLeaseFenced(); Map result = new HashMap<>(); int templateId; Template template; @@ -377,6 +383,7 @@ public Map checkAllRelatedTemplate(PartialPath pathPattern) { public List