From eba6ac4b1db27022cbb63d142ade0b214065ff2c Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Wed, 21 Oct 2026 12:37:05 +0800 Subject: [PATCH 1/2] [To dev/1.3] [Test] Check receiver data after stopping pipes (#18941) Backport the five applicable tree-model lifecycle and protocol assertions, adapting the test paths to the 1.3 layout. Omit the table-model changes, which do not apply to this branch. Restore the consistency helper polling loop using the existing master behavior. The 1.3 helper previously configured failFast without a terminal condition and never queried the receiver. (cherry picked from commit c535723a17a9492dc4479de5af7ab0c02864b509) --- .../org/apache/iotdb/db/it/utils/TestUtils.java | 9 +++++---- .../pipe/it/autocreate/IoTDBPipeLifeCycleIT.java | 13 ++++++++----- .../pipe/it/autocreate/IoTDBPipeProtocolIT.java | 4 ++-- 3 files changed, 15 insertions(+), 11 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java index b2a7ba4d22e8..e15e24f85e2c 100644 --- a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/TestUtils.java +++ b/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(); diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeLifeCycleIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeLifeCycleIT.java index 7044b9942627..79eb02b4ec40 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeLifeCycleIT.java +++ b/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),", diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeProtocolIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeProtocolIT.java index c9ef7a9da95f..3f406091ffe9 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeProtocolIT.java +++ b/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),", From f53fda837ae156fadfe35c0207283c4543ad4f1d Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Wed, 21 Oct 2026 14:35:54 +0800 Subject: [PATCH 2/2] Fix pipe inclusion assertions exposed by continuous polling --- .../pipe/it/manual/IoTDBPipeInclusionIT.java | 22 +++++++++++++++---- .../it/manual/IoTDBPipeMetaHistoricalIT.java | 2 +- 2 files changed, 19 insertions(+), 5 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/manual/IoTDBPipeInclusionIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/manual/IoTDBPipeInclusionIT.java index 1385829578fd..394cf62a5838 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/manual/IoTDBPipeInclusionIT.java +++ b/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,", diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/manual/IoTDBPipeMetaHistoricalIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/manual/IoTDBPipeMetaHistoricalIT.java index 0827d3621790..baa295c3578f 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/manual/IoTDBPipeMetaHistoricalIT.java +++ b/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);