PipeConsensus: apply IoTConsensusV2 as cluster mode for integration test

.github/workflows/pipe-it-2cluster.yml

@@ -37,7 +37,7 @@ jobs:
       matrix:
         java: [17]
         # StrongConsistencyClusterMode is ignored now because RatisConsensus has not been supported yet.
-        cluster: [LightWeightStandaloneMode, ScalableSingleNodeMode, HighPerformanceMode]
+        cluster: [PipeConsensusBatchMode]
         os: [ ubuntu-latest ]
     runs-on: ${{ matrix.os }}
     steps:
@@ -74,22 +74,9 @@ jobs:
       matrix:
         java: [17]
         # StrongConsistencyClusterMode is ignored now because RatisConsensus has not been supported yet.
-        cluster1: [LightWeightStandaloneMode, ScalableSingleNodeMode, HighPerformanceMode]
-        cluster2: [LightWeightStandaloneMode, ScalableSingleNodeMode, HighPerformanceMode]
+        cluster1: [PipeConsensusBatchMode]
+        cluster2: [LightWeightStandaloneMode, ScalableSingleNodeMode, HighPerformanceMode, PipeConsensusBatchMode]
         os: [ ubuntu-latest ]
-        exclude:
-          - cluster1: LightWeightStandaloneMode
-            cluster2: LightWeightStandaloneMode
-          - cluster1: LightWeightStandaloneMode
-            cluster2: ScalableSingleNodeMode
-          - cluster1: ScalableSingleNodeMode
-            cluster2: LightWeightStandaloneMode
-          - cluster1: ScalableSingleNodeMode
-            cluster2: HighPerformanceMode
-          - cluster1: HighPerformanceMode
-            cluster2: LightWeightStandaloneMode
-          - cluster1: HighPerformanceMode
-            cluster2: HighPerformanceMode
     runs-on: ${{ matrix.os }}
     steps:
       - uses: actions/checkout@v4
@@ -125,8 +112,8 @@ jobs:
       matrix:
         java: [ 17 ]
         # StrongConsistencyClusterMode is ignored now because RatisConsensus has not been supported yet.
-        cluster1: [ LightWeightStandaloneMode, ScalableSingleNodeMode, HighPerformanceMode ]
-        cluster2: [ ScalableSingleNodeMode ]
+        cluster1: [ PipeConsensusBatchMode ]
+        cluster2: [ PipeConsensusBatchMode ]
         os: [ ubuntu-latest ]
     runs-on: ${{ matrix.os }}
     steps:
@@ -163,8 +150,8 @@ jobs:
       matrix:
         java: [ 17 ]
         # do not use HighPerformanceMode here, otherwise some tests will cause the GH runner to receive a shutdown signal
-        cluster1: [ ScalableSingleNodeMode ]
-        cluster2: [ ScalableSingleNodeMode ]
+        cluster1: [ PipeConsensusBatchMode ]
+        cluster2: [ PipeConsensusBatchMode ]
         os: [ ubuntu-latest ]
     runs-on: ${{ matrix.os }}
     steps:
@@ -201,8 +188,8 @@ jobs:
       matrix:
         java: [ 17 ]
         # do not use HighPerformanceMode here, otherwise some tests will cause the GH runner to receive a shutdown signal
-        cluster1: [ ScalableSingleNodeMode ]
-        cluster2: [ ScalableSingleNodeMode ]
+        cluster1: [ PipeConsensusBatchMode ]
+        cluster2: [ PipeConsensusBatchMode ]
         os: [ ubuntu-latest ]
     runs-on: ${{ matrix.os }}
     steps:

integration-test/pom.xml

@@ -49,14 +49,14 @@
         <scalableSingleNodeMode.configNodeConsensus>Ratis</scalableSingleNodeMode.configNodeConsensus>
         <scalableSingleNodeMode.configNodeNumber>1</scalableSingleNodeMode.configNodeNumber>
         <scalableSingleNodeMode.dataNodeNumber>1</scalableSingleNodeMode.dataNodeNumber>
