From bb29a71608bd0c81eda92237bde246be6fa59c2a Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Thu, 22 Oct 2026 17:48:34 +0800 Subject: [PATCH 1/4] [Bug] Recognize IoTConsensusV2 class name aliases on ConfigNode ConfigNode compared the raw value of data_region_consensus_protocol_class with ConsensusFactory.IOT_CONSENSUS_V2, while DataNode normalizes the configured value through ConsensusFactory.normalizeConsensusProtocolClass before using it. When the configuration used the legacy org.apache.iotdb.consensus.pipe.PipeConsensus name or the implementation class name org.apache.iotdb.consensus.pipe.IoTConsensusV2, ConfigNode concluded that IoTConsensusV2 was not configured and therefore: - createInitialConsensusPipes() never created the __consensus.* pipes on database creation; - checkAndRepairConsensusPipes() returned immediately, so PipeMetaSyncer could never repair missing pipes; - RouteBalancer disabled the automatic data region leader balance. Since the IoTConsensusV2 write path only writes the local state machine, the missing pipes silently degraded the data region replication factor to one: no error was reported and syncLag stayed at zero. Normalize the value once at the configuration boundary so CONF never holds an alias, log the normalization when the configuration file uses an alias, keep the comparison sites and the RouteBalancer cache defensive, and normalize the value sent by a joining ConfigNode so a mixed-version cluster is not rejected. --- .../confignode/i18n/ConfigNodeMessages.java | 3 ++ .../confignode/i18n/ConfigNodeMessages.java | 3 ++ .../confignode/conf/ConfigNodeConfig.java | 9 +++- .../confignode/conf/ConfigNodeDescriptor.java | 20 +++++++- .../confignode/manager/ConfigManager.java | 10 ++-- .../manager/load/balancer/RouteBalancer.java | 4 +- .../procedure/env/RegionMaintainHandler.java | 21 +++++++-- .../confignode/conf/ConfigNodeConfigTest.java | 25 ++++++++++ ...egionMaintainHandlerConsensusPipeTest.java | 47 +++++++++++++++++++ 9 files changed, 131 insertions(+), 11 deletions(-) diff --git a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java index 2b6d03294b00f..79d077c9336dd 100644 --- a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java +++ b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java @@ -447,6 +447,9 @@ public final class ConfigNodeMessages { public static final String SUCCESSFULLY_ON_DATANODE = "Successfully {} on DataNode: {}"; public static final String SUCCESSFULLY_REMOVE_CONFIGNODE_CURRENT_CONFIGNODEGROUP = "Successfully remove ConfigNode: {}. Current ConfigNodeGroup: {}"; + public static final String + LOG_NORMALIZE_DATA_REGION_CONSENSUS_PROTOCOL_CLASS_FROM_ARG_TO_ARG_FOR_COMPATIBILITY_6E662489 = + "Normalize data_region_consensus_protocol_class from {} to {} for compatibility."; public static final String SUCCESSFULLY_SETUP_INTERNAL_SERVICES = "Successfully setup internal services."; public static final String SUCCESSFULLY_UPDATE_NODE_S_VERSION = diff --git a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java index 11eaa6324a946..90dbd70ca5a32 100644 --- a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java +++ b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java @@ -415,6 +415,9 @@ public final class ConfigNodeMessages { public static final String SUCCESSFULLY_ON_DATANODE = "在 DataNode: {} 上成功执行 {}"; public static final String SUCCESSFULLY_REMOVE_CONFIGNODE_CURRENT_CONFIGNODEGROUP = "成功移除 ConfigNode: {}。当前 ConfigNodeGroup: {}"; + public static final String + LOG_NORMALIZE_DATA_REGION_CONSENSUS_PROTOCOL_CLASS_FROM_ARG_TO_ARG_FOR_COMPATIBILITY_6E662489 = + "为兼容性将 data_region_consensus_protocol_class 从 {} 规范化为 {}。"; public static final String SUCCESSFULLY_SETUP_INTERNAL_SERVICES = "成功初始化内部服务。"; public static final String SUCCESSFULLY_UPDATE_NODE_S_VERSION = "成功更新 Node {} 的版本。"; public static final String SYSTEMPROPERTIES_NORMALIZE_FROM_TO_FOR_COMPATIBILITY = 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 c5facfb371ece..8094b8274dd9a 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 @@ -642,7 +642,14 @@ public String getDataRegionConsensusProtocolClass() { } public void setDataRegionConsensusProtocolClass(String dataRegionConsensusProtocolClass) { - this.dataRegionConsensusProtocolClass = dataRegionConsensusProtocolClass; + // DataNode normalizes the legacy PipeConsensus / implementation class names before it + // dispatches the consensus implementation, but ConfigNode used to keep and compare the raw + // value. Keeping the in-memory value canonical makes every ConfigNode decision (consensus + // pipe creation and repair, leader balance, ...) agree with DataNode, no matter whether the + // value came from iotdb-system.properties, confignode-system.properties or programmatic + // configuration. + this.dataRegionConsensusProtocolClass = + ConsensusFactory.normalizeConsensusProtocolClass(dataRegionConsensusProtocolClass); } public String getIotConsensusV2Mode() { diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java index e9ab7657c50a6..af1f19808e7d0 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java @@ -36,6 +36,7 @@ import org.apache.iotdb.confignode.manager.load.balancer.router.priority.IPriorityBalancer; import org.apache.iotdb.confignode.manager.load.cache.IFailureDetector; import org.apache.iotdb.confignode.manager.partition.RegionGroupExtensionPolicy; +import org.apache.iotdb.consensus.ConsensusFactory; import org.apache.iotdb.metrics.config.MetricConfigDescriptor; import org.apache.iotdb.metrics.utils.NodeType; @@ -52,6 +53,7 @@ import java.net.UnknownHostException; import java.nio.charset.StandardCharsets; import java.util.Collections; +import java.util.Objects; import java.util.Optional; public class ConfigNodeDescriptor { @@ -213,9 +215,23 @@ private void loadProperties(TrimProperties properties) throws BadNodeUrlExceptio properties.getProperty( "schema_replication_factor", String.valueOf(conf.getSchemaReplicationFactor())))); - conf.setDataRegionConsensusProtocolClass( + final String dataRegionConsensusProtocolClass = properties.getProperty( - "data_region_consensus_protocol_class", conf.getDataRegionConsensusProtocolClass())); + "data_region_consensus_protocol_class", conf.getDataRegionConsensusProtocolClass()); + final String normalizedDataRegionConsensusProtocolClass = + ConsensusFactory.normalizeConsensusProtocolClass(dataRegionConsensusProtocolClass); + if (!Objects.equals( + dataRegionConsensusProtocolClass, normalizedDataRegionConsensusProtocolClass)) { + // DataNode already normalizes these names, so ConfigNode must do the same. Otherwise it + // treats an aliased IoTConsensusV2 configuration as "not V2", never creates the + // __consensus.* pipes that V2 replication depends on, and the degradation stays silent. + LOGGER.warn( + ConfigNodeMessages + .LOG_NORMALIZE_DATA_REGION_CONSENSUS_PROTOCOL_CLASS_FROM_ARG_TO_ARG_FOR_COMPATIBILITY_6E662489, + dataRegionConsensusProtocolClass, + normalizedDataRegionConsensusProtocolClass); + } + conf.setDataRegionConsensusProtocolClass(normalizedDataRegionConsensusProtocolClass); conf.setIotConsensusV2Mode( properties.getProperty("iot_consensus_v2_mode", conf.getIotConsensusV2Mode())); diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java index 8b0b2d4751fe0..7659475d8418c 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java @@ -268,6 +268,7 @@ import org.apache.iotdb.confignode.rpc.thrift.TTimeSlotList; import org.apache.iotdb.confignode.rpc.thrift.TUnsetSchemaTemplateReq; import org.apache.iotdb.confignode.rpc.thrift.TUnsubscribeReq; +import org.apache.iotdb.consensus.ConsensusFactory; import org.apache.iotdb.consensus.common.DataSet; import org.apache.iotdb.consensus.exception.ConsensusException; import org.apache.iotdb.db.schemaengine.template.TemplateAlterOperationType; @@ -1517,9 +1518,12 @@ public TSStatus checkConfigNodeGlobalConfig(TConfigNodeRegisterReq req) { .EXCEPTION_REJECT_REGISTER_PLEASE_ENSURE_THAT_THE_PARAMETER_ARG_IS_CONSISTENT_WITH_THE_SEED_CONFIGNODE_C072CE7C, ManagerMessages.MESSAGE_CONFIG_NODE_CONSENSUS_PROTOCOL_CLASS_D0F437AF)); } - if (!clusterParameters - .getDataRegionConsensusProtocolClass() - .equals(CONF.getDataRegionConsensusProtocolClass())) { + // Normalize the joining ConfigNode's value as well, so a legacy IoTConsensusV2 class name sent + // by a not-yet-upgraded ConfigNode is not mistaken for a different consensus protocol. + if (!Objects.equals( + ConsensusFactory.normalizeConsensusProtocolClass( + clusterParameters.getDataRegionConsensusProtocolClass()), + CONF.getDataRegionConsensusProtocolClass())) { return errorStatus.setMessage( String.format( ManagerMessages diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RouteBalancer.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RouteBalancer.java index 63575a9a2c11a..65d1061624d4b 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RouteBalancer.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RouteBalancer.java @@ -84,7 +84,9 @@ public class RouteBalancer implements IClusterStatusSubscriber { private static final String SCHEMA_REGION_CONSENSUS_PROTOCOL_CLASS = CONF.getSchemaRegionConsensusProtocolClass(); private static final String DATA_REGION_CONSENSUS_PROTOCOL_CLASS = - CONF.getDataRegionConsensusProtocolClass(); + // Normalize here as well so the cached value can never disable DataRegion leader balance + // merely because the configuration used a legacy IoTConsensusV2 class name. + ConsensusFactory.normalizeConsensusProtocolClass(CONF.getDataRegionConsensusProtocolClass()); private static final boolean IS_ENABLE_AUTO_LEADER_BALANCE_FOR_DATA_REGION = (CONF.isEnableAutoLeaderBalanceForRatisConsensus() && ConsensusFactory.RATIS_CONSENSUS.equals(DATA_REGION_CONSENSUS_PROTOCOL_CLASS)) diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java index 04100d05f9cb0..1f2eb6e8119a4 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java @@ -56,6 +56,7 @@ import org.apache.iotdb.confignode.manager.partition.PartitionManager; import org.apache.iotdb.confignode.procedure.exception.ProcedureException; import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq; +import org.apache.iotdb.consensus.ConsensusFactory; import org.apache.iotdb.consensus.pipe.consensuspipe.ConsensusPipeName; import org.apache.iotdb.mpp.rpc.thrift.TCreatePeerReq; import org.apache.iotdb.mpp.rpc.thrift.TMaintainPeerReq; @@ -178,7 +179,7 @@ public TSStatus createNewRegionPeer(TConsensusGroupId regionId, TDataNodeLocatio List currentPeerNodes; if (TConsensusGroupType.DataRegion.equals(regionId.getType()) && (IOT_CONSENSUS.equals(CONF.getDataRegionConsensusProtocolClass()) - || IOT_CONSENSUS_V2.equals(CONF.getDataRegionConsensusProtocolClass()))) { + || isIoTConsensusV2Configured())) { // parameter of createPeer for MultiLeader should be all peers; callers (e.g. // AddRegionPeerProcedure) may have already inserted destDataNode into the partition // table before reaching here, so append only when not already present to avoid @@ -556,7 +557,7 @@ public void createConsensusPipesForAddPeer( * are activated. */ public void createInitialConsensusPipes(CreateRegionGroupsPlan persistPlan) { - if (!IOT_CONSENSUS_V2.equals(CONF.getDataRegionConsensusProtocolClass())) { + if (!isIoTConsensusV2Configured()) { return; } @@ -624,7 +625,7 @@ public void dropConsensusPipesForRemovePeer( * Creates missing pipes, drops unexpected pipes, and restarts stopped pipes. */ public void checkAndRepairConsensusPipes() { - if (!IOT_CONSENSUS_V2.equals(CONF.getDataRegionConsensusProtocolClass())) { + if (!isIoTConsensusV2Configured()) { return; } @@ -710,7 +711,19 @@ private static TDataNodeLocation findLocationByNodeId( private boolean isIoTConsensusV2DataRegion(TConsensusGroupId regionId) { return TConsensusGroupType.DataRegion.equals(regionId.getType()) - && IOT_CONSENSUS_V2.equals(CONF.getDataRegionConsensusProtocolClass()); + && isIoTConsensusV2Configured(); + } + + /** + * Compares the configured protocol through the same normalization DataNode applies, so a legacy + * PipeConsensus / implementation class name in the configuration is still recognized as + * IoTConsensusV2. Without this, consensus pipes would silently never be created or repaired, and + * V2 replication (which only writes the local state machine) would stop working. + */ + private static boolean isIoTConsensusV2Configured() { + return IOT_CONSENSUS_V2.equals( + ConsensusFactory.normalizeConsensusProtocolClass( + CONF.getDataRegionConsensusProtocolClass())); } private TCreatePipeReq buildConsensusPipeReq( diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/conf/ConfigNodeConfigTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/conf/ConfigNodeConfigTest.java index da016c5e16324..f635a5e89e5f1 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/conf/ConfigNodeConfigTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/conf/ConfigNodeConfigTest.java @@ -20,12 +20,37 @@ package org.apache.iotdb.confignode.conf; import org.apache.iotdb.commons.snapshot.SnapshotStreamFactory; +import org.apache.iotdb.consensus.ConsensusFactory; import org.junit.Assert; import org.junit.Test; public class ConfigNodeConfigTest { + @Test + public void testDataRegionConsensusProtocolClassNormalizesLegacyNames() { + final ConfigNodeConfig configNodeConfig = new ConfigNodeConfig(); + + // Both the old PipeConsensus name and the implementation class name must be stored in their + // canonical form, otherwise ConfigNode compares them against IOT_CONSENSUS_V2 and concludes + // that IoTConsensusV2 is not configured. + configNodeConfig.setDataRegionConsensusProtocolClass(ConsensusFactory.LEGACY_IOT_CONSENSUS_V2); + Assert.assertEquals( + ConsensusFactory.IOT_CONSENSUS_V2, configNodeConfig.getDataRegionConsensusProtocolClass()); + + configNodeConfig.setDataRegionConsensusProtocolClass(ConsensusFactory.REAL_IOT_CONSENSUS_V2); + Assert.assertEquals( + ConsensusFactory.IOT_CONSENSUS_V2, configNodeConfig.getDataRegionConsensusProtocolClass()); + + // Canonical and unrelated values must be preserved. + configNodeConfig.setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS_V2); + Assert.assertEquals( + ConsensusFactory.IOT_CONSENSUS_V2, configNodeConfig.getDataRegionConsensusProtocolClass()); + configNodeConfig.setDataRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS); + Assert.assertEquals( + ConsensusFactory.RATIS_CONSENSUS, configNodeConfig.getDataRegionConsensusProtocolClass()); + } + @Test public void testSnapshotBufferSizeMaxDefault() { final ConfigNodeConfig configNodeConfig = new ConfigNodeConfig(); diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandlerConsensusPipeTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandlerConsensusPipeTest.java index a99d6f52fb819..939bc98784c25 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandlerConsensusPipeTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandlerConsensusPipeTest.java @@ -30,6 +30,7 @@ import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStatus; import org.apache.iotdb.confignode.client.sync.CnToDnSyncRequestType; import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor; +import org.apache.iotdb.confignode.consensus.request.write.region.CreateRegionGroupsPlan; import org.apache.iotdb.confignode.manager.ConfigManager; import org.apache.iotdb.confignode.manager.ProcedureManager; import org.apache.iotdb.confignode.manager.load.LoadManager; @@ -49,6 +50,8 @@ import java.util.Map; import static org.apache.iotdb.consensus.ConsensusFactory.IOT_CONSENSUS_V2; +import static org.apache.iotdb.consensus.ConsensusFactory.LEGACY_IOT_CONSENSUS_V2; +import static org.apache.iotdb.consensus.ConsensusFactory.REAL_IOT_CONSENSUS_V2; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; @@ -179,6 +182,50 @@ public void testCreatesMissingPipes() { verify(procedureManager, never()).startConsensusPipe(any()); } + @Test + public void testCreatesMissingPipesWhenLegacyConsensusProtocolAliasConfigured() { + // iotdb-system.properties may keep the pre-rename IoTConsensusV2 class name. ConfigNode must + // treat it exactly like the canonical name; otherwise the periodic self-healing entry point + // never creates the consensus pipes that V2 replication depends on. + ConfigNodeDescriptor.getInstance() + .getConf() + .setDataRegionConsensusProtocolClass(LEGACY_IOT_CONSENSUS_V2); + + TDataNodeLocation loc1 = makeLocation(1, "127.0.0.1", 40010); + TDataNodeLocation loc2 = makeLocation(2, "127.0.0.2", 40010); + TRegionReplicaSet replicaSet = makeReplicaSet(100, loc1, loc2); + + TConsensusGroupId groupId = new TConsensusGroupId(TConsensusGroupType.DataRegion, 100); + Map replicaSets = new HashMap<>(); + replicaSets.put(groupId, replicaSet); + when(partitionManager.getAllReplicaSetsMap(TConsensusGroupType.DataRegion)) + .thenReturn(replicaSets); + when(pipeTaskCoordinator.getConsensusPipeStatusMap()).thenReturn(Collections.emptyMap()); + + handler.checkAndRepairConsensusPipes(); + + verify(procedureManager, times(2)).createConsensusPipeAsync(any(TCreatePipeReq.class)); + } + + @Test + public void testCreatesInitialPipesWhenImplementationClassNameConfigured() { + // Writing the implementation class name into the configuration is a supported way to select + // IoTConsensusV2, so database creation must also create the initial consensus pipes. + ConfigNodeDescriptor.getInstance() + .getConf() + .setDataRegionConsensusProtocolClass(REAL_IOT_CONSENSUS_V2); + + TDataNodeLocation loc1 = makeLocation(1, "127.0.0.1", 40010); + TDataNodeLocation loc2 = makeLocation(2, "127.0.0.2", 40010); + TRegionReplicaSet replicaSet = makeReplicaSet(100, loc1, loc2); + CreateRegionGroupsPlan plan = new CreateRegionGroupsPlan(); + plan.addRegionGroup("root.test", replicaSet); + + handler.createInitialConsensusPipes(plan); + + verify(procedureManager, times(2)).createConsensusPipeAsync(any(TCreatePipeReq.class)); + } + @Test public void testDropsUnexpectedPipes() { when(partitionManager.getAllReplicaSetsMap(TConsensusGroupType.DataRegion)) From beb645335a024dd291af3f4a30bbc1534431e08d Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Fri, 6 Nov 2026 10:00:21 +0800 Subject: [PATCH 2/4] [ConfigNode] Reject an unrecognized data_region_consensus_protocol_class at startup ConfigNode decides whether it has to create consensus pipes by comparing data_region_consensus_protocol_class with the canonical IoTConsensusV2 class name. An unrecognized value therefore silently disables IoTConsensusV2 replication: data_replication_factor=2 degrades to a single replica without any error, and the pipe based syncLag metric keeps reporting 0. Fail the startup check with a ConfigurationException that lists the accepted classes instead of starting a cluster that cannot replicate. --- .../confignode/i18n/ConfigNodeMessages.java | 4 + .../confignode/i18n/ConfigNodeMessages.java | 4 + .../conf/ConfigNodeStartupCheck.java | 32 +++++++ .../conf/ConfigNodeStartupCheckTest.java | 88 +++++++++++++++++++ 4 files changed, 128 insertions(+) create mode 100644 iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/conf/ConfigNodeStartupCheckTest.java diff --git a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java index 79d077c9336dd..d9a319039bfde 100644 --- a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java +++ b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java @@ -628,6 +628,10 @@ private ConfigNodeMessages() {} public static final String EXCEPTION_MESSAGE_E81C4E4F = "or"; public static final String EXCEPTION_AVAILABLE_ONLY_DATA_REPLICATION_FACTOR_SET_1_71748D3D = "available only when data_replication_factor is set to 1"; public static final String EXCEPTION_ARG_ARG_6E068B23 = "%s or %s"; + public static final String EXCEPTION_ARG_ARG_ARG_OR_ARG_E950ED19 = "%s, %s, %s or %s"; + public static final String + EXCEPTION_UNRECOGNIZED_DATA_REGION_CONSENSUS_PROTOCOL_CLASS_IS_SET_DAB37EAF = + "an unrecognized data_region_consensus_protocol_class is set"; public static final String EXCEPTION_SCHEMAREGION_DOESN_T_SUPPORT_ORG_APACHE_IOTDB_CONSENSUS_IOT_IOTCONSENSUS_84350FD1 = "the SchemaRegion doesn't support org.apache.iotdb.consensus.iot.IoTConsensus"; public static final String EXCEPTION_SCHEMAREGION_DOESN_T_SUPPORT_ORG_APACHE_IOTDB_CONSENSUS_IOT_IOTCONSENSUSV2_BA353C6D = "the SchemaRegion doesn't support org.apache.iotdb.consensus.iot.IoTConsensusV2"; public static final String EXCEPTION_GREEDY_MIN_COST_FLOW_HASH_C07DA2EE = "GREEDY or MIN_COST_FLOW or HASH"; diff --git a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java index 90dbd70ca5a32..bd5803f32f33d 100644 --- a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java +++ b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java @@ -608,6 +608,10 @@ private ConfigNodeMessages() {} public static final String EXCEPTION_AVAILABLE_ONLY_DATA_REPLICATION_FACTOR_SET_1_71748D3D = "仅在 data_replication_factor 设置为 1 时可用"; public static final String EXCEPTION_ARG_ARG_6E068B23 = "%s 或 %s"; + public static final String EXCEPTION_ARG_ARG_ARG_OR_ARG_E950ED19 = "%s、%s、%s 或 %s"; + public static final String + EXCEPTION_UNRECOGNIZED_DATA_REGION_CONSENSUS_PROTOCOL_CLASS_IS_SET_DAB37EAF = + "设置了无法识别的 data_region_consensus_protocol_class"; public static final String EXCEPTION_SCHEMAREGION_DOESN_T_SUPPORT_ORG_APACHE_IOTDB_CONSENSUS_IOT_IOTCONSENSUS_84350FD1 = "SchemaRegion 不支持 org.apache.iotdb.consensus.iot.IoTConsensus"; diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeStartupCheck.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeStartupCheck.java index fb7d674d0822a..7ad8c84555946 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeStartupCheck.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeStartupCheck.java @@ -173,6 +173,13 @@ private void checkGlobalConfig() throws ConfigurationException { .EXCEPTION_SCHEMAREGION_DOESN_T_SUPPORT_ORG_APACHE_IOTDB_CONSENSUS_IOT_IOTCONSENSUSV2_BA353C6D); } + // The data region consensus protocol is limited. An unrecognized value is not cosmetic: + // ConfigNode compares this value to decide whether it has to create consensus pipes, so a class + // name it does not know silently disables IoTConsensusV2 replication. Fail fast and print the + // accepted values; IoTConsensusV2 must be written as + // org.apache.iotdb.consensus.iot.IoTConsensusV2. + checkDataRegionConsensusProtocolClass(); + // The leader distribution policy is limited if (!AbstractLeaderBalancer.GREEDY_POLICY.equals(CONF.getLeaderDistributionPolicy()) && !AbstractLeaderBalancer.CFS_POLICY.equals(CONF.getLeaderDistributionPolicy()) @@ -226,6 +233,31 @@ private void checkGlobalConfig() throws ConfigurationException { } } + /** + * Rejects a data_region_consensus_protocol_class that this release cannot act on. Package-private + * so that the accepted / rejected values can be unit tested without booting a ConfigNode. + */ + static void checkDataRegionConsensusProtocolClass() throws ConfigurationException { + final String dataRegionConsensusProtocolClass = CONF.getDataRegionConsensusProtocolClass(); + if (ConsensusFactory.SIMPLE_CONSENSUS.equals(dataRegionConsensusProtocolClass) + || ConsensusFactory.RATIS_CONSENSUS.equals(dataRegionConsensusProtocolClass) + || ConsensusFactory.IOT_CONSENSUS.equals(dataRegionConsensusProtocolClass) + || ConsensusFactory.IOT_CONSENSUS_V2.equals(dataRegionConsensusProtocolClass)) { + return; + } + throw new ConfigurationException( + ConfigNodeMessages.DATA_REGION_CONSENSUS_PROTOCOL_CLASS, + dataRegionConsensusProtocolClass, + String.format( + ConfigNodeMessages.EXCEPTION_ARG_ARG_ARG_OR_ARG_E950ED19, + ConsensusFactory.SIMPLE_CONSENSUS, + ConsensusFactory.RATIS_CONSENSUS, + ConsensusFactory.IOT_CONSENSUS, + ConsensusFactory.IOT_CONSENSUS_V2), + ConfigNodeMessages + .EXCEPTION_UNRECOGNIZED_DATA_REGION_CONSENSUS_PROTOCOL_CLASS_IS_SET_DAB37EAF); + } + private void createDirsIfNecessary() throws IOException { // If systemDir does not exist, create systemDir File systemDir = new File(CONF.getSystemDir()); diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/conf/ConfigNodeStartupCheckTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/conf/ConfigNodeStartupCheckTest.java new file mode 100644 index 0000000000000..d31a72baf3b08 --- /dev/null +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/conf/ConfigNodeStartupCheckTest.java @@ -0,0 +1,88 @@ +/* + * 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.conf; + +import org.apache.iotdb.commons.exception.ConfigurationException; +import org.apache.iotdb.consensus.ConsensusFactory; + +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +public class ConfigNodeStartupCheckTest { + + private final ConfigNodeConfig conf = ConfigNodeDescriptor.getInstance().getConf(); + private String originalDataRegionConsensusProtocolClass; + + @Before + public void setUp() { + originalDataRegionConsensusProtocolClass = conf.getDataRegionConsensusProtocolClass(); + } + + @After + public void tearDown() { + conf.setDataRegionConsensusProtocolClass(originalDataRegionConsensusProtocolClass); + } + + @Test + public void testAcceptsEverySupportedConsensusProtocolClass() throws ConfigurationException { + for (final String protocolClass : + new String[] { + ConsensusFactory.SIMPLE_CONSENSUS, + ConsensusFactory.RATIS_CONSENSUS, + ConsensusFactory.IOT_CONSENSUS, + ConsensusFactory.IOT_CONSENSUS_V2 + }) { + conf.setDataRegionConsensusProtocolClass(protocolClass); + ConfigNodeStartupCheck.checkDataRegionConsensusProtocolClass(); + } + } + + @Test + public void testAcceptsLegacyAliasesBecauseTheyAreNormalized() throws ConfigurationException { + // Both aliases used to be accepted by DataNode but silently ignored by ConfigNode, so they must + // keep passing this check: the value is normalized before it is compared. + for (final String alias : + new String[] { + ConsensusFactory.LEGACY_IOT_CONSENSUS_V2, ConsensusFactory.REAL_IOT_CONSENSUS_V2 + }) { + conf.setDataRegionConsensusProtocolClass(alias); + ConfigNodeStartupCheck.checkDataRegionConsensusProtocolClass(); + Assert.assertEquals( + ConsensusFactory.IOT_CONSENSUS_V2, conf.getDataRegionConsensusProtocolClass()); + } + } + + @Test + public void testRejectsUnrecognizedConsensusProtocolClass() { + // An unknown class name used to be accepted silently and then degraded the whole cluster to one + // replica, so startup must fail and point at the accepted values. + conf.setDataRegionConsensusProtocolClass("org.apache.iotdb.consensus.pipe.PipeConsensusV3"); + try { + ConfigNodeStartupCheck.checkDataRegionConsensusProtocolClass(); + Assert.fail("An unrecognized data_region_consensus_protocol_class must be rejected"); + } catch (final ConfigurationException e) { + Assert.assertTrue(e.getMessage().contains("data_region_consensus_protocol_class")); + Assert.assertTrue(e.getMessage().contains(ConsensusFactory.IOT_CONSENSUS_V2)); + Assert.assertTrue(e.getMessage().contains(ConsensusFactory.RATIS_CONSENSUS)); + } + } +} From 606eef2c0c60e293d9727df4d9075380f419833d Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Fri, 6 Nov 2026 10:36:20 +0800 Subject: [PATCH 3/4] [ConfigNode] Reconcile consensus pipes as soon as the pipe count mismatches The reconciliation ran on a fixed round cadence, so a region that lost its consensus pipes could stay unreplicated for up to five sync intervals even after the starvation fix made the check reachable again. Probe the pipe count on every sync round instead: the number of ordered replica pairs over the DataRegion replica sets is exactly the number of consensus pipes IoTConsensusV2 requires, and comparing it with the number of existing consensus pipes neither builds a pipe name nor copies a replica set (the new PartitionManager#getReplicaPairCount iterates the partition tables in place). A mismatch means at least one region has lost its replication channel, so the full reconciliation runs immediately and keeps running every round until the counts match again. Equal counts still trigger the full check periodically, because a same-size but different pipe set cannot be detected by counting. The probe is skipped entirely when IoTConsensusV2 is not configured, and a single-replica region expects no pipe at all, so single-replica clusters never report a mismatch. --- .../confignode/i18n/ManagerMessages.java | 2 + .../confignode/i18n/ManagerMessages.java | 2 + .../manager/partition/PartitionManager.java | 13 +++ .../coordinator/runtime/PipeMetaSyncer.java | 24 +++++- .../coordinator/task/PipeTaskCoordinator.java | 4 + .../partition/DatabasePartitionTable.java | 19 +++++ .../persistence/partition/PartitionInfo.java | 17 ++++ .../persistence/partition/RegionGroup.java | 7 ++ .../persistence/pipe/PipeTaskInfo.java | 20 +++++ .../procedure/env/RegionMaintainHandler.java | 20 +++++ .../runtime/PipeMetaSyncerTest.java | 79 +++++++++++++++++++ ...egionMaintainHandlerConsensusPipeTest.java | 39 +++++++++ 12 files changed, 245 insertions(+), 1 deletion(-) create mode 100644 iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeMetaSyncerTest.java 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 a6df250eed70d..9d8466f8719d9 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 @@ -125,6 +125,8 @@ public final class ManagerMessages { "Failed to alter pipe {}. Result status: {}."; public static final String FAILED_TO_CHECK_AND_REPAIR_CONSENSUS_PIPES = "Failed to check and repair consensus pipes"; + public static final String LOG_FAILED_TO_PROBE_THE_NUMBER_OF_CONSENSUS_PIPES_FALL_BACK_TO_THE_PERIODIC_CONSENSUS_PIPE_CHECK_E3E527AF = + "Failed to probe the number of consensus pipes, fall back to the periodic consensus pipe check."; public static final String FAILED_TO_CHECK_PASSWORD_FOR_PIPE = "Failed to check password for pipe %s."; public static final String FAILED_TO_CLOSE_CONSUMER_IN_CONSUMER_GROUP_RESULT_STATUS = 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 34aa75da47957..a636768f9461f 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 @@ -124,6 +124,8 @@ public final class ManagerMessages { "修改 pipe {} 失败。结果状态：{}。"; public static final String FAILED_TO_CHECK_AND_REPAIR_CONSENSUS_PIPES = "检查并修复 consensus pipe 失败"; + public static final String LOG_FAILED_TO_PROBE_THE_NUMBER_OF_CONSENSUS_PIPES_FALL_BACK_TO_THE_PERIODIC_CONSENSUS_PIPE_CHECK_E3E527AF = + "探测 consensus pipe 数量失败，将回退到定期检查 consensus pipe。"; public static final String FAILED_TO_CHECK_PASSWORD_FOR_PIPE = "检查 pipe %s 的密码失败。"; public static final String FAILED_TO_CLOSE_CONSUMER_IN_CONSUMER_GROUP_RESULT_STATUS = diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java index cacc7ceadcbd9..162ff69154978 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java @@ -865,6 +865,19 @@ public Map getAllReplicaSetsMap(TConsensus Collectors.toMap(TRegionReplicaSet::getRegionId, regionReplicaSet -> regionReplicaSet)); } + /** + * Only leader use this interface. + * + * @param type The specified TConsensusGroupType + * @return The number of ordered replica pairs of all Regions with the specified + * TConsensusGroupType, i.e. the sum of N * (N - 1) over those Regions. Unlike {@link + * #getAllReplicaSetsMap(TConsensusGroupType)}, this does not copy any replica set, so it is + * cheap enough to be called on every pipe meta sync round. + */ + public long getReplicaPairCount(TConsensusGroupType type) { + return partitionInfo.getReplicaPairCount(type); + } + /** * Only leader use this interface. * diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeMetaSyncer.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeMetaSyncer.java index 0d348e16417e6..aab6272dc47d2 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeMetaSyncer.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeMetaSyncer.java @@ -118,7 +118,11 @@ private synchronized void sync() { pipeAutoRestartRoundCounter.set(0); } - if (consensusPipeCheckRoundCounter.incrementAndGet() >= CONSENSUS_PIPE_CHECK_INTERVAL_ROUND) { + // A count mismatch means a region lost its replication channel, so reconcile now instead of + // waiting for the periodic round. Equal counts are not proof of correctness, so the periodic + // round is kept. + if (hasConsensusPipeCountMismatch() + || consensusPipeCheckRoundCounter.incrementAndGet() >= CONSENSUS_PIPE_CHECK_INTERVAL_ROUND) { consensusPipeCheckRoundCounter.set(0); checkAndRepairConsensusPipes(); } @@ -176,6 +180,24 @@ private boolean autoRestartWithLock() { } } + private boolean hasConsensusPipeCountMismatch() { + try { + return configManager + .getProcedureManager() + .getEnv() + .getRegionMaintainHandler() + .hasConsensusPipeCountMismatch(); + } catch (Exception e) { + // The probe must never break the syncer; the periodic check still runs. + PipeLogger.log( + LOGGER::warn, + e, + ManagerMessages + .LOG_FAILED_TO_PROBE_THE_NUMBER_OF_CONSENSUS_PIPES_FALL_BACK_TO_THE_PERIODIC_CONSENSUS_PIPE_CHECK_E3E527AF); + return false; + } + } + private void checkAndRepairConsensusPipes() { try { configManager diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/task/PipeTaskCoordinator.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/task/PipeTaskCoordinator.java index feb78adb15ce0..d32c2f1dba0ba 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/task/PipeTaskCoordinator.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/task/PipeTaskCoordinator.java @@ -461,6 +461,10 @@ public Map getConsensusPipeStatusMap() { return pipeTaskInfo.getConsensusPipeStatusMap(); } + public int getConsensusPipeCount() { + return pipeTaskInfo.getConsensusPipeCount(); + } + /** Caller should ensure that the method is called in the write lock of {@link #pipeTaskInfo}. */ public void updateLastSyncedVersion() { pipeTaskInfo.updateLastSyncedVersion(); diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/DatabasePartitionTable.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/DatabasePartitionTable.java index 3e3170af19f6e..4d393362ed031 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/DatabasePartitionTable.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/DatabasePartitionTable.java @@ -140,6 +140,25 @@ public List getAllReplicaSets(TConsensusGroupType type) { return result; } + /** + * Counts the ordered replica pairs of all Regions with the specified TConsensusGroupType, i.e. + * the sum of N * (N - 1) over those Regions, without copying any replica set. Callers use it to + * probe, on a hot path, how many consensus pipes the partition table requires. + * + * @param type The specified TConsensusGroupType + * @return The number of ordered replica pairs of the specified Region type + */ + public long getReplicaPairCount(TConsensusGroupType type) { + long count = 0; + for (RegionGroup regionGroup : regionGroupMap.values()) { + if (type.equals(regionGroup.getId().getType())) { + final long replicaNum = regionGroup.getReplicaNum(); + count += replicaNum * (replicaNum - 1); + } + } + return count; + } + /** * Get all RegionGroups currently owned by the specified Database. * diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/PartitionInfo.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/PartitionInfo.java index 900ab7ab4ae0c..531b0fabda045 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/PartitionInfo.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/PartitionInfo.java @@ -738,6 +738,23 @@ public List getAllReplicaSets(TConsensusGroupType type) { return result; } + /** + * Only leader use this interface. + * + * @param type The specified TConsensusGroupType + * @return The number of ordered replica pairs of all Regions with the specified + * TConsensusGroupType, i.e. the sum of N * (N - 1) over those Regions + */ + public long getReplicaPairCount(TConsensusGroupType type) { + long count = 0; + for (DatabasePartitionTable databasePartitionTable : databasePartitionTables.values()) { + if (databasePartitionTable.isNotPreDeleted()) { + count += databasePartitionTable.getReplicaPairCount(type); + } + } + return count; + } + /** * Only leader use this interface. * diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/RegionGroup.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/RegionGroup.java index 6f9860b6bce61..8baadf3901296 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/RegionGroup.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/RegionGroup.java @@ -75,6 +75,13 @@ public synchronized TRegionReplicaSet getReplicaSet() { return replicaSet.deepCopy(); } + /** + * @return the number of replicas of this Region, without copying the replica set + */ + public synchronized int getReplicaNum() { + return replicaSet.getDataNodeLocationsSize(); + } + /** * Update the DataNodeLocation in TRegionReplicaSet if necessary. * diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java index 269be5b4dee11..9b90ea7006e98 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java @@ -877,6 +877,26 @@ public long generateUniqueCreationTime(final String pipeName) { } } + /** + * Counts the consensus pipes without materializing their names. Callers use this to probe whether + * a consensus pipe reconciliation is needed: building {@link #getConsensusPipeStatusMap()} is far + * more expensive than counting, and the probe runs on every pipe meta sync round. + */ + public int getConsensusPipeCount() { + acquireReadLock(); + try { + int count = 0; + for (final PipeMeta pipeMeta : pipeMetaKeeper.getPipeMetaList()) { + if (PipeType.CONSENSUS.equals(pipeMeta.getStaticMeta().getPipeType())) { + ++count; + } + } + return count; + } finally { + releaseReadLock(); + } + } + public Map getConsensusPipeStatusMap() { acquireReadLock(); try { diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java index 1f2eb6e8119a4..3d3d5180c2997 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java @@ -619,6 +619,26 @@ public void dropConsensusPipesForRemovePeer( } } + /** + * Cheap probe used by PipeMetaSyncer on every round to decide whether a full {@link + * #checkAndRepairConsensusPipes()} is needed. It only compares how many consensus pipes the + * replica sets require (N * (N - 1) per region) with how many actually exist, so it neither + * builds a pipe name nor copies a replica set. A mismatch is a sufficient reason to reconcile + * immediately; equal counts are still reconciled periodically to catch same-size but different + * pipe sets. + * + *

Regions with a single replica expect no pipe at all, so single-replica clusters never report + * a mismatch. + */ + public boolean hasConsensusPipeCountMismatch() { + if (!isIoTConsensusV2Configured()) { + return false; + } + + return configManager.getPartitionManager().getReplicaPairCount(TConsensusGroupType.DataRegion) + != configManager.getPipeManager().getPipeTaskCoordinator().getConsensusPipeCount(); + } + /** * Periodically called by PipeMetaSyncer to reconcile expected consensus pipes (derived from * PartitionManager replica sets) with actually existing consensus pipes (from PipeTaskInfo). diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeMetaSyncerTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeMetaSyncerTest.java new file mode 100644 index 0000000000000..ff55ccafc39c1 --- /dev/null +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeMetaSyncerTest.java @@ -0,0 +1,79 @@ +/* + * 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.manager.pipe.coordinator.runtime; + +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.confignode.manager.ConfigManager; +import org.apache.iotdb.confignode.manager.ProcedureManager; +import org.apache.iotdb.confignode.manager.pipe.coordinator.PipeManager; +import org.apache.iotdb.confignode.manager.pipe.coordinator.task.PipeTaskCoordinator; +import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv; +import org.apache.iotdb.confignode.procedure.env.RegionMaintainHandler; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.junit.Test; +import org.mockito.Mockito; + +import java.lang.reflect.Method; + +public class PipeMetaSyncerTest { + + @Test + public void testConsensusPipeCountMismatchTriggersCheckImmediately() throws Exception { + // A count mismatch means a region lost its replication channel, so the check runs on this round + // even though the periodic interval has not elapsed, and keeps running until the counts match. + final ConfigManager configManager = Mockito.mock(ConfigManager.class); + final ProcedureManager procedureManager = Mockito.mock(ProcedureManager.class); + final ConfigNodeProcedureEnv env = Mockito.mock(ConfigNodeProcedureEnv.class); + final RegionMaintainHandler regionMaintainHandler = Mockito.mock(RegionMaintainHandler.class); + final PipeManager pipeManager = Mockito.mock(PipeManager.class); + final PipeTaskCoordinator pipeTaskCoordinator = Mockito.mock(PipeTaskCoordinator.class); + Mockito.when(configManager.getProcedureManager()).thenReturn(procedureManager); + Mockito.when(configManager.getPipeManager()).thenReturn(pipeManager); + Mockito.when(pipeManager.getPipeTaskCoordinator()).thenReturn(pipeTaskCoordinator); + Mockito.when(procedureManager.getEnv()).thenReturn(env); + Mockito.when(env.getRegionMaintainHandler()).thenReturn(regionMaintainHandler); + // A failure status keeps the syncer from skipping the following rounds. + Mockito.when(procedureManager.pipeMetaSync()) + .thenReturn(new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode())); + Mockito.when(regionMaintainHandler.hasConsensusPipeCountMismatch()).thenReturn(true); + + final PipeMetaSyncer pipeMetaSyncer = new PipeMetaSyncer(configManager); + + sync(pipeMetaSyncer); + Mockito.verify(regionMaintainHandler, Mockito.times(1)).checkAndRepairConsensusPipes(); + + sync(pipeMetaSyncer); + Mockito.verify(regionMaintainHandler, Mockito.times(2)).checkAndRepairConsensusPipes(); + + // The counts match again: the next 4 rounds stay below the interval and must not check. + Mockito.when(regionMaintainHandler.hasConsensusPipeCountMismatch()).thenReturn(false); + for (int i = 0; i < 4; ++i) { + sync(pipeMetaSyncer); + } + Mockito.verify(regionMaintainHandler, Mockito.times(2)).checkAndRepairConsensusPipes(); + } + + private static void sync(final PipeMetaSyncer pipeMetaSyncer) throws Exception { + final Method sync = PipeMetaSyncer.class.getDeclaredMethod("sync"); + sync.setAccessible(true); + sync.invoke(pipeMetaSyncer); + } +} diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandlerConsensusPipeTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandlerConsensusPipeTest.java index 939bc98784c25..a9f056ad8ab1c 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandlerConsensusPipeTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandlerConsensusPipeTest.java @@ -373,6 +373,45 @@ public void testEmptyReplicaSetsAndEmptyPipes() { verify(procedureManager, never()).startConsensusPipe(any()); } + @Test + public void testConsensusPipeCountMismatchWhenRegionHasNoPipe() { + // A 2-replica region requires 2 consensus pipes. Having none of them is the exact degradation + // the cheap probe has to detect, without building any pipe name. + when(partitionManager.getReplicaPairCount(TConsensusGroupType.DataRegion)).thenReturn(2L); + when(pipeTaskCoordinator.getConsensusPipeCount()).thenReturn(0); + + assertTrue(handler.hasConsensusPipeCountMismatch()); + } + + @Test + public void testNoConsensusPipeCountMismatchWhenAllPipesExist() { + when(partitionManager.getReplicaPairCount(TConsensusGroupType.DataRegion)).thenReturn(2L); + when(pipeTaskCoordinator.getConsensusPipeCount()).thenReturn(2); + + assertFalse(handler.hasConsensusPipeCountMismatch()); + } + + @Test + public void testNoConsensusPipeCountMismatchForSingleReplicaRegion() { + // A single-replica region has no follower and therefore expects no consensus pipe at all, so a + // single-replica cluster must never be reported as degraded. + when(partitionManager.getReplicaPairCount(TConsensusGroupType.DataRegion)).thenReturn(0L); + when(pipeTaskCoordinator.getConsensusPipeCount()).thenReturn(0); + + assertFalse(handler.hasConsensusPipeCountMismatch()); + } + + @Test + public void testConsensusPipeCountProbeIsSkippedWhenNotIoTConsensusV2() { + ConfigNodeDescriptor.getInstance() + .getConf() + .setDataRegionConsensusProtocolClass("org.apache.iotdb.consensus.ratis.RatisConsensus"); + + assertFalse(handler.hasConsensusPipeCountMismatch()); + + verify(partitionManager, never()).getReplicaPairCount(any()); + } + private static class TestRegionMaintainHandler extends RegionMaintainHandler { private boolean lastUseFullRetry; private TEndPoint lastEndPoint; From 5888a325a3080d8f46b09114de9e0e2f2a3ab048 Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Fri, 6 Nov 2026 11:10:26 +0800 Subject: [PATCH 4/4] [ConfigNode] Tighten comments and document the new tests Keep production comments to what the code does not already say: the long explanations belonged in the commit messages, not next to the statements. Give every new test a short comment stating the scenario it covers and the behaviour it asserts. --- .../confignode/conf/ConfigNodeStartupCheck.java | 11 ++--------- .../manager/partition/PartitionManager.java | 6 ++---- .../pipe/coordinator/runtime/PipeMetaSyncer.java | 3 ++- .../partition/DatabasePartitionTable.java | 7 ++----- .../persistence/partition/PartitionInfo.java | 4 ++-- .../confignode/persistence/pipe/PipeTaskInfo.java | 5 ++--- .../procedure/env/RegionMaintainHandler.java | 14 +++++--------- .../conf/ConfigNodeStartupCheckTest.java | 7 +++---- .../RegionMaintainHandlerConsensusPipeTest.java | 15 ++++++--------- 9 files changed, 26 insertions(+), 46 deletions(-) diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeStartupCheck.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeStartupCheck.java index 7ad8c84555946..16712c09ffb1e 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeStartupCheck.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeStartupCheck.java @@ -173,11 +173,7 @@ private void checkGlobalConfig() throws ConfigurationException { .EXCEPTION_SCHEMAREGION_DOESN_T_SUPPORT_ORG_APACHE_IOTDB_CONSENSUS_IOT_IOTCONSENSUSV2_BA353C6D); } - // The data region consensus protocol is limited. An unrecognized value is not cosmetic: - // ConfigNode compares this value to decide whether it has to create consensus pipes, so a class - // name it does not know silently disables IoTConsensusV2 replication. Fail fast and print the - // accepted values; IoTConsensusV2 must be written as - // org.apache.iotdb.consensus.iot.IoTConsensusV2. + // An unrecognized class silently disables IoTConsensusV2 replication, so fail fast. checkDataRegionConsensusProtocolClass(); // The leader distribution policy is limited @@ -233,10 +229,7 @@ private void checkGlobalConfig() throws ConfigurationException { } } - /** - * Rejects a data_region_consensus_protocol_class that this release cannot act on. Package-private - * so that the accepted / rejected values can be unit tested without booting a ConfigNode. - */ + /** Rejects a data_region_consensus_protocol_class that this release cannot act on. */ static void checkDataRegionConsensusProtocolClass() throws ConfigurationException { final String dataRegionConsensusProtocolClass = CONF.getDataRegionConsensusProtocolClass(); if (ConsensusFactory.SIMPLE_CONSENSUS.equals(dataRegionConsensusProtocolClass) diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java index 162ff69154978..488a1e00f58e4 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java @@ -869,10 +869,8 @@ public Map getAllReplicaSetsMap(TConsensus * Only leader use this interface. * * @param type The specified TConsensusGroupType - * @return The number of ordered replica pairs of all Regions with the specified - * TConsensusGroupType, i.e. the sum of N * (N - 1) over those Regions. Unlike {@link - * #getAllReplicaSetsMap(TConsensusGroupType)}, this does not copy any replica set, so it is - * cheap enough to be called on every pipe meta sync round. + * @return Sum of N * (N - 1) over the Regions of the specified type, without copying any replica + * set, so it is cheap enough to call on every pipe meta sync round */ public long getReplicaPairCount(TConsensusGroupType type) { return partitionInfo.getReplicaPairCount(type); diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeMetaSyncer.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeMetaSyncer.java index aab6272dc47d2..dc80e4e4d7fba 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeMetaSyncer.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeMetaSyncer.java @@ -122,7 +122,8 @@ private synchronized void sync() { // waiting for the periodic round. Equal counts are not proof of correctness, so the periodic // round is kept. if (hasConsensusPipeCountMismatch() - || consensusPipeCheckRoundCounter.incrementAndGet() >= CONSENSUS_PIPE_CHECK_INTERVAL_ROUND) { + || consensusPipeCheckRoundCounter.incrementAndGet() + >= CONSENSUS_PIPE_CHECK_INTERVAL_ROUND) { consensusPipeCheckRoundCounter.set(0); checkAndRepairConsensusPipes(); } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/DatabasePartitionTable.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/DatabasePartitionTable.java index 4d393362ed031..65f347927a04d 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/DatabasePartitionTable.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/DatabasePartitionTable.java @@ -141,12 +141,9 @@ public List getAllReplicaSets(TConsensusGroupType type) { } /** - * Counts the ordered replica pairs of all Regions with the specified TConsensusGroupType, i.e. - * the sum of N * (N - 1) over those Regions, without copying any replica set. Callers use it to - * probe, on a hot path, how many consensus pipes the partition table requires. - * * @param type The specified TConsensusGroupType - * @return The number of ordered replica pairs of the specified Region type + * @return Sum of N * (N - 1) over the Regions of the specified type, i.e. their ordered replica + * pairs, without copying any replica set */ public long getReplicaPairCount(TConsensusGroupType type) { long count = 0; diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/PartitionInfo.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/PartitionInfo.java index 531b0fabda045..094ad1fe470bc 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/PartitionInfo.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/partition/PartitionInfo.java @@ -742,8 +742,8 @@ public List getAllReplicaSets(TConsensusGroupType type) { * Only leader use this interface. * * @param type The specified TConsensusGroupType - * @return The number of ordered replica pairs of all Regions with the specified - * TConsensusGroupType, i.e. the sum of N * (N - 1) over those Regions + * @return Sum of N * (N - 1) over the Regions of the specified type, without copying any replica + * set */ public long getReplicaPairCount(TConsensusGroupType type) { long count = 0; diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java index 9b90ea7006e98..9d038c19d5039 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java @@ -878,9 +878,8 @@ public long generateUniqueCreationTime(final String pipeName) { } /** - * Counts the consensus pipes without materializing their names. Callers use this to probe whether - * a consensus pipe reconciliation is needed: building {@link #getConsensusPipeStatusMap()} is far - * more expensive than counting, and the probe runs on every pipe meta sync round. + * Counts the consensus pipes without materializing their names. It is called on every pipe meta + * sync round, where building {@link #getConsensusPipeStatusMap()} would be far too expensive. */ public int getConsensusPipeCount() { acquireReadLock(); diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java index 3d3d5180c2997..25b96eb931605 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java @@ -620,15 +620,11 @@ public void dropConsensusPipesForRemovePeer( } /** - * Cheap probe used by PipeMetaSyncer on every round to decide whether a full {@link - * #checkAndRepairConsensusPipes()} is needed. It only compares how many consensus pipes the - * replica sets require (N * (N - 1) per region) with how many actually exist, so it neither - * builds a pipe name nor copies a replica set. A mismatch is a sufficient reason to reconcile - * immediately; equal counts are still reconciled periodically to catch same-size but different - * pipe sets. - * - *

Regions with a single replica expect no pipe at all, so single-replica clusters never report - * a mismatch. + * Cheap probe: compares how many consensus pipes the replica sets require (N * (N - 1) per + * region) with how many exist, without building a pipe name. A mismatch means a region lost its + * replication channel, so PipeMetaSyncer reconciles immediately; equal counts are still + * reconciled periodically, because a same-size but different pipe set cannot be detected by + * counting. */ public boolean hasConsensusPipeCountMismatch() { if (!isIoTConsensusV2Configured()) { diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/conf/ConfigNodeStartupCheckTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/conf/ConfigNodeStartupCheckTest.java index d31a72baf3b08..764ba4020dd21 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/conf/ConfigNodeStartupCheckTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/conf/ConfigNodeStartupCheckTest.java @@ -44,6 +44,7 @@ public void tearDown() { @Test public void testAcceptsEverySupportedConsensusProtocolClass() throws ConfigurationException { + // The four supported classes must all pass the startup check. for (final String protocolClass : new String[] { ConsensusFactory.SIMPLE_CONSENSUS, @@ -58,8 +59,7 @@ public void testAcceptsEverySupportedConsensusProtocolClass() throws Configurati @Test public void testAcceptsLegacyAliasesBecauseTheyAreNormalized() throws ConfigurationException { - // Both aliases used to be accepted by DataNode but silently ignored by ConfigNode, so they must - // keep passing this check: the value is normalized before it is compared. + // Both aliases must pass and be stored in canonical form. for (final String alias : new String[] { ConsensusFactory.LEGACY_IOT_CONSENSUS_V2, ConsensusFactory.REAL_IOT_CONSENSUS_V2 @@ -73,8 +73,7 @@ public void testAcceptsLegacyAliasesBecauseTheyAreNormalized() throws Configurat @Test public void testRejectsUnrecognizedConsensusProtocolClass() { - // An unknown class name used to be accepted silently and then degraded the whole cluster to one - // replica, so startup must fail and point at the accepted values. + // An unknown class must fail with a message that names the accepted values. conf.setDataRegionConsensusProtocolClass("org.apache.iotdb.consensus.pipe.PipeConsensusV3"); try { ConfigNodeStartupCheck.checkDataRegionConsensusProtocolClass(); diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandlerConsensusPipeTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandlerConsensusPipeTest.java index a9f056ad8ab1c..9a68a1ecd0f3b 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandlerConsensusPipeTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandlerConsensusPipeTest.java @@ -184,9 +184,7 @@ public void testCreatesMissingPipes() { @Test public void testCreatesMissingPipesWhenLegacyConsensusProtocolAliasConfigured() { - // iotdb-system.properties may keep the pre-rename IoTConsensusV2 class name. ConfigNode must - // treat it exactly like the canonical name; otherwise the periodic self-healing entry point - // never creates the consensus pipes that V2 replication depends on. + // The pre-rename class name must create pipes exactly like the canonical name. ConfigNodeDescriptor.getInstance() .getConf() .setDataRegionConsensusProtocolClass(LEGACY_IOT_CONSENSUS_V2); @@ -209,8 +207,7 @@ public void testCreatesMissingPipesWhenLegacyConsensusProtocolAliasConfigured() @Test public void testCreatesInitialPipesWhenImplementationClassNameConfigured() { - // Writing the implementation class name into the configuration is a supported way to select - // IoTConsensusV2, so database creation must also create the initial consensus pipes. + // Selecting V2 by its implementation class name must also create the initial pipes. ConfigNodeDescriptor.getInstance() .getConf() .setDataRegionConsensusProtocolClass(REAL_IOT_CONSENSUS_V2); @@ -375,8 +372,7 @@ public void testEmptyReplicaSetsAndEmptyPipes() { @Test public void testConsensusPipeCountMismatchWhenRegionHasNoPipe() { - // A 2-replica region requires 2 consensus pipes. Having none of them is the exact degradation - // the cheap probe has to detect, without building any pipe name. + // A 2-replica region requires 2 pipes; having none is the degradation the probe must detect. when(partitionManager.getReplicaPairCount(TConsensusGroupType.DataRegion)).thenReturn(2L); when(pipeTaskCoordinator.getConsensusPipeCount()).thenReturn(0); @@ -385,6 +381,7 @@ public void testConsensusPipeCountMismatchWhenRegionHasNoPipe() { @Test public void testNoConsensusPipeCountMismatchWhenAllPipesExist() { + // 2 required, 2 present. when(partitionManager.getReplicaPairCount(TConsensusGroupType.DataRegion)).thenReturn(2L); when(pipeTaskCoordinator.getConsensusPipeCount()).thenReturn(2); @@ -393,8 +390,7 @@ public void testNoConsensusPipeCountMismatchWhenAllPipesExist() { @Test public void testNoConsensusPipeCountMismatchForSingleReplicaRegion() { - // A single-replica region has no follower and therefore expects no consensus pipe at all, so a - // single-replica cluster must never be reported as degraded. + // A single-replica region has no follower and expects no pipe, so it is never degraded. when(partitionManager.getReplicaPairCount(TConsensusGroupType.DataRegion)).thenReturn(0L); when(pipeTaskCoordinator.getConsensusPipeCount()).thenReturn(0); @@ -403,6 +399,7 @@ public void testNoConsensusPipeCountMismatchForSingleReplicaRegion() { @Test public void testConsensusPipeCountProbeIsSkippedWhenNotIoTConsensusV2() { + // A non-V2 cluster must not even pay for the probe. ConfigNodeDescriptor.getInstance() .getConf() .setDataRegionConsensusProtocolClass("org.apache.iotdb.consensus.ratis.RatisConsensus");