fix tests

Author: jt2594838

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

@@ -67,4 +67,10 @@ public ConfigNodeConfig setMetricPrometheusReporterPassword(String password) {
     properties.setProperty("metric_prometheus_reporter_password", password);
     return this;
   }
+
+  @Override
+  public ConfigNodeConfig setEnableAutoLeaderBalanceForIotConsensus(boolean enable) {
+    properties.setProperty("enable_auto_leader_balance_for_iot_consensus", String.valueOf(enable));
+    return this;
+  }
 }

integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteConfigNodeConfig.java

@@ -38,4 +38,9 @@ public ConfigNodeConfig setMetricPrometheusReporterUsername(String username) {
   public ConfigNodeConfig setMetricPrometheusReporterPassword(String password) {
     return this;
   }
+
+  @Override
+  public ConfigNodeConfig setEnableAutoLeaderBalanceForIotConsensus(boolean enable) {
+    return this;
+  }
 }

integration-test/src/main/java/org/apache/iotdb/itbase/env/ConfigNodeConfig.java

@@ -29,4 +29,6 @@ public interface ConfigNodeConfig {
   ConfigNodeConfig setMetricPrometheusReporterUsername(String username);
 
   ConfigNodeConfig setMetricPrometheusReporterPassword(String password);
+
+  ConfigNodeConfig setEnableAutoLeaderBalanceForIotConsensus(boolean enable);
 }

integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeExtractorIT.java

@@ -73,12 +73,14 @@ public void setUp() {
         .setEnableSeqSpaceCompaction(false)
         .setEnableUnseqSpaceCompaction(false)
         .setEnableCrossSpaceCompaction(false);
+    senderEnv.getConfig().getConfigNodeConfig().setEnableAutoLeaderBalanceForIotConsensus(false);
     receiverEnv
         .getConfig()
         .getCommonConfig()
         .setAutoCreateSchemaEnabled(true)
         .setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
         .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS);
+    receiverEnv.getConfig().getConfigNodeConfig().setEnableAutoLeaderBalanceForIotConsensus(false);
 
     // 10 min, assert that the operations will not time out
     senderEnv.getConfig().getCommonConfig().setDnConnectionTimeoutMs(600000);

integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/AbstractSubscriptionTripleIT.java

@@ -76,6 +76,10 @@ protected void setUpConfig() {
         .getConfig()
         .getCommonConfig()
         .setSubscriptionPrefetchTsFileBatchMaxSizeInBytes(32 * 1024);
+
+    sender.getConfig().getConfigNodeConfig().setEnableAutoLeaderBalanceForIotConsensus(false);
+    receiver1.getConfig().getConfigNodeConfig().setEnableAutoLeaderBalanceForIotConsensus(false);
+    receiver2.getConfig().getConfigNodeConfig().setEnableAutoLeaderBalanceForIotConsensus(false);
   }
 
   @Override

integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/autocommit/IoTDBTestAutoCommitFalseDataSetPullConsumerIT.java

@@ -126,8 +126,9 @@ private void insert_data(long timestamp)
     session_src.insertTablet(tablet);
   }
 
-  private void consume_data_noCommit(SubscriptionTreePullConsumer consumer, Session session)
+  private int consume_data_noCommit(SubscriptionTreePullConsumer consumer, Session session)
       throws InterruptedException, StatementExecutionException, IoTDBConnectionException {
+    int rowConsumed = 0;
     while (true) {
       Thread.sleep(1000);
       List<SubscriptionMessage> messages = consumer.poll(Duration.ofMillis(10000));
@@ -149,11 +150,15 @@ private void consume_data_noCommit(SubscriptionTreePullConsumer consumer, Sessio
                   .map(IMeasurementSchema::getMeasurementName)
                   .collect(Collectors.toList()));
           session.insertTablet(tablet);
+          rowConsumed +=
+              (int)
+                  Arrays.stream(tablet.getTimestamps(), 0, tablet.getRowSize()).distinct().count();
           System.out.println(
               FORMAT.format(new Date()) + " consume data no commit:" + tablet.getRowSize());
         }
       }
     }
+    return rowConsumed;
   }
 
   @Test
@@ -183,13 +188,14 @@ public void do_test()
 
     // Subscribe and then write data
     insert_data(System.currentTimeMillis());
-    consume_data_noCommit(consumer, session_dest2);
+    // a leader change may cause a progress-rollback, resulting in consuming more data
+    int dest2Consumed = consume_data_noCommit(consumer, session_dest2);
     System.out.println("src sql1: " + getCount(session_src, sql1));
     System.out.println("dest sql1: " + getCount(session_dest, sql1));
     System.out.println("dest2 sql1: " + getCount(session_dest2, sql1));
     check_count(4, sql1, "dest consume subscription data 2:s_0");
-    check_count2(4, sql1, "dest2 consumption subscription data 2:s_0");
-    check_count2(4, sql2, "dest2 consumption subscription data 2:s_1");
+    check_count2(dest2Consumed, sql1, "dest2 consumption subscription data 2:s_0");
+    check_count2(dest2Consumed, sql2, "dest2 consumption subscription data 2:s_1");
 
     //        insert_data(1706659300000L); //2024-01-31 08:00:00+08:00
     // Will consume again
@@ -199,6 +205,6 @@ public void do_test()
     System.out.println("dest2 sql1: " + getCount(session_dest2, sql1));
     check_count(4, sql1, "dest consumption subscription before data3:s_0");
     check_count(4, sql2, "dest consume subscription before data3:s_1");
-    check_count2(4, sql2, "dest2 consumption subscription before count 3 data:s_1");
+    check_count2(dest2Consumed, sql2, "dest2 consumption subscription before count 3 data:s_1");
   }
 }