From e71b3ce62b119340904cc70d6a3a2efe843078c6 Mon Sep 17 00:00:00 2001 From: Yaobin Chen Date: Mon, 20 Jul 2026 16:53:22 +0800 Subject: [PATCH 1/6] support high availability of DeactivateTemplateProcedure --- .../it/schema/IoTDBDeviceTemplateIT.java | 97 +++++++++++++++++++ .../schema/DeactivateTemplateProcedure.java | 37 ++----- 2 files changed, 107 insertions(+), 27 deletions(-) create mode 100644 integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDeviceTemplateIT.java 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..206aeb5a3b50 --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDeviceTemplateIT.java @@ -0,0 +1,97 @@ +/* + * 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.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 java.sql.Connection; +import java.sql.ResultSet; +import java.sql.Statement; + +@RunWith(IoTDBTestRunner.class) +@Category({ClusterIT.class}) +public class IoTDBDeviceTemplateIT { + + private static final String DATABASE = "root.device_template_ha"; + private static final String TEMPLATE = "temp1"; + private static final String DEVICE = DATABASE + ".dev01"; + + @Test + public void testDeactivateTemplateWithOneDataNodeDown() throws Exception { + 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); + + 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()) { + statement.execute("CREATE DATABASE " + DATABASE); + statement.execute("CREATE DEVICE TEMPLATE " + TEMPLATE + " (s1 INT32, s2 INT64)"); + statement.execute("SET DEVICE TEMPLATE " + TEMPLATE + " TO " + DATABASE); + statement.execute("CREATE TIMESERIES OF DEVICE TEMPLATE ON " + DEVICE); + statement.execute("INSERT INTO " + DEVICE + "(time, s1, s2) VALUES(1, 1, 1)"); + + victimDataNode.stop(); + Assert.assertFalse("victim DataNode should be stopped", victimDataNode.isAlive()); + + statement.execute("DEACTIVATE DEVICE TEMPLATE " + TEMPLATE + " FROM " + DEVICE); + + try (final ResultSet resultSet = + statement.executeQuery("SHOW PATHS USING DEVICE TEMPLATE " + TEMPLATE)) { + Assert.assertFalse("the device template should be deactivated", resultSet.next()); + } + try (final ResultSet resultSet = + statement.executeQuery("SHOW PATHS SET DEVICE TEMPLATE " + TEMPLATE)) { + Assert.assertTrue("the device template should remain set", resultSet.next()); + Assert.assertEquals(DATABASE, resultSet.getString(1)); + Assert.assertFalse(resultSet.next()); + } + } + } finally { + EnvFactory.getEnv().cleanClusterEnvironment(); + } + } +} 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); From 3a607b2e26bd17f5ae1d8cb548e7ea9110b97990 Mon Sep 17 00:00:00 2001 From: Yaobin Chen Date: Tue, 21 Jul 2026 12:00:44 +0800 Subject: [PATCH 2/6] register the clusterTemplateManager to the metadataLeaseManager --- .../confignode/manager/ConfigManager.java | 3 +- .../lease/MetadataLeaseManager.java | 59 ++++++++++++------- .../table/DataNodeTableCache.java | 36 ++++------- .../db/schemaengine/table/ITableCache.java | 2 +- .../template/ClusterTemplateManager.java | 48 ++++++++++++--- .../org/apache/iotdb/db/service/DataNode.java | 2 +- .../lease/MetadataLeaseManagerTest.java | 35 ++++++----- .../lease/MetadataLeaseTestUtils.java | 13 ++-- .../src/main/thrift/confignode.thrift | 1 + 9 files changed, 118 insertions(+), 81 deletions(-) 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/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..c6d5e1b51669 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