-        <scalableSingleNodeMode.dataRegionConsensus>IoT</scalableSingleNodeMode.dataRegionConsensus>
+        <scalableSingleNodeMode.dataRegionConsensus>IoTV2</scalableSingleNodeMode.dataRegionConsensus>
         <scalableSingleNodeMode.dataRegionReplicaNumber>1</scalableSingleNodeMode.dataRegionReplicaNumber>
         <scalableSingleNodeMode.schemaRegionConsensus>Ratis</scalableSingleNodeMode.schemaRegionConsensus>
         <scalableSingleNodeMode.schemaRegionReplicaNumber>1</scalableSingleNodeMode.schemaRegionReplicaNumber>
         <highPerformanceMode.configNodeConsensus>Ratis</highPerformanceMode.configNodeConsensus>
         <highPerformanceMode.configNodeNumber>1</highPerformanceMode.configNodeNumber>
         <highPerformanceMode.dataNodeNumber>3</highPerformanceMode.dataNodeNumber>
-        <highPerformanceMode.dataRegionConsensus>IoT</highPerformanceMode.dataRegionConsensus>
+        <highPerformanceMode.dataRegionConsensus>IoTV2</highPerformanceMode.dataRegionConsensus>
         <highPerformanceMode.dataRegionReplicaNumber>2</highPerformanceMode.dataRegionReplicaNumber>
         <highPerformanceMode.schemaRegionConsensus>Ratis</highPerformanceMode.schemaRegionConsensus>
         <highPerformanceMode.schemaRegionReplicaNumber>3</highPerformanceMode.schemaRegionReplicaNumber>
@@ -67,6 +67,13 @@
         <strongConsistencyClusterMode.dataRegionReplicaNumber>3</strongConsistencyClusterMode.dataRegionReplicaNumber>
         <strongConsistencyClusterMode.schemaRegionConsensus>Ratis</strongConsistencyClusterMode.schemaRegionConsensus>
         <strongConsistencyClusterMode.schemaRegionReplicaNumber>3</strongConsistencyClusterMode.schemaRegionReplicaNumber>
+        <pipeConsensusBatchMode.configNodeConsensus>Ratis</pipeConsensusBatchMode.configNodeConsensus>
+        <pipeConsensusBatchMode.configNodeNumber>1</pipeConsensusBatchMode.configNodeNumber>
+        <pipeConsensusBatchMode.dataNodeNumber>3</pipeConsensusBatchMode.dataNodeNumber>
+        <pipeConsensusBatchMode.dataRegionConsensus>IoTV2</pipeConsensusBatchMode.dataRegionConsensus>
+        <pipeConsensusBatchMode.dataRegionReplicaNumber>2</pipeConsensusBatchMode.dataRegionReplicaNumber>
+        <pipeConsensusBatchMode.schemaRegionConsensus>Ratis</pipeConsensusBatchMode.schemaRegionConsensus>
+        <pipeConsensusBatchMode.schemaRegionReplicaNumber>3</pipeConsensusBatchMode.schemaRegionReplicaNumber>
     </properties>
     <dependencies>
         <dependency>
@@ -314,6 +321,13 @@
                                 <strongConsistencyClusterMode.dataRegionConsensus>${strongConsistencyClusterMode.dataRegionConsensus}</strongConsistencyClusterMode.dataRegionConsensus>
                                 <strongConsistencyClusterMode.schemaRegionReplicaNumber>${strongConsistencyClusterMode.schemaRegionReplicaNumber}</strongConsistencyClusterMode.schemaRegionReplicaNumber>
                                 <strongConsistencyClusterMode.dataRegionReplicaNumber>${strongConsistencyClusterMode.dataRegionReplicaNumber}</strongConsistencyClusterMode.dataRegionReplicaNumber>
