From 54fa9a1b369faaea1c0b93ad9e1820f4531266b3 Mon Sep 17 00:00:00 2001 From: Yaobin Chen Date: Thu, 16 Jul 2026 18:10:56 +0800 Subject: [PATCH 1/4] modify the read logic of metadata lease timeout --- .../cluster/config/MppConfigNodeConfig.java | 6 ++ .../cluster/config/MppSharedCommonConfig.java | 1 - .../remote/config/RemoteConfigNodeConfig.java | 5 ++ .../iotdb/itbase/env/ConfigNodeConfig.java | 2 + .../it/schema/IoTDBTableDDLHAIT.java | 63 ++++++++++++++-- .../it/schema/IoTDBTableMetaLeaseIT.java | 75 +++++++++++++++++++ .../confignode/conf/ConfigNodeConfig.java | 15 ++++ .../confignode/conf/ConfigNodeDescriptor.java | 5 ++ .../confignode/manager/ConfigManager.java | 4 +- .../manager/lease/ClusterCachePropagator.java | 4 +- .../manager/lease/DataNodeContactTracker.java | 23 +++++- .../load/service/HeartbeatService.java | 7 ++ .../confignode/manager/node/NodeManager.java | 2 + .../iotdb/db/i18n/DataNodeSchemaMessages.java | 4 + .../iotdb/db/i18n/DataNodeSchemaMessages.java | 4 + .../impl/DataNodeInternalRPCServiceImpl.java | 6 ++ .../lease/MetadataLeaseManager.java | 19 +++-- .../table/DataNodeTableCache.java | 7 ++ .../org/apache/iotdb/db/service/DataNode.java | 9 +++ .../lease/MetadataLeaseTestUtils.java | 2 +- .../compaction/AbstractCompactionTest.java | 8 +- .../iotdb/db/utils/EnvironmentUtils.java | 3 +- .../iotdb/commons/conf/CommonConfig.java | 15 ---- .../iotdb/commons/conf/CommonDescriptor.java | 5 -- .../src/main/thrift/confignode.thrift | 2 + .../src/main/thrift/datanode.thrift | 1 + 26 files changed, 252 insertions(+), 45 deletions(-) create mode 100644 integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBTableMetaLeaseIT.java 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 4385097ce045b..8da0f4e02b249 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 48544b901d4a5..36d06aefe889d 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 c16722f4bd943..36fbdcbfc0e86 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 4af35f6f56a9d..aec8f0f233543 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 396a918d173bc..68f6c11811214 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 0000000000000..4f532fc92b21d --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBTableMetaLeaseIT.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.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.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 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); + 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 d79ad04b4d682..f49525ea7da7b 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 ef56f9c6125f2..16dfd4150f1d0 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/ConfigManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java index fee78f5568a04..9b1b75a7cd5ea 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,9 @@ public TDataNodeLeaseRecoveryResp reloadCacheAfterLeaseRecovery() { } return new TDataNodeLeaseRecoveryResp() .setStatus(RpcUtils.SUCCESS_STATUS) - .setTableInfo(clusterSchemaManager.getAllTableInfoForDataNodeActivation()); + .setTableInfo(clusterSchemaManager.getAllTableInfoForDataNodeActivation()) + .setFenceThresholdMs( + ConfigNodeDescriptor.getInstance().getConf().getMetadataLeaseFenceMs()); } @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 2e23ac679e900..d0cae2aeedf98 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 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 fc01c9f98f664..ef068f59e6cd0 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 lastSuccessfulResponseNanos = new ConcurrentHashMap<>(); + private final Map 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 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 239a68801e209..ef732145f22d0 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 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 13e476c0037c0..293f1942ff5a9 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 855d42bd5a7a6..f6971840f5ae5 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 ba99a6e8cd33e..9a2ab87eab6d9 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 afea35affd8f9..ebf9845aeaec0 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 a1e7937758332..623af9a795088 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 pullMetaList; private final LongSupplier nanoClock; - private final LongSupplier fenceThresholdMs; + private volatile long fenceThresholdMs; private volatile long lastConfigNodeHeartbeatNanos; @@ -95,7 +94,7 @@ private enum MetadataState { private MetadataLeaseManager() { this( System::nanoTime, - () -> CommonDescriptor.getInstance().getConfig().getMetadataLeaseFenceMs(), + 0, defaultClearCacheList(), defaultPullMetaList(), IoTDBThreadPoolFactory.newCachedThreadPool(RELOAD_TABLE_METADATA_CACHE.getName()), @@ -117,7 +116,7 @@ private static List defaultPullMetaList() { MetadataLeaseManager( final LongSupplier nanoClock, - final LongSupplier fenceThresholdMs, + final long fenceThresholdMs, final List clearCacheList, final List pullMetaList, final ExecutorService pullExecutorService, @@ -251,7 +250,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 +287,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 +303,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/schemaengine/table/DataNodeTableCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/table/DataNodeTableCache.java index 5b90104de838f..9c1020287b729 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 @@ -158,6 +158,13 @@ public void reloadTableCacheAfterLeaseRecovery() { if (resp.isSetTableInfo()) { init(resp.getTableInfo()); } + if (!resp.isSetFenceThresholdMs()) { + throw new RuntimeException(DataNodeSchemaMessages.FAILED_TO_GET_FENCE_THRESHOLD_FROM_CN); + } + MetadataLeaseManager.getInstance().updateFenceThresholdMs(resp.getFenceThresholdMs()); + LOGGER.info( + DataNodeSchemaMessages.UPDATED_METADATA_LEASE_FENCE_THRESHOLD, + resp.getFenceThresholdMs()); } catch (final ClientManagerException | TException e) { throw new RuntimeException(FAILED_TO_REFRESH_CACHE_FROM_CN, e); } 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 0489adb7646a5..1328aa9e87315 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 6631c0500a7eb..d5adca86662b3 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 @@ -67,7 +67,7 @@ static MetadataLeaseManager newManager( final MetadataLeaseManager.MetadataAction pullAction) { return new MetadataLeaseManager( nowNanos, - () -> T_FENCE_MS, + T_FENCE_MS, Collections.singletonList(clearAction), Collections.singletonList(pullAction), MoreExecutors.newDirectExecutorService(), 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 89e7a64070765..951585605df82 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 83e0106e2aeb6..873c83d86387f 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 a52112283009c..91393bf9fd2b4 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 9933997622a4f..e326c42c239f3 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 4b0b995c16778..81673de1e60e3 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 allUserDefinedServiceInfo + 12: required i64 fenceThresholdMs } struct TDataNodeRegisterReq { @@ -162,6 +163,7 @@ struct TDataNodeRestartResp { struct TDataNodeLeaseRecoveryResp{ 1: required common.TSStatus status 2: optional binary tableInfo + 3: optional i64 fenceThresholdMs } struct TDataNodeRemoveReq { diff --git a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift index 054b1628c6af0..cc5e0b7dc2511 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 { From 3b8e1ac39bc58df5d4b237f7e2cbd58d5c2c9e0d Mon Sep 17 00:00:00 2001 From: Yaobin Chen Date: Thu, 16 Jul 2026 18:49:54 +0800 Subject: [PATCH 2/4] fix the read logic --- .../org/apache/iotdb/confignode/manager/ConfigManager.java | 4 +--- .../iotdb/db/schemaengine/table/DataNodeTableCache.java | 7 ------- .../thrift-confignode/src/main/thrift/confignode.thrift | 3 +-- 3 files changed, 2 insertions(+), 12 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 9b1b75a7cd5ea..fee78f5568a04 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,9 +3308,7 @@ public TDataNodeLeaseRecoveryResp reloadCacheAfterLeaseRecovery() { } return new TDataNodeLeaseRecoveryResp() .setStatus(RpcUtils.SUCCESS_STATUS) - .setTableInfo(clusterSchemaManager.getAllTableInfoForDataNodeActivation()) - .setFenceThresholdMs( - ConfigNodeDescriptor.getInstance().getConf().getMetadataLeaseFenceMs()); + .setTableInfo(clusterSchemaManager.getAllTableInfoForDataNodeActivation()); } @Override 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 9c1020287b729..5b90104de838f 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 @@ -158,13 +158,6 @@ public void reloadTableCacheAfterLeaseRecovery() { if (resp.isSetTableInfo()) { init(resp.getTableInfo()); } - if (!resp.isSetFenceThresholdMs()) { - throw new RuntimeException(DataNodeSchemaMessages.FAILED_TO_GET_FENCE_THRESHOLD_FROM_CN); - } - MetadataLeaseManager.getInstance().updateFenceThresholdMs(resp.getFenceThresholdMs()); - LOGGER.info( - DataNodeSchemaMessages.UPDATED_METADATA_LEASE_FENCE_THRESHOLD, - resp.getFenceThresholdMs()); } catch (final ClientManagerException | TException e) { throw new RuntimeException(FAILED_TO_REFRESH_CACHE_FROM_CN, e); } diff --git a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift index 81673de1e60e3..7aa75731545e5 100644 --- a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift +++ b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift @@ -129,7 +129,7 @@ struct TRuntimeConfiguration { 10: optional bool enableSeparationOfAdminPowers // use 'optional' here to support rolling upgrade 11: optional list allUserDefinedServiceInfo - 12: required i64 fenceThresholdMs + 12: optional i64 fenceThresholdMs } struct TDataNodeRegisterReq { @@ -163,7 +163,6 @@ struct TDataNodeRestartResp { struct TDataNodeLeaseRecoveryResp{ 1: required common.TSStatus status 2: optional binary tableInfo - 3: optional i64 fenceThresholdMs } struct TDataNodeRemoveReq { From 882503688aa8293fe6006b6641d0f46878a2525a Mon Sep 17 00:00:00 2001 From: Yaobin Chen Date: Thu, 16 Jul 2026 19:04:32 +0800 Subject: [PATCH 3/4] fix --- .../lease/MetadataLeaseManager.java | 5 +---- .../lease/MetadataLeaseTestUtils.java | 18 ++++++++++-------- 2 files changed, 11 insertions(+), 12 deletions(-) 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 623af9a795088..9f00901da5688 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 @@ -73,7 +73,7 @@ interface MetadataAction { private final List pullMetaList; private final LongSupplier nanoClock; - private volatile long fenceThresholdMs; + private volatile long fenceThresholdMs = 20000; private volatile long lastConfigNodeHeartbeatNanos; @@ -94,7 +94,6 @@ private enum MetadataState { private MetadataLeaseManager() { this( System::nanoTime, - 0, defaultClearCacheList(), defaultPullMetaList(), IoTDBThreadPoolFactory.newCachedThreadPool(RELOAD_TABLE_METADATA_CACHE.getName()), @@ -116,14 +115,12 @@ private static List defaultPullMetaList() { MetadataLeaseManager( final LongSupplier nanoClock, - final long fenceThresholdMs, final List clearCacheList, final List 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. 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 d5adca86662b3..e59873bf36a87 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; } } From 53f6dceb449c5c9db907e1513974e11e44662c52 Mon Sep 17 00:00:00 2001 From: Yaobin Chen Date: Fri, 17 Jul 2026 09:35:34 +0800 Subject: [PATCH 4/4] fix IT --- .../it/schema/IoTDBTableMetaLeaseIT.java | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) 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 index 4f532fc92b21d..8fb421490f4d4 100644 --- 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 @@ -24,6 +24,7 @@ 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; @@ -31,6 +32,7 @@ import org.junit.runner.RunWith; import java.io.IOException; +import java.util.concurrent.TimeUnit; import static org.junit.Assert.assertTrue; @@ -68,8 +70,14 @@ public void testDnRestartLogsFenceThreshold() throws Exception { EnvFactory.getEnv().shutdownDataNode(0); dataNodeWrapper.clearLogContent(); EnvFactory.getEnv().startDataNode(0); - assertTrue( - "DN restart log should contain updated fence threshold", - dataNodeWrapper.logContains(EXPECTED_LOG)); + Awaitility.await() + .atMost(10, TimeUnit.SECONDS) + .pollInterval(1, TimeUnit.SECONDS) + .untilAsserted( + () -> + assertTrue( + "DN restart log should contain updated fence threshold", + dataNodeWrapper.logContains(EXPECTED_LOG))); + ; } }