[Bug] Recognize IoTConsensusV2 class name aliases on ConfigNode

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";

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 =

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";

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 =

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() {

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()));

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());

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

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))

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.
    *

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

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();

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.
    *

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.
    *

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.
    *

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 {