luoluoyuyu · Oct 22, 2026
diff --git a/...core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java b/...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/...core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java b/...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
@@ -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/...-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java b/...-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
@@ -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/...ignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RouteBalancer.java b/...ignode/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/...ignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java b/...ignode/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<TDataNodeLocation> 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/...-core/confignode/src/test/java/org/apache/iotdb/confignode/conf/ConfigNodeConfigTest.java b/...-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/...ava/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandlerConsensusPipeTest.java b/...ava/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<TConsensusGroupId, TRegionReplicaSet> 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))