From 99a446f15f77e8997e45de46b1a7c6fda35710c3 Mon Sep 17 00:00:00 2001 From: luoluoyuyu Date: Thu, 22 Oct 2026 17:48:34 +0800 Subject: [PATCH] [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 cfde56c6419da..78a838bd0c93c 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 @@ -445,6 +445,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 ddf8be67cefc6..69b723253e76e 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 @@ -414,6 +414,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 a08b4e87ba187..cad6c0ff2d457 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 @@ -267,6 +267,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; @@ -1513,9 +1514,12 @@ public TSStatus checkConfigNodeGlobalConfig(TConfigNodeRegisterReq req) { + ManagerMessages.MESSAGE_CONFIG_NODE_CONSENSUS_PROTOCOL_CLASS_D0F437AF + errorSuffix); } - 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( errorPrefix + ManagerMessages.MESSAGE_DATA_REGION_CONSENSUS_PROTOCOL_CLASS_AB025B20 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))