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 abd499334f0 [Subscription] Bound pending WAL gap replay per round
(#18766)
abd499334f0 is described below
commit abd499334f024eb36379582d6c3b60b2690ed062
Author: Caideyipi <[email protected]>
AuthorDate: Wed Sep 30 10:16:08 2026 +0800
[Subscription] Bound pending WAL gap replay per round (#18766)
* [Subscription] Bound pending WAL gap replay per round
* [Subscription] Stabilize WAL gap counter regression test
---
.../consensus/ConsensusPrefetchingQueue.java | 16 ++-
.../consensus/ConsensusPrefetchingQueueTest.java | 108 ++++++++++++++++++---
2 files changed, 108 insertions(+), 16 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java
index d415656986b..149d968796e 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java
@@ -1481,7 +1481,12 @@ public class ConsensusPrefetchingQueue {
try {
batchResult =
accumulateFromPending(
- batch, lingerBatch, observedSeekGeneration, maxTablets,
maxBatchBytes);
+ batch,
+ lingerBatch,
+ observedSeekGeneration,
+ maxWalEntries,
+ maxTablets,
+ maxBatchBytes);
} finally {
pendingEntries.release(batch);
}
@@ -1758,6 +1763,7 @@ public class ConsensusPrefetchingQueue {
final List<IndexedConsensusRequest> batch,
final DeliveryBatchState lingerBatch,
final long expectedSeekGeneration,
+ final int maxWalEntries,
final int maxTablets,
final long maxBatchBytes) {
@@ -1784,6 +1790,7 @@ public class ConsensusPrefetchingQueue {
searchIndex,
lingerBatch,
expectedSeekGeneration,
+ maxWalEntries,
maxTablets,
maxBatchBytes);
if (gapFillResult != MaterializationResult.SUCCESS) {
@@ -1850,6 +1857,7 @@ public class ConsensusPrefetchingQueue {
final long toIndex,
final DeliveryBatchState batchState,
final long expectedSeekGeneration,
+ final int maxWalEntries,
final int maxTablets,
final long maxBatchBytes) {
pendingWalGapRetryRequested = false;
@@ -1857,9 +1865,13 @@ public class ConsensusPrefetchingQueue {
if (seekGeneration.get() != expectedSeekGeneration || isClosed) {
return MaterializationResult.STALE;
}
+ // A pending-queue jump can span millions of WAL entries when real-time
delivery has overflowed.
+ // Keep gap recovery within the normal per-round budget so sparse topic
filtering cannot hold
+ // the
+ // queue read lock indefinitely and block unsubscribe from acquiring the
close write lock.
final MaterializationResult pumpResult =
pumpFromSubscriptionWAL(
- batchState, expectedSeekGeneration, Integer.MAX_VALUE, maxTablets,
maxBatchBytes);
+ batchState, expectedSeekGeneration, maxWalEntries, maxTablets,
maxBatchBytes);
if (pumpResult != MaterializationResult.SUCCESS) {
return pumpResult;
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java
index 69ece3840ef..9d85d6b9d57 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java
@@ -1178,27 +1178,44 @@ public class ConsensusPrefetchingQueueTest {
}
@Test
- public void testWalReplayCountsOnlyUnavailableSearchIndexes() throws
Exception {
+ public void testPendingGapReplayHonorsPerRoundWalEntryLimit() throws
Exception {
final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
- final File systemDir = temporaryFolder.newFolder("wal-replay-gap-counter");
+ final int originalBatchMaxWalEntries =
+
CommonDescriptor.getInstance().getConfig().getSubscriptionConsensusBatchMaxWalEntries();
+ final File systemDir =
temporaryFolder.newFolder("bounded-pending-gap-replay");
ConsensusPrefetchingQueue queue = null;
try {
- final DataRegionId regionId = new DataRegionId(9);
- final FakeConsensusReqReader reader = new FakeConsensusReqReader();
- reader.currentSearchIndex = 4L;
+
CommonDescriptor.getInstance().getConfig().setSubscriptionConsensusBatchMaxWalEntries(2);
+
+ final DataRegionId regionId = new DataRegionId(12);
+ final WALNode walNode = mock(WALNode.class);
+ when(walNode.getCurrentSearchIndex()).thenReturn(5L);
+ when(walNode.getLogDirectory()).thenReturn(systemDir);
final IoTConsensusServerImpl serverImpl =
mock(IoTConsensusServerImpl.class);
- when(serverImpl.getConsensusReqReader()).thenReturn(reader);
+ when(serverImpl.getConsensusReqReader()).thenReturn(walNode);
when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new
WriterSafeFrontierTracker());
+ final AtomicInteger conversionCount = new AtomicInteger();
final ConsensusLogToTabletConverter converter =
mock(ConsensusLogToTabletConverter.class);
-
when(converter.convert(any())).thenReturn(Collections.singletonList(createTablet()));
+ when(converter.convert(any()))
+ .thenAnswer(
+ ignored -> {
+ conversionCount.incrementAndGet();
+ return Collections.emptyList();
+ });
when(converter.getDatabaseName()).thenReturn("db");
- final Iterator<IndexedConsensusRequest> retainedWalEntries =
- Arrays.asList(createRequest(1L), createRequest(4L)).iterator();
- final ProgressWALIterator walIterator = mock(ProgressWALIterator.class);
- when(walIterator.hasNext()).thenAnswer(ignored ->
retainedWalEntries.hasNext());
- when(walIterator.next()).thenAnswer(ignored ->
retainedWalEntries.next());
+ final Iterator<IndexedConsensusRequest> walEntries =
+ Arrays.asList(
+ createRequest(1L),
+ createRequest(2L),
+ createRequest(3L),
+ createRequest(4L),
+ createRequest(5L))
+ .iterator();
+ final ProgressWALIterator iterator = mock(ProgressWALIterator.class);
+ when(iterator.hasNext()).thenAnswer(ignored -> walEntries.hasNext());
+ when(iterator.next()).thenAnswer(ignored -> walEntries.next());
queue =
new ConsensusPrefetchingQueue(
@@ -1220,14 +1237,77 @@ public class ConsensusPrefetchingQueueTest {
@Override
protected ProgressWALIterator createSubscriptionWALIterator(
final long startSearchIndex) {
- return walIterator;
+ return iterator;
}
};
+ queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(16L *
1024 * 1024));
assertNull(queue.poll("consumer"));
+ assertTrue(pendingEntries(queue).offer(createRequest(5L)));
+
+ queue.drivePrefetchOnce();
+ assertEquals(2, conversionCount.get());
+ assertEquals(3L, queue.getCurrentReadSearchIndex());
+ assertTrue(pendingEntries(queue).isEmpty());
+
queue.drivePrefetchOnce();
+ assertEquals(4, conversionCount.get());
+ assertEquals(5L, queue.getCurrentReadSearchIndex());
+
+ queue.drivePrefetchOnce();
+ assertEquals(5, conversionCount.get());
+ assertEquals(6L, queue.getCurrentReadSearchIndex());
+ } finally {
+ if (queue != null) {
+ queue.close();
+ }
+ CommonDescriptor.getInstance()
+ .getConfig()
+
.setSubscriptionConsensusBatchMaxWalEntries(originalBatchMaxWalEntries);
+
IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
+ }
+ }
+
+ @Test
+ public void testWalReplayCountsOnlyUnavailableSearchIndexes() throws
Exception {
+ final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
+ final File systemDir = temporaryFolder.newFolder("wal-replay-gap-counter");
+ ConsensusPrefetchingQueue queue = null;
+ try {
+ final DataRegionId regionId = new DataRegionId(9);
+ final FakeConsensusReqReader reader = new FakeConsensusReqReader();
+ final IoTConsensusServerImpl serverImpl =
mock(IoTConsensusServerImpl.class);
+ when(serverImpl.getConsensusReqReader()).thenReturn(reader);
+ when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new
WriterSafeFrontierTracker());
+
+ queue =
+ new ConsensusPrefetchingQueue(
+ "consumerGroup",
+ "topic",
+ TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE,
+ regionId,
+ serverImpl,
+ new SubscriptionWalRetentionPolicy(
+ "topic",
+ SubscriptionWalRetentionPolicy.UNBOUNDED,
+ SubscriptionWalRetentionPolicy.UNBOUNDED),
+ mock(ConsensusLogToTabletConverter.class),
+ newCommitManager(systemDir),
+ new RegionProgress(Collections.emptyMap()),
+ 1L,
+ 1L,
+ true);
+
+ final Method advanceWalReplayCursorIfPresent =
+ ConsensusPrefetchingQueue.class.getDeclaredMethod(
+ "advanceWalReplayCursorIfPresent",
IndexedConsensusRequest.class);
+ advanceWalReplayCursorIfPresent.setAccessible(true);
+
+ advanceWalReplayCursorIfPresent.invoke(queue, createRequest(1L));
+ assertEquals(0L, queue.getWalGapSkippedEntries());
+ assertEquals(2L, queue.getCurrentReadSearchIndex());
- assertEquals(2L, queue.getWalPathAcceptedEntries());
+ advanceWalReplayCursorIfPresent.invoke(queue, createRequest(4L));
assertEquals(2L, queue.getWalGapSkippedEntries());
assertEquals(5L, queue.getCurrentReadSearchIndex());
} finally {
