modify the read logic of metadata lease timeout

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;
+  }
 }

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;
   }
 

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;
+  }
 }

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);
 }

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")) {

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)));
+    ;
+  }
+}

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;

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(

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);

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() {