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..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
@@ -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 =
@@ -625,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/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/ConfigNodeMessages.java b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java
index 11eaa6324a946..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
@@ -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 =
@@ -605,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/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/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/conf/ConfigNodeStartupCheck.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeStartupCheck.java
index fb7d674d0822a..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,6 +173,9 @@ private void checkGlobalConfig() throws ConfigurationException {
               .EXCEPTION_SCHEMAREGION_DOESN_T_SUPPORT_ORG_APACHE_IOTDB_CONSENSUS_IOT_IOTCONSENSUSV2_BA353C6D);
     }
 
+    // An unrecognized class silently disables IoTConsensusV2 replication, so fail fast.
+    checkDataRegionConsensusProtocolClass();
+
     // The leader distribution policy is limited
     if (!AbstractLeaderBalancer.GREEDY_POLICY.equals(CONF.getLeaderDistributionPolicy())
         && !AbstractLeaderBalancer.CFS_POLICY.equals(CONF.getLeaderDistributionPolicy())
@@ -226,6 +229,28 @@ private void checkGlobalConfig() throws ConfigurationException {
     }
   }
 
+  /** 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)
+        || 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/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/manager/partition/PartitionManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java
index cacc7ceadcbd9..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
@@ -865,6 +865,17 @@ public Map<TConsensusGroupId, TRegionReplicaSet> getAllReplicaSetsMap(TConsensus
             Collectors.toMap(TRegionReplicaSet::getRegionId, regionReplicaSet -> regionReplicaSet));
   }
 
+  /**
+   * Only leader use this interface.
+   *
+   * @param type The specified TConsensusGroupType
+   * @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);
+  }
+
   /**
    * 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..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
@@ -118,7 +118,12 @@ 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 +181,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<String, PipeStatus> 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..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
@@ -140,6 +140,22 @@ public List<TRegionReplicaSet> getAllReplicaSets(TConsensusGroupType type) {
     return result;
   }
 
+  /**
+   * @param type The specified TConsensusGroupType
+   * @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;
+    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..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
@@ -738,6 +738,23 @@ public List<TRegionReplicaSet> getAllReplicaSets(TConsensusGroupType type) {
     return result;
   }
 
+  /**
+   * Only leader use this interface.
+   *
+   * @param type The specified TConsensusGroupType
+   * @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;
+    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..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
@@ -877,6 +877,25 @@ public long generateUniqueCreationTime(final String pipeName) {
     }
   }
 
+  /**
+   * 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();
+    try {
+      int count = 0;
+      for (final PipeMeta pipeMeta : pipeMetaKeeper.getPipeMetaList()) {
+        if (PipeType.CONSENSUS.equals(pipeMeta.getStaticMeta().getPipeType())) {
+          ++count;
+        }
+      }
+      return count;
+    } finally {
+      releaseReadLock();
+    }
+  }
+
   public Map<String, PipeStatus> 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 04100d05f9cb0..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
@@ -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;
     }
 
@@ -618,13 +619,29 @@ public void dropConsensusPipesForRemovePeer(
     }
   }
 
+  /**
+   * 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()) {
+      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).
    * 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 +727,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/conf/ConfigNodeStartupCheckTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/conf/ConfigNodeStartupCheckTest.java
new file mode 100644
index 0000000000000..764ba4020dd21
--- /dev/null
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/conf/ConfigNodeStartupCheckTest.java
@@ -0,0 +1,87 @@
+/*
+ * 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 {
+    // The four supported classes must all pass the startup check.
+    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 must pass and be stored in canonical form.
+    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 must fail with a message that names 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));
+    }
+  }
+}
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 a99d6f52fb819..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
@@ -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,47 @@ public void testCreatesMissingPipes() {
     verify(procedureManager, never()).startConsensusPipe(any());
   }
 
+  @Test
+  public void testCreatesMissingPipesWhenLegacyConsensusProtocolAliasConfigured() {
+    // The pre-rename class name must create pipes exactly like the canonical name.
+    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() {
+    // Selecting V2 by its implementation class name must also create the initial 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))
@@ -326,6 +370,45 @@ public void testEmptyReplicaSetsAndEmptyPipes() {
     verify(procedureManager, never()).startConsensusPipe(any());
   }
 
+  @Test
+  public void testConsensusPipeCountMismatchWhenRegionHasNoPipe() {
+    // 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);
+
+    assertTrue(handler.hasConsensusPipeCountMismatch());
+  }
+
+  @Test
+  public void testNoConsensusPipeCountMismatchWhenAllPipesExist() {
+    // 2 required, 2 present.
+    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 expects no pipe, so it is never degraded.
+    when(partitionManager.getReplicaPairCount(TConsensusGroupType.DataRegion)).thenReturn(0L);
+    when(pipeTaskCoordinator.getConsensusPipeCount()).thenReturn(0);
+
+    assertFalse(handler.hasConsensusPipeCountMismatch());
+  }
+
+  @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");
+
+    assertFalse(handler.hasConsensusPipeCountMismatch());
+
+    verify(partitionManager, never()).getReplicaPairCount(any());
+  }
+
   private static class TestRegionMaintainHandler extends RegionMaintainHandler {
     private boolean lastUseFullRetry;
     private TEndPoint lastEndPoint;