Skip to content

[Fix] Keep heartbeat alive when metadata lease is fenced #18343

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Caideyipi wants to merge 2 commits into
base: master
Choose a base branch
from
Open

[Fix] Keep heartbeat alive when metadata lease is fenced #18343

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7dc9d05
[Fix] Keep heartbeat alive when metadata lease is fenced
Caideyipi Jul 29, 2026
48b5374
Stabilize metadata lease heartbeat test
Caideyipi Jul 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,9 @@ public final class DataNodeSchemaMessages {
"SchemaRegion(id = {}) has been deleted, skiped";
public static final String FAILED_TO_GET_TABLE_FOR_TIMESERIES_COUNT =
"Failed to get table {}.{} when calculating the time series number. Maybe the cluster is restarting or the table is being dropped.";
public static final String
LOG_METADATA_LEASE_IS_FENCED_SKIP_REPORTING_SCHEMA_USAGE_IN_THIS_HEARTBEAT_81D36975 =
"Metadata lease is fenced. Skip reporting schema usage in this heartbeat.";
public static final String TREE_VIEW_TABLE_CANNOT_BE_WRITTEN_OR_DELETED =
"The table %s.%s is a view from tree, cannot be written or deleted from";
public static final String PEER_IS_SHUTTING_DOWN = "Peer is shutting down now.";
Expand Down
3 changes: 3 additions & 0 deletions iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,9 @@ public final class DataNodeSchemaMessages {
public static final String SCHEMA_REGION_ALREADY_DELETED = "SchemaRegion(id = {}) 已被删除，已跳过";
public static final String FAILED_TO_GET_TABLE_FOR_TIMESERIES_COUNT =
"计算时间序列数量时获取表 {}.{} 失败，可能是集群正在重启或表正在被删除。";
public static final String
LOG_METADATA_LEASE_IS_FENCED_SKIP_REPORTING_SCHEMA_USAGE_IN_THIS_HEARTBEAT_81D36975 =
"元数据租约已被隔离，本次心跳跳过上报 schema 使用量。";
public static final String TREE_VIEW_TABLE_CANNOT_BE_WRITTEN_OR_DELETED =
"表 %s.%s 是树模型视图，不能写入或删除";
public static final String PEER_IS_SHUTTING_DOWN = "节点正在关闭中。";
Expand Down
31 changes: 26 additions & 5 deletions ...rc/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,7 @@
import org.apache.iotdb.commons.enums.DataPartitionTableGeneratorState;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.commons.exception.MetadataLeaseFencedException;
import org.apache.iotdb.commons.i18n.PipeMessages;
import org.apache.iotdb.commons.partition.DataPartitionTable;
import org.apache.iotdb.commons.partition.DatabaseScopedDataPartitionTable;
Expand Down Expand Up @@ -117,6 +118,7 @@
import org.apache.iotdb.db.consensus.SchemaRegionConsensusImpl;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.i18n.DataNodeMiscMessages;
import org.apache.iotdb.db.i18n.DataNodeSchemaMessages;
import org.apache.iotdb.db.partition.DataPartitionTableGenerator;
import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
import org.apache.iotdb.db.protocol.client.ConfigNodeInfo;
Expand Down Expand Up @@ -2355,19 +2357,38 @@ public TDataNodeHeartbeatResp getDataNodeHeartBeat(TDataNodeHeartbeatReq req) th
if (commonConfig.getStatusReason() != null) {
resp.setStatusReason(commonConfig.getStatusReason());
}
MetadataLeaseFencedException schemaUsageCollectionException = null;
if (req.getSchemaRegionIds() != null) {
spaceQuotaManager.updateSpaceQuotaUsage(req.getSpaceQuotaUsage());
resp.setRegionDeviceUsageMap(
schemaEngine.countDeviceNumBySchemaRegion(req.getSchemaRegionIds()));
resp.setRegionSeriesUsageMap(
schemaEngine.countTimeSeriesNumBySchemaRegion(req.getSchemaRegionIds()));
try {
resp.setRegionDeviceUsageMap(
schemaEngine.countDeviceNumBySchemaRegion(req.getSchemaRegionIds()));
resp.setRegionSeriesUsageMap(
schemaEngine.countTimeSeriesNumBySchemaRegion(req.getSchemaRegionIds()));
} catch (final MetadataLeaseFencedException e) {
schemaUsageCollectionException = e;
}
}
if (req.getDataRegionIds() != null) {
spaceQuotaManager.setDataRegionIds(req.getDataRegionIds());
resp.setRegionDisk(spaceQuotaManager.getRegionDisk());
}
// Update schema quota if necessary
SchemaEngine.getInstance().updateAndFillSchemaCountMap(req, resp);
try {
SchemaEngine.getInstance().updateAndFillSchemaCountMap(req, resp);
} catch (final MetadataLeaseFencedException e) {
if (schemaUsageCollectionException == null) {
schemaUsageCollectionException = e;
}
}
if (schemaUsageCollectionException != null) {
resp.unsetRegionDeviceUsageMap();
resp.unsetRegionSeriesUsageMap();
LOGGER.warn(
DataNodeSchemaMessages
.LOG_METADATA_LEASE_IS_FENCED_SKIP_REPORTING_SCHEMA_USAGE_IN_THIS_HEARTBEAT_81D36975,
schemaUsageCollectionException);
}

// Update pipe meta if necessary
if (req.isNeedPipeMetaList()) {
Expand Down
28 changes: 22 additions & 6 deletions iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/SchemaEngine.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -447,11 +447,21 @@ public void updateAndFillSchemaCountMap(
schemaQuotaManager.updateRemain(
req.getTimeSeriesQuotaRemain(),
req.isSetDeviceQuotaRemain() ? req.getDeviceQuotaRemain() : -1);
// Build both maps on snapshots and publish them together only after all counts succeed, so a
// failure cannot leave a partially updated heartbeat response.
Map<Integer, Long> regionDeviceUsageMap =
resp.getRegionDeviceUsageMap() == null
? null
: new HashMap<>(resp.getRegionDeviceUsageMap());
Map<Integer, Long> regionSeriesUsageMap =
resp.getRegionSeriesUsageMap() == null
? null
: new HashMap<>(resp.getRegionSeriesUsageMap());
if (schemaQuotaManager.isDeviceLimit()) {
if (resp.getRegionDeviceUsageMap() == null) {
resp.setRegionDeviceUsageMap(new HashMap<>());
if (regionDeviceUsageMap == null) {
regionDeviceUsageMap = new HashMap<>();
}
final Map<Integer, Long> tmp = resp.getRegionDeviceUsageMap();
final Map<Integer, Long> tmp = regionDeviceUsageMap;
SchemaRegionConsensusImpl.getInstance().getAllConsensusGroupIds().stream()
.filter(
consensusGroupId ->
Expand All @@ -467,10 +477,10 @@ public void updateAndFillSchemaCountMap(
.orElse(0L)));
}
if (schemaQuotaManager.isMeasurementLimit()) {
if (resp.getRegionSeriesUsageMap() == null) {
resp.setRegionSeriesUsageMap(new HashMap<>());
if (regionSeriesUsageMap == null) {
regionSeriesUsageMap = new HashMap<>();
}
final Map<Integer, Long> tmp = resp.getRegionSeriesUsageMap();
final Map<Integer, Long> tmp = regionSeriesUsageMap;
SchemaRegionConsensusImpl.getInstance().getAllConsensusGroupIds().stream()
.filter(
consensusGroupId ->
Expand All @@ -490,6 +500,12 @@ public void updateAndFillSchemaCountMap(
.map(this::getTimeSeriesNumber4Quota)
.orElse(0L)));
}
if (regionDeviceUsageMap != null) {
resp.setRegionDeviceUsageMap(regionDeviceUsageMap);
}
if (regionSeriesUsageMap != null) {
resp.setRegionSeriesUsageMap(regionSeriesUsageMap);
}
}

public ISchemaEngineStatistics getSchemaEngineStatistics() {
Expand Down
73 changes: 73 additions & 0 deletions ...atanode/src/test/java/org/apache/iotdb/db/service/DataNodeInternalRPCServiceImplTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
import org.apache.iotdb.commons.consensus.DataRegionId;
import org.apache.iotdb.commons.consensus.SchemaRegionId;
import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.commons.exception.MetadataLeaseFencedException;
import org.apache.iotdb.commons.path.MeasurementPath;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
Expand All @@ -50,10 +51,14 @@
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metadata.write.CreateMultiTimeSeriesNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.metadata.write.CreateTimeSeriesNode;
import org.apache.iotdb.db.schemaengine.SchemaEngine;
import org.apache.iotdb.db.schemaengine.lease.MetadataLeaseManager;
import org.apache.iotdb.db.schemaengine.rescon.DataNodeSchemaQuotaManager;
import org.apache.iotdb.db.service.DataNode.DataNodeContext;
import org.apache.iotdb.db.storageengine.dataregion.DataRegion;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import org.apache.iotdb.db.utils.EnvironmentUtils;
import org.apache.iotdb.mpp.rpc.thrift.TDataNodeHeartbeatReq;
import org.apache.iotdb.mpp.rpc.thrift.TDataNodeHeartbeatResp;
import org.apache.iotdb.mpp.rpc.thrift.TPlanNode;
import org.apache.iotdb.mpp.rpc.thrift.TSendBatchPlanNodeReq;
import org.apache.iotdb.mpp.rpc.thrift.TSendBatchPlanNodeResp;
Expand All @@ -63,6 +68,7 @@
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
import org.awaitility.Awaitility;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Assert;
Expand All @@ -73,6 +79,8 @@

import java.io.File;
import java.io.IOException;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
Expand All @@ -81,6 +89,8 @@
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import static org.mockito.Mockito.when;

Expand Down Expand Up @@ -229,6 +239,69 @@ public void testRejectLoad4NonActiveImpl() throws LoadFileException {
.setActive(true);
}

@Test
public void testHeartbeatSkipsSchemaUsageWhenMetadataLeaseIsFenced() throws Exception {
final MetadataLeaseManager leaseManager = MetadataLeaseManager.getInstance();
final Field hasPullTaskNowRefField =
MetadataLeaseManager.class.getDeclaredField("hasPullTaskNowRef");
hasPullTaskNowRefField.setAccessible(true);
final AtomicBoolean hasPullTaskNowRef =
(AtomicBoolean) hasPullTaskNowRefField.get(leaseManager);
final Map<String, Long> tableDeviceNumberMap =
SchemaEngine.getInstance()
.getSchemaRegion(new SchemaRegionId(0))
.getSchemaRegionStatistics()
.getTable2DevicesNumMap();

try {
Awaitility.await()
.atMost(10, TimeUnit.SECONDS)
.until(() -> SchemaRegionConsensusImpl.getInstance().isLeader(new SchemaRegionId(0)));
hasPullTaskNowRef.set(true);
tableDeviceNumberMap.put("table", 1L);
leaseManager.recoveryLeaseForTest(false);
final Method checkLeaseStatus =
MetadataLeaseManager.class.getDeclaredMethod("checkLeaseStatus");
checkLeaseStatus.setAccessible(true);
checkLeaseStatus.invoke(leaseManager);
Assert.assertTrue(leaseManager.isFenced());

final TDataNodeHeartbeatReq req =
new TDataNodeHeartbeatReq()
.setHeartbeatTimestamp(1L)
.setNeedJudgeLeader(false)
.setNeedSamplingLoad(false)
.setTimeSeriesQuotaRemain(10L)
.setDeviceQuotaRemain(10L)
.setLogicalClock(0L);
final Map<Integer, Long> previousDeviceCountMap = new HashMap<>();
previousDeviceCountMap.put(-1, 1L);
final Map<Integer, Long> previousSeriesCountMap = new HashMap<>();
previousSeriesCountMap.put(-1, 2L);
final TDataNodeHeartbeatResp schemaCountResp =
new TDataNodeHeartbeatResp()
.setRegionDeviceUsageMap(new HashMap<>(previousDeviceCountMap))
.setRegionSeriesUsageMap(new HashMap<>(previousSeriesCountMap));

Assert.assertThrows(
MetadataLeaseFencedException.class,
() -> SchemaEngine.getInstance().updateAndFillSchemaCountMap(req, schemaCountResp));
Assert.assertEquals(previousDeviceCountMap, schemaCountResp.getRegionDeviceUsageMap());
Assert.assertEquals(previousSeriesCountMap, schemaCountResp.getRegionSeriesUsageMap());

final TDataNodeHeartbeatResp resp = dataNodeInternalRPCServiceImpl.getDataNodeHeartBeat(req);

Assert.assertEquals(req.getHeartbeatTimestamp(), resp.getHeartbeatTimestamp());
Assert.assertFalse(resp.isSetRegionDeviceUsageMap());
Assert.assertFalse(resp.isSetRegionSeriesUsageMap());
} finally {
tableDeviceNumberMap.remove("table");
hasPullTaskNowRef.set(false);
leaseManager.recoveryLeaseForTest(true);
DataNodeSchemaQuotaManager.getInstance().updateRemain(-1, -1);
}
}

@Test
public void testCreateAlignedTimeSeries() throws MetadataException {
CreateAlignedTimeSeriesNode createAlignedTimeSeriesNode =
Expand Down
Loading