Merge branch 'master' into strong_password

# Conflicts:
#	integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteCommonConfig.java

Author: jt2594838

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

@@ -527,6 +527,23 @@ public CommonConfig setQueryMemoryProportion(String queryMemoryProportion) {
   }
 
   @Override
+  public CommonConfig setSubscriptionPrefetchTsFileBatchMaxDelayInMs(
+      int subscriptionPrefetchTsFileBatchMaxDelayInMs) {
+    setProperty(
+        "subscription_prefetch_ts_file_batch_max_delay_in_ms",
+        String.valueOf(subscriptionPrefetchTsFileBatchMaxDelayInMs));
+    return this;
+  }
+
+  @Override
+  public CommonConfig setSubscriptionPrefetchTsFileBatchMaxSizeInBytes(
+      int subscriptionPrefetchTsFileBatchMaxSizeInBytes) {
+    setProperty(
+        "subscription_prefetch_ts_file_batch_max_size_in_bytes",
+        String.valueOf(subscriptionPrefetchTsFileBatchMaxSizeInBytes));
+    return this;
+  }
+
   public CommonConfig setSubscriptionEnabled(boolean subscriptionEnabled) {
     setProperty("subscription_enabled", String.valueOf(subscriptionEnabled));
     return this;

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

@@ -542,6 +542,25 @@ public CommonConfig setQueryMemoryProportion(String queryMemoryProportion) {
   }
 
   @Override
+  public CommonConfig setSubscriptionPrefetchTsFileBatchMaxDelayInMs(
+      int subscriptionPrefetchTsFileBatchMaxDelayInMs) {
+    dnConfig.setSubscriptionPrefetchTsFileBatchMaxDelayInMs(
+        subscriptionPrefetchTsFileBatchMaxDelayInMs);
+    cnConfig.setSubscriptionPrefetchTsFileBatchMaxDelayInMs(
+        subscriptionPrefetchTsFileBatchMaxDelayInMs);
+    return this;
+  }
+
+  @Override
+  public CommonConfig setSubscriptionPrefetchTsFileBatchMaxSizeInBytes(
+      int subscriptionPrefetchTsFileBatchMaxSizeInBytes) {
+    dnConfig.setSubscriptionPrefetchTsFileBatchMaxSizeInBytes(
+        subscriptionPrefetchTsFileBatchMaxSizeInBytes);
+    cnConfig.setSubscriptionPrefetchTsFileBatchMaxSizeInBytes(
+        subscriptionPrefetchTsFileBatchMaxSizeInBytes);
+    return this;
+  }
+
   public CommonConfig setSubscriptionEnabled(boolean subscriptionEnabled) {
     dnConfig.setSubscriptionEnabled(subscriptionEnabled);
     cnConfig.setSubscriptionEnabled(subscriptionEnabled);

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

@@ -386,6 +386,18 @@ public CommonConfig setEnforceStrongPassword(boolean enforceStrongPassword) {
     return this;
   }
 
+  @Override
+  public CommonConfig setSubscriptionPrefetchTsFileBatchMaxDelayInMs(
+      int subscriptionPrefetchTsFileBatchMaxDelayInMs) {
+    return this;
+  }
+
+  @Override
+  public CommonConfig setSubscriptionPrefetchTsFileBatchMaxSizeInBytes(
+      int subscriptionPrefetchTsFileBatchMaxSizeInBytes) {
+    return this;
+  }
+
   @Override
   public CommonConfig setSubscriptionEnabled(boolean subscriptionEnabled) {
     return this;

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

@@ -168,6 +168,12 @@ CommonConfig setPipeConnectorRequestSliceThresholdBytes(
 
   CommonConfig setQueryMemoryProportion(String queryMemoryProportion);
 
+  CommonConfig setSubscriptionPrefetchTsFileBatchMaxDelayInMs(
+      int subscriptionPrefetchTsFileBatchMaxDelayInMs);
+
+  CommonConfig setSubscriptionPrefetchTsFileBatchMaxSizeInBytes(
+      int subscriptionPrefetchTsFileBatchMaxSizeInBytes);
+
   CommonConfig setSubscriptionEnabled(boolean subscriptionEnabled);
 
   default CommonConfig setDefaultStorageGroupLevel(int defaultStorageGroupLevel) {

integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeExtractorIT.java

@@ -980,6 +980,9 @@ public void testTableModeSQLSupportNowFunc() {
 
   private void assertTimeseriesCountOnReceiver(BaseEnv receiverEnv, int count) {
     TestUtils.assertDataEventuallyOnEnv(
-        receiverEnv, "count timeseries root.db.**", "count(timeseries),", Collections.singleton(count + ","));
+        receiverEnv,
+        "count timeseries root.db.**",
+        "count(timeseries),",
+        Collections.singleton(count + ","));
   }
 }

integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipePermissionIT.java

@@ -243,7 +243,8 @@ public void testReceiverPermission() throws Exception {
 
     try (final SyncConfigNodeIServiceClient client =
         (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
-      if (!TestUtils.tryExecuteNonQueryWithRetry(receiverEnv, "create user testUser 'passwD@123456'")) {
+      if (!TestUtils.tryExecuteNonQueryWithRetry(
+          receiverEnv, "create user testUser 'passwD@123456'")) {
         return;
       }
 

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

@@ -1044,7 +1044,10 @@ private void assertTimeseriesCountOnReceiver(BaseEnv receiverEnv, int count) {
     // for system password history
     count += 2;
     TestUtils.assertDataEventuallyOnEnv(
-        receiverEnv, "count timeseries root.**", "count(timeseries),", Collections.singleton(count + ","));
+        receiverEnv,
+        "count timeseries root.**",
+        "count(timeseries),",
+        Collections.singleton(count + ","));
   }
 
   private void assertPipeCount(int count) throws Exception {