This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 6889b1e3cf7 modify the read logic of metadata lease timeout (#18228)
6889b1e3cf7 is described below
commit 6889b1e3cf7b5f8a923c26a3abef8e10530e2cfc
Author: CYB <[email protected]>
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 a5211228300..91393bf9fd2 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
@@ -479,13 +479,6 @@ public class CommonConfig {
private volatile long remoteWriteMaxRetryDurationInMs = 60000;
- // The DataNode self-fences its ConfigNode-pushed metadata caches
(table/tree schema, template,
- // TTL, permission, ...) if it has not received a ConfigNode heartbeat
within this duration. Kept
- // aligned with the failure detector threshold so a partitioned DataNode
stops trusting stale
- // caches around the same time the cluster would consider it dead. Also used
by the ConfigNode to
- // derive how long it must wait before treating an unreachable DataNode as
safely fenced.
- private volatile long metadataLeaseFenceMs = 20_000;
-
private final RateLimiter querySamplingRateLimiter = RateLimiter.create(160);
// if querySamplingRateLimiter < 0, means that there is no rate limit, we
need to full sample all
// the queries
@@ -2994,14 +2987,6 @@ public 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 {
