[To dev/1.3] [Test] Fix asynchronous assertions in pipe integration tests (#18941) (#18973)

Author: Caideyipi

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