+                                <pipeConsensusBatchMode.configNodeNumber>${pipeConsensusBatchMode.configNodeNumber}</pipeConsensusBatchMode.configNodeNumber>
+                                <pipeConsensusBatchMode.dataNodeNumber>${pipeConsensusBatchMode.dataNodeNumber}</pipeConsensusBatchMode.dataNodeNumber>
+                                <pipeConsensusBatchMode.configNodeConsensus>${pipeConsensusBatchMode.configNodeConsensus}</pipeConsensusBatchMode.configNodeConsensus>
+                                <pipeConsensusBatchMode.schemaRegionConsensus>${pipeConsensusBatchMode.schemaRegionConsensus}</pipeConsensusBatchMode.schemaRegionConsensus>
+                                <pipeConsensusBatchMode.dataRegionConsensus>${pipeConsensusBatchMode.dataRegionConsensus}</pipeConsensusBatchMode.dataRegionConsensus>
+                                <pipeConsensusBatchMode.schemaRegionReplicaNumber>${pipeConsensusBatchMode.schemaRegionReplicaNumber}</pipeConsensusBatchMode.schemaRegionReplicaNumber>
+                                <pipeConsensusBatchMode.dataRegionReplicaNumber>${pipeConsensusBatchMode.dataRegionReplicaNumber}</pipeConsensusBatchMode.dataRegionReplicaNumber>
                             </systemPropertyVariables>
                             <summaryFile>target/failsafe-reports/failsafe-summary-IT.xml</summaryFile>
                         </configuration>

integration-test/src/main/java/org/apache/iotdb/it/env/cluster/ClusterConstant.java

