From b0045ffbc0aa5125e0ffd8370eb54bfacb303bee Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Tue, 20 Oct 2026 18:50:09 +0800 Subject: [PATCH 1/2] [Test] Align WAL replay and pipe permission assertions with runtime behavior --- .../dual/treemodel/manual/IoTDBPipePermissionIT.java | 11 +++++++++-- .../ConsensusPrefetchingQueueWalBackpressureTest.java | 8 +++++--- 2 files changed, 14 insertions(+), 5 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipePermissionIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipePermissionIT.java index b9fc3906292f..3b4cf97e8a1a 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipePermissionIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipePermissionIT.java @@ -21,6 +21,7 @@ import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient; +import org.apache.iotdb.commons.schema.column.ColumnHeaderConstant; import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq; import org.apache.iotdb.consensus.ConsensusFactory; import org.apache.iotdb.db.it.utils.TestUtils; @@ -452,7 +453,10 @@ public void testSourcePermission() { // root.test / root.db shall not be transferred TestUtils.assertDataAlwaysOnEnv( - receiverEnv, "count databases", "count,", Collections.singleton("0,")); + receiverEnv, + "count databases", + ColumnHeaderConstant.COUNT_DATABASE + ",", + Collections.singleton("0,")); // Test snapshot filter TestUtils.executeNonQueries( @@ -506,7 +510,10 @@ public void testSourcePermission() { // Shall be transferred // The root.test may or may not be transferred due to delayed start of the config source TestUtils.assertDataEventuallyOnEnv( - receiverEnv, "count databases root.test1", "count,", Collections.singleton("1,")); + receiverEnv, + "count databases root.test1", + ColumnHeaderConstant.COUNT_DATABASE + ",", + Collections.singleton("1,")); // Write some data TestUtils.executeNonQueries( diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueWalBackpressureTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueWalBackpressureTest.java index adcf18a50d82..e7699eb5e205 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueWalBackpressureTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueWalBackpressureTest.java @@ -373,11 +373,13 @@ private void assertWalGapRecoveryResumesRealtime( deliveredLocalSeq, event.getCommitContext().getWriterProgress().getLocalSeq()); assertTrue(queue.ack("consumer", event.getCommitContext())); } - assertEquals( - followerBeforePending ? oneTabletBytes : 0L, memoryManager.getUsedMemorySizeInBytes()); + // Ordered replay retains blocked writer heads as raw requests until they can be emitted. + assertEquals(0L, memoryManager.getUsedMemorySizeInBytes()); } assertEquals(requestCount, deliveredLocalSeq); if (followerBeforePending) { + assertTrue(queue.getRetainedRequestBytes() > 0L); + assertEquals(queue.getRetainedRequestBytes(), requestMemoryBlock.getUsedMemoryInBytes()); // The replicated writer's later timestamp lets local requests drain in order. Removing // the local writer then releases that final follower entry without reading it again. queue.setActiveWriterNodeIds(Collections.singleton(WRITER_NODE_ID + 1)); @@ -411,7 +413,7 @@ private void assertWalGapRecoveryResumesRealtime( missingWalGap ? Arrays.asList(2L, 3L, 4L, 5L, 6L, 7L, 8L) : (followerBeforePending - ? Arrays.asList(1L, -1L, 2L, 3L, 4L, 5L, 6L, 7L, 8L) + ? Arrays.asList(1L, 2L, 3L, 4L, 5L, 6L, 7L, 8L, -1L) : Arrays.asList(1L, 2L, 3L, 4L, 5L, 6L, 7L, 8L)), conversions); } From 179cf8d470b2b0bde6ad5ac335a7ba22cb7f683a Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Tue, 20 Oct 2026 19:08:02 +0800 Subject: [PATCH 2/2] [Test] Keep existing pipe database-count expectations --- .../dual/treemodel/manual/IoTDBPipePermissionIT.java | 11 ++--------- 1 file changed, 2 insertions(+), 9 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipePermissionIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipePermissionIT.java index 3b4cf97e8a1a..b9fc3906292f 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipePermissionIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipePermissionIT.java @@ -21,7 +21,6 @@ import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient; -import org.apache.iotdb.commons.schema.column.ColumnHeaderConstant; import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq; import org.apache.iotdb.consensus.ConsensusFactory; import org.apache.iotdb.db.it.utils.TestUtils; @@ -453,10 +452,7 @@ public void testSourcePermission() { // root.test / root.db shall not be transferred TestUtils.assertDataAlwaysOnEnv( - receiverEnv, - "count databases", - ColumnHeaderConstant.COUNT_DATABASE + ",", - Collections.singleton("0,")); + receiverEnv, "count databases", "count,", Collections.singleton("0,")); // Test snapshot filter TestUtils.executeNonQueries( @@ -510,10 +506,7 @@ public void testSourcePermission() { // Shall be transferred // The root.test may or may not be transferred due to delayed start of the config source TestUtils.assertDataEventuallyOnEnv( - receiverEnv, - "count databases root.test1", - ColumnHeaderConstant.COUNT_DATABASE + ",", - Collections.singleton("1,")); + receiverEnv, "count databases root.test1", "count,", Collections.singleton("1,")); // Write some data TestUtils.executeNonQueries(