diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppConfigNodeConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppConfigNodeConfig.java
index 4385097ce045..8da0f4e02b24 100644
--- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppConfigNodeConfig.java
+++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppConfigNodeConfig.java
@@ -73,4 +73,10 @@ public ConfigNodeConfig setLeaderDistributionPolicy(String policy) {
     properties.setProperty("leader_distribution_policy", policy);
     return this;
   }
+
+  @Override
+  public ConfigNodeConfig setMetadataLeaseFenceMs(long metadataLeaseFenceMs) {
+    properties.setProperty("metadata_lease_fence_ms", String.valueOf(metadataLeaseFenceMs));
+    return this;
+  }
 }
diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java
index 48544b901d4a..36d06aefe889 100644
--- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java
+++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppSharedCommonConfig.java
@@ -64,7 +64,6 @@ public CommonConfig setMemtableSizeThreshold(long memtableSizeThreshold) {
   @Override
   public CommonConfig setMetadataLeaseFenceMs(long metadataLeaseFenceMs) {
     cnConfig.setMetadataLeaseFenceMs(metadataLeaseFenceMs);
-    dnConfig.setMetadataLeaseFenceMs(metadataLeaseFenceMs);
     return this;
   }
 
diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteConfigNodeConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteConfigNodeConfig.java
index c16722f4bd94..36fbdcbfc0e8 100644
--- a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteConfigNodeConfig.java
+++ b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteConfigNodeConfig.java
@@ -43,4 +43,9 @@ public ConfigNodeConfig setMetricPrometheusReporterPassword(String password) {
   public ConfigNodeConfig setLeaderDistributionPolicy(String policy) {
     return this;
   }
+
+  @Override
+  public ConfigNodeConfig setMetadataLeaseFenceMs(long metadataLeaseFenceMs) {
+    return this;
+  }
 }
diff --git a/integration-test/src/main/java/org/apache/iotdb/itbase/env/ConfigNodeConfig.java b/integration-test/src/main/java/org/apache/iotdb/itbase/env/ConfigNodeConfig.java
index 4af35f6f56a9..aec8f0f23354 100644
--- a/integration-test/src/main/java/org/apache/iotdb/itbase/env/ConfigNodeConfig.java
+++ b/integration-test/src/main/java/org/apache/iotdb/itbase/env/ConfigNodeConfig.java
@@ -31,4 +31,6 @@ public interface ConfigNodeConfig {
   ConfigNodeConfig setMetricPrometheusReporterPassword(String password);
 
   ConfigNodeConfig setLeaderDistributionPolicy(String policy);
+
+  ConfigNodeConfig setMetadataLeaseFenceMs(long metadataLeaseFenceMs);
 }
diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBTableDDLHAIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBTableDDLHAIT.java
index 396a918d173b..68f6c1181121 100644
--- a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBTableDDLHAIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBTableDDLHAIT.java
@@ -21,6 +21,7 @@
 
 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;
@@ -53,16 +54,19 @@ public class IoTDBTableDDLHAIT {
   private final String tableName = "table_ddl_ha";
   private final String createdAfterDownTableName = "table_ddl_ha_created_after_down";
 
+  private final String sourceTableName = "SOURCE_TABLE";
+
   private static void initCluster() {
     EnvFactory.getEnv()
         .getConfig()
         .getCommonConfig()
-        .setMetadataLeaseFenceMs(20000)
         .setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
         .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
         .setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS)
         .setSchemaReplicationFactor(3)
-        .setDataReplicationFactor(3);
+        .setDataReplicationFactor(2);
+
+    EnvFactory.getEnv().getConfig().getConfigNodeConfig().setMetadataLeaseFenceMs(20000);
     EnvFactory.getEnv().initClusterEnvironment(1, 3);
   }
 
@@ -83,6 +87,14 @@ private void preTableData(Statement statement, String databaseName, String table
     statement.execute("CREATE TABLE TABLE1 (dev STRING TAG, s1 INT32 FIELD)");
     statement.execute(
         "INSERT INTO TABLE1 (time, dev, s1) VALUES(1, 'dev01', 1), (2, 'dev02', 2), (3, 'dev03', 3)");
+
+    // ready for the drop database
+    statement.execute(
+        "CREATE TABLE " + sourceTableName + " (dev STRING TAG, s1 INT32 FIELD, s4 INT32 FIELD)");
+    statement.execute(
+        "INSERT INTO "
+            + sourceTableName
+            + " (time, dev, s1) VALUES(1, 'dev01', 1), (2, 'dev02', 2), (3, 'dev03', 3)");
   }
 
   @Test
@@ -93,7 +105,11 @@ public void testHAWithOneDataNodeIsDown() throws Exception {
       final DataNodeWrapper victimDataNode = EnvFactory.getEnv().getDataNodeWrapper(2);
       try (final Connection connection =
               EnvFactory.getEnv()
-                  .getConnection(liveDataNode, "root", "root", BaseEnv.TABLE_SQL_DIALECT);
+                  .getConnection(
+                      liveDataNode,
+                      SessionConfig.DEFAULT_USER,
+                      SessionConfig.DEFAULT_PASSWORD,
+                      BaseEnv.TABLE_SQL_DIALECT);
           final Statement statement = connection.createStatement()) {
         preTableData(statement, databaseName, tableName);
 
@@ -103,6 +119,7 @@ public void testHAWithOneDataNodeIsDown() throws Exception {
         Assert.assertFalse("victim DataNode should be stopped", victimDataNode.isAlive());
 
         executeSpecificSql(databaseName, tableName, statement, createdAfterDownTableName);
+        testDropDeviceForTreeModel(statement);
       }
     } finally {
       cleanCluster();
@@ -119,14 +136,22 @@ public void testHAWithOneDataNodeIsReadOnly() throws Exception {
       // Prepare data first so the table exists on all DataNodes.
       try (final Connection connection =
               EnvFactory.getEnv()
-                  .getConnection(liveDataNode, "root", "root", BaseEnv.TABLE_SQL_DIALECT);
+                  .getConnection(
+                      liveDataNode,
+                      SessionConfig.DEFAULT_USER,
+                      SessionConfig.DEFAULT_PASSWORD,
+                      BaseEnv.TABLE_SQL_DIALECT);
           final Statement statement = connection.createStatement()) {
         preTableData(statement, databaseName, tableName);
 
         // try to set the DN to readOnly status
         try (final Connection victimConn =
                 EnvFactory.getEnv()
-                    .getConnection(victimDataNode, "root", "root", BaseEnv.TABLE_SQL_DIALECT);
+                    .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");
@@ -217,6 +242,28 @@ public void executeSpecificSql(
         "DROP DATABASE must succeed with one DataNode down");
   }
 
+  private void testDropDeviceForTreeModel(Statement statement) throws Exception {
+    final String treeDevicePath = "root.db" + ".dev01";
+    LOGGER.info("8. start to test high availability of deleting tree model device procedure");
+    statement.execute("SET SQL_DIALECT=tree");
+    assertStatementEffect(
+        statement,
+        "INSERT INTO " + treeDevicePath + "(s1, s2, s3) VALUES(1, 2, 3)",
+        () ->
+            timeSeriesExists(statement, treeDevicePath + ".s1")
+                && timeSeriesExists(statement, treeDevicePath + ".s2")
+                && timeSeriesExists(statement, treeDevicePath + ".s3"),
+        "Tree model device data must be writable with one DataNode down");
+    assertStatementEffect(
+        statement,
+        "DELETE TIMESERIES " + treeDevicePath + ".**",
+        () ->
+            !timeSeriesExists(statement, treeDevicePath + ".s1")
+                && !timeSeriesExists(statement, treeDevicePath + ".s2")
+                && !timeSeriesExists(statement, treeDevicePath + ".s3"),
+        "DELETE TIMESERIES must delete the tree model device with one DataNode down");
+  }
+
   private void assertStatementEffect(
       final Statement statement,
       final String sql,
@@ -254,6 +301,12 @@ private boolean columnHasType(
     return false;
   }
 
+  private boolean timeSeriesExists(final Statement statement, final String path) throws Exception {
+    try (final ResultSet resultSet = statement.executeQuery("SHOW TIMESERIES " + path)) {
+      return resultSet.next();
+    }
+  }
+
   private boolean tableHasTtl(final Statement statement, final String tableName, final String ttl)
       throws Exception {
     try (final ResultSet resultSet = statement.executeQuery("SHOW TABLES")) {
diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBTableMetaLeaseIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBTableMetaLeaseIT.java
new file mode 100644
index 000000000000..8fb421490f4d
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBTableMetaLeaseIT.java
@@ -0,0 +1,83 @@
+/*
+ * 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.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.TableLocalStandaloneIT;
+
+import org.awaitility.Awaitility;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertTrue;
+
+@Category({TableLocalStandaloneIT.class})
+@RunWith(IoTDBTestRunner.class)
+public class IoTDBTableMetaLeaseIT {
+
+  private static final long CUSTOM_FENCE_MS = 20101L;
+  private static final String EXPECTED_LOG =
+      "Updated metadata lease fence threshold to " + CUSTOM_FENCE_MS + " ms";
+
+  private static DataNodeWrapper dataNodeWrapper;
+
+  @BeforeClass
+  public static void setUp() throws Exception {
+    EnvFactory.getEnv().getConfig().getConfigNodeConfig().setMetadataLeaseFenceMs(CUSTOM_FENCE_MS);
+    EnvFactory.getEnv().initClusterEnvironment();
+    dataNodeWrapper = EnvFactory.getEnv().getDataNodeWrapper(0);
+  }
+
+  @AfterClass
+  public static void tearDown() {
+    EnvFactory.getEnv().cleanClusterEnvironment();
+  }
+
+  @Test
+  public void testDnStartupLogsFenceThreshold() throws IOException {
+    assertTrue(
+        "DN startup log should contain updated fence threshold",
+        dataNodeWrapper.logContains(EXPECTED_LOG));
+  }
+
+  @Test
+  public void testDnRestartLogsFenceThreshold() throws Exception {
+    EnvFactory.getEnv().shutdownDataNode(0);
+    dataNodeWrapper.clearLogContent();
+    EnvFactory.getEnv().startDataNode(0);
+    Awaitility.await()
+        .atMost(10, TimeUnit.SECONDS)
+        .pollInterval(1, TimeUnit.SECONDS)
+        .untilAsserted(
+            () ->
+                assertTrue(
+                    "DN restart log should contain updated fence threshold",
+                    dataNodeWrapper.logContains(EXPECTED_LOG)));
+    ;
+  }
+}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConfig.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConfig.java
index d79ad04b4d68..f49525ea7da7 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConfig.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConfig.java
@@ -351,6 +351,13 @@ public class ConfigNodeConfig {
 
   private long forceWalPeriodForConfigNodeSimpleInMs = 100;
 
+  // The DataNode self-fences its ConfigNode-pushed metadata caches (table/tree schema, template,
+  // TTL, permission, ...) if it has not received a ConfigNode heartbeat within this duration. Kept
+  // aligned with the failure detector threshold so a partitioned DataNode stops trusting stale
+  // caches around the same time the cluster would consider it dead. Also used by the ConfigNode to
+  // derive how long it must wait before treating an unreachable DataNode as safely fenced.
+  private long metadataLeaseFenceMs = 20_000;
+
   public ConfigNodeConfig() {
     // empty constructor
   }
@@ -1261,6 +1268,14 @@ public void setForceWalPeriodForConfigNodeSimpleInMs(long forceWalPeriodForConfi
     this.forceWalPeriodForConfigNodeSimpleInMs = forceWalPeriodForConfigNodeSimpleInMs;
   }
 
+  public long getMetadataLeaseFenceMs() {
+    return metadataLeaseFenceMs;
+  }
+
+  public void setMetadataLeaseFenceMs(long metadataLeaseFenceMs) {
+    this.metadataLeaseFenceMs = metadataLeaseFenceMs;
+  }
+
   public String getConfigMessage() {
     StringBuilder configMessage = new StringBuilder();
     String configContent;
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java
index ef56f9c6125f..16dfd4150f1d 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java
@@ -312,6 +312,11 @@ private void loadProperties(TrimProperties properties) throws BadNodeUrlExceptio
                 "failure_detector_phi_acceptable_pause_in_ms",
                 String.valueOf(conf.getFailureDetectorPhiAcceptablePauseInMs()))));
 
+    conf.setMetadataLeaseFenceMs(
+        Long.parseLong(
+            properties.getProperty(
+                "metadata_lease_fence_ms", String.valueOf(conf.getMetadataLeaseFenceMs()))));
+
     conf.setEnableTopologyProbing(
         Boolean.parseBoolean(
             properties.getProperty(
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 2e23ac679e90..d0cae2aeedf9 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
@@ -21,7 +21,7 @@
 
 import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
 import org.apache.iotdb.common.rpc.thrift.TSStatus;
-import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
 import org.apache.iotdb.confignode.manager.lease.MetadataBroadcastVerdict.DataNodeState;
 import org.apache.iotdb.confignode.manager.lease.MetadataBroadcastVerdict.Verdict;
 import org.apache.iotdb.rpc.TSStatusCode;
@@ -76,7 +76,7 @@ public ClusterCachePropagator(final Map<Integer, TDataNodeLocation> registeredDa
         registeredDataNodes,
         nodeId -> DataNodeContactTracker.getInstance().getMillisSinceLastSuccessfulResponse(nodeId),
         () ->
-            CommonDescriptor.getInstance().getConfig().getMetadataLeaseFenceMs()
+            ConfigNodeDescriptor.getInstance().getConf().getMetadataLeaseFenceMs()
                 + DEFAULT_PROCEED_MARGIN_MS,
         System::nanoTime,
         Thread::sleep);
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/lease/DataNodeContactTracker.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/lease/DataNodeContactTracker.java
index fc01c9f98f66..ef068f59e6cd 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/lease/DataNodeContactTracker.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/lease/DataNodeContactTracker.java
@@ -19,7 +19,7 @@
 
 package org.apache.iotdb.confignode.manager.lease;
 
-import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
 
 import java.util.Collection;
 import java.util.Map;
@@ -50,6 +50,8 @@ public class DataNodeContactTracker {
 
   private final Map<Integer, Long> lastSuccessfulResponseNanos = new ConcurrentHashMap<>();
 
+  private final Map<Integer, Long> lastSentFenceThresholdMs = new ConcurrentHashMap<>();
+
   private DataNodeContactTracker() {
     this(System::nanoTime);
   }
@@ -61,6 +63,7 @@ private DataNodeContactTracker() {
   /** Record that a successful heartbeat response from the DataNode was just received. */
   public void recordSuccessfulResponse(final int dataNodeId) {
     lastSuccessfulResponseNanos.put(dataNodeId, nanoClock.getAsLong());
+    lastSentFenceThresholdMs.put(dataNodeId, getCurrentFenceThresholdMs());
   }
 
   /**
@@ -79,7 +82,7 @@ public long getMillisSinceLastSuccessfulResponse(final int dataNodeId) {
 
   public boolean isDataNodeFenced(final int dataNodeId) {
     return getMillisSinceLastSuccessfulResponse(dataNodeId)
-        > (CommonDescriptor.getInstance().getConfig().getMetadataLeaseFenceMs()
+        > (ConfigNodeDescriptor.getInstance().getConf().getMetadataLeaseFenceMs()
             + DEFAULT_PROCEED_MARGIN_MS);
   }
 
@@ -92,10 +95,26 @@ public void onLeadershipAcquired(final Collection<Integer> registeredDataNodeIds
     for (final Integer dataNodeId : registeredDataNodeIds) {
       lastSuccessfulResponseNanos.put(dataNodeId, now);
     }
+    lastSentFenceThresholdMs.clear();
   }
 
   public void removeDataNode(final int dataNodeId) {
     lastSuccessfulResponseNanos.remove(dataNodeId);
+    lastSentFenceThresholdMs.remove(dataNodeId);
+  }
+
+  /** The current fence threshold value from the ConfigNode's own configuration. */
+  public long getCurrentFenceThresholdMs() {
+    return ConfigNodeDescriptor.getInstance().getConf().getMetadataLeaseFenceMs();
+  }
+
+  /**
+   * Returns true if this DataNode has already received the current (confirmed by a successful
+   * heartbeat response sent after that value was loaded).
+   */
+  public boolean hasDeliveredCurrentFenceThreshold(final int dataNodeId) {
+    final Long lastSent = lastSentFenceThresholdMs.get(dataNodeId);
+    return lastSent != null && lastSent == getCurrentFenceThresholdMs();
   }
 
   public static DataNodeContactTracker getInstance() {
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/HeartbeatService.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/HeartbeatService.java
index 239a68801e20..ef732145f22d 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/HeartbeatService.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/HeartbeatService.java
@@ -38,6 +38,7 @@
 import org.apache.iotdb.confignode.i18n.ManagerMessages;
 import org.apache.iotdb.confignode.manager.IManager;
 import org.apache.iotdb.confignode.manager.consensus.ConsensusManager;
+import org.apache.iotdb.confignode.manager.lease.DataNodeContactTracker;
 import org.apache.iotdb.confignode.manager.load.cache.LoadCache;
 import org.apache.iotdb.confignode.manager.load.cache.node.ConfigNodeHeartbeatCache;
 import org.apache.iotdb.confignode.manager.node.NodeManager;
@@ -266,6 +267,7 @@ protected ConfigNodeHeartbeatHandler getConfigNodeHeartbeatHandler(int configNod
   private void pingRegisteredDataNodes(
       TDataNodeHeartbeatReq heartbeatReq, List<TDataNodeConfiguration> registeredDataNodes) {
     // Send heartbeat requests
+    final DataNodeContactTracker contactTracker = DataNodeContactTracker.getInstance();
     for (TDataNodeConfiguration dataNodeInfo : registeredDataNodes) {
       int dataNodeId = dataNodeInfo.getLocation().getDataNodeId();
       if (loadCache.checkAndSetHeartbeatProcessing(dataNodeId)) {
@@ -284,6 +286,11 @@ private void pingRegisteredDataNodes(
               configManager.getPipeManager().getPipeRuntimeCoordinator());
       configManager.getClusterQuotaManager().updateSpaceQuotaUsage();
       addConfigNodeLocationsToReq(dataNodeId, heartbeatReq);
+      if (contactTracker.hasDeliveredCurrentFenceThreshold(dataNodeId)) {
+        heartbeatReq.setFenceThresholdMsIsSet(false);
+      } else {
+        heartbeatReq.setFenceThresholdMs(contactTracker.getCurrentFenceThresholdMs());
+      }
       AsyncDataNodeHeartbeatClientPool.getInstance()
           .getDataNodeHeartBeat(
               dataNodeInfo.getLocation().getInternalEndPoint(), heartbeatReq, handler);
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
index 13e476c0037c..293f1942ff5a 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
@@ -303,6 +303,8 @@ private TRuntimeConfiguration getRuntimeConfiguration(int dataNodeId) {
           runtimeConfiguration.setClusterId(getClusterManager().getClusterId());
           runtimeConfiguration.setAuditConfig(getAuditConfig());
           runtimeConfiguration.setSuperUserName(getPermissionManager().getUserName(0));
+          runtimeConfiguration.setFenceThresholdMs(
+              ConfigNodeDescriptor.getInstance().getConf().getMetadataLeaseFenceMs());
           return runtimeConfiguration;
         } catch (AuthException e) {
           // This will never reach, definitely
diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java
index 855d42bd5a7a..f6971840f5ae 100644
--- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java
+++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java
@@ -574,6 +574,8 @@ public final class DataNodeSchemaMessages {
       "commit delete table {}.{} successfully.";
   public static final String FAILED_TO_REFRESH_CACHE_FROM_CN =
       "Failed to refresh DataNodeTableCache from ConfigNode";
+  public static final String FAILED_TO_GET_FENCE_THRESHOLD_FROM_CN =
+      "Failed to get fence threshold from ConfigNode during lease recovery";
   public static final String INTERRUPTED_ACQUIRE_SEMAPHORE_GET_TABLES =
       "Interrupted when trying to acquire semaphore when trying to get tables from configNode, ignore.";
   public static final String UPDATE_TABLE_BY_FETCH_WITH_DETAIL =
@@ -604,6 +606,8 @@ public final class DataNodeSchemaMessages {
   public static final String FAILED_TO_PULL_OR_INIT_METADATA = "Failed to pull or init metadata.";
   public static final String METADATA_LEASE_IS_FENCED =
       "Metadata lease is fenced. The local metadata cache is unavailable.";
+  public static final String UPDATED_METADATA_LEASE_FENCE_THRESHOLD =
+      "Updated metadata lease fence threshold to {} ms";
 
   // ======================== ClusterTemplateManager ========================
 
diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java
index ba99a6e8cd33..9a2ab87eab6d 100644
--- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java
+++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java
@@ -569,6 +569,8 @@ public final class DataNodeSchemaMessages {
   public static final String COMMIT_DELETE_TABLE_SUCCESS = "提交删除表 {}.{} 操作成功。";
   public static final String FAILED_TO_REFRESH_CACHE_FROM_CN =
       "从configNode拉取元数据更新DataNodeTableCache失败";
+  public static final String FAILED_TO_GET_FENCE_THRESHOLD_FROM_CN =
+      "在租约恢复期间未能从 ConfigNode 获取隔离阈值";
   public static final String INTERRUPTED_ACQUIRE_SEMAPHORE_GET_TABLES =
       "尝试获取信号量以从 ConfigNode 获取表时被中断，已忽略。";
   public static final String UPDATE_TABLE_BY_FETCH_WITH_DETAIL = "获取表 {}.{} 信息, {}";
@@ -599,6 +601,8 @@ public final class DataNodeSchemaMessages {
       "无法将元数据状态 {} 标记为 PULLING，因为已有其他元数据拉取线程正在运行。";
   public static final String FAILED_TO_PULL_OR_INIT_METADATA = "拉取元数据失败。";
   public static final String METADATA_LEASE_IS_FENCED = "元数据租约已过期, 本地缓存不可用";
+  public static final String UPDATED_METADATA_LEASE_FENCE_THRESHOLD =
+      "元数据租约隔离阈值已更新为 {} 毫秒";
 
   private DataNodeSchemaMessages() {}
   // ---------------------------------------------------------------------------
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
index afea35affd8f..ebf9845aeaec 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
@@ -2292,6 +2292,12 @@ private PathPatternTree filterPathPatternTree(PathPatternTree patternTree, Strin
   public TDataNodeHeartbeatResp getDataNodeHeartBeat(TDataNodeHeartbeatReq req) throws TException {
     TDataNodeHeartbeatResp resp = new TDataNodeHeartbeatResp();
 
+    // Update the fence threshold if the ConfigNode has sent a new value,
+    // then renew the metadata lease based on the latest threshold.
+    if (req.isSetFenceThresholdMs()) {
+      MetadataLeaseManager.getInstance().updateFenceThresholdMs(req.getFenceThresholdMs());
+    }
+
     // Renew the metadata lease: receiving a ConfigNode heartbeat means this DataNode is still in
     // contact with the cluster and may keep trusting its ConfigNode-pushed metadata caches.
     MetadataLeaseManager.getInstance().triggerCheckWithHeartBeat();
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 a1e793775833..9f00901da568 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
@@ -21,7 +21,6 @@
 
 import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
 import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
-import org.apache.iotdb.commons.conf.CommonDescriptor;
 import org.apache.iotdb.commons.exception.MetadataLeaseFencedException;
 import org.apache.iotdb.commons.utils.TestOnly;
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
@@ -74,7 +73,7 @@ interface MetadataAction {
   private final List<MetadataAction> pullMetaList;
 
   private final LongSupplier nanoClock;
-  private final LongSupplier fenceThresholdMs;
+  private volatile long fenceThresholdMs = 20000;
 
   private volatile long lastConfigNodeHeartbeatNanos;
 
@@ -95,7 +94,6 @@ private enum MetadataState {
   private MetadataLeaseManager() {
     this(
         System::nanoTime,
-        () -> CommonDescriptor.getInstance().getConfig().getMetadataLeaseFenceMs(),
         defaultClearCacheList(),
         defaultPullMetaList(),
         IoTDBThreadPoolFactory.newCachedThreadPool(RELOAD_TABLE_METADATA_CACHE.getName()),
@@ -117,14 +115,12 @@ private static List<MetadataAction> defaultPullMetaList() {
 
   MetadataLeaseManager(
       final LongSupplier nanoClock,
-      final LongSupplier fenceThresholdMs,
       final List<MetadataAction> clearCacheList,
       final List<MetadataAction> pullMetaList,
       final ExecutorService pullExecutorService,
       final long checkDnLeaseStatusIntervalMs,
       final ScheduledExecutorService checkLeaseStatusExecutor) {
     this.nanoClock = nanoClock;
-    this.fenceThresholdMs = fenceThresholdMs;
     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.
@@ -251,7 +247,7 @@ private static void rethrowUnchecked(final Throwable t) {
   }
 
   private boolean hasOutOfLease() {
-    return getMillisSinceLastConfigNodeHeartbeat() > fenceThresholdMs.getAsLong();
+    return getMillisSinceLastConfigNodeHeartbeat() > fenceThresholdMs;
   }
 
   /** Milliseconds elapsed since the last ConfigNode heartbeat was received (never negative). */
@@ -288,6 +284,14 @@ public void failIfMetadataLeaseFenced() {
     }
   }
 
+  /**
+   * Update the fence threshold from the ConfigNode. Called during startup (from
+   * TRuntimeConfiguration), heartbeat, and lease recovery.
+   */
+  public void updateFenceThresholdMs(final long newValue) {
+    this.fenceThresholdMs = newValue;
+  }
+
   /** Force the lease to appear expired, for tests that exercise fail-closed behavior. */
   @TestOnly
   public void recoveryLeaseForTest(boolean recovery) {
@@ -296,7 +300,7 @@ public void recoveryLeaseForTest(boolean recovery) {
       metadataStateRef.set(MetadataState.NORMAL, 0);
     } else {
       this.lastConfigNodeHeartbeatNanos =
-          nanoClock.getAsLong() - (fenceThresholdMs.getAsLong() + 1_000L) * 1_000_000L;
+          nanoClock.getAsLong() - (fenceThresholdMs + 1_000L) * 1_000_000L;
     }
   }
 
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 0489adb7646a..1328aa9e8731 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
@@ -80,6 +80,7 @@
 import org.apache.iotdb.db.consensus.DataRegionConsensusImpl;
 import org.apache.iotdb.db.consensus.SchemaRegionConsensusImpl;
 import org.apache.iotdb.db.i18n.DataNodeMiscMessages;
+import org.apache.iotdb.db.i18n.DataNodeSchemaMessages;
 import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
 import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
 import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager;
@@ -99,6 +100,7 @@
 import org.apache.iotdb.db.queryengine.plan.planner.distribution.SourceRewriter;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.LogicalQueryPlan;
 import org.apache.iotdb.db.schemaengine.SchemaEngine;
+import org.apache.iotdb.db.schemaengine.lease.MetadataLeaseManager;
 import org.apache.iotdb.db.schemaengine.schemaregion.attribute.update.GeneralRegionAttributeSecurityService;
 import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache;
 import org.apache.iotdb.db.schemaengine.template.ClusterTemplateManager;
@@ -531,6 +533,13 @@ protected void storeRuntimeConfigurations(
 
     /* Store superuser name */
     AuthorityChecker.setSuperUser(runtimeConfiguration.getSuperUserName());
+
+    /* Store metadata lease fence threshold from ConfigNode */
+    MetadataLeaseManager.getInstance()
+        .updateFenceThresholdMs(runtimeConfiguration.getFenceThresholdMs());
+    logger.info(
+        DataNodeSchemaMessages.UPDATED_METADATA_LEASE_FENCE_THRESHOLD,
+        runtimeConfiguration.getFenceThresholdMs());
   }
 
   /**
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/lease/MetadataLeaseTestUtils.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/lease/MetadataLeaseTestUtils.java
index 6631c0500a7e..e59873bf36a8 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/lease/MetadataLeaseTestUtils.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/lease/MetadataLeaseTestUtils.java
@@ -65,13 +65,15 @@ static MetadataLeaseManager newManager(
       final LongSupplier nowNanos,
       final MetadataLeaseManager.MetadataAction clearAction,
       final MetadataLeaseManager.MetadataAction pullAction) {
-    return new MetadataLeaseManager(
-        nowNanos,
-        () -> T_FENCE_MS,
-        Collections.singletonList(clearAction),
-        Collections.singletonList(pullAction),
-        MoreExecutors.newDirectExecutorService(),
-        500L,
-        null);
+    final MetadataLeaseManager manager =
+        new MetadataLeaseManager(
+            nowNanos,
+            Collections.singletonList(clearAction),
+            Collections.singletonList(pullAction),
+            MoreExecutors.newDirectExecutorService(),
+            500L,
+            null);
+    manager.updateFenceThresholdMs(T_FENCE_MS);
+    return manager;
   }
 }
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/AbstractCompactionTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/AbstractCompactionTest.java
index 89e7a6407076..951585605df8 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/AbstractCompactionTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/AbstractCompactionTest.java
@@ -19,7 +19,6 @@
 
 package org.apache.iotdb.db.storageengine.dataregion.compaction;
 
-import org.apache.iotdb.commons.conf.CommonDescriptor;
 import org.apache.iotdb.commons.conf.IoTDBConstant;
 import org.apache.iotdb.commons.exception.IllegalPathException;
 import org.apache.iotdb.commons.exception.MetadataException;
@@ -156,9 +155,6 @@ public class AbstractCompactionTest {
   private final long oldLongestExpiredTime =
       IoTDBDescriptor.getInstance().getConfig().getMaxExpiredTime();
 
-  private final long oldMetadataLeaseFenceMs =
-      CommonDescriptor.getInstance().getConfig().getMetadataLeaseFenceMs();
-
   protected static File STORAGE_GROUP_DIR =
       new File(
           TestConstant.BASE_OUTPUT_PATH
@@ -205,7 +201,7 @@ public class AbstractCompactionTest {
 
   public void setUp()
       throws IOException, WriteProcessException, MetadataException, InterruptedException {
-    CommonDescriptor.getInstance().getConfig().setMetadataLeaseFenceMs(Long.MAX_VALUE);
+    MetadataLeaseManager.getInstance().updateFenceThresholdMs(Long.MAX_VALUE);
     MetadataLeaseManager.getInstance().recoveryLeaseForTest(true);
     fileCount = 0;
     if (!SEQ_DIRS.exists()) {
@@ -502,7 +498,7 @@ public void tearDown() throws IOException, StorageEngineException {
         .getConfig()
         .setInnerCompactionTaskSelectionModsFileThreshold(oldModsFileSize);
     IoTDBDescriptor.getInstance().getConfig().setMaxExpiredTime(oldLongestExpiredTime);
-    CommonDescriptor.getInstance().getConfig().setMetadataLeaseFenceMs(oldMetadataLeaseFenceMs);
+    MetadataLeaseManager.getInstance().updateFenceThresholdMs(20_000);
     MetadataLeaseManager.getInstance().recoveryLeaseForTest(true);
     TSFileDescriptor.getInstance().getConfig().setGroupSizeInByte(oldChunkGroupSize);
 
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/EnvironmentUtils.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/EnvironmentUtils.java
index 83e0106e2aeb..873c83d86387 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/EnvironmentUtils.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/EnvironmentUtils.java
@@ -30,6 +30,7 @@
 import org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext;
 import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext;
 import org.apache.iotdb.db.schemaengine.SchemaEngine;
+import org.apache.iotdb.db.schemaengine.lease.MetadataLeaseManager;
 import org.apache.iotdb.db.storageengine.StorageEngine;
 import org.apache.iotdb.db.storageengine.buffer.BloomFilterCache;
 import org.apache.iotdb.db.storageengine.buffer.ChunkCache;
@@ -353,7 +354,7 @@ public FileVisitResult postVisitDirectory(Path dir, IOException exc) throws IOEx
   public static void envSetUp() {
     logger.debug("EnvironmentUtil setup...");
     config.setThriftServerAwaitTimeForStopService(60);
-    CommonDescriptor.getInstance().getConfig().setMetadataLeaseFenceMs(Long.MAX_VALUE);
+    MetadataLeaseManager.getInstance().updateFenceThresholdMs(Long.MAX_VALUE);
     createAllDir();
 
     try {
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
index a52112283009..91393bf9fd2b 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
@@ -479,13 +479,6 @@ public class CommonConfig {
 
   private volatile long remoteWriteMaxRetryDurationInMs = 60000;
 
-  // The DataNode self-fences its ConfigNode-pushed metadata caches (table/tree schema, template,
-  // TTL, permission, ...) if it has not received a ConfigNode heartbeat within this duration. Kept
-  // aligned with the failure detector threshold so a partitioned DataNode stops trusting stale
-  // caches around the same time the cluster would consider it dead. Also used by the ConfigNode to
-  // derive how long it must wait before treating an unreachable DataNode as safely fenced.
-  private volatile long metadataLeaseFenceMs = 20_000;
-
   private final RateLimiter querySamplingRateLimiter = RateLimiter.create(160);
   // if querySamplingRateLimiter < 0, means that there is no rate limit, we need to full sample all
   // the queries
@@ -2994,14 +2987,6 @@ public void setRemoteWriteMaxRetryDurationInMs(long remoteWriteMaxRetryDurationI
     this.remoteWriteMaxRetryDurationInMs = remoteWriteMaxRetryDurationInMs;
   }
 
-  public long getMetadataLeaseFenceMs() {
-    return metadataLeaseFenceMs;
-  }
-
-  public void setMetadataLeaseFenceMs(long metadataLeaseFenceMs) {
-    this.metadataLeaseFenceMs = metadataLeaseFenceMs;
-  }
-
   public int getArenaNum() {
     return arenaNum;
   }
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java
index 9933997622a4..e326c42c239f 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java
@@ -341,11 +341,6 @@ public void loadCommonProps(TrimProperties properties) throws IOException {
             properties.getProperty(
                 "path_log_max_size", String.valueOf(config.getPathLogMaxSize()))));
 
-    config.setMetadataLeaseFenceMs(
-        Long.parseLong(
-            properties.getProperty(
-                "metadata_lease_fence_ms", String.valueOf(config.getMetadataLeaseFenceMs()))));
-
     loadRetryProperties(properties);
     loadBinaryAllocatorProps(properties);
   }
diff --git a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
index 4b0b995c1677..7aa75731545e 100644
--- a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
+++ b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
@@ -129,6 +129,7 @@ struct TRuntimeConfiguration {
   10: optional bool enableSeparationOfAdminPowers
   // use 'optional' here to support rolling upgrade
   11: optional list<common.TExternalServiceEntry> allUserDefinedServiceInfo
+  12: optional i64 fenceThresholdMs
 }
 
 struct TDataNodeRegisterReq {
diff --git a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
index 054b1628c6af..cc5e0b7dc251 100644
--- a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
+++ b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
@@ -289,6 +289,7 @@ struct TDataNodeHeartbeatReq {
   // Using 8 bit to represent 8 bool
   // lowest bit: enable separation of admin powers
   16: optional byte booleanVariables1
+  17: optional i64 fenceThresholdMs
 }
 
 struct TDataNodeActivation {