From 9a5f2392a5d4a72e0bdbb92de2126ff472f6d344 Mon Sep 17 00:00:00 2001 From: Yongzao <532741407@qq.com> Date: Wed, 20 May 2026 11:27:23 +0800 Subject: [PATCH 1/5] Add ConfigNode ReadOnly state with DiskFull/DiskCrash heartbeat self-check ConfigNode now reports its own NodeStatus.ReadOnly when its critical directories (systemDir, consensusDir) are unwritable or near-full, mirroring the existing DataNode behavior. NodeStatus reasons are extended with a new DISK_CRASH constant alongside DISK_FULL, and the ConfigNode heartbeat carries status/statusReason back to the leader. - node-commons: new DiskChecker utility (probe + state-machine apply), with priority DiskCrash > DiskFull and recovery to Running when the reason was disk-related. i18n messages added in en + zh. - thrift-confignode: TConfigNodeHeartbeatResp gains optional status and statusReason fields (forward-compatible). - confignode: leader self-checks before fanning out heartbeats; follower self-checks on receive and reports back; cache reads from CommonConfig for the leader's self entry, otherwise from the sample. - datanode: FolderManager exposes a static hasAnyAbnormalFolder() aggregator; sampleDiskLoad treats any ABNORMAL folder as DiskCrash (which wins over DiskFull) and reuses DiskChecker.apply. --- .../confignode/conf/ConfigNodeConfig.java | 10 + .../cache/node/ConfigNodeHeartbeatCache.java | 48 ++-- .../load/cache/node/NodeHeartbeatSample.java | 9 +- .../load/service/HeartbeatService.java | 8 + .../thrift/ConfigNodeRPCServiceProcessor.java | 9 + .../impl/DataNodeInternalRPCServiceImpl.java | 25 +- .../iotdb/commons/i18n/CommonMessages.java | 10 + .../iotdb/commons/i18n/CommonMessages.java | 10 + .../iotdb/commons/cluster/DiskChecker.java | 146 ++++++++++++ .../iotdb/commons/cluster/NodeStatus.java | 1 + .../iotdb/commons/disk/FolderManager.java | 39 ++- .../commons/cluster/DiskCheckerTest.java | 224 ++++++++++++++++++ .../src/main/thrift/confignode.thrift | 7 + 13 files changed, 517 insertions(+), 29 deletions(-) create mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/cluster/DiskChecker.java create mode 100644 iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/cluster/DiskCheckerTest.java 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 f49525ea7da7..944463e697af 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 @@ -41,6 +41,7 @@ import java.io.File; import java.lang.reflect.Field; import java.util.Arrays; +import java.util.List; public class ConfigNodeConfig { @@ -523,6 +524,15 @@ public void setConsensusDir(String consensusDir) { this.consensusDir = consensusDir; } + /** + * Directories whose loss would render this ConfigNode unable to serve. Used by the periodic + * disk-health check on both leader (in HeartbeatService loop) and followers (in the + * heartbeat-receive path). + */ + public List getCriticalDirs() { + return Arrays.asList(systemDir, consensusDir); + } + public String getConfigNodeConsensusProtocolClass() { return configNodeConsensusProtocolClass; } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/ConfigNodeHeartbeatCache.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/ConfigNodeHeartbeatCache.java index 4d675e1e8a44..7228fca648cf 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/ConfigNodeHeartbeatCache.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/ConfigNodeHeartbeatCache.java @@ -20,6 +20,7 @@ package org.apache.iotdb.confignode.manager.load.cache.node; import org.apache.iotdb.commons.cluster.NodeStatus; +import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor; import org.apache.iotdb.confignode.manager.load.cache.AbstractHeartbeatSample; @@ -49,28 +50,39 @@ public ConfigNodeHeartbeatCache(int configNodeId, NodeStatistics statistics) { @Override public synchronized void updateCurrentStatistics(boolean forceUpdate) { - // Skip itself and the Removing status can not be updated - if (nodeId == CURRENT_NODE_ID || NodeStatus.Removing.equals(getNodeStatus())) { + // Removing status can not be updated + if (NodeStatus.Removing.equals(getNodeStatus())) { return; } - NodeHeartbeatSample lastSample; - // Update Node status - NodeStatus status; long currentNanoTime = System.nanoTime(); - final List heartbeatHistory; - synchronized (slidingWindow) { - lastSample = (NodeHeartbeatSample) getLastSample(); - heartbeatHistory = Collections.unmodifiableList(slidingWindow); + NodeStatus status; + String statusReason; + + if (nodeId == CURRENT_NODE_ID) { + // Self entry: heartbeat loop never sends to itself, so mirror the status that + // this ConfigNode's local disk-check / startup wrote into CommonConfig. + status = CommonDescriptor.getInstance().getConfig().getNodeStatus(); + statusReason = CommonDescriptor.getInstance().getConfig().getStatusReason(); + } else { + NodeHeartbeatSample lastSample; + final List heartbeatHistory; + synchronized (slidingWindow) { + lastSample = (NodeHeartbeatSample) getLastSample(); + heartbeatHistory = Collections.unmodifiableList(slidingWindow); - if (lastSample == null) { - /* First heartbeat not received from this ConfigNode, status is UNKNOWN */ - status = NodeStatus.Unknown; - } else if (!failureDetector.isAvailable(nodeId, heartbeatHistory)) { - /* Failure detector decides that this ConfigNode is UNKNOWN */ - status = NodeStatus.Unknown; - } else { - status = lastSample.getStatus(); + if (lastSample == null) { + /* First heartbeat not received from this ConfigNode, status is UNKNOWN */ + status = NodeStatus.Unknown; + statusReason = null; + } else if (!failureDetector.isAvailable(nodeId, heartbeatHistory)) { + /* Failure detector decides that this ConfigNode is UNKNOWN */ + status = NodeStatus.Unknown; + statusReason = null; + } else { + status = lastSample.getStatus(); + statusReason = lastSample.getStatusReason(); + } } } @@ -79,6 +91,6 @@ public synchronized void updateCurrentStatistics(boolean forceUpdate) { // TODO: Construct load score module long loadScore = NodeStatus.isNormalStatus(status) ? 0 : Long.MAX_VALUE; - currentStatistics.set(new NodeStatistics(currentNanoTime, status, null, loadScore)); + currentStatistics.set(new NodeStatistics(currentNanoTime, status, statusReason, loadScore)); } } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/NodeHeartbeatSample.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/NodeHeartbeatSample.java index 8217593f5d67..5ead20d291f4 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/NodeHeartbeatSample.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/NodeHeartbeatSample.java @@ -74,8 +74,13 @@ public NodeHeartbeatSample(TAIHeartbeatResp heartbeatResp) { /** Constructor for ConfigNode sample. */ public NodeHeartbeatSample(TConfigNodeHeartbeatResp heartbeatResp) { super(heartbeatResp.getTimestamp()); - this.status = NodeStatus.Running; - this.statusReason = null; + // Old ConfigNodes don't populate status/statusReason — fall back to Running/null + // so a rolling upgrade leaves the leader's view of legacy peers unchanged. + this.status = + heartbeatResp.isSetStatus() + ? NodeStatus.parse(heartbeatResp.getStatus()) + : NodeStatus.Running; + this.statusReason = heartbeatResp.isSetStatusReason() ? heartbeatResp.getStatusReason() : null; this.loadSample = null; } 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 ef732145f22d..be2ae1982cf2 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 @@ -24,9 +24,11 @@ import org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation; import org.apache.iotdb.common.rpc.thrift.TDataNodeConfiguration; import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.commons.cluster.DiskChecker; import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory; import org.apache.iotdb.commons.concurrent.ThreadName; import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil; +import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.pipe.config.PipeConfig; import org.apache.iotdb.confignode.client.async.AsyncAINodeHeartbeatClientPool; import org.apache.iotdb.confignode.client.async.AsyncConfigNodeHeartbeatClientPool; @@ -142,6 +144,12 @@ private void heartbeatLoopBody() { .ifPresent( consensusManager -> { if (getConsensusManager().isLeader()) { + // Leader self-checks its own disk health before fanning out heartbeats. + // Followers run the same check when receiving each heartbeat request + // (see ConfigNodeRPCServiceProcessor#getConfigNodeHeartBeat). + DiskChecker.checkAndApply( + ConfigNodeDescriptor.getInstance().getConf().getCriticalDirs(), + CommonDescriptor.getInstance().getConfig().getDiskSpaceWarningThreshold()); // Send heartbeat requests to all the registered ConfigNodes pingRegisteredConfigNodes( genConfigNodeHeartbeatReq(), getNodeManager().getRegisteredConfigNodes()); diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java index b6fe2c0fc219..92e3221ddf13 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java @@ -40,6 +40,7 @@ import org.apache.iotdb.commons.auth.entity.PrivilegeModelType; import org.apache.iotdb.commons.auth.entity.PrivilegeType; import org.apache.iotdb.commons.auth.entity.PrivilegeUnion; +import org.apache.iotdb.commons.cluster.DiskChecker; import org.apache.iotdb.commons.conf.CommonConfig; import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.consensus.ConsensusGroupId; @@ -1114,6 +1115,14 @@ public TRegionRouteMapResp getLatestRegionRouteMap() { public TConfigNodeHeartbeatResp getConfigNodeHeartBeat(TConfigNodeHeartbeatReq heartbeatReq) { TConfigNodeHeartbeatResp resp = new TConfigNodeHeartbeatResp(); resp.setTimestamp(heartbeatReq.getTimestamp()); + // Follower self-check: probe critical dirs each time the leader pings us. + // The leader runs the same check in its HeartbeatService loop. + DiskChecker.checkAndApply( + configNodeConfig.getCriticalDirs(), commonConfig.getDiskSpaceWarningThreshold()); + resp.setStatus(commonConfig.getNodeStatus().getStatus()); + if (commonConfig.getStatusReason() != null) { + resp.setStatusReason(commonConfig.getStatusReason()); + } return resp; } 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 cde5b09b4578..bbf67a8354d2 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 @@ -49,6 +49,7 @@ import org.apache.iotdb.commons.audit.UserEntity; import org.apache.iotdb.commons.auth.entity.PrivilegeType; import org.apache.iotdb.commons.client.request.AsyncRequestContext; +import org.apache.iotdb.commons.cluster.DiskChecker; import org.apache.iotdb.commons.cluster.NodeStatus; import org.apache.iotdb.commons.concurrent.Await; import org.apache.iotdb.commons.concurrent.AwaitTimeoutException; @@ -219,6 +220,7 @@ import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileManager; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; +import org.apache.iotdb.db.storageengine.rescon.disk.FolderManager; import org.apache.iotdb.db.storageengine.rescon.quotas.DataNodeSpaceQuotaManager; import org.apache.iotdb.db.storageengine.rescon.quotas.DataNodeThrottleQuotaManager; import org.apache.iotdb.db.subscription.agent.SubscriptionAgent; @@ -2555,12 +2557,20 @@ private void sampleDiskLoad(TLoadSample loadSample) { SYSTEM) .getValue(); + // Derive the disk status: an ABNORMAL data folder observed by any FolderManager wins over + // a full-disk reading, and a low free-space ratio still drives DISK_FULL when nothing is + // crashed. DiskChecker.apply then performs the actual NodeStatus transition (including the + // "ReadOnly(DiskFull|DiskCrash) -> Running" recovery for the DiskFull path). + DiskChecker.DiskStatus diskStatus = DiskChecker.DiskStatus.NORMAL; + if (FolderManager.hasAnyAbnormalFolder()) { + diskStatus = DiskChecker.DiskStatus.DISK_CRASH; + } if (availableDisk != 0 && totalDisk != 0) { double freeDiskRatio = availableDisk / totalDisk; loadSample.setFreeDiskSpace(availableDisk); loadSample.setDiskUsageRate(1d - freeDiskRatio); - // Reset NodeStatus if necessary - if (freeDiskRatio < commonConfig.getDiskSpaceWarningThreshold()) { + if (diskStatus == DiskChecker.DiskStatus.NORMAL + && freeDiskRatio < commonConfig.getDiskSpaceWarningThreshold()) { LOGGER.warn( DataNodeMiscMessages .MISC_LOG_THE_AVAILABLE_DISK_SPACE_IS_THE_TOTAL_DISK_SPACE_IS_AND_4506856F, @@ -2568,14 +2578,13 @@ private void sampleDiskLoad(TLoadSample loadSample) { RamUsageEstimator.humanReadableUnits((long) totalDisk), freeDiskRatio, commonConfig.getDiskSpaceWarningThreshold()); - commonConfig.setNodeStatus(NodeStatus.ReadOnly); - commonConfig.setStatusReason(NodeStatus.DISK_FULL); - } else if (NodeStatus.ReadOnly.equals(commonConfig.getNodeStatus()) - && NodeStatus.DISK_FULL.equals(commonConfig.getStatusReason())) { - commonConfig.setNodeStatus(NodeStatus.Running); - commonConfig.setStatusReason(null); + diskStatus = DiskChecker.DiskStatus.DISK_FULL; } + } else if (diskStatus == DiskChecker.DiskStatus.NORMAL) { + // Metrics not available yet — fall back to no-op so we don't churn the status. + return; } + DiskChecker.apply(diskStatus); } @Override diff --git a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java index ae97d09ddc25..510ef3b89db9 100644 --- a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java @@ -37,6 +37,16 @@ public final class CommonMessages { public static final String NODE_STATUS_NOT_EXIST = "NodeStatus %s doesn't exist."; public static final String UNKNOWN_NODE_STATUS = "Unknown NodeStatus %s."; + // --- disk health --- + public static final String DISK_FULL_SET_READ_ONLY = + "Free disk space ratio is below the configured threshold; set node status to ReadOnly(DiskFull)."; + public static final String DISK_CRASH_SET_READ_ONLY = + "Detected unwritable disk directory; set node status to ReadOnly(DiskCrash)."; + public static final String DISK_CRASH_PROBE_FAILED = + "Disk health probe write failed for directory {}."; + public static final String DISK_RECOVERED_SET_RUNNING = + "Disk health recovered (previous reason: {}); set node status to Running."; + // --- consensus --- public static final String UNRECOGNIZED_CONSENSUS_GROUP_ID = "Unrecognized ConsensusGroupId: %s"; diff --git a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java index 71c45ccaf3fc..8414a8eb916c 100644 --- a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java @@ -36,6 +36,16 @@ public final class CommonMessages { public static final String NODE_STATUS_NOT_EXIST = "NodeStatus %s 不存在。"; public static final String UNKNOWN_NODE_STATUS = "未知 NodeStatus %s。"; + // --- disk health --- + public static final String DISK_FULL_SET_READ_ONLY = + "磁盘剩余空间比例低于配置阈值，将节点状态设为 ReadOnly(DiskFull)。"; + public static final String DISK_CRASH_SET_READ_ONLY = + "检测到不可写的磁盘目录，将节点状态设为 ReadOnly(DiskCrash)。"; + public static final String DISK_CRASH_PROBE_FAILED = + "对目录 {} 进行磁盘健康探测时写入失败。"; + public static final String DISK_RECOVERED_SET_RUNNING = + "磁盘健康已恢复（先前原因：{}），将节点状态设为 Running。"; + // --- consensus --- public static final String UNRECOGNIZED_CONSENSUS_GROUP_ID = "无法识别的 ConsensusGroupId：%s"; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/cluster/DiskChecker.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/cluster/DiskChecker.java new file mode 100644 index 000000000000..4632ddf28dd5 --- /dev/null +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/cluster/DiskChecker.java @@ -0,0 +1,146 @@ +/* + * 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.commons.cluster; + +import org.apache.iotdb.commons.conf.CommonConfig; +import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.i18n.CommonMessages; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.List; + +/** + * Shared utility used by both ConfigNode and DataNode to evaluate the health of a set of critical + * directories and, optionally, drive the global {@link NodeStatus} on {@link CommonConfig}. + * + *

Only transitions between {@link NodeStatus#Running} and {@link NodeStatus#ReadOnly} with + * reason {@link NodeStatus#DISK_FULL}/{@link NodeStatus#DISK_CRASH} are managed here. Other + * ReadOnly reasons (e.g. manual) are left untouched. + */ +public class DiskChecker { + + private static final Logger LOGGER = LoggerFactory.getLogger(DiskChecker.class); + + private static final byte[] PROBE_PAYLOAD = new byte[] {0x01}; + private static final String PROBE_PREFIX = "iotdb-disk-probe-"; + private static final String PROBE_SUFFIX = ".tmp"; + + public enum DiskStatus { + NORMAL, + DISK_FULL, + DISK_CRASH + } + + private DiskChecker() {} + + /** + * Evaluate the supplied directories. A single unwritable directory yields {@link + * DiskStatus#DISK_CRASH}; any directory whose usable/total ratio is below the threshold yields + * {@link DiskStatus#DISK_FULL} (when no crash is detected); otherwise {@link DiskStatus#NORMAL}. + */ + public static DiskStatus check(List dirs, double freeRatioThreshold) { + if (dirs == null || dirs.isEmpty()) { + return DiskStatus.NORMAL; + } + boolean anyFull = false; + for (String dir : dirs) { + if (dir == null || dir.isEmpty()) { + continue; + } + File f = new File(dir); + if (!f.isDirectory()) { + LOGGER.warn(CommonMessages.DISK_CRASH_PROBE_FAILED, dir); + return DiskStatus.DISK_CRASH; + } + try { + Path probe = Files.createTempFile(Paths.get(dir), PROBE_PREFIX, PROBE_SUFFIX); + try { + Files.write(probe, PROBE_PAYLOAD); + } finally { + Files.deleteIfExists(probe); + } + } catch (IOException e) { + LOGGER.warn(CommonMessages.DISK_CRASH_PROBE_FAILED, dir, e); + return DiskStatus.DISK_CRASH; + } + long total = f.getTotalSpace(); + long usable = f.getUsableSpace(); + if (total > 0 && (double) usable / total < freeRatioThreshold) { + anyFull = true; + } + } + return anyFull ? DiskStatus.DISK_FULL : DiskStatus.NORMAL; + } + + /** + * Run {@link #check} and apply the result to {@link CommonConfig}. See class javadoc for + * transition rules. + */ + public static void checkAndApply(List dirs, double freeRatioThreshold) { + apply(check(dirs, freeRatioThreshold)); + } + + /** Visible for tests; package-public callers should prefer {@link #checkAndApply}. */ + public static void apply(DiskStatus result) { + CommonConfig config = CommonDescriptor.getInstance().getConfig(); + NodeStatus currentStatus = config.getNodeStatus(); + String currentReason = config.getStatusReason(); + boolean currentlyFull = + NodeStatus.ReadOnly.equals(currentStatus) && NodeStatus.DISK_FULL.equals(currentReason); + boolean currentlyCrash = + NodeStatus.ReadOnly.equals(currentStatus) && NodeStatus.DISK_CRASH.equals(currentReason); + + switch (result) { + case DISK_CRASH: + if (!currentlyCrash) { + LOGGER.warn(CommonMessages.DISK_CRASH_SET_READ_ONLY); + config.setNodeStatus(NodeStatus.ReadOnly); + config.setStatusReason(NodeStatus.DISK_CRASH); + } + break; + case DISK_FULL: + // DiskCrash has higher priority — do not downgrade an existing crash to full. + if (currentlyCrash) { + return; + } + if (!currentlyFull) { + LOGGER.warn(CommonMessages.DISK_FULL_SET_READ_ONLY); + config.setNodeStatus(NodeStatus.ReadOnly); + config.setStatusReason(NodeStatus.DISK_FULL); + } + break; + case NORMAL: + default: + if (currentlyFull || currentlyCrash) { + LOGGER.info(CommonMessages.DISK_RECOVERED_SET_RUNNING, currentReason); + config.setNodeStatus(NodeStatus.Running); + config.setStatusReason(null); + } + break; + } + } +} diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/cluster/NodeStatus.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/cluster/NodeStatus.java index 518a9faaed2e..e37b44edafda 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/cluster/NodeStatus.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/cluster/NodeStatus.java @@ -35,6 +35,7 @@ public enum NodeStatus { /** Only query statements are permitted */ ReadOnly("ReadOnly"); public static final String DISK_FULL = "DiskFull"; + public static final String DISK_CRASH = "DiskCrash"; private final String status; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/disk/FolderManager.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/disk/FolderManager.java index a7707077f84d..06234bc23a9c 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/disk/FolderManager.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/disk/FolderManager.java @@ -36,6 +36,7 @@ import org.slf4j.LoggerFactory; import java.io.IOException; +import java.lang.ref.WeakReference; import java.nio.file.FileStore; import java.nio.file.Files; import java.nio.file.Path; @@ -43,10 +44,20 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.CopyOnWriteArrayList; public class FolderManager { private static final Logger logger = LoggerFactory.getLogger(FolderManager.class); + /** + * Registry of every live {@link FolderManager} instance so the DataNode heartbeat path can ask + * "is any folder anywhere on this node currently ABNORMAL?" without each subsystem having to push + * state into a central reporter. Weak references avoid keeping short-lived managers alive (e.g. + * those created per snapshot/load). + */ + private static final List> ALL_INSTANCES = + new CopyOnWriteArrayList<>(); + /** Represents the operational states of a data folder. */ public enum FolderState { /** Indicates the folder is functioning normally with no issues. */ @@ -69,6 +80,7 @@ public FolderManager(List folders, DirectoryStrategyType type) throws DiskSpaceInsufficientException { this.folders = folders; folders.forEach(dir -> foldersStates.put(dir, FolderState.HEALTHY)); + ALL_INSTANCES.add(new WeakReference<>(this)); switch (type) { case SEQUENCE_STRATEGY: this.selectStrategy = new SequenceStrategy(); @@ -94,7 +106,7 @@ public FolderManager(List folders, DirectoryStrategyType type) } } - public void updateFolderState(String folder, FolderState state) { + public synchronized void updateFolderState(String folder, FolderState state) { foldersStates.replace(folder, state); selectStrategy.updateFolderState(folder, state); } @@ -189,4 +201,29 @@ public String getFirstFolderOfSameDisk(String pathStr) { } return null; } + + /** + * Walks every live FolderManager instance and reports whether any folder is currently {@link + * FolderState#ABNORMAL}. Used by the DataNode heartbeat path to derive a {@code + * NodeStatus.ReadOnly(DiskCrash)} signal from already-observed write failures. + * + *

Stale (GC'd) weak references are pruned as a side effect. + */ + public static boolean hasAnyAbnormalFolder() { + for (WeakReference reference : ALL_INSTANCES) { + FolderManager folderManager = reference.get(); + if (folderManager == null) { + continue; + } + synchronized (folderManager) { + for (FolderState state : folderManager.foldersStates.values()) { + if (state == FolderState.ABNORMAL) { + return true; + } + } + } + } + ALL_INSTANCES.removeIf(ref -> ref.get() == null); + return false; + } } diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/cluster/DiskCheckerTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/cluster/DiskCheckerTest.java new file mode 100644 index 000000000000..fde492c1090d --- /dev/null +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/cluster/DiskCheckerTest.java @@ -0,0 +1,224 @@ +/* + * 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.commons.cluster; + +import org.apache.iotdb.commons.conf.CommonConfig; +import org.apache.iotdb.commons.conf.CommonDescriptor; + +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.Collections; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +public class DiskCheckerTest { + + @Rule public TemporaryFolder tmp = new TemporaryFolder(); + + private NodeStatus savedStatus; + private String savedReason; + + @Before + public void setUp() { + CommonConfig config = CommonDescriptor.getInstance().getConfig(); + savedStatus = config.getNodeStatus(); + savedReason = config.getStatusReason(); + config.setNodeStatus(NodeStatus.Running); + config.setStatusReason(null); + } + + @After + public void tearDown() { + CommonConfig config = CommonDescriptor.getInstance().getConfig(); + config.setNodeStatus(savedStatus); + config.setStatusReason(savedReason); + } + + @Test + public void checkReturnsNormalForWritableDirectoryWithSpace() throws Exception { + File dir = tmp.newFolder(); + // threshold 0.0 → any positive usable space passes + assertEquals( + DiskChecker.DiskStatus.NORMAL, + DiskChecker.check(Collections.singletonList(dir.getAbsolutePath()), 0.0)); + } + + @Test + public void checkReturnsDiskFullWhenBelowThreshold() throws Exception { + File dir = tmp.newFolder(); + // threshold > 1 forces every directory to be reported as full + assertEquals( + DiskChecker.DiskStatus.DISK_FULL, + DiskChecker.check(Collections.singletonList(dir.getAbsolutePath()), 2.0)); + } + + @Test + public void checkReturnsDiskCrashWhenDirectoryMissing() { + String missing = new File(tmp.getRoot(), "does-not-exist").getAbsolutePath(); + assertEquals( + DiskChecker.DiskStatus.DISK_CRASH, + DiskChecker.check(Collections.singletonList(missing), 0.0)); + } + + @Test + public void checkReturnsDiskCrashWhenPathIsAFile() throws Exception { + File file = tmp.newFile(); + assertEquals( + DiskChecker.DiskStatus.DISK_CRASH, + DiskChecker.check(Collections.singletonList(file.getAbsolutePath()), 0.0)); + } + + @Test + public void checkPrioritizesCrashOverFull() throws Exception { + File healthy = tmp.newFolder(); + String missing = new File(tmp.getRoot(), "missing").getAbsolutePath(); + // Even with a "full" threshold the missing dir trumps it. + assertEquals( + DiskChecker.DiskStatus.DISK_CRASH, + DiskChecker.check(java.util.Arrays.asList(healthy.getAbsolutePath(), missing), 2.0)); + } + + @Test + public void checkSkipsNullAndEmptyEntries() throws Exception { + File dir = tmp.newFolder(); + assertEquals( + DiskChecker.DiskStatus.NORMAL, + DiskChecker.check(java.util.Arrays.asList(null, "", dir.getAbsolutePath()), 0.0)); + } + + @Test + public void applyDiskFullSetsReadOnlyFromRunning() { + DiskChecker.apply(DiskChecker.DiskStatus.DISK_FULL); + CommonConfig config = CommonDescriptor.getInstance().getConfig(); + assertEquals(NodeStatus.ReadOnly, config.getNodeStatus()); + assertEquals(NodeStatus.DISK_FULL, config.getStatusReason()); + } + + @Test + public void applyDiskCrashSetsReadOnlyFromRunning() { + DiskChecker.apply(DiskChecker.DiskStatus.DISK_CRASH); + CommonConfig config = CommonDescriptor.getInstance().getConfig(); + assertEquals(NodeStatus.ReadOnly, config.getNodeStatus()); + assertEquals(NodeStatus.DISK_CRASH, config.getStatusReason()); + } + + @Test + public void applyDiskCrashUpgradesFromDiskFull() { + DiskChecker.apply(DiskChecker.DiskStatus.DISK_FULL); + DiskChecker.apply(DiskChecker.DiskStatus.DISK_CRASH); + CommonConfig config = CommonDescriptor.getInstance().getConfig(); + assertEquals(NodeStatus.ReadOnly, config.getNodeStatus()); + assertEquals(NodeStatus.DISK_CRASH, config.getStatusReason()); + } + + @Test + public void applyDiskFullDoesNotDowngradeDiskCrash() { + DiskChecker.apply(DiskChecker.DiskStatus.DISK_CRASH); + DiskChecker.apply(DiskChecker.DiskStatus.DISK_FULL); + CommonConfig config = CommonDescriptor.getInstance().getConfig(); + assertEquals(NodeStatus.ReadOnly, config.getNodeStatus()); + assertEquals( + "DiskCrash must outrank DiskFull", NodeStatus.DISK_CRASH, config.getStatusReason()); + } + + @Test + public void applyNormalRecoversFromDiskFull() { + DiskChecker.apply(DiskChecker.DiskStatus.DISK_FULL); + DiskChecker.apply(DiskChecker.DiskStatus.NORMAL); + CommonConfig config = CommonDescriptor.getInstance().getConfig(); + assertEquals(NodeStatus.Running, config.getNodeStatus()); + assertNull(config.getStatusReason()); + } + + @Test + public void applyNormalRecoversFromDiskCrash() { + DiskChecker.apply(DiskChecker.DiskStatus.DISK_CRASH); + DiskChecker.apply(DiskChecker.DiskStatus.NORMAL); + CommonConfig config = CommonDescriptor.getInstance().getConfig(); + assertEquals(NodeStatus.Running, config.getNodeStatus()); + assertNull(config.getStatusReason()); + } + + @Test + public void applyLeavesNonDiskReadOnlyReasonUntouched() { + CommonConfig config = CommonDescriptor.getInstance().getConfig(); + config.setNodeStatus(NodeStatus.ReadOnly); + config.setStatusReason("ManualMaintenance"); + + DiskChecker.apply(DiskChecker.DiskStatus.NORMAL); + assertEquals(NodeStatus.ReadOnly, config.getNodeStatus()); + assertEquals("ManualMaintenance", config.getStatusReason()); + + DiskChecker.apply(DiskChecker.DiskStatus.DISK_FULL); + // DISK_FULL only fires when not already DiskFull/DiskCrash — it does take over here, + // mirroring the existing behavior for the legacy sampleDiskLoad path. + assertEquals(NodeStatus.ReadOnly, config.getNodeStatus()); + assertEquals(NodeStatus.DISK_FULL, config.getStatusReason()); + } + + @Test + public void applyIsIdempotentForRepeatedDiskCrash() { + DiskChecker.apply(DiskChecker.DiskStatus.DISK_CRASH); + NodeStatus before = CommonDescriptor.getInstance().getConfig().getNodeStatus(); + String reasonBefore = CommonDescriptor.getInstance().getConfig().getStatusReason(); + DiskChecker.apply(DiskChecker.DiskStatus.DISK_CRASH); + assertEquals(before, CommonDescriptor.getInstance().getConfig().getNodeStatus()); + assertEquals(reasonBefore, CommonDescriptor.getInstance().getConfig().getStatusReason()); + } + + @Test + public void checkAndApplyDrivesStatusEndToEnd() throws Exception { + File healthy = tmp.newFolder(); + DiskChecker.checkAndApply(Collections.singletonList(healthy.getAbsolutePath()), 0.0); + assertEquals(NodeStatus.Running, CommonDescriptor.getInstance().getConfig().getNodeStatus()); + + String missing = new File(tmp.getRoot(), "still-missing").getAbsolutePath(); + DiskChecker.checkAndApply(Collections.singletonList(missing), 0.0); + assertEquals(NodeStatus.ReadOnly, CommonDescriptor.getInstance().getConfig().getNodeStatus()); + assertEquals( + NodeStatus.DISK_CRASH, CommonDescriptor.getInstance().getConfig().getStatusReason()); + + DiskChecker.checkAndApply(Collections.singletonList(healthy.getAbsolutePath()), 0.0); + assertEquals(NodeStatus.Running, CommonDescriptor.getInstance().getConfig().getNodeStatus()); + assertNull(CommonDescriptor.getInstance().getConfig().getStatusReason()); + } + + @Test + public void emptyDirListIsNormal() { + assertEquals(DiskChecker.DiskStatus.NORMAL, DiskChecker.check(Collections.emptyList(), 1.0)); + assertEquals(DiskChecker.DiskStatus.NORMAL, DiskChecker.check(null, 1.0)); + } + + @Test + public void smokeProbeFileIsDeleted() throws Exception { + File dir = tmp.newFolder(); + DiskChecker.check(Collections.singletonList(dir.getAbsolutePath()), 0.0); + File[] leftovers = dir.listFiles(); + assertTrue( + "Disk probe should clean up its temp file", leftovers == null || leftovers.length == 0); + } +} diff --git a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift index 1111740f759a..a8c3341e5976 100644 --- a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift +++ b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift @@ -543,6 +543,13 @@ struct TConfigNodeHeartbeatResp { 1: required i64 timestamp 2: optional string activateStatus 3: optional common.TLicense license + // Reported ConfigNode status (e.g. Running, ReadOnly). Optional for forward + // compatibility — old ConfigNodes do not populate this field, in which case + // the leader falls back to assuming Running. + 4: optional string status + // Optional human/machine readable reason accompanying ReadOnly status, + // e.g. NodeStatus.DISK_FULL or NodeStatus.DISK_CRASH. + 5: optional string statusReason } struct TAddConsensusGroupReq { From d783e8b1a245ba0bac5ea889ff64913a36f2a50e Mon Sep 17 00:00:00 2001 From: Yongzao <532741407@qq.com> Date: Wed, 20 May 2026 16:51:41 +0800 Subject: [PATCH 2/5] Drive ConfigNode ReadOnly into Ratis: step down, demote priority, passive crash detect Three changes that let ReadOnly state actually shape Raft behavior on ConfigNode: - Utils.rejectWrite / stallApply now match ConfigRegion in addition to DataRegion, so a ReadOnly ConfigNode leader hits the same forceStepDownLeader path that DataRegion leaders already use. Comment at RatisConsensus.write updated. - New NodeStatus.priorityForStatus maps Running=0, ReadOnly(DiskFull)=-1, ReadOnly(DiskCrash)=-2. HeartbeatService runs a reconciliation step on the leader (same cadence as the load-sampling pass, after async fanout) that pushes each ConfigNode peer's desired priority into Ratis. Unknown/Removing/manual ReadOnly are left empty so transient blips do not churn the group config. IConsensus gains a default-no-op reconfigurePeerPriorities; RatisConsensus overrides it to rebuild the peer list and call sendReconfiguration. - Replace DiskChecker.check (active testWrite probe) with a passive observer threaded through Ratis. ApplicationStateMachineProxy gains a diskFailureListener parameter and fires it from the applyTransaction catch when Utils.isDiskFailure matches the cause (IOError / FileSystemException). RatisConsensus also tags IOException out of writeLocallyWithRetry / writeRemotelyWithRetry so log-write failures register as DiskCrash. DiskChecker keeps only checkFreeRatio (for the DiskFull path) and apply (for the state machine); DiskCrash is now sticky on both DataNode and ConfigNode until restart. DiskCheckerTest trimmed to drop testWrite-specific cases and to assert that NORMAL no longer recovers DiskCrash; 14 cases pass. --- .../confignode/i18n/ManagerMessages.java | 2 + .../confignode/i18n/ManagerMessages.java | 2 + .../manager/load/cache/LoadCache.java | 10 ++ .../load/cache/node/BaseNodeCache.java | 7 ++ .../load/service/HeartbeatService.java | 57 ++++++++++-- .../thrift/ConfigNodeRPCServiceProcessor.java | 10 +- .../apache/iotdb/consensus/IConsensus.java | 18 ++++ .../ratis/ApplicationStateMachineProxy.java | 10 +- .../iotdb/consensus/ratis/RatisConsensus.java | 65 ++++++++++++- .../iotdb/consensus/ratis/utils/Utils.java | 30 +++++- .../iotdb/commons/i18n/CommonMessages.java | 2 - .../iotdb/commons/i18n/CommonMessages.java | 2 - .../iotdb/commons/cluster/DiskChecker.java | 70 ++++++-------- .../iotdb/commons/cluster/NodeStatus.java | 31 +++++++ .../commons/cluster/DiskCheckerTest.java | 93 ++++++++----------- 15 files changed, 292 insertions(+), 117 deletions(-) diff --git a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java index 10ea4763af9a..a0541ca7c465 100644 --- a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java +++ b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java @@ -239,6 +239,8 @@ public final class ManagerMessages { "Heartbeat service is started successfully."; public static final String HEARTBEAT_SERVICE_IS_STOPPED_SUCCESSFULLY = "Heartbeat service is stopped successfully."; + public static final String RECONFIGURE_PEER_PRIORITIES_FAILED = + "Failed to reconfigure ConfigNode peer priorities to {}."; public static final String INCORRECT_VERSION_OF = "Incorrect version of "; public static final String INIT_CONSENSUSMANAGER_SUCCESSFULLY_WHEN_RESTARTED = "Init ConsensusManager successfully when restarted"; diff --git a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java index 499e922f7a7d..c7353f9c0f4d 100644 --- a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java +++ b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java @@ -237,6 +237,8 @@ public final class ManagerMessages { "心跳服务已成功启动。"; public static final String HEARTBEAT_SERVICE_IS_STOPPED_SUCCESSFULLY = "心跳服务已成功停止。"; + public static final String RECONFIGURE_PEER_PRIORITIES_FAILED = + "重新配置 ConfigNode 节点优先级 {} 失败。"; public static final String INCORRECT_VERSION_OF = "版本不正确："; public static final String INIT_CONSENSUSMANAGER_SUCCESSFULLY_WHEN_RESTARTED = "重启时成功初始化 ConsensusManager"; diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/LoadCache.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/LoadCache.java index 3416246811ed..dcd12145dad3 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/LoadCache.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/LoadCache.java @@ -547,6 +547,16 @@ public String getNodeStatusWithReason(int nodeId) { .orElseGet(() -> NodeStatus.Unknown.getStatus() + "(NoHeartbeat)"); } + /** + * @return The raw {@code statusReason} string for {@code nodeId}, or {@code null} when no cache + * entry exists yet or no reason has been reported. + */ + public String getNodeStatusReason(int nodeId) { + return Optional.ofNullable(nodeCacheMap.get(nodeId)) + .map(BaseNodeCache::getStatusReason) + .orElse(null); + } + /** * Get all Node's current status with reason. * diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/BaseNodeCache.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/BaseNodeCache.java index 4eeb96344ccd..434d5bbf3fe2 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/BaseNodeCache.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/BaseNodeCache.java @@ -66,4 +66,11 @@ public String getNodeStatusWithReason() { ? statistics.getStatus().getStatus() : statistics.getStatus().getStatus() + "(" + statistics.getStatusReason() + ")"; } + + /** + * @return The raw reason string (may be {@code null}) accompanying the current NodeStatus. + */ + public String getStatusReason() { + return ((NodeStatistics) currentStatistics.get()).getStatusReason(); + } } 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 be2ae1982cf2..312e435904b9 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 @@ -25,6 +25,7 @@ import org.apache.iotdb.common.rpc.thrift.TDataNodeConfiguration; import org.apache.iotdb.common.rpc.thrift.TEndPoint; import org.apache.iotdb.commons.cluster.DiskChecker; +import org.apache.iotdb.commons.cluster.NodeStatus; import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory; import org.apache.iotdb.commons.concurrent.ThreadName; import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil; @@ -45,6 +46,7 @@ import org.apache.iotdb.confignode.manager.load.cache.node.ConfigNodeHeartbeatCache; import org.apache.iotdb.confignode.manager.node.NodeManager; import org.apache.iotdb.confignode.rpc.thrift.TConfigNodeHeartbeatReq; +import org.apache.iotdb.consensus.exception.ConsensusException; import org.apache.iotdb.db.protocol.client.ConfigNodeInfo; import org.apache.iotdb.mpp.rpc.thrift.TDataNodeHeartbeatReq; @@ -52,7 +54,9 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.Optional; import java.util.Set; import java.util.concurrent.Future; @@ -144,12 +148,6 @@ private void heartbeatLoopBody() { .ifPresent( consensusManager -> { if (getConsensusManager().isLeader()) { - // Leader self-checks its own disk health before fanning out heartbeats. - // Followers run the same check when receiving each heartbeat request - // (see ConfigNodeRPCServiceProcessor#getConfigNodeHeartBeat). - DiskChecker.checkAndApply( - ConfigNodeDescriptor.getInstance().getConf().getCriticalDirs(), - CommonDescriptor.getInstance().getConfig().getDiskSpaceWarningThreshold()); // Send heartbeat requests to all the registered ConfigNodes pingRegisteredConfigNodes( genConfigNodeHeartbeatReq(), getNodeManager().getRegisteredConfigNodes()); @@ -158,10 +156,57 @@ private void heartbeatLoopBody() { genHeartbeatReq(), getNodeManager().getRegisteredDataNodes()); // Send heartbeat requests to all the registered AINodes pingRegisteredAINodes(genAIHeartbeatReq(), getNodeManager().getRegisteredAINodes()); + // Sample free-space on the same cadence DataNode samples its load. Runs after + // the async heartbeat dispatches so the OS call does not delay fanout. DiskCrash + // is observed passively by the Ratis write-path, not polled here. + if (iterationIndex % LOAD_SAMPLING_INTERVAL == 0) { + DiskChecker.checkFreeRatioAndApply( + ConfigNodeDescriptor.getInstance().getConf().getCriticalDirs(), + CommonDescriptor.getInstance().getConfig().getDiskSpaceWarningThreshold()); + reconcileConfigNodePeerPriorities(); + } } }); } + /** + * Push Ratis peer priorities to reflect each ConfigNode's current {@link NodeStatus} (see {@link + * NodeStatus#priorityForStatus}). Only fires on the leader; only acts when at least one peer's + * desired priority differs from the live group configuration. {@code Unknown}/{@code Removing} + * peers and non-disk {@code ReadOnly} reasons are left untouched so transient blips and + * operator-driven states cannot churn the group config. + */ + private void reconcileConfigNodePeerPriorities() { + Map desired = new HashMap<>(); + for (TConfigNodeLocation peer : getNodeManager().getRegisteredConfigNodes()) { + int nodeId = peer.getConfigNodeId(); + NodeStatus status; + String reason; + if (nodeId == ConfigNodeHeartbeatCache.CURRENT_NODE_ID) { + // The leader's own cache entry mirrors CommonConfig; reading from CommonConfig directly + // sidesteps any ordering between the local disk-check and the next cache refresh. + status = CommonDescriptor.getInstance().getConfig().getNodeStatus(); + reason = CommonDescriptor.getInstance().getConfig().getStatusReason(); + } else { + status = loadCache.getNodeStatus(nodeId); + reason = loadCache.getNodeStatusReason(nodeId); + } + NodeStatus.priorityForStatus(status, reason) + .ifPresent(priority -> desired.put(nodeId, priority)); + } + if (desired.isEmpty()) { + return; + } + try { + configManager + .getConsensusManager() + .getConsensusImpl() + .reconfigurePeerPriorities(ConsensusManager.DEFAULT_CONSENSUS_GROUP_ID, desired); + } catch (ConsensusException e) { + LOGGER.warn(ManagerMessages.RECONFIGURE_PEER_PRIORITIES_FAILED, desired, e); + } + } + protected TDataNodeHeartbeatReq genHeartbeatReq() { /* Generate heartbeat request */ TDataNodeHeartbeatReq heartbeatReq = new TDataNodeHeartbeatReq(); diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java index 92e3221ddf13..7ed189206acd 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java @@ -1115,10 +1115,12 @@ public TRegionRouteMapResp getLatestRegionRouteMap() { public TConfigNodeHeartbeatResp getConfigNodeHeartBeat(TConfigNodeHeartbeatReq heartbeatReq) { TConfigNodeHeartbeatResp resp = new TConfigNodeHeartbeatResp(); resp.setTimestamp(heartbeatReq.getTimestamp()); - // Follower self-check: probe critical dirs each time the leader pings us. - // The leader runs the same check in its HeartbeatService loop. - DiskChecker.checkAndApply( - configNodeConfig.getCriticalDirs(), commonConfig.getDiskSpaceWarningThreshold()); + // Sample free-space on the same cadence DataNode samples its load. DiskCrash is observed + // passively from the Ratis write-path on this node, not polled here. + if (heartbeatReceivedCounter.getAndIncrement() % HeartbeatService.LOAD_SAMPLING_INTERVAL == 0) { + DiskChecker.checkFreeRatioAndApply( + configNodeConfig.getCriticalDirs(), commonConfig.getDiskSpaceWarningThreshold()); + } resp.setStatus(commonConfig.getNodeStatus().getStatus()); if (commonConfig.getStatusReason() != null) { resp.setStatusReason(commonConfig.getStatusReason()); diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/IConsensus.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/IConsensus.java index c462aa3a046e..20b2af0d1ba2 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/IConsensus.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/IConsensus.java @@ -167,6 +167,24 @@ public interface IConsensus { */ void resetPeerList(ConsensusGroupId groupId, List correctPeers) throws ConsensusException; + /** + * Adjust the leader-election priority of one or more peers in {@code groupId}. + * + *

Implementations that have no concept of per-peer priority (e.g. Simple, IoT consensus) + * should leave the default no-op. The Ratis implementation rewrites the group configuration so a + * peer with a lower priority is less likely to win subsequent elections — this is the lever the + * cluster uses to demote a {@code ReadOnly} ConfigNode. + * + * @param groupId the consensus group whose peer priorities should be updated + * @param nodeIdToPriority desired priorities keyed by node id; peers not in the map keep their + * current priority + * @throws ConsensusException if the underlying reconfiguration fails + */ + default void reconfigurePeerPriorities( + ConsensusGroupId groupId, Map nodeIdToPriority) throws ConsensusException { + // no-op default + } + // management API /** diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/ApplicationStateMachineProxy.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/ApplicationStateMachineProxy.java index 67b1fdf7da90..c18129f93ac5 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/ApplicationStateMachineProxy.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/ApplicationStateMachineProxy.java @@ -70,17 +70,20 @@ public class ApplicationStateMachineProxy extends BaseStateMachine { private final RaftGroupId groupId; private final TConsensusGroupType consensusGroupType; private final BiConsumer leaderChangeListener; + private final BiConsumer diskFailureListener; ApplicationStateMachineProxy(IStateMachine stateMachine, RaftGroupId id) { - this(stateMachine, id, null); + this(stateMachine, id, null, null); } ApplicationStateMachineProxy( IStateMachine stateMachine, RaftGroupId id, - BiConsumer onLeaderChanged) { + BiConsumer onLeaderChanged, + BiConsumer onDiskFailure) { this.applicationStateMachine = stateMachine; this.leaderChangeListener = onLeaderChanged; + this.diskFailureListener = onDiskFailure; this.groupId = id; snapshotStorage = new SnapshotStorage(applicationStateMachine, groupId); consensusGroupType = Utils.getConsensusGroupTypeFromPrefix(groupId.toString()); @@ -161,6 +164,9 @@ && waitBeforeRetry()) { break; } catch (Throwable rte) { logger.error(RatisMessages.STATEMACHINE_RUNTIME_EXCEPTION, rte); + if (Utils.isDiskFailure(rte) && diskFailureListener != null) { + diskFailureListener.accept(groupId, rte); + } ret = new ResponseMessage( new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode()) diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java index b7369546e7cd..4711726bcaec 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java @@ -28,6 +28,7 @@ import org.apache.iotdb.commons.client.IClientPoolFactory; import org.apache.iotdb.commons.client.exception.ClientManagerException; import org.apache.iotdb.commons.client.property.ClientPoolProperty; +import org.apache.iotdb.commons.cluster.DiskChecker; import org.apache.iotdb.commons.consensus.ConsensusGroupId; import org.apache.iotdb.commons.request.IConsensusRequest; import org.apache.iotdb.commons.service.metric.MetricService; @@ -234,7 +235,8 @@ public RatisConsensus(ConsensusConfig config, IStateMachine.Registry registry) { registry.apply( Utils.fromRaftGroupIdToConsensusGroupId(raftGroupId)), raftGroupId, - this::onLeaderChanged)) + this::onLeaderChanged, + this::onDiskFailure)) .build()); } @@ -329,9 +331,9 @@ public TSStatus write(ConsensusGroupId groupId, IConsensusRequest request) throw new ConsensusGroupNotExistException(groupId); } - // current Peer is group leader and in ReadOnly State - // We only judge dataRegions here, because schema write when readOnly is handled at - // RegionWriteExecutor + // Current peer is group leader and in ReadOnly state. SchemaRegion writes are gated at + // RegionWriteExecutor; here we step down DataRegion and ConfigRegion leaders so that a + // healthy peer can take over and continue serving writes. if (isLeader(groupId) && Utils.rejectWrite(consensusGroupType)) { try { forceStepDownLeader(raftGroup); @@ -369,6 +371,7 @@ && waitUntilLeaderReady(raftGroupId)) { } catch (GroupMismatchException e) { throw new ConsensusGroupNotExistException(groupId); } catch (Exception e) { + maybeReportDiskFailure(raftGroupId, e); throw new RatisRequestFailedException(e); } } @@ -386,6 +389,7 @@ && waitUntilLeaderReady(raftGroupId)) { } catch (GroupMismatchException e) { throw new ConsensusGroupNotExistException(groupId); } catch (Exception e) { + maybeReportDiskFailure(raftGroupId, e); throw new RatisRequestFailedException(e); } @@ -965,6 +969,39 @@ private RatisClient getRaftClient(RaftGroup group) throws ClientManagerException } } + @Override + public void reconfigurePeerPriorities( + ConsensusGroupId groupId, Map nodeIdToPriority) throws ConsensusException { + if (nodeIdToPriority == null || nodeIdToPriority.isEmpty()) { + return; + } + RaftGroupId raftGroupId = Utils.fromConsensusGroupIdToRaftGroupId(groupId); + RaftGroup group = getGroupInfo(raftGroupId); + if (group == null || !group.getPeers().contains(myself)) { + throw new ConsensusGroupNotExistException(groupId); + } + boolean changed = false; + List newPeers = new ArrayList<>(group.getPeers().size()); + for (RaftPeer p : group.getPeers()) { + Integer desired = nodeIdToPriority.get(Utils.fromRaftPeerIdToNodeId(p.getId())); + if (desired == null || desired == p.getPriority()) { + newPeers.add(p); + continue; + } + newPeers.add( + RaftPeer.newBuilder() + .setId(p.getId()) + .setAddress(p.getAddress()) + .setPriority(desired) + .build()); + changed = true; + } + if (!changed) { + return; + } + sendReconfiguration(RaftGroup.valueOf(raftGroupId, newPeers)); + } + private RatisClient getConfigurationRaftClient(RaftGroup group) throws ClientManagerException { try { return reconfigurationClientManager.borrowClient(group); @@ -991,6 +1028,26 @@ private RaftClientReply sendReconfiguration(RaftGroup newGroupConf) return reply; } + /** + * Called from the state machine apply path and from the write-path catch blocks when a Ratis + * operation surfaces a disk-level error (see {@link Utils#isDiskFailure(Throwable)}). Drives the + * node to {@code ReadOnly(DiskCrash)} via {@link DiskChecker#apply}, which encapsulates the + * priority rules (DiskCrash wins over DiskFull, etc.). + */ + private void onDiskFailure(RaftGroupId groupId, Throwable cause) { + logger.error( + "Disk failure observed in Ratis group {}; marking node ReadOnly(DiskCrash).", + groupId, + cause); + DiskChecker.apply(DiskChecker.DiskStatus.DISK_CRASH); + } + + private void maybeReportDiskFailure(RaftGroupId groupId, Throwable cause) { + if (Utils.isDiskFailure(cause)) { + onDiskFailure(groupId, cause); + } + } + private void onLeaderChanged(RaftGroupMemberId groupMemberId, RaftPeerId leaderId) { Optional.ofNullable(canServeStaleRead) .ifPresent( diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/utils/Utils.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/utils/Utils.java index 7b052cebf979..1102e6ecb329 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/utils/Utils.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/utils/Utils.java @@ -238,7 +238,10 @@ public static TConsensusGroupType getConsensusGroupTypeFromPrefix(String prefix) } public static boolean rejectWrite(TConsensusGroupType type) { - return type == TConsensusGroupType.DataRegion && config.isReadOnly(); + // SchemaRegion writes are gated by RegionWriteExecutor at a higher level, so we only need + // to short-circuit DataRegion and ConfigRegion Ratis writes here. + return (type == TConsensusGroupType.DataRegion || type == TConsensusGroupType.ConfigRegion) + && config.isReadOnly(); } /** @@ -248,7 +251,30 @@ public static boolean rejectWrite(TConsensusGroupType type) { * still allow statemachine to apply while rejecting new client write requests. */ public static boolean stallApply(TConsensusGroupType type) { - return type == TConsensusGroupType.DataRegion && config.isReadOnly() && !config.isStopping(); + return (type == TConsensusGroupType.DataRegion || type == TConsensusGroupType.ConfigRegion) + && config.isReadOnly() + && !config.isStopping(); + } + + /** + * Treat a throwable (including any wrapped cause) as a disk-level failure when it carries an + * {@link java.io.IOError} or a {@link java.nio.file.FileSystemException}. Both are concrete + * filesystem-layer signals: {@code IOError} is thrown by the JVM for unrecoverable storage + * errors, and {@code FileSystemException} surfaces I/O syscalls failing on a specific file. Plain + * {@link java.io.IOException} is intentionally excluded — it is also raised by network code and + * would otherwise misclassify transient connectivity failures as disk crashes. + */ + public static boolean isDiskFailure(Throwable t) { + for (Throwable cur = t; cur != null; cur = cur.getCause()) { + if (cur instanceof java.io.IOError || cur instanceof java.nio.file.FileSystemException) { + return true; + } + // Guard against pathological self-referencing cause chains + if (cur.getCause() == cur) { + break; + } + } + return false; } /** return the max wait duration for retry */ diff --git a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java index 510ef3b89db9..cacd8ac15f25 100644 --- a/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/en/org/apache/iotdb/commons/i18n/CommonMessages.java @@ -42,8 +42,6 @@ public final class CommonMessages { "Free disk space ratio is below the configured threshold; set node status to ReadOnly(DiskFull)."; public static final String DISK_CRASH_SET_READ_ONLY = "Detected unwritable disk directory; set node status to ReadOnly(DiskCrash)."; - public static final String DISK_CRASH_PROBE_FAILED = - "Disk health probe write failed for directory {}."; public static final String DISK_RECOVERED_SET_RUNNING = "Disk health recovered (previous reason: {}); set node status to Running."; diff --git a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java index 8414a8eb916c..33f610e769fc 100644 --- a/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java +++ b/iotdb-core/node-commons/src/main/i18n/zh/org/apache/iotdb/commons/i18n/CommonMessages.java @@ -41,8 +41,6 @@ public final class CommonMessages { "磁盘剩余空间比例低于配置阈值，将节点状态设为 ReadOnly(DiskFull)。"; public static final String DISK_CRASH_SET_READ_ONLY = "检测到不可写的磁盘目录，将节点状态设为 ReadOnly(DiskCrash)。"; - public static final String DISK_CRASH_PROBE_FAILED = - "对目录 {} 进行磁盘健康探测时写入失败。"; public static final String DISK_RECOVERED_SET_RUNNING = "磁盘健康已恢复（先前原因：{}），将节点状态设为 Running。"; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/cluster/DiskChecker.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/cluster/DiskChecker.java index 4632ddf28dd5..682f9ad4a97a 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/cluster/DiskChecker.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/cluster/DiskChecker.java @@ -27,15 +27,21 @@ import org.slf4j.LoggerFactory; import java.io.File; -import java.io.IOException; -import java.nio.file.Files; -import java.nio.file.Path; -import java.nio.file.Paths; import java.util.List; /** - * Shared utility used by both ConfigNode and DataNode to evaluate the health of a set of critical - * directories and, optionally, drive the global {@link NodeStatus} on {@link CommonConfig}. + * Shared utility that drives the global {@link NodeStatus} on {@link CommonConfig} based on disk + * health signals. + * + *

Two sources feed it: + * + *

    + *
  • {@link #checkFreeRatioAndApply} polls usable / total space across critical directories and + * sets {@code ReadOnly(DISK_FULL)} when the ratio drops below a threshold. + *
  • {@link #apply}{@code (DISK_CRASH)} is called from passive failure observers — Ratis + * write-path catch blocks on ConfigNode and {@code FolderManager.ABNORMAL} aggregation on + * DataNode — when a real write IO error has already occurred. + *
* *

Only transitions between {@link NodeStatus#Running} and {@link NodeStatus#ReadOnly} with * reason {@link NodeStatus#DISK_FULL}/{@link NodeStatus#DISK_CRASH} are managed here. Other @@ -45,10 +51,6 @@ public class DiskChecker { private static final Logger LOGGER = LoggerFactory.getLogger(DiskChecker.class); - private static final byte[] PROBE_PAYLOAD = new byte[] {0x01}; - private static final String PROBE_PREFIX = "iotdb-disk-probe-"; - private static final String PROBE_SUFFIX = ".tmp"; - public enum DiskStatus { NORMAL, DISK_FULL, @@ -58,53 +60,39 @@ public enum DiskStatus { private DiskChecker() {} /** - * Evaluate the supplied directories. A single unwritable directory yields {@link - * DiskStatus#DISK_CRASH}; any directory whose usable/total ratio is below the threshold yields - * {@link DiskStatus#DISK_FULL} (when no crash is detected); otherwise {@link DiskStatus#NORMAL}. + * Evaluate the usable/total space ratio of each directory. Any directory whose ratio is below + * {@code freeRatioThreshold} yields {@link DiskStatus#DISK_FULL}; otherwise {@link + * DiskStatus#NORMAL}. This method never returns {@link DiskStatus#DISK_CRASH} — crash detection + * is driven by the Ratis write-path observer on ConfigNode and by {@code FolderManager} on + * DataNode, both of which call {@link #apply} directly. */ - public static DiskStatus check(List dirs, double freeRatioThreshold) { + public static DiskStatus checkFreeRatio(List dirs, double freeRatioThreshold) { if (dirs == null || dirs.isEmpty()) { return DiskStatus.NORMAL; } - boolean anyFull = false; for (String dir : dirs) { if (dir == null || dir.isEmpty()) { continue; } File f = new File(dir); - if (!f.isDirectory()) { - LOGGER.warn(CommonMessages.DISK_CRASH_PROBE_FAILED, dir); - return DiskStatus.DISK_CRASH; - } - try { - Path probe = Files.createTempFile(Paths.get(dir), PROBE_PREFIX, PROBE_SUFFIX); - try { - Files.write(probe, PROBE_PAYLOAD); - } finally { - Files.deleteIfExists(probe); - } - } catch (IOException e) { - LOGGER.warn(CommonMessages.DISK_CRASH_PROBE_FAILED, dir, e); - return DiskStatus.DISK_CRASH; - } long total = f.getTotalSpace(); long usable = f.getUsableSpace(); if (total > 0 && (double) usable / total < freeRatioThreshold) { - anyFull = true; + return DiskStatus.DISK_FULL; } } - return anyFull ? DiskStatus.DISK_FULL : DiskStatus.NORMAL; + return DiskStatus.NORMAL; } - /** - * Run {@link #check} and apply the result to {@link CommonConfig}. See class javadoc for - * transition rules. - */ - public static void checkAndApply(List dirs, double freeRatioThreshold) { - apply(check(dirs, freeRatioThreshold)); + /** Convenience: run {@link #checkFreeRatio} and apply the result to {@link CommonConfig}. */ + public static void checkFreeRatioAndApply(List dirs, double freeRatioThreshold) { + apply(checkFreeRatio(dirs, freeRatioThreshold)); } - /** Visible for tests; package-public callers should prefer {@link #checkAndApply}. */ + /** + * Apply a precomputed status to {@link CommonConfig}. Priority is {@code DiskCrash > DiskFull > + * Normal}; recovery to {@code Running} only fires when the active reason was disk-related. + */ public static void apply(DiskStatus result) { CommonConfig config = CommonDescriptor.getInstance().getConfig(); NodeStatus currentStatus = config.getNodeStatus(); @@ -135,7 +123,9 @@ public static void apply(DiskStatus result) { break; case NORMAL: default: - if (currentlyFull || currentlyCrash) { + // DiskCrash is sticky — only a restart clears it. The free-ratio probe can recover + // DiskFull alone because free-space reappearing is the literal inverse of running low. + if (currentlyFull) { LOGGER.info(CommonMessages.DISK_RECOVERED_SET_RUNNING, currentReason); config.setNodeStatus(NodeStatus.Running); config.setStatusReason(null); diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/cluster/NodeStatus.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/cluster/NodeStatus.java index e37b44edafda..ff64a666f98e 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/cluster/NodeStatus.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/cluster/NodeStatus.java @@ -21,6 +21,8 @@ import org.apache.iotdb.commons.i18n.CommonMessages; +import java.util.OptionalInt; + /** Node status for showing cluster */ public enum NodeStatus { /** Node running properly */ @@ -61,6 +63,35 @@ public static boolean isNormalStatus(NodeStatus status) { return status != null && status.equals(NodeStatus.Running); } + /** + * Map a node's {@link NodeStatus} (and optional reason) to the Ratis peer priority that should + * govern its candidacy in leader elections. + * + * 

+   *   Running                          →   0   (full candidate)
+   *   ReadOnly + {@link #DISK_FULL}    →  -1   (out-rank healthy peers but ahead of crashed)
+   *   ReadOnly + {@link #DISK_CRASH}   →  -2   (most degraded — last choice)
+   *   anything else                    →  empty (priority must not be changed)
+   *
+ * + * Returning {@link OptionalInt#empty()} for Unknown/Removing/manual ReadOnly keeps transient + * blips and operator-driven states from rewriting peer priorities. + */ + public static OptionalInt priorityForStatus(NodeStatus status, String statusReason) { + if (Running.equals(status)) { + return OptionalInt.of(0); + } + if (ReadOnly.equals(status)) { + if (DISK_CRASH.equals(statusReason)) { + return OptionalInt.of(-2); + } + if (DISK_FULL.equals(statusReason)) { + return OptionalInt.of(-1); + } + } + return OptionalInt.empty(); + } + public static boolean isReadable(NodeStatus status) { switch (status) { case Running: diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/cluster/DiskCheckerTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/cluster/DiskCheckerTest.java index fde492c1090d..84284b8806a3 100644 --- a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/cluster/DiskCheckerTest.java +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/cluster/DiskCheckerTest.java @@ -29,11 +29,11 @@ import org.junit.rules.TemporaryFolder; import java.io.File; +import java.util.Arrays; import java.util.Collections; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNull; -import static org.junit.Assert.assertTrue; public class DiskCheckerTest { @@ -58,57 +58,41 @@ public void tearDown() { config.setStatusReason(savedReason); } + // -- checkFreeRatio ------------------------------------------------------------------------ + @Test - public void checkReturnsNormalForWritableDirectoryWithSpace() throws Exception { + public void checkFreeRatioReturnsNormalWhenRatioMeetsThreshold() throws Exception { File dir = tmp.newFolder(); - // threshold 0.0 → any positive usable space passes assertEquals( DiskChecker.DiskStatus.NORMAL, - DiskChecker.check(Collections.singletonList(dir.getAbsolutePath()), 0.0)); + DiskChecker.checkFreeRatio(Collections.singletonList(dir.getAbsolutePath()), 0.0)); } @Test - public void checkReturnsDiskFullWhenBelowThreshold() throws Exception { + public void checkFreeRatioReturnsDiskFullWhenAnyDirBelowThreshold() throws Exception { File dir = tmp.newFolder(); - // threshold > 1 forces every directory to be reported as full + // threshold above 1.0 forces any directory with positive total space to register as full assertEquals( DiskChecker.DiskStatus.DISK_FULL, - DiskChecker.check(Collections.singletonList(dir.getAbsolutePath()), 2.0)); + DiskChecker.checkFreeRatio(Collections.singletonList(dir.getAbsolutePath()), 2.0)); } @Test - public void checkReturnsDiskCrashWhenDirectoryMissing() { - String missing = new File(tmp.getRoot(), "does-not-exist").getAbsolutePath(); + public void checkFreeRatioSkipsNullAndEmptyEntries() throws Exception { + File dir = tmp.newFolder(); assertEquals( - DiskChecker.DiskStatus.DISK_CRASH, - DiskChecker.check(Collections.singletonList(missing), 0.0)); + DiskChecker.DiskStatus.NORMAL, + DiskChecker.checkFreeRatio(Arrays.asList(null, "", dir.getAbsolutePath()), 0.0)); } @Test - public void checkReturnsDiskCrashWhenPathIsAFile() throws Exception { - File file = tmp.newFile(); + public void emptyDirListIsNormal() { assertEquals( - DiskChecker.DiskStatus.DISK_CRASH, - DiskChecker.check(Collections.singletonList(file.getAbsolutePath()), 0.0)); + DiskChecker.DiskStatus.NORMAL, DiskChecker.checkFreeRatio(Collections.emptyList(), 1.0)); + assertEquals(DiskChecker.DiskStatus.NORMAL, DiskChecker.checkFreeRatio(null, 1.0)); } - @Test - public void checkPrioritizesCrashOverFull() throws Exception { - File healthy = tmp.newFolder(); - String missing = new File(tmp.getRoot(), "missing").getAbsolutePath(); - // Even with a "full" threshold the missing dir trumps it. - assertEquals( - DiskChecker.DiskStatus.DISK_CRASH, - DiskChecker.check(java.util.Arrays.asList(healthy.getAbsolutePath(), missing), 2.0)); - } - - @Test - public void checkSkipsNullAndEmptyEntries() throws Exception { - File dir = tmp.newFolder(); - assertEquals( - DiskChecker.DiskStatus.NORMAL, - DiskChecker.check(java.util.Arrays.asList(null, "", dir.getAbsolutePath()), 0.0)); - } + // -- apply() state machine ----------------------------------------------------------------- @Test public void applyDiskFullSetsReadOnlyFromRunning() { @@ -155,12 +139,14 @@ public void applyNormalRecoversFromDiskFull() { } @Test - public void applyNormalRecoversFromDiskCrash() { + public void applyNormalDoesNotRecoverFromDiskCrash() { DiskChecker.apply(DiskChecker.DiskStatus.DISK_CRASH); DiskChecker.apply(DiskChecker.DiskStatus.NORMAL); CommonConfig config = CommonDescriptor.getInstance().getConfig(); - assertEquals(NodeStatus.Running, config.getNodeStatus()); - assertNull(config.getStatusReason()); + // DiskCrash is sticky: a free-ratio probe (the only source of NORMAL) cannot prove writes + // work again, so the node stays ReadOnly(DiskCrash) until restart. + assertEquals(NodeStatus.ReadOnly, config.getNodeStatus()); + assertEquals(NodeStatus.DISK_CRASH, config.getStatusReason()); } @Test @@ -190,35 +176,32 @@ public void applyIsIdempotentForRepeatedDiskCrash() { assertEquals(reasonBefore, CommonDescriptor.getInstance().getConfig().getStatusReason()); } - @Test - public void checkAndApplyDrivesStatusEndToEnd() throws Exception { - File healthy = tmp.newFolder(); - DiskChecker.checkAndApply(Collections.singletonList(healthy.getAbsolutePath()), 0.0); - assertEquals(NodeStatus.Running, CommonDescriptor.getInstance().getConfig().getNodeStatus()); + // -- checkFreeRatioAndApply ---------------------------------------------------------------- - String missing = new File(tmp.getRoot(), "still-missing").getAbsolutePath(); - DiskChecker.checkAndApply(Collections.singletonList(missing), 0.0); + @Test + public void checkFreeRatioAndApplyDrivesStatusEndToEnd() throws Exception { + File dir = tmp.newFolder(); + // Threshold above 1.0 -> always "DiskFull" + DiskChecker.checkFreeRatioAndApply(Collections.singletonList(dir.getAbsolutePath()), 2.0); assertEquals(NodeStatus.ReadOnly, CommonDescriptor.getInstance().getConfig().getNodeStatus()); assertEquals( - NodeStatus.DISK_CRASH, CommonDescriptor.getInstance().getConfig().getStatusReason()); + NodeStatus.DISK_FULL, CommonDescriptor.getInstance().getConfig().getStatusReason()); - DiskChecker.checkAndApply(Collections.singletonList(healthy.getAbsolutePath()), 0.0); + // Threshold 0.0 -> always "Normal" -> recovery + DiskChecker.checkFreeRatioAndApply(Collections.singletonList(dir.getAbsolutePath()), 0.0); assertEquals(NodeStatus.Running, CommonDescriptor.getInstance().getConfig().getNodeStatus()); assertNull(CommonDescriptor.getInstance().getConfig().getStatusReason()); } @Test - public void emptyDirListIsNormal() { - assertEquals(DiskChecker.DiskStatus.NORMAL, DiskChecker.check(Collections.emptyList(), 1.0)); - assertEquals(DiskChecker.DiskStatus.NORMAL, DiskChecker.check(null, 1.0)); - } - - @Test - public void smokeProbeFileIsDeleted() throws Exception { + public void checkFreeRatioAndApplyDoesNotClearDiskCrash() throws Exception { File dir = tmp.newFolder(); - DiskChecker.check(Collections.singletonList(dir.getAbsolutePath()), 0.0); - File[] leftovers = dir.listFiles(); - assertTrue( - "Disk probe should clean up its temp file", leftovers == null || leftovers.length == 0); + // Simulate a Ratis-passive DiskCrash signal. + DiskChecker.apply(DiskChecker.DiskStatus.DISK_CRASH); + // Subsequent healthy free-ratio polling must keep the node in ReadOnly(DiskCrash). + DiskChecker.checkFreeRatioAndApply(Collections.singletonList(dir.getAbsolutePath()), 0.0); + assertEquals(NodeStatus.ReadOnly, CommonDescriptor.getInstance().getConfig().getNodeStatus()); + assertEquals( + NodeStatus.DISK_CRASH, CommonDescriptor.getInstance().getConfig().getStatusReason()); } } From f7103d9c6c04321f5725b9f6962fe767c5e5419e Mon Sep 17 00:00:00 2001 From: libo Date: Thu, 21 May 2026 10:27:05 +0800 Subject: [PATCH 3/5] Trigger leader balance on long-term WAL write blocking Mark DataNode as ReadOnly(WALBlocked) when WAL write blocking persists, and let ConfigNode move Region leaders away from the blocked DataNode. Add UT and IT coverage for WAL block status and leader balance behavior. --- .../env/cluster/config/MppCommonConfig.java | 13 + .../cluster/config/MppSharedCommonConfig.java | 14 ++ .../env/remote/config/RemoteCommonConfig.java | 10 + .../apache/iotdb/itbase/env/CommonConfig.java | 4 + ...egionGroupLeaderBalanceWithWALBlockIT.java | 232 ++++++++++++++++++ .../impl/DataNodeInternalRPCServiceImpl.java | 9 + .../dataregion/wal/WALManager.java | 50 ++++ .../dataregion/wal/WALWriteBlockStatus.java | 43 ++++ .../utils/MemoryControlledWALEntryQueue.java | 42 ++-- .../dataregion/wal/WALManagerTest.java | 32 +++ .../wal/WALWriteBlockStatusTest.java | 93 +++++++ 11 files changed, 527 insertions(+), 15 deletions(-) create mode 100644 integration-test/src/test/java/org/apache/iotdb/confignode/it/load/IoTDBRegionGroupLeaderBalanceWithWALBlockIT.java create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALWriteBlockStatus.java create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALWriteBlockStatusTest.java diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java index ec67e5f451bd..0e9216a5c982 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppCommonConfig.java @@ -372,6 +372,19 @@ public CommonConfig setWalBufferSize(int walBufferSize) { return this; } + @Override + public CommonConfig setCheckPeriodWhenInsertBlocked(int checkPeriodWhenInsertBlocked) { + setProperty("check_period_when_insert_blocked", String.valueOf(checkPeriodWhenInsertBlocked)); + return this; + } + + @Override + public CommonConfig setMaxWaitingTimeWhenInsertBlocked(int maxWaitingTimeWhenInsertBlocked) { + setProperty( + "max_waiting_time_when_insert_blocked", String.valueOf(maxWaitingTimeWhenInsertBlocked)); + return this; + } + @Override public CommonConfig setDegreeOfParallelism(int degreeOfParallelism) { setProperty("degree_of_query_parallelism", String.valueOf(degreeOfParallelism)); 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 36d06aefe889..cff20f9cff87 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 @@ -373,6 +373,20 @@ public CommonConfig setWalBufferSize(int walBufferSize) { return this; } + @Override + public CommonConfig setCheckPeriodWhenInsertBlocked(int checkPeriodWhenInsertBlocked) { + cnConfig.setCheckPeriodWhenInsertBlocked(checkPeriodWhenInsertBlocked); + dnConfig.setCheckPeriodWhenInsertBlocked(checkPeriodWhenInsertBlocked); + return this; + } + + @Override + public CommonConfig setMaxWaitingTimeWhenInsertBlocked(int maxWaitingTimeWhenInsertBlocked) { + cnConfig.setMaxWaitingTimeWhenInsertBlocked(maxWaitingTimeWhenInsertBlocked); + dnConfig.setMaxWaitingTimeWhenInsertBlocked(maxWaitingTimeWhenInsertBlocked); + return this; + } + @Override public CommonConfig setDegreeOfParallelism(int degreeOfParallelism) { cnConfig.setDegreeOfParallelism(degreeOfParallelism); diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java index 752dcd009db0..f40d422892a2 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java @@ -263,6 +263,16 @@ public CommonConfig setWalBufferSize(int walBufferSize) { return this; } + @Override + public CommonConfig setCheckPeriodWhenInsertBlocked(int checkPeriodWhenInsertBlocked) { + return this; + } + + @Override + public CommonConfig setMaxWaitingTimeWhenInsertBlocked(int maxWaitingTimeWhenInsertBlocked) { + return this; + } + @Override public CommonConfig setDegreeOfParallelism(int degreeOfParallelism) { return this; diff --git a/integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java b/integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java index 0ad3c23af16f..b52429d2155b 100644 --- a/integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/itbase/env/CommonConfig.java @@ -119,6 +119,10 @@ CommonConfig setEnableAutoLeaderBalanceForRatisConsensus( CommonConfig setWalBufferSize(int walBufferSize); + CommonConfig setCheckPeriodWhenInsertBlocked(int checkPeriodWhenInsertBlocked); + + CommonConfig setMaxWaitingTimeWhenInsertBlocked(int maxWaitingTimeWhenInsertBlocked); + CommonConfig setDegreeOfParallelism(int degreeOfParallelism); CommonConfig setDataRatisTriggerSnapshotThreshold(long threshold); diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/load/IoTDBRegionGroupLeaderBalanceWithWALBlockIT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/load/IoTDBRegionGroupLeaderBalanceWithWALBlockIT.java new file mode 100644 index 000000000000..0ccdf62f3863 --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/confignode/it/load/IoTDBRegionGroupLeaderBalanceWithWALBlockIT.java @@ -0,0 +1,232 @@ +/* + * 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.confignode.it.load; + +import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType; +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot; +import org.apache.iotdb.common.rpc.thrift.TSetConfigurationReq; +import org.apache.iotdb.common.rpc.thrift.TTimePartitionSlot; +import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient; +import org.apache.iotdb.commons.cluster.NodeStatus; +import org.apache.iotdb.commons.cluster.RegionRoleType; +import org.apache.iotdb.commons.pipe.config.constant.SystemConstant; +import org.apache.iotdb.confignode.rpc.thrift.TDataPartitionReq; +import org.apache.iotdb.confignode.rpc.thrift.TDataPartitionTableResp; +import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema; +import org.apache.iotdb.confignode.rpc.thrift.TRegionInfo; +import org.apache.iotdb.confignode.rpc.thrift.TShowClusterResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowRegionReq; +import org.apache.iotdb.confignode.rpc.thrift.TShowRegionResp; +import org.apache.iotdb.confignode.rpc.thrift.TTimeSlotList; +import org.apache.iotdb.consensus.ConsensusFactory; +import org.apache.iotdb.it.env.EnvFactory; +import org.apache.iotdb.it.framework.IoTDBTestRunner; +import org.apache.iotdb.itbase.category.ClusterIT; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeUnit; + +@RunWith(IoTDBTestRunner.class) +@Category({ClusterIT.class}) +public class IoTDBRegionGroupLeaderBalanceWithWALBlockIT { + + private static final String TEST_SCHEMA_REGION_CONSENSUS_PROTOCOL_CLASS = + ConsensusFactory.RATIS_CONSENSUS; + private static final String TEST_DATA_REGION_CONSENSUS_PROTOCOL_CLASS = + ConsensusFactory.IOT_CONSENSUS; + private static final int TEST_REPLICATION_FACTOR = 3; + private static final int TEST_DATA_NODE_NUM = 3; + private static final int DATABASE_NUM = 3; + private static final int RETRY_NUM = 60; + + private static final String DATABASE = "root.wal_block_db"; + private static final String WAL_THROTTLE_THRESHOLD_IN_BYTE = "wal_throttle_threshold_in_byte"; + private static final String WAL_BLOCKED_STATUS = NodeStatus.ReadOnly.getStatus() + "(WALBlocked)"; + + @Before + public void setUp() { + EnvFactory.getEnv() + .getConfig() + .getCommonConfig() + .setEnableAutoLeaderBalanceForRatisConsensus(true) + .setEnableAutoLeaderBalanceForIoTConsensus(true) + .setSchemaRegionConsensusProtocolClass(TEST_SCHEMA_REGION_CONSENSUS_PROTOCOL_CLASS) + .setDataRegionConsensusProtocolClass(TEST_DATA_REGION_CONSENSUS_PROTOCOL_CLASS) + .setSchemaReplicationFactor(TEST_REPLICATION_FACTOR) + .setDataReplicationFactor(TEST_REPLICATION_FACTOR) + .setCheckPeriodWhenInsertBlocked(50) + .setMaxWaitingTimeWhenInsertBlocked(2000); + EnvFactory.getEnv().initClusterEnvironment(1, TEST_DATA_NODE_NUM); + } + + @After + public void tearDown() { + EnvFactory.getEnv().cleanClusterEnvironment(); + } + + @Test + public void testRegionLeaderBalanceWhenWalLongTermBlocked() throws Exception { + try (SyncConfigNodeIServiceClient client = + (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) { + createDataRegionGroups(client); + waitUntil( + "all DataNodes have balanced DataRegion leaders", + () -> isLeaderDistributionBalanced(client)); + + TRegionInfo targetLeader = findAnyDataRegionLeader(client); + triggerLongTermWalBlockingOnDataNode(client, targetLeader.getDataNodeId()); + + waitUntil( + "target leader DataNode becomes ReadOnly because of long-term WAL blocking", + () -> + WAL_BLOCKED_STATUS.equals( + getNodeStatusWithReason(client, targetLeader.getDataNodeId()))); + waitUntil( + "Region leaders are moved away from ReadOnly DataNodes", + () -> hasNoLeaderOnReadOnlyDataNode(client, targetLeader.getDataNodeId())); + } + } + + private void createDataRegionGroups(SyncConfigNodeIServiceClient client) throws Exception { + for (int i = 0; i < DATABASE_NUM; i++) { + TSStatus status = client.setDatabase(new TDatabaseSchema(DATABASE + i)); + Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode()); + + Map seriesSlotMap = new HashMap<>(); + seriesSlotMap.put( + new TSeriesPartitionSlot(1), + new TTimeSlotList() + .setTimePartitionSlots(Collections.singletonList(new TTimePartitionSlot(100)))); + Map> databaseSlotsMap = new HashMap<>(); + databaseSlotsMap.put(DATABASE + i, seriesSlotMap); + + TDataPartitionTableResp dataPartitionTableResp = + client.getOrCreateDataPartitionTable(new TDataPartitionReq(databaseSlotsMap)); + Assert.assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), + dataPartitionTableResp.getStatus().getCode()); + } + } + + private boolean isLeaderDistributionBalanced(SyncConfigNodeIServiceClient client) + throws Exception { + Map leaderCounter = new HashMap<>(); + for (TRegionInfo regionInfo : getUserDataRegionInfoList(client)) { + if (RegionRoleType.Leader.getRoleType().equals(regionInfo.getRoleType())) { + leaderCounter.merge(regionInfo.getDataNodeId(), 1, Integer::sum); + } + } + if (leaderCounter.size() != TEST_DATA_NODE_NUM) { + return false; + } + for (Integer leaderCount : leaderCounter.values()) { + if (leaderCount != DATABASE_NUM / TEST_DATA_NODE_NUM) { + return false; + } + } + return true; + } + + private TRegionInfo findAnyDataRegionLeader(SyncConfigNodeIServiceClient client) + throws Exception { + for (TRegionInfo regionInfo : getUserDataRegionInfoList(client)) { + if (RegionRoleType.Leader.getRoleType().equals(regionInfo.getRoleType())) { + return regionInfo; + } + } + throw new AssertionError("DataRegion leader not found"); + } + + private void triggerLongTermWalBlockingOnDataNode( + SyncConfigNodeIServiceClient client, int dataNodeId) throws Exception { + Map configItems = new HashMap<>(); + // The throttle threshold used by WALManager is 80% of this value, so 1 makes it 0 and + // deterministically triggers long-term WAL blocking on the target DataNode heartbeat. + configItems.put(WAL_THROTTLE_THRESHOLD_IN_BYTE, "1"); + TSStatus status = client.setConfiguration(new TSetConfigurationReq(configItems, dataNodeId)); + Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode()); + } + + private String getNodeStatusWithReason(SyncConfigNodeIServiceClient client, int dataNodeId) + throws Exception { + TShowClusterResp showClusterResp = client.showCluster(); + Assert.assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), showClusterResp.getStatus().getCode()); + return showClusterResp.getNodeStatus().get(dataNodeId); + } + + private boolean hasNoLeaderOnReadOnlyDataNode( + SyncConfigNodeIServiceClient client, int readOnlyDataNodeId) throws Exception { + for (TRegionInfo regionInfo : getUserDataRegionInfoList(client)) { + if (RegionRoleType.Leader.getRoleType().equals(regionInfo.getRoleType())) { + if (regionInfo.getDataNodeId() == readOnlyDataNodeId + || NodeStatus.ReadOnly.getStatus().equals(regionInfo.getStatus())) { + return false; + } + } + } + return true; + } + + private List getUserDataRegionInfoList(SyncConfigNodeIServiceClient client) + throws Exception { + TShowRegionResp showRegionResp = client.showRegion(new TShowRegionReq()); + Assert.assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), showRegionResp.getStatus().getCode()); + + List result = new ArrayList<>(); + for (TRegionInfo regionInfo : showRegionResp.getRegionInfoList()) { + if (TConsensusGroupType.DataRegion.equals(regionInfo.getConsensusGroupId().getType()) + && !regionInfo.getDatabase().startsWith(SystemConstant.SYSTEM_DATABASE) + && !regionInfo.getDatabase().startsWith(SystemConstant.AUDIT_DATABASE)) { + result.add(regionInfo); + } + } + return result; + } + + private void waitUntil(String condition, WaitCondition waitCondition) throws Exception { + for (int retry = 0; retry < RETRY_NUM; retry++) { + if (waitCondition.evaluate()) { + return; + } + TimeUnit.SECONDS.sleep(1); + } + Assert.fail("Failed to wait until " + condition); + } + + @FunctionalInterface + private interface WaitCondition { + boolean evaluate() throws Exception; + } +} 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 bbf67a8354d2..f2ecca222aa8 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 @@ -221,6 +221,8 @@ import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileManager; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.iotdb.db.storageengine.rescon.disk.FolderManager; +import org.apache.iotdb.db.storageengine.dataregion.wal.WALManager; +import org.apache.iotdb.db.storageengine.dataregion.wal.WALWriteBlockStatus; import org.apache.iotdb.db.storageengine.rescon.quotas.DataNodeSpaceQuotaManager; import org.apache.iotdb.db.storageengine.rescon.quotas.DataNodeThrottleQuotaManager; import org.apache.iotdb.db.subscription.agent.SubscriptionAgent; @@ -2372,6 +2374,8 @@ public TDataNodeHeartbeatResp getDataNodeHeartBeat(TDataNodeHeartbeatReq req) th .forEach((key, value) -> regionRawDataSize.put(Integer.parseInt(key), value.getLeft())); resp.setDataRegionRawDataSize(regionRawDataSize); } + AuthorityChecker.getAuthorityFetcher().refreshToken(); + updateWALBlockedStatus(); resp.setHeartbeatTimestamp(req.getHeartbeatTimestamp()); resp.setStatus(commonConfig.getNodeStatus().getStatus()); // Advertise that this DataNode supports metadata-lease self-fencing, so the ConfigNode may @@ -2437,6 +2441,11 @@ public TDataNodeHeartbeatResp getDataNodeHeartBeat(TDataNodeHeartbeatReq req) th return resp; } + private void updateWALBlockedStatus() { + WALWriteBlockStatus.updateStatus( + commonConfig, WALManager.getInstance().isLongTermWriteBlocked()); + } + @Override public TSStatus updateRegionCache(TRegionRouteReq req) { boolean result = ClusterPartitionFetcher.getInstance().updateRegionCache(req); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java index 8fafee304b2b..8f7924a7f100 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java @@ -52,6 +52,7 @@ import java.util.concurrent.Future; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import static org.apache.iotdb.commons.conf.IoTDBConstant.FILE_NAME_SEPARATOR; @@ -69,6 +70,9 @@ public class WALManager implements IService { private final AtomicLong totalDiskUsage = new AtomicLong(); // total number of wal files private final AtomicLong totalFileNum = new AtomicLong(); + private final AtomicLong walThrottleStartTimeInMs = new AtomicLong(-1); + private final AtomicLong walBufferQueueBlockedStartTimeInMs = new AtomicLong(-1); + private final AtomicInteger walBufferQueueBlockedWriterCount = new AtomicInteger(0); private WALManager() { if (config.getDataRegionConsensusProtocolClass().equals(ConsensusFactory.IOT_CONSENSUS) @@ -237,6 +241,49 @@ public boolean shouldThrottle() { return getTotalDiskUsage() >= getThrottleThreshold(); } + public boolean isLongTermWriteBlocked() { + return isLongTermWalThrottle() || isLongTermWalBufferQueueBlocked(); + } + + private boolean isLongTermWalThrottle() { + if (!shouldThrottle()) { + walThrottleStartTimeInMs.set(-1); + return false; + } + return isLongTermBlocked(walThrottleStartTimeInMs); + } + + private boolean isLongTermWalBufferQueueBlocked() { + if (walBufferQueueBlockedWriterCount.get() <= 0) { + walBufferQueueBlockedStartTimeInMs.set(-1); + return false; + } + return isLongTermBlocked(walBufferQueueBlockedStartTimeInMs); + } + + private boolean isLongTermBlocked(AtomicLong blockStartTimeInMs) { + long currentTimeInMs = System.currentTimeMillis(); + long blockStartTime = blockStartTimeInMs.get(); + if (blockStartTime < 0) { + blockStartTimeInMs.compareAndSet(-1, currentTimeInMs); + blockStartTime = blockStartTimeInMs.get(); + } + return currentTimeInMs - blockStartTime >= config.getMaxWaitingTimeWhenInsertBlocked(); + } + + public void markWalBufferQueueBlocked() { + if (walBufferQueueBlockedWriterCount.getAndIncrement() == 0) { + walBufferQueueBlockedStartTimeInMs.compareAndSet(-1, System.currentTimeMillis()); + } + } + + public void markWalBufferQueueAvailable() { + if (walBufferQueueBlockedWriterCount.decrementAndGet() <= 0) { + walBufferQueueBlockedWriterCount.set(0); + walBufferQueueBlockedStartTimeInMs.set(-1); + } + } + public long getThrottleThreshold() { return (long) (config.getThrottleThreshold() * 0.8); } @@ -323,6 +370,9 @@ public void syncDeleteOutdatedFilesInWALNodes() { public void clear() { totalDiskUsage.set(0); + walThrottleStartTimeInMs.set(-1); + walBufferQueueBlockedStartTimeInMs.set(-1); + walBufferQueueBlockedWriterCount.set(0); walNodesManager.clear(); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALWriteBlockStatus.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALWriteBlockStatus.java new file mode 100644 index 000000000000..70dad8a9d379 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALWriteBlockStatus.java @@ -0,0 +1,43 @@ +/* + * 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.dataregion.wal; + +import org.apache.iotdb.commons.cluster.NodeStatus; +import org.apache.iotdb.commons.conf.CommonConfig; + +public final class WALWriteBlockStatus { + + public static final String WAL_BLOCKED = "WALBlocked"; + + private WALWriteBlockStatus() {} + + public static void updateStatus(CommonConfig commonConfig, boolean longTermWriteBlocked) { + if (longTermWriteBlocked) { + if (NodeStatus.Running.equals(commonConfig.getNodeStatus())) { + commonConfig.setNodeStatus(NodeStatus.ReadOnly); + commonConfig.setStatusReason(WAL_BLOCKED); + } + } else if (NodeStatus.ReadOnly.equals(commonConfig.getNodeStatus()) + && WAL_BLOCKED.equals(commonConfig.getStatusReason())) { + commonConfig.setNodeStatus(NodeStatus.Running); + commonConfig.setStatusReason(null); + } + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/MemoryControlledWALEntryQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/MemoryControlledWALEntryQueue.java index 0974aef999a3..7ab08b9c86a7 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/MemoryControlledWALEntryQueue.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/MemoryControlledWALEntryQueue.java @@ -21,6 +21,7 @@ import org.apache.iotdb.commons.exception.IoTDBRuntimeException; import org.apache.iotdb.db.i18n.StorageEngineMessages; +import org.apache.iotdb.db.storageengine.dataregion.wal.WALManager; import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntry; import org.apache.iotdb.db.storageengine.rescon.memory.SystemInfo; @@ -51,24 +52,35 @@ public WALEntry poll(long timeout, TimeUnit unit) throws InterruptedException { public void put(WALEntry e) throws InterruptedException { long elementSize = getElementSize(e); - synchronized (nonFullCondition) { - while (!SystemInfo.getInstance().getWalBufferQueueMemoryBlock().allocate(elementSize)) { - if (elementSize - > SystemInfo.getInstance().getWalBufferQueueMemoryBlock().getTotalMemorySizeInBytes()) { - throw new IoTDBRuntimeException( - String.format( - StorageEngineMessages - .STORAGE_EXCEPTION_THE_ELEMENT_SIZE_OF_WALENTRY_S_IS_LARGER_THAN_THE_TOTAL_E494520D, - elementSize, - SystemInfo.getInstance() - .getWalBufferQueueMemoryBlock() - .getTotalMemorySizeInBytes()), - WAL_ENTRY_TOO_LARGE.getStatusCode()); + boolean blocked = false; + try { + synchronized (nonFullCondition) { + while (!SystemInfo.getInstance().getWalBufferQueueMemoryBlock().allocate(elementSize)) { + if (elementSize + > SystemInfo.getInstance().getWalBufferQueueMemoryBlock().getTotalMemorySizeInBytes()) { + throw new IoTDBRuntimeException( + String.format( + StorageEngineMessages + .STORAGE_EXCEPTION_THE_ELEMENT_SIZE_OF_WALENTRY_S_IS_LARGER_THAN_THE_TOTAL_E494520D, + elementSize, + SystemInfo.getInstance() + .getWalBufferQueueMemoryBlock() + .getTotalMemorySizeInBytes()), + WAL_ENTRY_TOO_LARGE.getStatusCode()); + } + if (!blocked) { + blocked = true; + WALManager.getInstance().markWalBufferQueueBlocked(); + } + nonFullCondition.wait(); } - nonFullCondition.wait(); + } + queue.put(e); + } finally { + if (blocked) { + WALManager.getInstance().markWalBufferQueueAvailable(); } } - queue.put(e); } public WALEntry take() throws InterruptedException { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManagerTest.java index 2017805ac7e2..1727d98522eb 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManagerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManagerTest.java @@ -44,6 +44,7 @@ import java.io.File; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; @@ -58,11 +59,13 @@ public class WALManagerTest { }; private String[] prevWalDirs; private String prevConsensus; + private int prevMaxWaitingTimeWhenInsertBlocked; @Before public void setUp() throws Exception { prevConsensus = config.getDataRegionConsensusProtocolClass(); prevWalDirs = commonConfig.getWalDirs(); + prevMaxWaitingTimeWhenInsertBlocked = config.getMaxWaitingTimeWhenInsertBlocked(); config.setDataRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS); commonConfig.setWalDirs(walDirs); EnvironmentUtils.envSetUp(); @@ -76,6 +79,7 @@ public void tearDown() throws Exception { } config.setDataRegionConsensusProtocolClass(prevConsensus); commonConfig.setWalDirs(prevWalDirs); + config.setMaxWaitingTimeWhenInsertBlocked(prevMaxWaitingTimeWhenInsertBlocked); } @Test @@ -118,6 +122,34 @@ public void testDeleteOutdatedWALFiles() throws IllegalPathException { } } + @Test + public void testLongTermWriteBlockedByWalThrottle() { + WALManager walManager = WALManager.getInstance(); + walManager.addTotalDiskUsage(walManager.getThrottleThreshold()); + + assertFalse(walManager.isLongTermWriteBlocked()); + + config.setMaxWaitingTimeWhenInsertBlocked(0); + assertTrue(walManager.isLongTermWriteBlocked()); + + walManager.clear(); + assertFalse(walManager.isLongTermWriteBlocked()); + } + + @Test + public void testLongTermWriteBlockedByWalBufferQueue() { + WALManager walManager = WALManager.getInstance(); + walManager.markWalBufferQueueBlocked(); + + assertFalse(walManager.isLongTermWriteBlocked()); + + config.setMaxWaitingTimeWhenInsertBlocked(0); + assertTrue(walManager.isLongTermWriteBlocked()); + + walManager.markWalBufferQueueAvailable(); + assertFalse(walManager.isLongTermWriteBlocked()); + } + private InsertRowNode getInsertRowNode() throws IllegalPathException { long time = 110L; TSDataType[] dataTypes = diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALWriteBlockStatusTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALWriteBlockStatusTest.java new file mode 100644 index 000000000000..2546457d456b --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALWriteBlockStatusTest.java @@ -0,0 +1,93 @@ +/* + * 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.dataregion.wal; + +import org.apache.iotdb.commons.cluster.NodeStatus; +import org.apache.iotdb.commons.conf.CommonConfig; + +import org.junit.Test; +import org.mockito.Mockito; + +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; + +public class WALWriteBlockStatusTest { + + @Test + public void testRunningNodeTurnsReadOnlyWhenWalBlocked() { + CommonConfig commonConfig = mockCommonConfig(NodeStatus.Running, null); + + WALWriteBlockStatus.updateStatus(commonConfig, true); + + assertEquals(NodeStatus.ReadOnly, commonConfig.getNodeStatus()); + assertEquals(WALWriteBlockStatus.WAL_BLOCKED, commonConfig.getStatusReason()); + } + + @Test + public void testWalBlockedReadOnlyNodeRecovers() { + CommonConfig commonConfig = + mockCommonConfig(NodeStatus.ReadOnly, WALWriteBlockStatus.WAL_BLOCKED); + + WALWriteBlockStatus.updateStatus(commonConfig, false); + + assertEquals(NodeStatus.Running, commonConfig.getNodeStatus()); + assertNull(commonConfig.getStatusReason()); + } + + @Test + public void testOtherReadOnlyReasonIsNotOverwrittenOrRecovered() { + CommonConfig commonConfig = mockCommonConfig(NodeStatus.ReadOnly, NodeStatus.DISK_FULL); + + WALWriteBlockStatus.updateStatus(commonConfig, true); + assertEquals(NodeStatus.ReadOnly, commonConfig.getNodeStatus()); + assertEquals(NodeStatus.DISK_FULL, commonConfig.getStatusReason()); + + WALWriteBlockStatus.updateStatus(commonConfig, false); + assertEquals(NodeStatus.ReadOnly, commonConfig.getNodeStatus()); + assertEquals(NodeStatus.DISK_FULL, commonConfig.getStatusReason()); + } + + private CommonConfig mockCommonConfig(NodeStatus initialStatus, String initialStatusReason) { + AtomicReference status = new AtomicReference<>(initialStatus); + AtomicReference statusReason = new AtomicReference<>(initialStatusReason); + CommonConfig commonConfig = Mockito.mock(CommonConfig.class); + + Mockito.when(commonConfig.getNodeStatus()).thenAnswer(invocation -> status.get()); + Mockito.when(commonConfig.getStatusReason()).thenAnswer(invocation -> statusReason.get()); + Mockito.doAnswer( + invocation -> { + status.set(invocation.getArgument(0)); + statusReason.set(null); + return null; + }) + .when(commonConfig) + .setNodeStatus(Mockito.any(NodeStatus.class)); + Mockito.doAnswer( + invocation -> { + statusReason.set(invocation.getArgument(0)); + return null; + }) + .when(commonConfig) + .setStatusReason(Mockito.any()); + return commonConfig; + } +} From be42aa9a0462e90b4e668c1d67d7e8d574e34b1b Mon Sep 17 00:00:00 2001 From: libo Date: Thu, 25 Jun 2026 19:45:15 +0800 Subject: [PATCH 4/5] Fix issues related to review comments --- ...egionGroupLeaderBalanceWithWALBlockIT.java | 116 ++++++++++++++++-- .../dataregion/wal/WALManager.java | 6 +- .../dataregion/wal/WALManagerTest.java | 27 ++++ 3 files changed, 141 insertions(+), 8 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/load/IoTDBRegionGroupLeaderBalanceWithWALBlockIT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/load/IoTDBRegionGroupLeaderBalanceWithWALBlockIT.java index 0ccdf62f3863..313caf205d2f 100644 --- a/integration-test/src/test/java/org/apache/iotdb/confignode/it/load/IoTDBRegionGroupLeaderBalanceWithWALBlockIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/confignode/it/load/IoTDBRegionGroupLeaderBalanceWithWALBlockIT.java @@ -37,7 +37,9 @@ import org.apache.iotdb.confignode.rpc.thrift.TShowRegionResp; import org.apache.iotdb.confignode.rpc.thrift.TTimeSlotList; import org.apache.iotdb.consensus.ConsensusFactory; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileUtils; 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.ClusterIT; import org.apache.iotdb.rpc.TSStatusCode; @@ -49,12 +51,21 @@ import org.junit.experimental.categories.Category; import org.junit.runner.RunWith; +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.sql.Connection; +import java.sql.Statement; import java.util.ArrayList; import java.util.Collections; +import java.util.Comparator; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; +import java.util.stream.Stream; @RunWith(IoTDBTestRunner.class) @Category({ClusterIT.class}) @@ -68,9 +79,15 @@ public class IoTDBRegionGroupLeaderBalanceWithWALBlockIT { private static final int TEST_DATA_NODE_NUM = 3; private static final int DATABASE_NUM = 3; private static final int RETRY_NUM = 60; + private static final int TEST_SERIES_PARTITION_SLOT = 0; + private static final long TEST_TIME_PARTITION_SLOT = 0; + private static final int WAL_FILE_SIZE_THRESHOLD_IN_BYTE = 1024; + private static final int WAL_PAYLOAD_REPEAT_COUNT = 128; private static final String DATABASE = "root.wal_block_db"; private static final String WAL_THROTTLE_THRESHOLD_IN_BYTE = "wal_throttle_threshold_in_byte"; + private static final String WAL_FILE_SIZE_THRESHOLD_IN_BYTE_CONFIG = + "wal_file_size_threshold_in_byte"; private static final String WAL_BLOCKED_STATUS = NodeStatus.ReadOnly.getStatus() + "(WALBlocked)"; @Before @@ -84,6 +101,7 @@ public void setUp() { .setDataRegionConsensusProtocolClass(TEST_DATA_REGION_CONSENSUS_PROTOCOL_CLASS) .setSchemaReplicationFactor(TEST_REPLICATION_FACTOR) .setDataReplicationFactor(TEST_REPLICATION_FACTOR) + .setSeriesSlotNum(1) .setCheckPeriodWhenInsertBlocked(50) .setMaxWaitingTimeWhenInsertBlocked(2000); EnvFactory.getEnv().initClusterEnvironment(1, TEST_DATA_NODE_NUM); @@ -104,7 +122,8 @@ public void testRegionLeaderBalanceWhenWalLongTermBlocked() throws Exception { () -> isLeaderDistributionBalanced(client)); TRegionInfo targetLeader = findAnyDataRegionLeader(client); - triggerLongTermWalBlockingOnDataNode(client, targetLeader.getDataNodeId()); + long walDiskUsage = generateWalTraffic(client, targetLeader); + triggerLongTermWalBlockingOnDataNode(client, targetLeader.getDataNodeId(), walDiskUsage); waitUntil( "target leader DataNode becomes ReadOnly because of long-term WAL blocking", @@ -124,9 +143,10 @@ private void createDataRegionGroups(SyncConfigNodeIServiceClient client) throws Map seriesSlotMap = new HashMap<>(); seriesSlotMap.put( - new TSeriesPartitionSlot(1), + new TSeriesPartitionSlot(TEST_SERIES_PARTITION_SLOT), new TTimeSlotList() - .setTimePartitionSlots(Collections.singletonList(new TTimePartitionSlot(100)))); + .setTimePartitionSlots( + Collections.singletonList(new TTimePartitionSlot(TEST_TIME_PARTITION_SLOT)))); Map> databaseSlotsMap = new HashMap<>(); databaseSlotsMap.put(DATABASE + i, seriesSlotMap); @@ -168,13 +188,95 @@ private TRegionInfo findAnyDataRegionLeader(SyncConfigNodeIServiceClient client) } private void triggerLongTermWalBlockingOnDataNode( - SyncConfigNodeIServiceClient client, int dataNodeId) throws Exception { + SyncConfigNodeIServiceClient client, int dataNodeId, long walDiskUsage) throws Exception { + Assert.assertTrue("No WAL traffic was generated on target DataNode", walDiskUsage > 0); + + Map configItems = new HashMap<>(); + configItems.put(WAL_THROTTLE_THRESHOLD_IN_BYTE, Long.toString(walDiskUsage)); + TSStatus status = client.setConfiguration(new TSetConfigurationReq(configItems, dataNodeId)); + Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode()); + } + + private long generateWalTraffic(SyncConfigNodeIServiceClient client, TRegionInfo targetLeader) + throws Exception { + int dataNodeId = targetLeader.getDataNodeId(); + long originalWalDiskUsage = countWalDiskUsage(dataNodeId); Map configItems = new HashMap<>(); - // The throttle threshold used by WALManager is 80% of this value, so 1 makes it 0 and - // deterministically triggers long-term WAL blocking on the target DataNode heartbeat. - configItems.put(WAL_THROTTLE_THRESHOLD_IN_BYTE, "1"); + configItems.put( + WAL_FILE_SIZE_THRESHOLD_IN_BYTE_CONFIG, Integer.toString(WAL_FILE_SIZE_THRESHOLD_IN_BYTE)); TSStatus status = client.setConfiguration(new TSetConfigurationReq(configItems, dataNodeId)); Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode()); + + DataNodeWrapper dataNodeWrapper = + EnvFactory.getEnv() + .dataNodeIdToWrapper(dataNodeId) + .orElseThrow(() -> new AssertionError("DataNode not found: " + dataNodeId)); + + try (Connection connection = + EnvFactory.getEnv().getConnectionWithSpecifiedDataNode(dataNodeWrapper); + Statement statement = connection.createStatement()) { + String payload = String.join("", Collections.nCopies(WAL_PAYLOAD_REPEAT_COUNT, "wal_block")); + String device = + targetLeader.getDatabase() + ".d" + targetLeader.getConsensusGroupId().getId(); + statement.execute("CREATE TIMESERIES " + device + ".s WITH DATATYPE=TEXT, ENCODING=PLAIN"); + for (int i = 0; i < 16; i++) { + statement.execute( + "INSERT INTO " + + device + + "(time,s) VALUES(" + + (TEST_TIME_PARTITION_SLOT + i) + + ", '" + + payload + + "')"); + } + } + + final long[] currentWalDiskUsage = new long[1]; + waitUntil( + "target DataNode generates WAL files", + () -> { + currentWalDiskUsage[0] = countWalDiskUsage(dataNodeId); + return currentWalDiskUsage[0] > originalWalDiskUsage; + }); + return currentWalDiskUsage[0]; + } + + private long countWalDiskUsage(int dataNodeId) throws IOException { + DataNodeWrapper dataNodeWrapper = + EnvFactory.getEnv() + .dataNodeIdToWrapper(dataNodeId) + .orElseThrow(() -> new AssertionError("DataNode not found: " + dataNodeId)); + Path walDir = new File(dataNodeWrapper.getWalDir()).toPath(); + if (!Files.exists(walDir)) { + return 0; + } + try (Stream paths = Files.walk(walDir)) { + Map> walFilesByDir = + paths + .filter(Files::isRegularFile) + .filter( + path -> + WALFileUtils.walFilenameFilter( + path.getParent().toFile(), path.getFileName().toString())) + .collect(Collectors.groupingBy(Path::getParent)); + long walDiskUsage = 0; + for (List walFiles : walFilesByDir.values()) { + if (walFiles.size() <= 1) { + continue; + } + Path currentWalFile = + Collections.max( + walFiles, + Comparator.comparingLong( + path -> WALFileUtils.parseVersionId(path.getFileName().toString()))); + for (Path walFile : walFiles) { + if (!walFile.equals(currentWalFile)) { + walDiskUsage += walFile.toFile().length(); + } + } + } + return walDiskUsage; + } } private String getNodeStatusWithReason(SyncConfigNodeIServiceClient client, int dataNodeId) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java index 8f7924a7f100..a579871ba616 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManager.java @@ -285,7 +285,11 @@ public void markWalBufferQueueAvailable() { } public long getThrottleThreshold() { - return (long) (config.getThrottleThreshold() * 0.8); + long configuredThrottleThreshold = config.getThrottleThreshold(); + if (configuredThrottleThreshold <= 0) { + return Long.MAX_VALUE; + } + return Math.max((long) (configuredThrottleThreshold * 0.8), 1); } public long getTotalDiskUsage() { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManagerTest.java index 1727d98522eb..0b43f1f21f3a 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManagerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALManagerTest.java @@ -59,12 +59,14 @@ public class WALManagerTest { }; private String[] prevWalDirs; private String prevConsensus; + private long prevThrottleThreshold; private int prevMaxWaitingTimeWhenInsertBlocked; @Before public void setUp() throws Exception { prevConsensus = config.getDataRegionConsensusProtocolClass(); prevWalDirs = commonConfig.getWalDirs(); + prevThrottleThreshold = config.getThrottleThreshold(); prevMaxWaitingTimeWhenInsertBlocked = config.getMaxWaitingTimeWhenInsertBlocked(); config.setDataRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS); commonConfig.setWalDirs(walDirs); @@ -79,6 +81,7 @@ public void tearDown() throws Exception { } config.setDataRegionConsensusProtocolClass(prevConsensus); commonConfig.setWalDirs(prevWalDirs); + config.setThrottleThreshold(prevThrottleThreshold); config.setMaxWaitingTimeWhenInsertBlocked(prevMaxWaitingTimeWhenInsertBlocked); } @@ -136,6 +139,30 @@ public void testLongTermWriteBlockedByWalThrottle() { assertFalse(walManager.isLongTermWriteBlocked()); } + @Test + public void testNonPositiveWalThrottleThresholdIsIgnored() { + WALManager walManager = WALManager.getInstance(); + config.setThrottleThreshold(0); + walManager.addTotalDiskUsage(1); + + assertEquals(Long.MAX_VALUE, walManager.getThrottleThreshold()); + assertFalse(walManager.shouldThrottle()); + assertFalse(walManager.isLongTermWriteBlocked()); + + walManager.clear(); + config.setThrottleThreshold(-1); + walManager.addTotalDiskUsage(1); + assertEquals(Long.MAX_VALUE, walManager.getThrottleThreshold()); + assertFalse(walManager.shouldThrottle()); + + walManager.clear(); + config.setThrottleThreshold(1); + assertEquals(1, walManager.getThrottleThreshold()); + walManager.addTotalDiskUsage(1); + assertTrue(walManager.shouldThrottle()); + walManager.clear(); + } + @Test public void testLongTermWriteBlockedByWalBufferQueue() { WALManager walManager = WALManager.getInstance(); From c6477c0941e0b8aa209f246ed80da364f0f163c1 Mon Sep 17 00:00:00 2001 From: libo Date: Thu, 13 Aug 2026 13:23:44 +0800 Subject: [PATCH 5/5] Improve ReadOnly status reason observability --- .../load/IoTDBRegionGroupLeaderBalanceWithWALBlockIT.java | 3 ++- .../confignode/manager/load/service/HeartbeatService.java | 2 +- .../service/thrift/ConfigNodeRPCServiceProcessor.java | 7 ++++++- .../en/org/apache/iotdb/consensus/i18n/RatisMessages.java | 3 +++ .../zh/org/apache/iotdb/consensus/i18n/RatisMessages.java | 3 +++ .../org/apache/iotdb/consensus/ratis/RatisConsensus.java | 3 ++- .../thrift/impl/DataNodeInternalRPCServiceImpl.java | 3 +-- .../storageengine/dataregion/wal/WALWriteBlockStatus.java | 6 ++---- .../wal/utils/MemoryControlledWALEntryQueue.java | 4 +++- .../dataregion/wal/WALWriteBlockStatusTest.java | 5 ++--- .../java/org/apache/iotdb/commons/cluster/NodeStatus.java | 1 + 11 files changed, 26 insertions(+), 14 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/load/IoTDBRegionGroupLeaderBalanceWithWALBlockIT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/load/IoTDBRegionGroupLeaderBalanceWithWALBlockIT.java index 313caf205d2f..5158cd1632e1 100644 --- a/integration-test/src/test/java/org/apache/iotdb/confignode/it/load/IoTDBRegionGroupLeaderBalanceWithWALBlockIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/confignode/it/load/IoTDBRegionGroupLeaderBalanceWithWALBlockIT.java @@ -88,7 +88,8 @@ public class IoTDBRegionGroupLeaderBalanceWithWALBlockIT { private static final String WAL_THROTTLE_THRESHOLD_IN_BYTE = "wal_throttle_threshold_in_byte"; private static final String WAL_FILE_SIZE_THRESHOLD_IN_BYTE_CONFIG = "wal_file_size_threshold_in_byte"; - private static final String WAL_BLOCKED_STATUS = NodeStatus.ReadOnly.getStatus() + "(WALBlocked)"; + private static final String WAL_BLOCKED_STATUS = + NodeStatus.ReadOnly.getStatus() + "(" + NodeStatus.WAL_BLOCKED + ")"; @Before public void setUp() { 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 312e435904b9..2c07212f10cd 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 @@ -159,7 +159,7 @@ private void heartbeatLoopBody() { // Sample free-space on the same cadence DataNode samples its load. Runs after // the async heartbeat dispatches so the OS call does not delay fanout. DiskCrash // is observed passively by the Ratis write-path, not polled here. - if (iterationIndex % LOAD_SAMPLING_INTERVAL == 0) { + if (heartbeatCounter.get() % 10 == 0) { DiskChecker.checkFreeRatioAndApply( ConfigNodeDescriptor.getInstance().getConf().getCriticalDirs(), CommonDescriptor.getInstance().getConfig().getDiskSpaceWarningThreshold()); diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java index 7ed189206acd..ea60ffd33447 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java @@ -253,12 +253,17 @@ import java.util.Map; import java.util.Set; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; /** ConfigNodeRPCServer exposes the interface that interacts with the DataNode */ public class ConfigNodeRPCServiceProcessor implements IConfigNodeRPCService.Iface { private static final Logger LOGGER = LoggerFactory.getLogger(ConfigNodeRPCServiceProcessor.class); + private static final int DISK_CHECK_INTERVAL_IN_HEARTBEATS = 10; + + private final AtomicLong heartbeatReceivedCounter = new AtomicLong(0); + protected final CommonConfig commonConfig; protected final ConfigNodeConfig configNodeConfig; protected final ConfigNode configNode; @@ -1117,7 +1122,7 @@ public TConfigNodeHeartbeatResp getConfigNodeHeartBeat(TConfigNodeHeartbeatReq h resp.setTimestamp(heartbeatReq.getTimestamp()); // Sample free-space on the same cadence DataNode samples its load. DiskCrash is observed // passively from the Ratis write-path on this node, not polled here. - if (heartbeatReceivedCounter.getAndIncrement() % HeartbeatService.LOAD_SAMPLING_INTERVAL == 0) { + if (heartbeatReceivedCounter.getAndIncrement() % DISK_CHECK_INTERVAL_IN_HEARTBEATS == 0) { DiskChecker.checkFreeRatioAndApply( configNodeConfig.getCriticalDirs(), commonConfig.getDiskSpaceWarningThreshold()); } diff --git a/iotdb-core/consensus/src/main/i18n/en/org/apache/iotdb/consensus/i18n/RatisMessages.java b/iotdb-core/consensus/src/main/i18n/en/org/apache/iotdb/consensus/i18n/RatisMessages.java index f980de691b26..0857f9107840 100644 --- a/iotdb-core/consensus/src/main/i18n/en/org/apache/iotdb/consensus/i18n/RatisMessages.java +++ b/iotdb-core/consensus/src/main/i18n/en/org/apache/iotdb/consensus/i18n/RatisMessages.java @@ -35,6 +35,9 @@ private RatisMessages() {} "null reply received in writeWithRetry for request "; public static final String LEADER_READ_ONLY_STEP_DOWN_FAILED = "leader {} read only, force step down failed due to, "; + public static final String + LOG_DISK_FAILURE_OBSERVED_IN_RATIS_GROUP_ARG_MARKING_NODE_READONLY_DISKCRASH_31E461D2 = + "Disk failure observed in Ratis group {}; marking node ReadOnly(DiskCrash)."; public static final String TRY_ADD_CONFLICTING_PEER = "{}: try to add a peer {} with conflicting id or address in {}"; public static final String IS_LEADER_REQUEST_FAILED = diff --git a/iotdb-core/consensus/src/main/i18n/zh/org/apache/iotdb/consensus/i18n/RatisMessages.java b/iotdb-core/consensus/src/main/i18n/zh/org/apache/iotdb/consensus/i18n/RatisMessages.java index fdf4660bf8f7..73cde2a8ce6c 100644 --- a/iotdb-core/consensus/src/main/i18n/zh/org/apache/iotdb/consensus/i18n/RatisMessages.java +++ b/iotdb-core/consensus/src/main/i18n/zh/org/apache/iotdb/consensus/i18n/RatisMessages.java @@ -34,6 +34,9 @@ private RatisMessages() {} "writeWithRetry 中收到空回复，请求为 "; public static final String LEADER_READ_ONLY_STEP_DOWN_FAILED = "leader {} 处于只读模式，强制降级失败，原因 "; + public static final String + LOG_DISK_FAILURE_OBSERVED_IN_RATIS_GROUP_ARG_MARKING_NODE_READONLY_DISKCRASH_31E461D2 = + "在 Ratis 组 {} 中发现磁盘故障，节点标记为 ReadOnly(DiskCrash)。"; public static final String TRY_ADD_CONFLICTING_PEER = "{}：尝试添加 ID 或地址冲突的 peer {} 到 {}"; public static final String IS_LEADER_REQUEST_FAILED = diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java index 4711726bcaec..44f481002d57 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java @@ -1036,7 +1036,8 @@ private RaftClientReply sendReconfiguration(RaftGroup newGroupConf) */ private void onDiskFailure(RaftGroupId groupId, Throwable cause) { logger.error( - "Disk failure observed in Ratis group {}; marking node ReadOnly(DiskCrash).", + RatisMessages + .LOG_DISK_FAILURE_OBSERVED_IN_RATIS_GROUP_ARG_MARKING_NODE_READONLY_DISKCRASH_31E461D2, groupId, cause); DiskChecker.apply(DiskChecker.DiskStatus.DISK_CRASH); 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 f2ecca222aa8..33b4941f0762 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 @@ -66,6 +66,7 @@ import org.apache.iotdb.commons.consensus.SchemaRegionId; import org.apache.iotdb.commons.consensus.index.ProgressIndex; import org.apache.iotdb.commons.consensus.index.ProgressIndexType; +import org.apache.iotdb.commons.disk.FolderManager; import org.apache.iotdb.commons.enums.DataPartitionTableGeneratorState; import org.apache.iotdb.commons.exception.IllegalPathException; import org.apache.iotdb.commons.exception.MetadataException; @@ -220,7 +221,6 @@ import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileManager; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; -import org.apache.iotdb.db.storageengine.rescon.disk.FolderManager; import org.apache.iotdb.db.storageengine.dataregion.wal.WALManager; import org.apache.iotdb.db.storageengine.dataregion.wal.WALWriteBlockStatus; import org.apache.iotdb.db.storageengine.rescon.quotas.DataNodeSpaceQuotaManager; @@ -2374,7 +2374,6 @@ public TDataNodeHeartbeatResp getDataNodeHeartBeat(TDataNodeHeartbeatReq req) th .forEach((key, value) -> regionRawDataSize.put(Integer.parseInt(key), value.getLeft())); resp.setDataRegionRawDataSize(regionRawDataSize); } - AuthorityChecker.getAuthorityFetcher().refreshToken(); updateWALBlockedStatus(); resp.setHeartbeatTimestamp(req.getHeartbeatTimestamp()); resp.setStatus(commonConfig.getNodeStatus().getStatus()); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALWriteBlockStatus.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALWriteBlockStatus.java index 70dad8a9d379..16facd9eafd4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALWriteBlockStatus.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALWriteBlockStatus.java @@ -24,18 +24,16 @@ public final class WALWriteBlockStatus { - public static final String WAL_BLOCKED = "WALBlocked"; - private WALWriteBlockStatus() {} public static void updateStatus(CommonConfig commonConfig, boolean longTermWriteBlocked) { if (longTermWriteBlocked) { if (NodeStatus.Running.equals(commonConfig.getNodeStatus())) { commonConfig.setNodeStatus(NodeStatus.ReadOnly); - commonConfig.setStatusReason(WAL_BLOCKED); + commonConfig.setStatusReason(NodeStatus.WAL_BLOCKED); } } else if (NodeStatus.ReadOnly.equals(commonConfig.getNodeStatus()) - && WAL_BLOCKED.equals(commonConfig.getStatusReason())) { + && NodeStatus.WAL_BLOCKED.equals(commonConfig.getStatusReason())) { commonConfig.setNodeStatus(NodeStatus.Running); commonConfig.setStatusReason(null); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/MemoryControlledWALEntryQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/MemoryControlledWALEntryQueue.java index 7ab08b9c86a7..63d16eb6db8a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/MemoryControlledWALEntryQueue.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/MemoryControlledWALEntryQueue.java @@ -57,7 +57,9 @@ public void put(WALEntry e) throws InterruptedException { synchronized (nonFullCondition) { while (!SystemInfo.getInstance().getWalBufferQueueMemoryBlock().allocate(elementSize)) { if (elementSize - > SystemInfo.getInstance().getWalBufferQueueMemoryBlock().getTotalMemorySizeInBytes()) { + > SystemInfo.getInstance() + .getWalBufferQueueMemoryBlock() + .getTotalMemorySizeInBytes()) { throw new IoTDBRuntimeException( String.format( StorageEngineMessages diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALWriteBlockStatusTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALWriteBlockStatusTest.java index 2546457d456b..c0dd2620a22f 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALWriteBlockStatusTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/WALWriteBlockStatusTest.java @@ -39,13 +39,12 @@ public void testRunningNodeTurnsReadOnlyWhenWalBlocked() { WALWriteBlockStatus.updateStatus(commonConfig, true); assertEquals(NodeStatus.ReadOnly, commonConfig.getNodeStatus()); - assertEquals(WALWriteBlockStatus.WAL_BLOCKED, commonConfig.getStatusReason()); + assertEquals(NodeStatus.WAL_BLOCKED, commonConfig.getStatusReason()); } @Test public void testWalBlockedReadOnlyNodeRecovers() { - CommonConfig commonConfig = - mockCommonConfig(NodeStatus.ReadOnly, WALWriteBlockStatus.WAL_BLOCKED); + CommonConfig commonConfig = mockCommonConfig(NodeStatus.ReadOnly, NodeStatus.WAL_BLOCKED); WALWriteBlockStatus.updateStatus(commonConfig, false); diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/cluster/NodeStatus.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/cluster/NodeStatus.java index ff64a666f98e..dc5b2cafdeba 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/cluster/NodeStatus.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/cluster/NodeStatus.java @@ -38,6 +38,7 @@ public enum NodeStatus { ReadOnly("ReadOnly"); public static final String DISK_FULL = "DiskFull"; public static final String DISK_CRASH = "DiskCrash"; + public static final String WAL_BLOCKED = "WALBlocked"; private final String status;