You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ja...@apache.org on 2026/08/24 02:07:00 UTC
(iotdb) branch rc/2.0.11 created (now b78205c05a6)
This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a change to branch rc/2.0.11
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at b78205c05a6 Fix max data region group quota calculation (#18263)
This branch includes the following new commits:
new ddf12bd5602 Fix duplicate TsFile metric updates (#18219)
new 219f28104ca fix: optimize reconnect exception as `TTransport.TTransportException` (#18200)
new d9f499f6b31 Fix concurrent access during TierManager folder reset (#18223)
new cdd3ba2f88f modify the read logic of metadata lease timeout (#18228)
new 9d3807a05fa Enable CRC32 intrinsic for DataNode (#18239)
new 1fd1616db55 Compatibility changes for internal settle compaction API (#18230)
new b78205c05a6 Fix max data region group quota calculation (#18263)
The 7 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iotdb) 04/07: modify the read logic of metadata lease timeout (#18228)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch rc/2.0.11
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit cdd3ba2f88fc20149df624abb6c175d2fb7ef250
Author: CYB <al...@gmail.com>
AuthorDate: Fri Jul 17 10:48:37 2026 +0800
modify the read logic of metadata lease timeout (#18228)
---
.../it/env/cluster/config/MppConfigNodeConfig.java | 6 ++
.../env/cluster/config/MppSharedCommonConfig.java | 1 -
.../env/remote/config/RemoteConfigNodeConfig.java | 5 ++
.../apache/iotdb/itbase/env/ConfigNodeConfig.java | 2 +
.../relational/it/schema/IoTDBTableDDLHAIT.java | 63 ++++++++++++++--
.../it/schema/IoTDBTableMetaLeaseIT.java | 83 ++++++++++++++++++++++
.../iotdb/confignode/conf/ConfigNodeConfig.java | 15 ++++
.../confignode/conf/ConfigNodeDescriptor.java | 5 ++
.../manager/lease/ClusterCachePropagator.java | 4 +-
.../manager/lease/DataNodeContactTracker.java | 23 +++++-
.../manager/load/service/HeartbeatService.java | 7 ++
.../iotdb/confignode/manager/node/NodeManager.java | 2 +
.../iotdb/db/i18n/DataNodeSchemaMessages.java | 4 ++
.../iotdb/db/i18n/DataNodeSchemaMessages.java | 4 ++
.../impl/DataNodeInternalRPCServiceImpl.java | 6 ++
.../schemaengine/lease/MetadataLeaseManager.java | 18 +++--
.../java/org/apache/iotdb/db/service/DataNode.java | 9 +++
.../schemaengine/lease/MetadataLeaseTestUtils.java | 18 ++---
.../compaction/AbstractCompactionTest.java | 8 +--
.../apache/iotdb/db/utils/EnvironmentUtils.java | 3 +-
.../apache/iotdb/commons/conf/CommonConfig.java | 15 ----
.../iotdb/commons/conf/CommonDescriptor.java | 5 --
.../src/main/thrift/confignode.thrift | 1 +
.../src/main/thrift/datanode.thrift | 1 +
24 files changed, 256 insertions(+), 52 deletions(-)
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 4385097ce04..8da0f4e02b2 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 class MppConfigNodeConfig extends MppBaseConfig implements ConfigNodeConf
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 48544b901d4..36d06aefe88 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 class MppSharedCommonConfig implements CommonConfig {
@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 c16722f4bd9..36fbdcbfc0e 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 class RemoteConfigNodeConfig implements ConfigNodeConfig {
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 4af35f6f56a..aec8f0f2335 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 396a918d173..68f6c118112 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 @@ package org.apache.iotdb.relational.it.schema;
import org.apache.iotdb.commons.cluster.NodeStatus;
import org.apache.iotdb.consensus.ConsensusFactory;
+import org.apache.iotdb.isession.SessionConfig;
import org.apache.iotdb.it.env.EnvFactory;
import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
import org.apache.iotdb.it.framework.IoTDBTestRunner;
@@ -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 @@ public class IoTDBTableDDLHAIT {
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 class IoTDBTableDDLHAIT {
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 class IoTDBTableDDLHAIT {
Assert.assertFalse("victim DataNode should be stopped", victimDataNode.isAlive());
executeSpecificSql(databaseName, tableName, statement, createdAfterDownTableName);
+ testDropDeviceForTreeModel(statement);
}
} finally {
cleanCluster();
@@ -119,14 +136,22 @@ public class IoTDBTableDDLHAIT {
// 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 class IoTDBTableDDLHAIT {
"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 @@ public class IoTDBTableDDLHAIT {
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 00000000000..8fb421490f4
--- /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 d79ad04b4d6..f49525ea7da 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 class ConfigNodeConfig {
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 ef56f9c6125..16dfd4150f1 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 @@ public class ConfigNodeDescriptor {
"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 2e23ac679e9..d0cae2aeedf 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 @@ package org.apache.iotdb.confignode.manager.lease;
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 class ClusterCachePropagator {
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 fc01c9f98f6..ef068f59e6c 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 @@ public class 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 class DataNodeContactTracker {
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 class DataNodeContactTracker {
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 239a68801e2..ef732145f22 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.conf.ConfigNodeDescriptor;
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 @@ public class HeartbeatService {
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 @@ public class HeartbeatService {
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 13e476c0037..293f1942ff5 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 @@ public class NodeManager {
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 855d42bd5a7..f6971840f5a 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 ba99a6e8cd3..9a2ab87eab6 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 afea35affd8..ebf9845aeae 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 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
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 a1e79377583..9f00901da56 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 @@ package org.apache.iotdb.db.schemaengine.lease;
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 @@ public class MetadataLeaseManager {
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 @@ public class MetadataLeaseManager {
private MetadataLeaseManager() {
this(
System::nanoTime,
- () -> CommonDescriptor.getInstance().getConfig().getMetadataLeaseFenceMs(),
defaultClearCacheList(),
defaultPullMetaList(),
IoTDBThreadPoolFactory.newCachedThreadPool(RELOAD_TABLE_METADATA_CACHE.getName()),
@@ -117,14 +115,12 @@ public class MetadataLeaseManager {
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 @@ public class MetadataLeaseManager {
}
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 class MetadataLeaseManager {
}
}
+ /**
+ * 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 class MetadataLeaseManager {
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 0489adb7646..1328aa9e873 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.conf.IoTDBStartCheck;
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.DistributionPla
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 @@ public class DataNode extends ServerCommandLine implements DataNodeMBean {
/* 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 6631c0500a7..e59873bf36a 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 @@ public final class MetadataLeaseTestUtils {
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 89e7a640707..951585605df 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 class AbstractCompactionTest {
.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 83e0106e2ae..873c83d8638 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.exception.StorageEngineException;
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 class EnvironmentUtils {
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 c811ae9f5f9..d83d0b5d9a8 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
@@ -477,13 +477,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
@@ -2964,14 +2957,6 @@ public class CommonConfig {
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 9933997622a..e326c42c239 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 class CommonDescriptor {
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 4b0b995c167..7aa75731545 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 054b1628c6a..cc5e0b7dc25 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 {
(iotdb) 01/07: Fix duplicate TsFile metric updates (#18219)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch rc/2.0.11
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit ddf12bd5602fc6ebb9f7dab9f1206cc56a10e7bf
Author: Jiang Tian <jt...@163.com>
AuthorDate: Thu Jul 16 09:28:56 2026 +0800
Fix duplicate TsFile metric updates (#18219)
* refactor addFile
* record whether a file is recorded by metric
* remove redundant arg
---
.../iotdb/db/service/metrics/FileMetrics.java | 8 +--
.../db/service/metrics/file/TsFileMetrics.java | 19 +++++--
.../db/storageengine/dataregion/DataRegion.java | 61 ++++----------------
.../execute/task/CrossSpaceCompactionTask.java | 4 +-
.../execute/task/InnerSpaceCompactionTask.java | 5 +-
.../task/InsertionCrossSpaceCompactionTask.java | 12 +---
.../execute/task/SettleCompactionTask.java | 3 +-
.../compaction/execute/utils/CompactionUtils.java | 17 ++----
.../dataregion/tsfile/TsFileResource.java | 17 ++++++
.../cross/InsertionCrossSpaceCompactionTest.java | 16 +-----
.../utils/CompactionUpdateFileCountTest.java | 66 ++++++++++++++++++++--
.../compaction/utils/CompactionUtilsTest.java | 2 +-
12 files changed, 122 insertions(+), 108 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/FileMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/FileMetrics.java
index 67cc75be749..a714adfe315 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/FileMetrics.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/FileMetrics.java
@@ -64,12 +64,12 @@ public class FileMetrics implements IMetricSet {
}
// region TsFile Related Metrics Update
- public void addTsFile(String database, String regionId, long size, boolean seq, String name) {
- TS_FILE_METRICS.addTsFile(database, regionId, size, seq, name);
+ public void addTsFile(TsFileResource tsFileResource) {
+ TS_FILE_METRICS.addTsFile(tsFileResource);
}
- public void deleteTsFile(boolean seq, List<TsFileResource> tsFileResourceList) {
- TS_FILE_METRICS.deleteFile(seq, tsFileResourceList);
+ public void deleteTsFile(List<TsFileResource> tsFileResourceList) {
+ TS_FILE_METRICS.deleteFile(tsFileResourceList);
}
public void deleteRegion(String database, String regionId) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/file/TsFileMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/file/TsFileMetrics.java
index 2d78e2f6baf..de0c0fd23d6 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/file/TsFileMetrics.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/file/TsFileMetrics.java
@@ -92,10 +92,17 @@ public class TsFileMetrics implements IMetricSet {
}
// region external update tsfile related metrics
- public void addTsFile(String database, String regionId, long size, boolean seq, String name) {
- updateGlobalTsFileCountAndSize(database, regionId, 1, size, seq);
+ public void addTsFile(TsFileResource tsFileResource) {
+ if (!tsFileResource.markAsRecordedByMetric()) {
+ return;
+ }
+ long size = tsFileResource.getTsFileSize();
+ boolean seq = tsFileResource.isSeq();
+ updateGlobalTsFileCountAndSize(
+ tsFileResource.getDatabaseName(), tsFileResource.getDataRegionId(), 1, size, seq);
try {
- TsFileNameGenerator.TsFileName tsFileName = TsFileNameGenerator.getTsFileName(name);
+ TsFileNameGenerator.TsFileName tsFileName =
+ TsFileNameGenerator.getTsFileName(tsFileResource.getTsFile().getName());
int level = tsFileName.getInnerCompactionCnt();
updateLevelTsFileCountAndSize(size, 1, seq, level);
} catch (IOException e) {
@@ -103,10 +110,14 @@ public class TsFileMetrics implements IMetricSet {
}
}
- public void deleteFile(boolean seq, List<TsFileResource> tsFileResourceList) {
+ public void deleteFile(List<TsFileResource> tsFileResourceList) {
for (TsFileResource tsFileResource : tsFileResourceList) {
+ if (!tsFileResource.markAsUnrecordedByMetric()) {
+ continue;
+ }
String name = tsFileResource.getTsFile().getName();
long size = tsFileResource.getTsFileSize();
+ boolean seq = tsFileResource.isSeq();
updateGlobalTsFileCountAndSize(
tsFileResource.getDatabaseName(), tsFileResource.getDataRegionId(), -1, -size, seq);
try {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
index 63a6269b190..7f29ab552df 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
@@ -628,13 +628,7 @@ public class DataRegion implements IDataRegionForQuery {
// tsFiles without resource file are unsealed
for (TsFileResource resource : value) {
if (resource.resourceFileExists()) {
- FileMetrics.getInstance()
- .addTsFile(
- resource.getDatabaseName(),
- resource.getDataRegionId(),
- resource.getTsFile().length(),
- true,
- resource.getTsFile().getName());
+ FileMetrics.getInstance().addTsFile(resource);
if (ModificationFile.getExclusiveMods(resource.getTsFile()).exists()) {
// update mods file metrics
resource.getExclusiveModFile();
@@ -662,13 +656,7 @@ public class DataRegion implements IDataRegionForQuery {
// tsFiles without resource file are unsealed
for (TsFileResource resource : unseqTsFiles) {
if (resource.resourceFileExists()) {
- FileMetrics.getInstance()
- .addTsFile(
- resource.getDatabaseName(),
- resource.getDataRegionId(),
- resource.getTsFile().length(),
- false,
- resource.getTsFile().getName());
+ FileMetrics.getInstance().addTsFile(resource);
} else {
WALRecoverListener recoverListener =
recoverUnsealedTsFile(resource, dataRegionRecoveryContext, false);
@@ -975,13 +963,7 @@ public class DataRegion implements IDataRegionForQuery {
}
updateDeviceLastFlushTime(tsFileResource);
tsFileResourceManager.registerSealedTsFileResource(tsFileResource);
- FileMetrics.getInstance()
- .addTsFile(
- tsFileResource.getDatabaseName(),
- tsFileResource.getDataRegionId(),
- tsFileResource.getTsFile().length(),
- recoverPerformer.isSequence(),
- tsFileResource.getTsFile().getName());
+ FileMetrics.getInstance().addTsFile(tsFileResource);
} else {
// the last file is not closed, continue writing to it
RestorableTsFileIOWriter writer = recoverPerformer.getWriter();
@@ -2302,7 +2284,7 @@ public class DataRegion implements IDataRegionForQuery {
tsFileResourceList.addAll(tsFileManager.getTsFileList(false));
tsFileResourceList.forEach(
x -> {
- FileMetrics.getInstance().deleteTsFile(x.isSeq(), Collections.singletonList(x));
+ FileMetrics.getInstance().deleteTsFile(Collections.singletonList(x));
try {
x.removeModFile();
} catch (IOException e) {
@@ -2578,13 +2560,7 @@ public class DataRegion implements IDataRegionForQuery {
closedTsFileResources.add(tsFileProcessor.getTsFileResource());
}
for (TsFileResource resource : closedTsFileResources) {
- FileMetrics.getInstance()
- .addTsFile(
- resource.getDatabaseName(),
- resource.getDataRegionId(),
- resource.getTsFileSize(),
- resource.isSeq(),
- resource.getTsFile().getName());
+ FileMetrics.getInstance().addTsFile(resource);
}
WritingMetrics.getInstance().recordActiveTimePartitionCount(-1);
logger.info(StorageEngineMessages.FILES_WERE_CLOSED, closedTsFileResources.size());
@@ -3708,8 +3684,7 @@ public class DataRegion implements IDataRegionForQuery {
tsFileManager.remove(tsFileResource, tsFileResource.isSeq());
tsFileResource.writeLock();
try {
- FileMetrics.getInstance()
- .deleteTsFile(tsFileResource.isSeq(), Collections.singletonList(tsFileResource));
+ FileMetrics.getInstance().deleteTsFile(Collections.singletonList(tsFileResource));
tsFileResource.remove();
logger.info(
StorageEngineMessages.REMOVE_TSFILE_DIRECTLY_WHEN_DELETE_DATA,
@@ -3752,8 +3727,7 @@ public class DataRegion implements IDataRegionForQuery {
tsFileManager.remove(tsFileResource, tsFileResource.isSeq());
tsFileResource.writeLock();
try {
- FileMetrics.getInstance()
- .deleteTsFile(tsFileResource.isSeq(), Collections.singletonList(tsFileResource));
+ FileMetrics.getInstance().deleteTsFile(Collections.singletonList(tsFileResource));
tsFileResource.remove();
logger.info(
StorageEngineMessages.REMOVE_TSFILE_DIRECTLY_WHEN_DELETE_DATA,
@@ -3855,13 +3829,7 @@ public class DataRegion implements IDataRegionForQuery {
}
if (!isValidateTsFileFailed) {
TsFileResource tsFileResource = tsFileProcessor.getTsFileResource();
- FileMetrics.getInstance()
- .addTsFile(
- tsFileResource.getDatabaseName(),
- tsFileResource.getDataRegionId(),
- tsFileResource.getTsFileSize(),
- tsFileProcessor.isSequence(),
- tsFileResource.getTsFile().getName());
+ FileMetrics.getInstance().addTsFile(tsFileResource);
}
}
@@ -4226,13 +4194,7 @@ public class DataRegion implements IDataRegionForQuery {
TableDiskUsageIndex.getInstance()
.write(databaseName, newTsFileResource.getTsFileID(), stringLongMap));
- FileMetrics.getInstance()
- .addTsFile(
- newTsFileResource.getDatabaseName(),
- newTsFileResource.getDataRegionId(),
- newTsFileResource.getTsFile().length(),
- false,
- newTsFileResource.getTsFile().getName());
+ FileMetrics.getInstance().addTsFile(newTsFileResource);
if (config.isEnableSeparateData()) {
final DataRegionId dataRegionId =
@@ -4645,8 +4607,7 @@ public class DataRegion implements IDataRegionForQuery {
if (sequenceResource.getTsFile().getName().equals(fileToBeUnloaded.getName())) {
unloadedTsFileResource = sequenceResource;
tsFileManager.remove(unloadedTsFileResource, true);
- FileMetrics.getInstance()
- .deleteTsFile(true, Collections.singletonList(unloadedTsFileResource));
+ FileMetrics.getInstance().deleteTsFile(Collections.singletonList(unloadedTsFileResource));
break;
}
}
@@ -4658,7 +4619,7 @@ public class DataRegion implements IDataRegionForQuery {
unloadedTsFileResource = unsequenceResource;
tsFileManager.remove(unloadedTsFileResource, false);
FileMetrics.getInstance()
- .deleteTsFile(false, Collections.singletonList(unloadedTsFileResource));
+ .deleteTsFile(Collections.singletonList(unloadedTsFileResource));
break;
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/CrossSpaceCompactionTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/CrossSpaceCompactionTask.java
index 1a6703e4b67..345bca19b31 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/CrossSpaceCompactionTask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/CrossSpaceCompactionTask.java
@@ -357,8 +357,8 @@ public class CrossSpaceCompactionTask extends AbstractCompactionTask {
throw new CompactionRecoverException(StorageEngineMessages.SOURCE_FILES_CANNOT_BE_DELETED);
}
if (recoverMemoryStatus) {
- FileMetrics.getInstance().deleteTsFile(true, selectedSequenceFiles);
- FileMetrics.getInstance().deleteTsFile(false, selectedUnsequenceFiles);
+ FileMetrics.getInstance().deleteTsFile(selectedSequenceFiles);
+ FileMetrics.getInstance().deleteTsFile(selectedUnsequenceFiles);
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/InnerSpaceCompactionTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/InnerSpaceCompactionTask.java
index acc2ff500a9..1b0fb3205be 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/InnerSpaceCompactionTask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/InnerSpaceCompactionTask.java
@@ -435,8 +435,7 @@ public class InnerSpaceCompactionTask extends AbstractCompactionTask {
isHoldingWriteLock[i] = true;
}
- CompactionUtils.deleteSourceTsFileAndUpdateFileMetrics(
- filesView.sourceFilesInLog, filesView.sequence);
+ CompactionUtils.deleteSourceTsFileAndUpdateFileMetrics(filesView.sourceFilesInLog);
updateTableSizeCache();
CompactionMetrics.getInstance().recordSummaryInfo(summary);
}
@@ -599,7 +598,7 @@ public class InnerSpaceCompactionTask extends AbstractCompactionTask {
throw new CompactionRecoverException(StorageEngineMessages.SOURCE_FILES_CANNOT_BE_DELETED);
}
if (recoverMemoryStatus) {
- FileMetrics.getInstance().deleteTsFile(filesView.sequence, filesView.sourceFilesInLog);
+ FileMetrics.getInstance().deleteTsFile(filesView.sourceFilesInLog);
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/InsertionCrossSpaceCompactionTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/InsertionCrossSpaceCompactionTask.java
index 6ae9791401a..60c1bdc106d 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/InsertionCrossSpaceCompactionTask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/InsertionCrossSpaceCompactionTask.java
@@ -371,18 +371,10 @@ public class InsertionCrossSpaceCompactionTask extends AbstractCompactionTask {
}
private void updateFileMetrics() {
- // Here the target file is used for updating metrics because the source file
- // has been deleted here.
// The statistics of the mods file can be left unchanged, as it does not
// differentiate between sequence or unsequence.
- FileMetrics.getInstance().deleteTsFile(false, Collections.singletonList(targetFile));
- FileMetrics.getInstance()
- .addTsFile(
- targetFile.getDatabaseName(),
- targetFile.getDataRegionId(),
- targetFile.getTsFileSize(),
- true,
- targetFile.getTsFile().getName());
+ FileMetrics.getInstance().deleteTsFile(Collections.singletonList(unseqFileToInsert));
+ FileMetrics.getInstance().addTsFile(targetFile);
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/SettleCompactionTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/SettleCompactionTask.java
index 388197e66e3..4de88103322 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/SettleCompactionTask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/task/SettleCompactionTask.java
@@ -226,8 +226,7 @@ public class SettleCompactionTask extends InnerSpaceCompactionTask {
.STORAGE_LOG_SETTLE_TASK_DELETES_FULLY_DIRTY_TSFILE_SUCCESSFULLY_18D81225,
resource.getTsFile().getAbsolutePath());
if (recoverMemoryStatus) {
- FileMetrics.getInstance()
- .deleteTsFile(resource.isSeq(), Collections.singletonList(resource));
+ FileMetrics.getInstance().deleteTsFile(Collections.singletonList(resource));
}
} else {
LOGGER.error(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionUtils.java
index 5798f8fd382..d26ecdb2129 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionUtils.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/CompactionUtils.java
@@ -268,13 +268,7 @@ public class CompactionUtils {
}
public static void addFilesToFileMetrics(TsFileResource resource) {
- FileMetrics.getInstance()
- .addTsFile(
- resource.getDatabaseName(),
- resource.getDataRegionId(),
- resource.getTsFile().length(),
- resource.isSeq(),
- resource.getTsFile().getName());
+ FileMetrics.getInstance().addTsFile(resource);
}
private static void updateOneTargetMods(TsFileResource targetFile, Set<ModEntry> modifications)
@@ -396,16 +390,15 @@ public class CompactionUtils {
public static void deleteSourceTsFileAndUpdateFileMetrics(
List<TsFileResource> sourceSeqResourceList, List<TsFileResource> sourceUnseqResourceList) {
- deleteSourceTsFileAndUpdateFileMetrics(sourceSeqResourceList, true);
- deleteSourceTsFileAndUpdateFileMetrics(sourceUnseqResourceList, false);
+ deleteSourceTsFileAndUpdateFileMetrics(sourceSeqResourceList);
+ deleteSourceTsFileAndUpdateFileMetrics(sourceUnseqResourceList);
}
- public static void deleteSourceTsFileAndUpdateFileMetrics(
- List<TsFileResource> resources, boolean seq) {
+ public static void deleteSourceTsFileAndUpdateFileMetrics(List<TsFileResource> resources) {
for (TsFileResource resource : resources) {
deleteTsFileResourceWithoutLock(resource);
}
- FileMetrics.getInstance().deleteTsFile(seq, resources);
+ FileMetrics.getInstance().deleteTsFile(resources);
}
public static void deleteTsFileResourceWithoutLock(TsFileResource resource) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java
index bffa3a0f1eb..a5ec8a555eb 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java
@@ -92,6 +92,7 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
@@ -159,6 +160,8 @@ public class TsFileResource implements PersistentResource, Cloneable {
private boolean isSeq;
+ private AtomicBoolean isRecordedByMetric = new AtomicBoolean(false);
+
private final FSFactory fsFactory = FSFactoryProducer.getFSFactory();
private DataRegion.SettleTsFileCallBack settleTsFileCallBack;
@@ -259,6 +262,7 @@ public class TsFileResource implements PersistentResource, Cloneable {
this.originTsFileResource = originTsFileResource;
this.tsFileID = originTsFileResource.tsFileID;
this.isSeq = originTsFileResource.isSeq;
+ this.isRecordedByMetric = originTsFileResource.isRecordedByMetric;
this.tierLevel = originTsFileResource.tierLevel;
}
@@ -1372,6 +1376,18 @@ public class TsFileResource implements PersistentResource, Cloneable {
return isSeq;
}
+ public boolean isRecordedByMetric() {
+ return isRecordedByMetric.get();
+ }
+
+ public boolean markAsRecordedByMetric() {
+ return isRecordedByMetric.compareAndSet(false, true);
+ }
+
+ public boolean markAsUnrecordedByMetric() {
+ return isRecordedByMetric.compareAndSet(true, false);
+ }
+
public int compareIndexDegradePriority(TsFileResource tsFileResource) {
int cmp = timeIndex.compareDegradePriority(tsFileResource.timeIndex);
return cmp == 0 ? file.getAbsolutePath().compareTo(tsFileResource.file.getAbsolutePath()) : cmp;
@@ -1665,6 +1681,7 @@ public class TsFileResource implements PersistentResource, Cloneable {
cloned.sharedModFileOffset = this.sharedModFileOffset;
cloned.compactionModFile = this.compactionModFile;
cloned.isSeq = this.isSeq;
+ cloned.isRecordedByMetric = this.isRecordedByMetric;
cloned.tsFileRepairStatus = this.tsFileRepairStatus;
cloned.settleTsFileCallBack = this.settleTsFileCallBack;
cloned.deviceTimeIndexRamSize = this.deviceTimeIndexRamSize;
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/InsertionCrossSpaceCompactionTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/InsertionCrossSpaceCompactionTest.java
index 0e0516a46c0..73c0bc2a721 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/InsertionCrossSpaceCompactionTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/cross/InsertionCrossSpaceCompactionTest.java
@@ -564,13 +564,7 @@ public class InsertionCrossSpaceCompactionTest extends AbstractCompactionTest {
TsFileResource unseqResource1 =
generateSingleNonAlignedSeriesFileWithDevices(
"2-2-0-0.tsfile", new String[] {"d1"}, new TimeRange[] {new TimeRange(1, 4)}, false);
- FileMetrics.getInstance()
- .addTsFile(
- unseqResource1.getDatabaseName(),
- unseqResource1.getDataRegionId(),
- unseqResource1.getTsFileSize(),
- false,
- unseqResource1.getTsFile().getName());
+ FileMetrics.getInstance().addTsFile(unseqResource1);
long seqFileNumBeforeCompaction = FileMetrics.getInstance().getFileCount(true);
long unseqFileNumBeforeCompaction = FileMetrics.getInstance().getFileCount(false);
@@ -588,13 +582,7 @@ public class InsertionCrossSpaceCompactionTest extends AbstractCompactionTest {
TsFileResource unseqResource2 =
generateSingleNonAlignedSeriesFileWithDevices(
"3-3-0-0.tsfile", new String[] {"d1"}, new TimeRange[] {new TimeRange(5, 6)}, false);
- FileMetrics.getInstance()
- .addTsFile(
- unseqResource2.getDatabaseName(),
- unseqResource2.getDataRegionId(),
- unseqResource2.getTsFileSize(),
- false,
- unseqResource2.getTsFile().getName());
+ FileMetrics.getInstance().addTsFile(unseqResource2);
seqFileNumBeforeCompaction = FileMetrics.getInstance().getFileCount(true);
unseqFileNumBeforeCompaction = FileMetrics.getInstance().getFileCount(false);
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionUpdateFileCountTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionUpdateFileCountTest.java
index 1740eb2f1c1..c79cd467603 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionUpdateFileCountTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionUpdateFileCountTest.java
@@ -27,6 +27,7 @@ import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.ReadChunkCompactionPerformer;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.CrossSpaceCompactionTask;
import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.task.InnerSpaceCompactionTask;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import org.apache.tsfile.exception.write.WriteProcessException;
import org.junit.After;
@@ -35,6 +36,9 @@ import org.junit.Before;
import org.junit.Test;
import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
public class CompactionUpdateFileCountTest extends AbstractCompactionTest {
@@ -53,10 +57,11 @@ public class CompactionUpdateFileCountTest extends AbstractCompactionTest {
public void testSeqSpaceCompactionFileMetric()
throws MetadataException, IOException, WriteProcessException {
registerTimeseriesInMManger(2, 3, false);
- long initSeqFileNum = FileMetrics.getInstance().getFileCount(true);
- long initUnSeqFileNum = FileMetrics.getInstance().getFileCount(false);
createFiles(1, 2, 3, 100, 1, 0, 50, 0, false, true);
createFiles(1, 2, 3, 50, 200, 30000, 50, 50, false, true);
+ seqResources.forEach(FileMetrics.getInstance()::addTsFile);
+ long initSeqFileNum = FileMetrics.getInstance().getFileCount(true);
+ long initUnSeqFileNum = FileMetrics.getInstance().getFileCount(false);
tsFileManager.addAll(seqResources, true);
InnerSpaceCompactionTask task =
new InnerSpaceCompactionTask(
@@ -70,10 +75,11 @@ public class CompactionUpdateFileCountTest extends AbstractCompactionTest {
public void testUnSeqSpaceCompactionFileMetric()
throws MetadataException, IOException, WriteProcessException {
registerTimeseriesInMManger(2, 3, false);
- long initSeqFileNum = FileMetrics.getInstance().getFileCount(true);
- long initUnSeqFileNum = FileMetrics.getInstance().getFileCount(false);
createFiles(1, 2, 3, 100, 1, 0, 50, 0, false, false);
createFiles(1, 2, 3, 50, 20, 10000, 50, 50, false, false);
+ unseqResources.forEach(FileMetrics.getInstance()::addTsFile);
+ long initSeqFileNum = FileMetrics.getInstance().getFileCount(true);
+ long initUnSeqFileNum = FileMetrics.getInstance().getFileCount(false);
tsFileManager.addAll(unseqResources, false);
InnerSpaceCompactionTask task =
new InnerSpaceCompactionTask(
@@ -87,10 +93,12 @@ public class CompactionUpdateFileCountTest extends AbstractCompactionTest {
public void testCrossSpaceCompactionFileMetric()
throws MetadataException, IOException, WriteProcessException {
registerTimeseriesInMManger(2, 3, false);
- long initSeqFileNum = FileMetrics.getInstance().getFileCount(true);
- long initUnSeqFileNum = FileMetrics.getInstance().getFileCount(false);
createFiles(1, 2, 3, 100, 1, 0, 50, 0, false, true);
createFiles(3, 2, 3, 50, 20, 10000, 50, 50, false, false);
+ seqResources.forEach(FileMetrics.getInstance()::addTsFile);
+ unseqResources.forEach(FileMetrics.getInstance()::addTsFile);
+ long initSeqFileNum = FileMetrics.getInstance().getFileCount(true);
+ long initUnSeqFileNum = FileMetrics.getInstance().getFileCount(false);
tsFileManager.addAll(seqResources, true);
tsFileManager.addAll(unseqResources, false);
CrossSpaceCompactionTask task =
@@ -106,4 +114,50 @@ public class CompactionUpdateFileCountTest extends AbstractCompactionTest {
Assert.assertEquals(initSeqFileNum, FileMetrics.getInstance().getFileCount(true));
Assert.assertEquals(initUnSeqFileNum - 3, FileMetrics.getInstance().getFileCount(false));
}
+
+ @Test
+ public void testRepeatedlyUpdateFileMetric()
+ throws MetadataException, IOException, WriteProcessException {
+ registerTimeseriesInMManger(2, 3, false);
+ createFiles(1, 2, 3, 100, 1, 0, 50, 0, false, true);
+ TsFileResource resource = seqResources.get(0);
+ long initSeqFileNum = FileMetrics.getInstance().getFileCount(true);
+
+ try {
+ FileMetrics.getInstance().addTsFile(resource);
+ FileMetrics.getInstance().addTsFile(resource);
+ Assert.assertTrue(resource.isRecordedByMetric());
+ Assert.assertEquals(initSeqFileNum + 1, FileMetrics.getInstance().getFileCount(true));
+
+ FileMetrics.getInstance().deleteTsFile(Collections.singletonList(resource));
+ FileMetrics.getInstance().deleteTsFile(Collections.singletonList(resource));
+ Assert.assertFalse(resource.isRecordedByMetric());
+ Assert.assertEquals(initSeqFileNum, FileMetrics.getInstance().getFileCount(true));
+ } finally {
+ FileMetrics.getInstance().deleteTsFile(Collections.singletonList(resource));
+ }
+ }
+
+ @Test
+ public void testDeleteFileMetricByResourceSequence()
+ throws MetadataException, IOException, WriteProcessException {
+ registerTimeseriesInMManger(2, 3, false);
+ createFiles(1, 2, 3, 100, 1, 0, 50, 0, false, true);
+ createFiles(1, 2, 3, 50, 200, 30000, 50, 50, false, false);
+ List<TsFileResource> resources = Arrays.asList(seqResources.get(0), unseqResources.get(0));
+ long initSeqFileNum = FileMetrics.getInstance().getFileCount(true);
+ long initUnSeqFileNum = FileMetrics.getInstance().getFileCount(false);
+
+ try {
+ resources.forEach(FileMetrics.getInstance()::addTsFile);
+ Assert.assertEquals(initSeqFileNum + 1, FileMetrics.getInstance().getFileCount(true));
+ Assert.assertEquals(initUnSeqFileNum + 1, FileMetrics.getInstance().getFileCount(false));
+
+ FileMetrics.getInstance().deleteTsFile(resources);
+ Assert.assertEquals(initSeqFileNum, FileMetrics.getInstance().getFileCount(true));
+ Assert.assertEquals(initUnSeqFileNum, FileMetrics.getInstance().getFileCount(false));
+ } finally {
+ FileMetrics.getInstance().deleteTsFile(resources);
+ }
+ }
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionUtilsTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionUtilsTest.java
index 2972c0ee929..97427b16321 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionUtilsTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/utils/CompactionUtilsTest.java
@@ -85,7 +85,7 @@ public class CompactionUtilsTest extends AbstractCompactionTest {
Assert.assertEquals(
modFileSizeBefore + totalModFileSize, FileMetrics.getInstance().getModFileSize());
- CompactionUtils.deleteSourceTsFileAndUpdateFileMetrics(new ArrayList<>(seqResources), true);
+ CompactionUtils.deleteSourceTsFileAndUpdateFileMetrics(new ArrayList<>(seqResources));
Assert.assertEquals(modFileNumBefore, FileMetrics.getInstance().getModFileNum());
Assert.assertEquals(modFileSizeBefore, FileMetrics.getInstance().getModFileSize());
(iotdb) 03/07: Fix concurrent access during TierManager folder reset (#18223)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch rc/2.0.11
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit d9f499f6b3122add46f76cac0a47e6612a9cddb6
Author: Jiang Tian <jt...@163.com>
AuthorDate: Thu Jul 16 17:01:58 2026 +0800
Fix concurrent access during TierManager folder reset (#18223)
* fix TierManager may access an empty list during re-initialization
* Fix concurrent access to tier directory maps
---
.../db/storageengine/rescon/disk/TierManager.java | 39 ++++--
.../storageengine/rescon/disk/TierManagerTest.java | 156 +++++++++++++++++++++
2 files changed, 183 insertions(+), 12 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/disk/TierManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/disk/TierManager.java
index 3e47ce442f7..87c5cf00273 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/disk/TierManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/rescon/disk/TierManager.java
@@ -62,21 +62,21 @@ public class TierManager {
/**
* seq folder manager of each storage tier, managing both data directories and multi-dir strategy
*/
- private final List<FolderManager> seqTiers = new ArrayList<>();
+ private volatile List<FolderManager> seqTiers = new ArrayList<>();
/**
* unSeq folder manager of each storage tier, managing both data directories and multi-dir
* strategy
*/
- private final List<FolderManager> unSeqTiers = new ArrayList<>();
+ private volatile List<FolderManager> unSeqTiers = new ArrayList<>();
- private final List<FolderManager> objectTiers = new ArrayList<>();
+ private volatile List<FolderManager> objectTiers = new ArrayList<>();
/** seq file folder's rawFsPath path -> tier level */
- private final Map<String, Integer> seqDir2TierLevel = new HashMap<>();
+ private volatile Map<String, Integer> seqDir2TierLevel = new HashMap<>();
/** unSeq file folder's rawFsPath path -> tier level */
- private final Map<String, Integer> unSeqDir2TierLevel = new HashMap<>();
+ private volatile Map<String, Integer> unSeqDir2TierLevel = new HashMap<>();
private List<String> objectDirs;
@@ -95,6 +95,15 @@ public class TierManager {
}
public synchronized void initFolders() {
+ initFolders(seqTiers, unSeqTiers, objectTiers, seqDir2TierLevel, unSeqDir2TierLevel);
+ }
+
+ private void initFolders(
+ List<FolderManager> seqTiers,
+ List<FolderManager> unSeqTiers,
+ List<FolderManager> objectTiers,
+ Map<String, Integer> seqDir2TierLevel,
+ Map<String, Integer> unSeqDir2TierLevel) {
directoryStrategyType =
DirectoryStrategyType.fromClassName(config.getMultiDirStrategyClassName());
@@ -204,13 +213,19 @@ public class TierManager {
public synchronized void resetFolders() {
long startTime = System.currentTimeMillis();
- seqTiers.clear();
- unSeqTiers.clear();
- objectTiers.clear();
- seqDir2TierLevel.clear();
- unSeqDir2TierLevel.clear();
-
- initFolders();
+ List<FolderManager> newSeqTiers = new ArrayList<>();
+ List<FolderManager> newUnSeqTiers = new ArrayList<>();
+ List<FolderManager> newObjectTiers = new ArrayList<>();
+ Map<String, Integer> newSeqDir2TierLevel = new HashMap<>();
+ Map<String, Integer> newUnSeqDir2TierLevel = new HashMap<>();
+
+ initFolders(
+ newSeqTiers, newUnSeqTiers, newObjectTiers, newSeqDir2TierLevel, newUnSeqDir2TierLevel);
+ seqTiers = newSeqTiers;
+ unSeqTiers = newUnSeqTiers;
+ objectTiers = newObjectTiers;
+ seqDir2TierLevel = newSeqDir2TierLevel;
+ unSeqDir2TierLevel = newUnSeqDir2TierLevel;
long endTime = System.currentTimeMillis();
logger.info(StorageEngineMessages.FOLDERS_RESET_SUCCESSFULLY, (endTime - startTime));
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/rescon/disk/TierManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/rescon/disk/TierManagerTest.java
new file mode 100644
index 00000000000..046e41dce48
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/rescon/disk/TierManagerTest.java
@@ -0,0 +1,156 @@
+/*
+ * 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.db.storageengine.rescon.disk;
+
+import org.apache.iotdb.commons.conf.IoTDBConstant;
+import org.apache.iotdb.db.conf.IoTDBConfig;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+public class TierManagerTest {
+
+ private static final int DATA_DIR_NUM = 16;
+ private static final int RESET_TIMES = 20;
+
+ @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ private final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
+ private String[][] originalTierDataDirs;
+ private TierManager tierManager;
+ private File sequenceFile;
+
+ @Before
+ public void setUp() throws Exception {
+ originalTierDataDirs = config.getTierDataDirs();
+ String[] dataDirs = new String[DATA_DIR_NUM];
+ for (int i = 0; i < DATA_DIR_NUM; i++) {
+ dataDirs[i] = new File(temporaryFolder.getRoot(), "data" + i).getAbsolutePath();
+ }
+ config.setTierDataDirs(new String[][] {dataDirs});
+ tierManager = TierManager.getInstance();
+ tierManager.resetFolders();
+ sequenceFile =
+ new File(dataDirs[0], IoTDBConstant.SEQUENCE_FOLDER_NAME + File.separator + "test.tsfile");
+ assertTrue(sequenceFile.createNewFile());
+ }
+
+ @After
+ public void tearDown() {
+ config.setTierDataDirs(originalTierDataDirs);
+ tierManager.resetFolders();
+ }
+
+ @Test
+ public void testConcurrentResetFoldersAndGetNextFolder() throws Exception {
+ List<Callable<?>> folderAccessors =
+ Arrays.asList(
+ () -> tierManager.getNextFolderForTsFile(0, true),
+ () -> tierManager.getNextFolderForTsFile(0, false),
+ tierManager::getNextFolderForObjectFile,
+ tierManager::getNextFolderForCopyToTargetFile,
+ () -> tierManager.getFolderManager(0, true),
+ () -> tierManager.getFolderManager(0, false),
+ () -> {
+ int tiersNum = tierManager.getTiersNum();
+ assertTrue(tiersNum > 0);
+ return tiersNum;
+ },
+ () -> assertFoldersAvailable(tierManager.getAllFilesFolders()),
+ () -> assertFoldersAvailable(tierManager.getAllLocalFilesFolders()),
+ () -> assertFoldersAvailable(tierManager.getAllSequenceFileFolders()),
+ () -> assertFoldersAvailable(tierManager.getAllLocalSequenceFileFolders()),
+ () -> assertFoldersAvailable(tierManager.getAllUnSequenceFileFolders()),
+ () -> assertFoldersAvailable(tierManager.getAllLocalUnSequenceFileFolders()),
+ () -> tierManager.getFileTierLevel(sequenceFile));
+ ExecutorService executorService = Executors.newFixedThreadPool(folderAccessors.size() + 1);
+ CountDownLatch startLatch = new CountDownLatch(1);
+ AtomicBoolean resetting = new AtomicBoolean(true);
+
+ Future<?> resetFuture =
+ executorService.submit(
+ () -> {
+ startLatch.await();
+ try {
+ for (int i = 0; i < RESET_TIMES; i++) {
+ tierManager.resetFolders();
+ }
+ } finally {
+ resetting.set(false);
+ }
+ return null;
+ });
+ List<AtomicInteger> accessorCounts = new ArrayList<>();
+ List<Future<?>> accessorFutures = new ArrayList<>();
+ for (Callable<?> folderAccessor : folderAccessors) {
+ AtomicInteger accessorCount = new AtomicInteger();
+ accessorCounts.add(accessorCount);
+ accessorFutures.add(
+ executorService.submit(
+ () -> {
+ startLatch.await();
+ do {
+ assertNotNull(folderAccessor.call());
+ accessorCount.incrementAndGet();
+ } while (resetting.get());
+ return null;
+ }));
+ }
+
+ startLatch.countDown();
+ try {
+ resetFuture.get(30, TimeUnit.SECONDS);
+ for (Future<?> accessorFuture : accessorFutures) {
+ accessorFuture.get(30, TimeUnit.SECONDS);
+ }
+ for (AtomicInteger accessorCount : accessorCounts) {
+ assertTrue(accessorCount.get() > 0);
+ }
+ } finally {
+ executorService.shutdownNow();
+ assertTrue(executorService.awaitTermination(10, TimeUnit.SECONDS));
+ }
+ }
+
+ private List<String> assertFoldersAvailable(List<String> folders) {
+ assertTrue(folders.size() >= DATA_DIR_NUM);
+ return folders;
+ }
+}
(iotdb) 07/07: Fix max data region group quota calculation (#18263)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch rc/2.0.11
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit b78205c05a661dc0bca28cade8a3330647a45d16
Author: libo <ze...@163.com>
AuthorDate: Tue Jul 21 16:07:59 2026 +0800
Fix max data region group quota calculation (#18263)
---
.../IoTDBDatabaseAutoDataRegionGroupQuotaIT.java | 91 ++++++++++++++++++++++
.../manager/schema/ClusterSchemaManager.java | 4 +-
2 files changed, 94 insertions(+), 1 deletion(-)
diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDatabaseAutoDataRegionGroupQuotaIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDatabaseAutoDataRegionGroupQuotaIT.java
new file mode 100644
index 00000000000..4d2fd8c780c
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDatabaseAutoDataRegionGroupQuotaIT.java
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.relational.it.schema;
+
+import org.apache.iotdb.consensus.ConsensusFactory;
+import org.apache.iotdb.db.it.utils.TestUtils;
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.TableClusterIT;
+import org.apache.iotdb.itbase.env.BaseEnv;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Collections;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({TableClusterIT.class})
+public class IoTDBDatabaseAutoDataRegionGroupQuotaIT {
+
+ private static final int CONFIG_NODE_NUM = 1;
+ private static final int DATA_NODE_NUM = 3;
+ private static final int SCHEMA_REPLICATION_FACTOR = 3;
+ private static final int DATA_REPLICATION_FACTOR = 1;
+ private static final int DATA_REGION_PER_DATA_NODE = 2;
+
+ @Before
+ public void setUp() throws Exception {
+ EnvFactory.getEnv()
+ .getConfig()
+ .getCommonConfig()
+ .setSchemaRegionGroupExtensionPolicy("CUSTOM")
+ .setDataRegionGroupExtensionPolicy("AUTO")
+ .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setDefaultSchemaRegionGroupNumPerDatabase(1)
+ .setDefaultDataRegionGroupNumPerDatabase(1)
+ .setSchemaReplicationFactor(SCHEMA_REPLICATION_FACTOR)
+ .setDataReplicationFactor(DATA_REPLICATION_FACTOR)
+ .setDataRegionPerDataNode(DATA_REGION_PER_DATA_NODE);
+ EnvFactory.getEnv().initClusterEnvironment(CONFIG_NODE_NUM, DATA_NODE_NUM);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ EnvFactory.getEnv().cleanClusterEnvironment();
+ }
+
+ @Test
+ public void testMaxDataRegionGroupNumUsesDataReplicationFactor() throws SQLException {
+ try (final Connection connection =
+ EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+ final Statement statement = connection.createStatement()) {
+ final int expectedMaxDataRegionGroupNum =
+ (int)
+ Math.ceil(
+ (double) DATA_REGION_PER_DATA_NODE * DATA_NODE_NUM / DATA_REPLICATION_FACTOR);
+
+ statement.execute("create database test_data_region_rf with(max_schema_region_group_num=2)");
+
+ TestUtils.assertResultSetEqual(
+ statement.executeQuery(
+ "select database, max_schema_region_group_num, max_data_region_group_num "
+ + "from information_schema.databases where database = 'test_data_region_rf'"),
+ "database,max_schema_region_group_num,max_data_region_group_num,",
+ Collections.singleton("test_data_region_rf,2," + expectedMaxDataRegionGroupNum + ","));
+ }
+ }
+}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java
index a66b88f0a10..357c14eca4a 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/schema/ClusterSchemaManager.java
@@ -603,7 +603,9 @@ public class ClusterSchemaManager {
? dataNodeNum
: (CONF.getDataRegionPerDataNode() == 0 ? totalCpuCoreNum : dataNodeNum),
databaseNum,
- databaseSchema.getSchemaReplicationFactor(),
+ (consensusGroupType == TConsensusGroupType.SchemaRegion)
+ ? databaseSchema.getSchemaReplicationFactor()
+ : databaseSchema.getDataReplicationFactor(),
allocatedRegionGroupCount);
LOGGER.info(
(consensusGroupType == TConsensusGroupType.SchemaRegion)
(iotdb) 02/07: fix: optimize reconnect exception as `TTransport.TTransportException` (#18200)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch rc/2.0.11
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 219f28104cad6b49db570d4039f0ccdf96c3b2d0
Author: Zeyu Zhang <11...@users.noreply.github.com>
AuthorDate: Thu Jul 16 13:59:12 2026 +0800
fix: optimize reconnect exception as `TTransport.TTransportException` (#18200)
---
iotdb-client/client-py/iotdb/Session.py | 7 +-
.../tests/unit/test_session_connection.py | 116 +++++++++++++++++++++
2 files changed, 121 insertions(+), 2 deletions(-)
diff --git a/iotdb-client/client-py/iotdb/Session.py b/iotdb-client/client-py/iotdb/Session.py
index 0a5cb43fd54..32cdccbe5b9 100644
--- a/iotdb-client/client-py/iotdb/Session.py
+++ b/iotdb-client/client-py/iotdb/Session.py
@@ -179,7 +179,7 @@ class Session(object):
self.__default_connection = self.init_connection(
self.__default_endpoint
)
- except Exception as e:
+ except IoTDBConnectionException as e:
if not self.reconnect():
if str(e).startswith("Could not connect to any of"):
error_msg = (
@@ -240,9 +240,12 @@ class Session(object):
session_id = open_resp.sessionId
statement_id = client.requestStatementId(session_id)
- except Exception as e:
+ except TTransport.TTransportException as e:
transport.close()
raise IoTDBConnectionException(e) from None
+ except Exception:
+ transport.close()
+ raise
if self.__zone_id is not None:
request = TSSetTimeZoneReq(session_id, self.__zone_id)
diff --git a/iotdb-client/client-py/tests/unit/test_session_connection.py b/iotdb-client/client-py/tests/unit/test_session_connection.py
new file mode 100644
index 00000000000..7a78c697fea
--- /dev/null
+++ b/iotdb-client/client-py/tests/unit/test_session_connection.py
@@ -0,0 +1,116 @@
+# 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.
+#
+
+import importlib
+from types import SimpleNamespace
+
+import pytest
+from thrift.transport import TTransport
+
+from iotdb.thrift.common.ttypes import TSStatus
+from iotdb.utils.exception import StatementExecutionException
+
+
+class FakeSocket:
+ def setTimeout(self, timeout):
+ self.timeout = timeout
+
+
+class FakeTransport:
+ def __init__(self, socket):
+ self.socket = socket
+ self.opened = False
+
+ def isOpen(self):
+ return self.opened
+
+ def open(self):
+ self.opened = True
+
+ def close(self):
+ self.opened = False
+
+
+@pytest.mark.parametrize(
+ "status_code,status_message",
+ [
+ (801, "Authentication failed."),
+ (822, "Account is blocked due to consecutive failed logins."),
+ ],
+)
+def test_session_does_not_retry_iotdb_status_error(
+ monkeypatch, status_code, status_message
+):
+ session_module = importlib.import_module("iotdb.Session")
+
+ class RejectingClient:
+ open_session_calls = 0
+
+ def __init__(self, protocol):
+ pass
+
+ def openSession(self, request):
+ RejectingClient.open_session_calls += 1
+ return SimpleNamespace(
+ status=TSStatus(code=status_code, message=status_message)
+ )
+
+ monkeypatch.setattr(
+ session_module.TSocket, "TSocket", lambda host, port: FakeSocket()
+ )
+ monkeypatch.setattr(session_module.TTransport, "TFramedTransport", FakeTransport)
+ monkeypatch.setattr(session_module, "Client", RejectingClient)
+
+ session = session_module.Session.init_from_node_urls(
+ ["127.0.0.1:6667"], user="test", password="wrong"
+ )
+
+ with pytest.raises(StatementExecutionException) as exc_info:
+ session.open()
+
+ assert str(exc_info.value) == f"{status_code}: {status_message}"
+ assert RejectingClient.open_session_calls == 1
+
+
+def test_session_retries_transport_error(monkeypatch):
+ session_module = importlib.import_module("iotdb.Session")
+
+ class FailingClient:
+ open_session_calls = 0
+
+ def __init__(self, protocol):
+ pass
+
+ def openSession(self, request):
+ FailingClient.open_session_calls += 1
+ raise TTransport.TTransportException(message="Network is unavailable.")
+
+ monkeypatch.setattr(
+ session_module.TSocket, "TSocket", lambda host, port: FakeSocket()
+ )
+ monkeypatch.setattr(session_module.TTransport, "TFramedTransport", FakeTransport)
+ monkeypatch.setattr(session_module, "Client", FailingClient)
+
+ session = session_module.Session.init_from_node_urls(
+ ["127.0.0.1:6667"], user="test", password="wrong"
+ )
+
+ with pytest.raises(session_module.IoTDBConnectionException):
+ session.open()
+
+ assert FailingClient.open_session_calls == 1 + session.RETRY_NUM
(iotdb) 06/07: Compatibility changes for internal settle compaction API (#18230)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch rc/2.0.11
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 1fd1616db5549f82d5e98dd5980e0a95d9840657
Author: shuwenwei <55...@users.noreply.github.com>
AuthorDate: Tue Jul 21 10:29:32 2026 +0800
Compatibility changes for internal settle compaction API (#18230)
---
.../selector/impl/SettleSelectorImpl.java | 43 +++++++++++++++++-----
1 file changed, 34 insertions(+), 9 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/impl/SettleSelectorImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/impl/SettleSelectorImpl.java
index cef6084d1ae..b08c032d483 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/impl/SettleSelectorImpl.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/selector/impl/SettleSelectorImpl.java
@@ -71,6 +71,7 @@ public class SettleSelectorImpl implements ISettleSelector {
private final TsFileManager tsFileManager;
private boolean isSeq;
private final CompactionScheduleContext context;
+ private final boolean ttlAuditEnabled;
public SettleSelectorImpl(
boolean heavySelect,
@@ -85,19 +86,28 @@ public class SettleSelectorImpl implements ISettleSelector {
this.timePartition = timePartition;
this.tsFileManager = tsFileManager;
this.context = context;
+ // this will be enabled in other branches
+ this.ttlAuditEnabled = false;
}
static class FileDirtyInfo {
DirtyStatus status;
long dirtyDataSize = 0;
+ Set<String> ttlTables = Collections.emptySet();
public FileDirtyInfo(DirtyStatus status) {
this.status = status;
}
- public FileDirtyInfo(DirtyStatus status, long dirtyDataSize) {
+ public FileDirtyInfo(DirtyStatus status, Set<String> ttlTables) {
+ this.status = status;
+ this.ttlTables = ttlTables;
+ }
+
+ public FileDirtyInfo(DirtyStatus status, long dirtyDataSize, Set<String> ttlTables) {
this.status = status;
this.dirtyDataSize = dirtyDataSize;
+ this.ttlTables = ttlTables;
}
}
@@ -106,15 +116,19 @@ public class SettleSelectorImpl implements ISettleSelector {
List<TsFileResource> fullyDirtyResources = new ArrayList<>();
List<TsFileResource> partiallyDirtyResources = new ArrayList<>();
long totalPartiallyDirtyFileSize = 0;
+ final Set<String> ttlTables = new HashSet<>();
- public void addFullyDirtyResource(TsFileResource resource) {
+ public void addFullyDirtyResource(TsFileResource resource, Set<String> fileTTLTables) {
fullyDirtyResources.add(resource);
+ ttlTables.addAll(fileTTLTables);
}
- public boolean addPartiallyDirtyResource(TsFileResource resource, long dirtyDataSize) {
+ public boolean addPartiallyDirtyResource(
+ TsFileResource resource, long dirtyDataSize, Set<String> fileTTLTables) {
partiallyDirtyResources.add(resource);
totalPartiallyDirtyFileSize += resource.getTsFileSize();
totalPartiallyDirtyFileSize -= dirtyDataSize;
+ ttlTables.addAll(fileTTLTables);
return checkHasReachedThreshold();
}
@@ -126,6 +140,10 @@ public class SettleSelectorImpl implements ISettleSelector {
return partiallyDirtyResources;
}
+ public Set<String> getTTLTables() {
+ return ttlTables;
+ }
+
public boolean checkHasReachedThreshold() {
return partiallyDirtyResources.size() >= config.getInnerCompactionCandidateFileNum()
|| totalPartiallyDirtyFileSize >= config.getTargetCompactionFileSize();
@@ -165,11 +183,12 @@ public class SettleSelectorImpl implements ISettleSelector {
switch (fileDirtyInfo.status) {
case FULLY_DIRTY:
- settleTaskResource.addFullyDirtyResource(resource);
+ settleTaskResource.addFullyDirtyResource(resource, fileDirtyInfo.ttlTables);
break;
case PARTIALLY_DIRTY:
shouldStop =
- settleTaskResource.addPartiallyDirtyResource(resource, fileDirtyInfo.dirtyDataSize);
+ settleTaskResource.addPartiallyDirtyResource(
+ resource, fileDirtyInfo.dirtyDataSize, fileDirtyInfo.ttlTables);
break;
case NOT_SATISFIED:
shouldStop = !settleTaskResource.getPartiallyDirtyResources().isEmpty();
@@ -231,6 +250,7 @@ public class SettleSelectorImpl implements ISettleSelector {
timeIndex = CompactionUtils.buildDeviceTimeIndex(resource);
}
Set<IDeviceID> deletedDevices = new HashSet<>();
+ Set<String> ttlTables = new HashSet<>();
boolean hasExpiredTooLong = false;
long currentTime = CommonDateTimeUtils.currentTime();
@@ -240,12 +260,17 @@ public class SettleSelectorImpl implements ISettleSelector {
long ttl;
String tableName = device.getTableName();
+ boolean hasSetTTL;
if (tableName.startsWith("root.")) {
ttl = DataNodeTTLCache.getInstance().getTTLForTree(device);
+ hasSetTTL = ttl != Long.MAX_VALUE;
} else {
ttl = DataNodeTTLCache.getInstance().getTTLForTable(storageGroupName, tableName);
+ hasSetTTL = ttl != Long.MAX_VALUE;
+ if (hasSetTTL && ttlAuditEnabled) {
+ ttlTables.add(tableName);
+ }
}
- boolean hasSetTTL = ttl != Long.MAX_VALUE;
long endTime = timeIndex.getEndTime(device).get();
boolean isDeleted =
@@ -256,7 +281,7 @@ public class SettleSelectorImpl implements ISettleSelector {
if (!isDeleted) {
// For devices with TTL set, all data must expire in order to meet the conditions for
// being selected.
- return new FileDirtyInfo(DirtyStatus.NOT_SATISFIED);
+ return new FileDirtyInfo(DirtyStatus.NOT_SATISFIED, ttlTables);
}
if (currentTime > endTime) {
@@ -281,13 +306,13 @@ public class SettleSelectorImpl implements ISettleSelector {
((double) deletedDevices.size()) / ((ArrayDeviceTimeIndex) timeIndex).getDevices().size();
if (deletedDeviceRatio == 1d) {
// the whole file is completely dirty
- return new FileDirtyInfo(DirtyStatus.FULLY_DIRTY);
+ return new FileDirtyInfo(DirtyStatus.FULLY_DIRTY, ttlTables);
}
hasExpiredTooLong = config.getMaxExpiredTime() != Long.MAX_VALUE && hasExpiredTooLong;
if (hasExpiredTooLong || deletedDeviceRatio >= config.getExpiredDataRatio()) {
// evaluate dirty data size in the tsfile
return new FileDirtyInfo(
- PARTIALLY_DIRTY, (long) (deletedDeviceRatio * resource.getTsFileSize()));
+ PARTIALLY_DIRTY, (long) (deletedDeviceRatio * resource.getTsFileSize()), ttlTables);
}
return new FileDirtyInfo(DirtyStatus.NOT_SATISFIED);
}
(iotdb) 05/07: Enable CRC32 intrinsic for DataNode (#18239)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch rc/2.0.11
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 9d3807a05fa2cfe81044bd1da40fe073431bcfb8
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Mon Jul 20 09:42:12 2026 +0800
Enable CRC32 intrinsic for DataNode (#18239)
---
scripts/conf/datanode-env.sh | 3 +++
scripts/conf/windows/datanode-env.bat | 2 ++
2 files changed, 5 insertions(+)
diff --git a/scripts/conf/datanode-env.sh b/scripts/conf/datanode-env.sh
index 85e880634c4..66a51a00fde 100755
--- a/scripts/conf/datanode-env.sh
+++ b/scripts/conf/datanode-env.sh
@@ -237,6 +237,9 @@ IOTDB_JMX_OPTS="$IOTDB_JMX_OPTS -Djdk.nio.maxCachedBufferSize=${MAX_CACHED_BUFFE
IOTDB_JMX_OPTS="$IOTDB_JMX_OPTS -XX:+CrashOnOutOfMemoryError"
IOTDB_JMX_OPTS="$IOTDB_JMX_OPTS -XX:+UseAdaptiveSizePolicy"
IOTDB_JMX_OPTS="$IOTDB_JMX_OPTS -Xss512k"
+# Enable the CRC32 intrinsic on compiled paths to avoid the JNI critical fallback.
+IOTDB_JMX_OPTS="$IOTDB_JMX_OPTS -XX:+UnlockDiagnosticVMOptions"
+IOTDB_JMX_OPTS="$IOTDB_JMX_OPTS -XX:+UseCRC32Intrinsics"
# these two options print safepoints with pauses longer than 1000ms to the standard output. You can see these logs via redirection when starting in the background like "start-datanode.sh > log_datanode_safepoint.log"
IOTDB_JMX_OPTS="$IOTDB_JMX_OPTS -XX:SafepointTimeoutDelay=1000"
IOTDB_JMX_OPTS="$IOTDB_JMX_OPTS -XX:+SafepointTimeout"
diff --git a/scripts/conf/windows/datanode-env.bat b/scripts/conf/windows/datanode-env.bat
index c9422670df5..c745e32eeec 100644
--- a/scripts/conf/windows/datanode-env.bat
+++ b/scripts/conf/windows/datanode-env.bat
@@ -126,6 +126,8 @@ set IOTDB_HEAP_OPTS=%IOTDB_HEAP_OPTS% -XX:+UseAdaptiveSizePolicy
set IOTDB_HEAP_OPTS=%IOTDB_HEAP_OPTS% -Xss512k
@REM options below try to optimize safepoint stw time.
set IOTDB_HEAP_OPTS=%IOTDB_HEAP_OPTS% -XX:+UnlockDiagnosticVMOptions
+@REM Enable the CRC32 intrinsic on compiled paths to avoid the JNI critical fallback.
+set IOTDB_HEAP_OPTS=%IOTDB_HEAP_OPTS% -XX:+UseCRC32Intrinsics
set IOTDB_HEAP_OPTS=%IOTDB_HEAP_OPTS% -XX:GuaranteedSafepointInterval=0
@REM these two options print safepoints with pauses longer than 1000ms to the standard output. You can see these logs via redirection when starting in the background like "start-datanode.sh > log_datanode_safepoint.txt"
set IOTDB_HEAP_OPTS=%IOTDB_HEAP_OPTS% -XX:SafepointTimeoutDelay=1000