This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new c535723a17a [Test] Fix asynchronous assertions in pipe integration
tests (#18941)
c535723a17a is described below
commit c535723a17a9492dc4479de5af7ab0c02864b509
Author: Caideyipi <[email protected]>
AuthorDate: Wed Oct 21 11:35:36 2026 +0800
[Test] Fix asynchronous assertions in pipe integration tests (#18941)
* [Test] Wait for asynchronous pipe attribute synchronization
* [Test] Check receiver data remains unchanged after stopping pipes
---
.../dual/tablemodel/manual/basic/IoTDBPipeLifeCycleIT.java | 3 ++-
.../dual/tablemodel/manual/basic/IoTDBPipeProtocolIT.java | 8 +++++++-
.../it/dual/tablemodel/manual/enhanced/IoTDBPipeMetaIT.java | 3 ++-
.../it/dual/treemodel/auto/basic/IoTDBPipeLifeCycleIT.java | 13 ++++++++-----
.../it/dual/treemodel/auto/basic/IoTDBPipeProtocolIT.java | 4 ++--
5 files changed, 21 insertions(+), 10 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeLifeCycleIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeLifeCycleIT.java
index afa6b38775e..be9f569e172 100644
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeLifeCycleIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeLifeCycleIT.java
@@ -539,7 +539,8 @@ public class IoTDBPipeLifeCycleIT extends
AbstractPipeTableModelDualManualIT {
TableModelUtils.insertData("test", "test", 200, 300, senderEnv);
- TableModelUtils.assertCountData("test", "test", 200, receiverEnv,
handleFailure);
+ // A stopped pipe must keep the receiver data unchanged.
+ TableModelUtils.assertCountDataAlwaysOnEnv("test", "test", 200,
receiverEnv);
}
}
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeProtocolIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeProtocolIT.java
index 9a6fe6ddf41..809278f998a 100644
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeProtocolIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/basic/IoTDBPipeProtocolIT.java
@@ -344,7 +344,13 @@ public class IoTDBPipeProtocolIT extends
AbstractPipeTableModelDualManualIT {
TableModelUtils.insertData("test", "test", 200, 300, senderEnv);
- TableModelUtils.assertData("test", "test", 0, 200, receiverEnv,
handleFailure);
+ // A stopped pipe must keep the receiver data unchanged.
+ TestUtils.assertDataAlwaysOnEnv(
+ receiverEnv,
+ TableModelUtils.getQuerySql("test"),
+ TableModelUtils.generateHeaderResults(),
+ TableModelUtils.generateExpectedResults(0, 200),
+ "test");
}
}
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeMetaIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeMetaIT.java
index 506f6facc94..f513a3e4f3a 100644
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeMetaIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeMetaIT.java
@@ -445,7 +445,8 @@ public class IoTDBPipeMetaIT extends
AbstractPipeTableModelDualManualIT {
receiverEnv.getDataNodeWrapperList().get(0).getIpAndPortString())),
BaseEnv.TABLE_SQL_DIALECT);
- TestUtils.assertDataAlwaysOnEnv(
+ // Schema snapshots and device attribute caches are updated asynchronously
after pipe creation.
+ TestUtils.assertDataEventuallyOnEnv(
receiverEnv,
"show devices from table1",
"a,b,c,",
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeLifeCycleIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeLifeCycleIT.java
index 4373345900b..1ef15e2577d 100644
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeLifeCycleIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeLifeCycleIT.java
@@ -322,7 +322,8 @@ public class IoTDBPipeLifeCycleIT extends
AbstractPipeDualTreeModelAutoIT {
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.db.**", "Time,root.db.d1.s1,",
expectedResSet);
}
}
@@ -387,7 +388,8 @@ public class IoTDBPipeLifeCycleIT extends
AbstractPipeDualTreeModelAutoIT {
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.db.**", "Time,root.db.d1.s1,",
expectedResSet);
}
}
@@ -452,7 +454,8 @@ public class IoTDBPipeLifeCycleIT extends
AbstractPipeDualTreeModelAutoIT {
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.db.**", "Time,root.db.d1.s1,",
expectedResSet);
}
}
@@ -659,8 +662,8 @@ public class IoTDBPipeLifeCycleIT extends
AbstractPipeDualTreeModelAutoIT {
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.db.**",
"count(root.db.d1.s1),",
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeProtocolIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeProtocolIT.java
index 5489c0691b3..ead03f73c6d 100644
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeProtocolIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeProtocolIT.java
@@ -334,8 +334,8 @@ public class IoTDBPipeProtocolIT extends
AbstractPipeDualTreeModelAutoIT {
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.db.**",
"count(root.db.d1.s1),",
