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 4c4572040dc [Test] Stabilize consensus subscription seek tests on cold
JVMs (#18972)
4c4572040dc is described below
commit 4c4572040dc93b6330b85aae9a4a5f0f52dd7e6f
Author: Caideyipi <[email protected]>
AuthorDate: Thu Oct 22 11:42:10 2026 +0800
[Test] Stabilize consensus subscription seek tests on cold JVMs (#18972)
---
.../ConsensusPrefetchingQueueSeekTest.java | 69 +++++++++++++++-------
1 file changed, 47 insertions(+), 22 deletions(-)
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueSeekTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueSeekTest.java
index bd361edb08a..d113e172ca3 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueSeekTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueSeekTest.java
@@ -22,11 +22,13 @@ package org.apache.iotdb.db.subscription.broker.consensus;
import org.apache.iotdb.commons.conf.CommonConfig;
import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.consensus.DataRegionId;
+import
org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeType;
import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest;
import org.apache.iotdb.consensus.iot.IoTConsensusServerImpl;
import org.apache.iotdb.consensus.iot.SubscriptionWalRetentionPolicy;
import org.apache.iotdb.consensus.iot.WriterSafeFrontierTracker;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode;
import org.apache.iotdb.db.queryengine.plan.statement.StatementTestUtils;
import org.apache.iotdb.db.storageengine.dataregion.wal.node.WALNode;
@@ -42,6 +44,7 @@ import
org.apache.iotdb.rpc.subscription.payload.poll.TabletsPayload;
import org.apache.tsfile.enums.ColumnCategory;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.write.record.Tablet;
+import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
@@ -74,6 +77,38 @@ public class ConsensusPrefetchingQueueSeekTest {
@Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
+ private CommonConfig config;
+ private ConsensusSubscriptionPrefetchExecutorManager manager;
+ private WALNode walNode;
+ private IoTConsensusServerImpl server;
+ private ConsensusLogToTabletConverter converter;
+
+ @Before
+ public void setUp() {
+ // Keep cold PowerMock/Mockito initialization outside the seek and replay
timeout.
+ config = CommonDescriptor.getInstance().getConfig();
+ // The worker first touches plan deserializers and the Pipe runtime while
materializing entries.
+ // Initialize them here so cold class loading does not consume the replay
polling deadline.
+ PlanNodeType.values();
+ PipeDataNodeAgent.runtime();
+ manager = mock(ConsensusSubscriptionPrefetchExecutorManager.class);
+
PowerMockito.mockStatic(ConsensusSubscriptionPrefetchExecutorManager.class);
+
PowerMockito.when(ConsensusSubscriptionPrefetchExecutorManager.getInstance())
+ .thenReturn(manager);
+
+ walNode = mock(WALNode.class);
+ server = mock(IoTConsensusServerImpl.class);
+ when(server.getConsensusReqReader()).thenReturn(walNode);
+ converter = mock(ConsensusLogToTabletConverter.class);
+ when(converter.getDatabaseName()).thenReturn("db");
+ when(converter.isTableModel()).thenReturn(true);
+ when(converter.convert(any()))
+ .thenAnswer(
+ invocation ->
+ Collections.singletonList(
+ createTablet(((InsertRowNode)
invocation.getArgument(0)).getTime())));
+ }
+
@Test(timeout = 10_000L)
public void testSeekToBeginningReplaysFirstLocalEntryWithoutWalGap() throws
Exception {
assertBeginningReplay(Collections.singletonList(createRequest(1L, 1L, 7)),
2L, 0L);
@@ -112,7 +147,6 @@ public class ConsensusPrefetchingQueueSeekTest {
final long expectedSkippedEntries,
final boolean multiWriter)
throws Exception {
- final CommonConfig config = CommonDescriptor.getInstance().getConfig();
final int originalBatchMaxDelay =
config.getSubscriptionConsensusBatchMaxDelayInMs();
final int originalPrefetchThreads =
config.getSubscriptionConsensusPrefetchExecutorMaxThreadNum();
@@ -128,18 +162,10 @@ public class ConsensusPrefetchingQueueSeekTest {
// Supply an isolated runtime even in builds where subscriptions are
disabled. Seek reset
// and replay still run through the real serial prefetch worker and the
public queue API.
executor = new ConsensusSubscriptionPrefetchExecutor();
- final ConsensusSubscriptionPrefetchExecutorManager manager =
- mock(ConsensusSubscriptionPrefetchExecutorManager.class);
-
PowerMockito.mockStatic(ConsensusSubscriptionPrefetchExecutorManager.class);
-
PowerMockito.when(ConsensusSubscriptionPrefetchExecutorManager.getInstance())
- .thenReturn(manager);
when(manager.getExecutor()).thenReturn(executor);
- final WALNode walNode = mock(WALNode.class);
when(walNode.getCurrentSearchIndex()).thenReturn(expectedNextIndex - 1L);
when(walNode.getLogDirectory()).thenReturn(systemDir);
- final IoTConsensusServerImpl server = mock(IoTConsensusServerImpl.class);
- when(server.getConsensusReqReader()).thenReturn(walNode);
final WriterSafeFrontierTracker tracker = new
WriterSafeFrontierTracker();
for (final IndexedConsensusRequest request : requests) {
tracker.recordAppliedProgress(
@@ -148,14 +174,6 @@ public class ConsensusPrefetchingQueueSeekTest {
3000L, request.getNodeId(), request.getProgressLocalSeq());
}
when(server.getWriterSafeFrontierTracker()).thenReturn(tracker);
- final ConsensusLogToTabletConverter converter =
mock(ConsensusLogToTabletConverter.class);
- when(converter.getDatabaseName()).thenReturn("db");
- when(converter.isTableModel()).thenReturn(true);
- when(converter.convert(any()))
- .thenAnswer(
- invocation ->
- Collections.singletonList(
- createTablet(((InsertRowNode)
invocation.getArgument(0)).getTime())));
queue =
new ConsensusPrefetchingQueue(
@@ -188,11 +206,18 @@ public class ConsensusPrefetchingQueueSeekTest {
request.getSearchIndex() < 0
|| request.getSearchIndex() >=
startSearchIndex)
.iterator();
- final ProgressWALIterator iterator =
mock(ProgressWALIterator.class);
- when(iterator.hasNext(any(ProgressWALIterator.ScanBudget.class)))
- .thenAnswer(ignored -> retainedEntries.hasNext());
- when(iterator.next()).thenAnswer(ignored ->
retainedEntries.next());
- return iterator;
+ // Do not generate a Mockito class on the worker during timed
replay.
+ return new ProgressWALIterator(systemDir, startSearchIndex) {
+ @Override
+ boolean hasNext(final ScanBudget scanBudget) {
+ return retainedEntries.hasNext();
+ }
+
+ @Override
+ public IndexedConsensusRequest next() {
+ return retainedEntries.next();
+ }
+ };
}
};
queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(16L *
1024 * 1024));
