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