@@ -52,6 +52,8 @@ public class ClusterConstant {
   public static final String SCALABLE_SINGLE_NODE_MODE = "ScalableSingleNodeMode";
   public static final String HIGH_PERFORMANCE_MODE = "HighPerformanceMode";
   public static final String STRONG_CONSISTENCY_CLUSTER_MODE = "StrongConsistencyClusterMode";
+  public static final String PIPE_CONSENSUS_BATCH_MODE = "PipeConsensusBatchMode";
+  public static final String PIPE_CONSENSUS_STREAM_MODE = "PipeConsensusStreamMode";
 
   // System arguments in pom.xml
   public static final String LIGHT_WEIGHT_STANDALONE_MODE_CONFIG_NODE_NUM =
@@ -114,6 +116,21 @@ public class ClusterConstant {
   public static final String STRONG_CONSISTENCY_CLUSTER_MODE_DATA_REGION_REPLICA_NUM =
       "strongConsistencyClusterMode.dataRegionReplicaNumber";
 
+  public static final String PIPE_CONSENSUS_BATCH_MODE_CONFIG_NODE_NUM =
+      "pipeConsensusBatchMode.configNodeNumber";
+  public static final String PIPE_CONSENSUS_BATCH_MODE_DATA_NODE_NUM =
+      "pipeConsensusBatchMode.dataNodeNumber";
+  public static final String PIPE_CONSENSUS_BATCH_MODE_CONFIG_NODE_CONSENSUS =
+      "pipeConsensusBatchMode.configNodeConsensus";
+  public static final String PIPE_CONSENSUS_BATCH_MODE_SCHEMA_REGION_CONSENSUS =
+      "pipeConsensusBatchMode.schemaRegionConsensus";
+  public static final String PIPE_CONSENSUS_BATCH_MODE_DATA_REGION_CONSENSUS =
+      "pipeConsensusBatchMode.dataRegionConsensus";
+  public static final String PIPE_CONSENSUS_BATCH_MODE_SCHEMA_REGION_REPLICA_NUM =
+      "pipeConsensusBatchMode.schemaRegionReplicaNumber";
+  public static final String PIPE_CONSENSUS_BATCH_MODE_DATA_REGION_REPLICA_NUM =
+      "pipeConsensusBatchMode.dataRegionReplicaNumber";
+
   // Property file names
   public static final String IOTDB_SYSTEM_PROPERTIES_FILE = "iotdb-system.properties";
 
@@ -125,6 +142,7 @@ public class ClusterConstant {
       "schema_region_consensus_protocol_class";
   public static final String DATA_REGION_CONSENSUS_PROTOCOL_CLASS =
       "data_region_consensus_protocol_class";
+  public static final String IOT_CONSENSUS_V2_MODE = "iot_consensus_v2_mode";
   public static final String SCHEMA_REPLICATION_FACTOR = "schema_replication_factor";
   public static final String DATA_REPLICATION_FACTOR = "data_replication_factor";
 

integration-test/src/main/java/org/apache/iotdb/it/env/cluster/EnvUtils.java

@@ -49,6 +49,9 @@
 import static org.apache.iotdb.it.env.cluster.ClusterConstant.LIGHT_WEIGHT_STANDALONE_MODE_CONFIG_NODE_NUM;
 import static org.apache.iotdb.it.env.cluster.ClusterConstant.LIGHT_WEIGHT_STANDALONE_MODE_DATA_NODE_NUM;
 import static org.apache.iotdb.it.env.cluster.ClusterConstant.LOCK_FILE_PATH;
+import static org.apache.iotdb.it.env.cluster.ClusterConstant.PIPE_CONSENSUS_BATCH_MODE;
+import static org.apache.iotdb.it.env.cluster.ClusterConstant.PIPE_CONSENSUS_BATCH_MODE_CONFIG_NODE_NUM;
+import static org.apache.iotdb.it.env.cluster.ClusterConstant.PIPE_CONSENSUS_BATCH_MODE_DATA_NODE_NUM;
 import static org.apache.iotdb.it.env.cluster.ClusterConstant.RATIS_CONSENSUS_STR;
 import static org.apache.iotdb.it.env.cluster.ClusterConstant.SCALABLE_SINGLE_NODE_MODE;
 import static org.apache.iotdb.it.env.cluster.ClusterConstant.SCALABLE_SINGLE_NODE_MODE_CONFIG_NODE_NUM;
@@ -142,6 +145,10 @@ private static Pair<Integer, Integer> getClusterNodesNum(final int index) {
           return new Pair<>(
               Integer.parseInt(System.getProperty(STRONG_CONSISTENCY_CLUSTER_MODE_CONFIG_NODE_NUM)),
               Integer.parseInt(System.getProperty(STRONG_CONSISTENCY_CLUSTER_MODE_DATA_NODE_NUM)));
+        case PIPE_CONSENSUS_BATCH_MODE:
+          return new Pair<>(
+              Integer.parseInt(System.getProperty(PIPE_CONSENSUS_BATCH_MODE_CONFIG_NODE_NUM)),
+              Integer.parseInt(System.getProperty(PIPE_CONSENSUS_BATCH_MODE_DATA_NODE_NUM)));
         default:
           // Print nothing to avoid polluting test outputs
           return null;

integration-test/src/main/java/org/apache/iotdb/it/env/cluster/node/AbstractNodeWrapper.java

@@ -20,6 +20,7 @@
 package org.apache.iotdb.it.env.cluster.node;
 
 import org.apache.iotdb.commons.conf.IoTDBConstant;
+import org.apache.iotdb.consensus.ConsensusFactory;
 import org.apache.iotdb.it.env.EnvFactory;
 import org.apache.iotdb.it.env.cluster.config.MppBaseConfig;
 import org.apache.iotdb.it.env.cluster.config.MppCommonConfig;
@@ -79,6 +80,7 @@
 import static org.apache.iotdb.it.env.cluster.ClusterConstant.HIGH_PERFORMANCE_MODE_SCHEMA_REGION_REPLICA_NUM;
 import static org.apache.iotdb.it.env.cluster.ClusterConstant.HYPHEN;
 import static org.apache.iotdb.it.env.cluster.ClusterConstant.INFLUXDB_RPC_PORT;
+import static org.apache.iotdb.it.env.cluster.ClusterConstant.IOT_CONSENSUS_V2_MODE;
 import static org.apache.iotdb.it.env.cluster.ClusterConstant.JAVA_CMD;
 import static org.apache.iotdb.it.env.cluster.ClusterConstant.LIGHT_WEIGHT_STANDALONE_MODE;
 import static org.apache.iotdb.it.env.cluster.ClusterConstant.LIGHT_WEIGHT_STANDALONE_MODE_CONFIG_NODE_CONSENSUS;
@@ -88,6 +90,12 @@
 import static org.apache.iotdb.it.env.cluster.ClusterConstant.LIGHT_WEIGHT_STANDALONE_MODE_SCHEMA_REGION_REPLICA_NUM;
 import static org.apache.iotdb.it.env.cluster.ClusterConstant.MQTT_HOST;
 import static org.apache.iotdb.it.env.cluster.ClusterConstant.MQTT_PORT;
+import static org.apache.iotdb.it.env.cluster.ClusterConstant.PIPE_CONSENSUS_BATCH_MODE;
+import static org.apache.iotdb.it.env.cluster.ClusterConstant.PIPE_CONSENSUS_BATCH_MODE_CONFIG_NODE_CONSENSUS;
+import static org.apache.iotdb.it.env.cluster.ClusterConstant.PIPE_CONSENSUS_BATCH_MODE_DATA_REGION_CONSENSUS;
+import static org.apache.iotdb.it.env.cluster.ClusterConstant.PIPE_CONSENSUS_BATCH_MODE_DATA_REGION_REPLICA_NUM;
+import static org.apache.iotdb.it.env.cluster.ClusterConstant.PIPE_CONSENSUS_BATCH_MODE_SCHEMA_REGION_CONSENSUS;
+import static org.apache.iotdb.it.env.cluster.ClusterConstant.PIPE_CONSENSUS_BATCH_MODE_SCHEMA_REGION_REPLICA_NUM;
 import static org.apache.iotdb.it.env.cluster.ClusterConstant.PIPE_LIB_DIR;
 import static org.apache.iotdb.it.env.cluster.ClusterConstant.REST_SERVICE_PORT;
 import static org.apache.iotdb.it.env.cluster.ClusterConstant.SCALABLE_SINGLE_NODE_MODE;
@@ -401,6 +409,29 @@ private void reloadClusterConfigurations() {
               DATA_REPLICATION_FACTOR,
               System.getProperty(STRONG_CONSISTENCY_CLUSTER_MODE_DATA_REGION_REPLICA_NUM));
           break;
+        case PIPE_CONSENSUS_BATCH_MODE:
+          clusterConfigProperties.setProperty(
+              CONFIG_NODE_CONSENSUS_PROTOCOL_CLASS,
+              fromConsensusAbbrToFullName(
+                  System.getProperty(PIPE_CONSENSUS_BATCH_MODE_CONFIG_NODE_CONSENSUS)));
+          clusterConfigProperties.setProperty(
+              SCHEMA_REGION_CONSENSUS_PROTOCOL_CLASS,
+              fromConsensusAbbrToFullName(
+                  System.getProperty(PIPE_CONSENSUS_BATCH_MODE_SCHEMA_REGION_CONSENSUS)));
+          clusterConfigProperties.setProperty(
+              DATA_REGION_CONSENSUS_PROTOCOL_CLASS,
+              fromConsensusAbbrToFullName(
+                  System.getProperty(PIPE_CONSENSUS_BATCH_MODE_DATA_REGION_CONSENSUS)));
+          clusterConfigProperties.setProperty(
+              SCHEMA_REPLICATION_FACTOR,
+              System.getProperty(PIPE_CONSENSUS_BATCH_MODE_SCHEMA_REGION_REPLICA_NUM));
+          clusterConfigProperties.setProperty(
+              DATA_REPLICATION_FACTOR,
+              System.getProperty(PIPE_CONSENSUS_BATCH_MODE_DATA_REGION_REPLICA_NUM));
+          // set mode
+          clusterConfigProperties.setProperty(
+              IOT_CONSENSUS_V2_MODE, ConsensusFactory.IOT_CONSENSUS_V2_BATCH_MODE);
+          break;
         default:
           // Print nothing to avoid polluting test outputs
       }