From c5324501aa4217d8c2d46cb1d5f5bdfc3dbfd504 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Tue, 20 Oct 2026 18:01:09 +0800 Subject: [PATCH 1/2] [Test] Wait for asynchronous pipe attribute synchronization --- .../it/dual/tablemodel/manual/enhanced/IoTDBPipeMetaIT.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) 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 506f6facc944e..f513a3e4f3ad8 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 void testAttributeSync() { 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,", From 507365924de535735e4628206744d8c091ad9157 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Tue, 20 Oct 2026 18:24:40 +0800 Subject: [PATCH 2/2] [Test] Check receiver data remains unchanged after stopping pipes --- .../manual/basic/IoTDBPipeLifeCycleIT.java | 3 ++- .../manual/basic/IoTDBPipeProtocolIT.java | 8 +++++++- .../treemodel/auto/basic/IoTDBPipeLifeCycleIT.java | 13 ++++++++----- .../treemodel/auto/basic/IoTDBPipeProtocolIT.java | 4 ++-- 4 files changed, 19 insertions(+), 9 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 afa6b38775e62..be9f569e1729b 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 void testReceiverAlreadyHaveTimeSeries() throws Exception { 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 9a6fe6ddf41d1..809278f998a60 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 @@ private void doTest() throws Exception { 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/treemodel/auto/basic/IoTDBPipeLifeCycleIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeLifeCycleIT.java index 4373345900b48..1ef15e2577d7d 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 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.db.**", "Time,root.db.d1.s1,", expectedResSet); } } @@ -387,7 +388,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.db.**", "Time,root.db.d1.s1,", expectedResSet); } } @@ -452,7 +454,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.db.**", "Time,root.db.d1.s1,", expectedResSet); } } @@ -659,8 +662,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.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 5489c0691b360..ead03f73c6dd1 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 @@ private void doTest() 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.db.**", "count(root.db.d1.s1),",