[To dev/1.3] [Test] Check receiver data after stopping pipes (#18941)

integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java

@@ -983,12 +983,12 @@ public static void assertDataAlwaysOnEnv(
       long consistentSeconds) {
     try (Connection connection = env.getConnection();
         Statement statement = connection.createStatement()) {
-      // Keep retrying if there are execution failures
       await()
           .pollInSameThread()
-          .pollDelay(1L, TimeUnit.SECONDS)
+          .pollDelay(0L, TimeUnit.SECONDS)
           .pollInterval(1L, TimeUnit.SECONDS)
-          .atMost(consistentSeconds, TimeUnit.SECONDS)
+          .during(consistentSeconds, TimeUnit.SECONDS)
+          .atMost(consistentSeconds + 5, TimeUnit.SECONDS)
           .failFast(
               () -> {
                 try {
@@ -997,7 +997,8 @@ public static void assertDataAlwaysOnEnv(
                 } catch (Exception e) {
                   Assert.fail();
                 }
-              });
+              })
+          .until(() -> true);
     } catch (Exception e) {
       e.printStackTrace();
       fail();

integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeLifeCycleIT.java

@@ -273,7 +273,8 @@ public void testLifeCycleLogMode() throws Exception {
           Arrays.asList("insert into root.db.d1(time, s1) values (3, 3)", "flush"),
           null);
 
-      TestUtils.assertDataEventuallyOnEnv(
+      // A stopped pipe must keep the receiver data unchanged.
+      TestUtils.assertDataAlwaysOnEnv(
           receiverEnv, "select * from root.**", "Time,root.db.d1.s1,", expectedResSet);
     }
   }
@@ -334,7 +335,8 @@ public void testLifeCycleFileMode() throws Exception {
           Arrays.asList("insert into root.db.d1(time, s1) values (3, 3)", "flush"),
           null);
 
-      TestUtils.assertDataEventuallyOnEnv(
+      // A stopped pipe must keep the receiver data unchanged.
+      TestUtils.assertDataAlwaysOnEnv(
           receiverEnv, "select * from root.**", "Time,root.db.d1.s1,", expectedResSet);
     }
   }
@@ -395,7 +397,8 @@ public void testLifeCycleHybridMode() throws Exception {
           Arrays.asList("insert into root.db.d1(time, s1) values (3, 3)", "flush"),
           null);
 
-      TestUtils.assertDataEventuallyOnEnv(
+      // A stopped pipe must keep the receiver data unchanged.
+      TestUtils.assertDataAlwaysOnEnv(
           receiverEnv, "select * from root.**", "Time,root.db.d1.s1,", expectedResSet);
     }
   }
@@ -587,8 +590,8 @@ public void testReceiverAlreadyHaveTimeSeries() throws Exception {
           Arrays.asList("insert into root.db.d1(time, s1) values (3, 3)", "flush"),
           null);
 
-      Thread.sleep(5000);
-      TestUtils.assertDataEventuallyOnEnv(
+      // A stopped pipe must keep the receiver data unchanged.
+      TestUtils.assertDataAlwaysOnEnv(
           receiverEnv,
           "select count(*) from root.**",
           "count(root.db.d1.s1),",

integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeProtocolIT.java

@@ -317,8 +317,8 @@ private void doTest() throws Exception {
       TestUtils.executeNonQueries(
           senderEnv, Arrays.asList("insert into root.db.d1(time,s1) values (3,3)", "flush"), null);
 
-      Thread.sleep(5000);
-      TestUtils.assertDataEventuallyOnEnv(
+      // A stopped pipe must keep the receiver data unchanged.
+      TestUtils.assertDataAlwaysOnEnv(
           receiverEnv,
           "select count(*) from root.**",
           "count(root.db.d1.s1),",

integration-test/src/test/java/org/apache/iotdb/pipe/it/manual/IoTDBPipeInclusionIT.java

@@ -23,6 +23,7 @@
 import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
 import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq;
 import org.apache.iotdb.db.it.utils.TestUtils;
+import org.apache.iotdb.db.queryengine.common.header.ColumnHeaderConstant;
 import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
 import org.apache.iotdb.it.framework.IoTDBTestRunner;
 import org.apache.iotdb.itbase.category.MultiClusterIT2ManualCreateSchema;
@@ -96,7 +97,10 @@ public void testPureSchemaInclusion() throws Exception {
           null);
 
       TestUtils.assertDataAlwaysOnEnv(
-          receiverEnv, "select * from root.**", "Time,", Collections.emptySet());
+          receiverEnv,
+          "select * from root.**",
+          "Time,root.ln.wf01.wt01.status,",
+          Collections.emptySet());
     }
   }
 
@@ -277,7 +281,10 @@ public void testAuthExclusion() throws Exception {
       TestUtils.executeNonQuery(senderEnv, "create user `ln_write_user` 'write_pwd'", null);
 
       TestUtils.assertDataAlwaysOnEnv(
-          receiverEnv, "list user", "user,", Collections.singleton("root,"));
+          receiverEnv,
+          "list user",
+          ColumnHeaderConstant.USER + ",",
+          Collections.singleton("root,"));
     }
   }
 
@@ -320,10 +327,17 @@ public void testAuthInclusionWithPattern() throws Exception {
               "GRANT READ_DATA, WRITE_DATA ON root.** TO USER ln_write_user;"),
           null);
 
-      TestUtils.assertDataAlwaysOnEnv(
+      TestUtils.assertDataEventuallyOnEnv(
           receiverEnv,
           "LIST PRIVILEGES OF USER ln_write_user",
-          "ROLE,PATH,PRIVILEGES,GRANT OPTION,",
+          ColumnHeaderConstant.ROLE
+              + ","
+              + ColumnHeaderConstant.PATH
+              + ","
+              + ColumnHeaderConstant.PRIVILEGES
+              + ","
+              + ColumnHeaderConstant.GRANT_OPTION
+              + ",",
           new HashSet<>(
               Arrays.asList(
                   ",root.**,MANAGE_USER,true,",

integration-test/src/test/java/org/apache/iotdb/pipe/it/manual/IoTDBPipeMetaHistoricalIT.java

@@ -259,7 +259,7 @@ public void testAuthInclusion() throws Exception {
           "Database,SchemaReplicationFactor,DataReplicationFactor,TimePartitionOrigin,TimePartitionInterval,",
           Collections.emptySet());
       TestUtils.assertDataAlwaysOnEnv(
-          receiverEnv, "select * from root.**", "Time", Collections.emptySet());
+          receiverEnv, "select * from root.**", "Time,", Collections.emptySet());
 
       TestUtils.executeNonQuery(senderEnv, "CREATE ROLE test", null);