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 4acf90c3d89 [Subscription] Resume realtime delivery after WAL gap
recovery (#18912)
4acf90c3d89 is described below
commit 4acf90c3d8956e79d141c63093c58a807b0bdf4b
Author: Caideyipi <[email protected]>
AuthorDate: Mon Oct 19 13:57:05 2026 +0800
[Subscription] Resume realtime delivery after WAL gap recovery (#18912)
---
.../consensus/ConsensusPrefetchingQueue.java | 23 ++-
.../ConsensusPrefetchingQueueDeliveryTest.java | 16 +-
...nsensusPrefetchingQueueWalBackpressureTest.java | 219 ++++++++++++++++++++-
3 files changed, 245 insertions(+), 13 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 6ed389ddafb..ae71c4fea2a 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
@@ -2286,14 +2286,15 @@ public class ConsensusPrefetchingQueue {
subscriptionWALIterator.advanceTo(fromIndex,
this::isWriterProgressCoveredForWalFastForward);
}
// 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.
+ // Stop replay at the pending boundary. Admitted requests should use
direct delivery instead of
+ // being replayed and discarded as duplicates. Gap recovery still shares
the normal scan budget,
+ // so sparse filtering cannot hold the read lock indefinitely and block
unsubscribe.
final MaterializationResult pumpResult =
pumpFromSubscriptionWAL(
batchState,
expectedSeekGeneration,
maxWalEntries,
+ toIndex,
maxTablets,
maxBatchBytes,
walScanBudget);
@@ -2353,6 +2354,7 @@ public class ConsensusPrefetchingQueue {
batchState,
expectedSeekGeneration,
maxWalEntries,
+ Long.MAX_VALUE,
maxTablets,
maxBatchBytes,
walScanBudget);
@@ -2380,6 +2382,7 @@ public class ConsensusPrefetchingQueue {
final DeliveryBatchState batchState,
final long expectedSeekGeneration,
final int maxWalEntries,
+ final long endSearchIndexExclusive,
final int maxTablets,
final long maxBatchBytes,
final ProgressWALIterator.ScanBudget walScanBudget) {
@@ -2406,6 +2409,14 @@ public class ConsensusPrefetchingQueue {
reportUnreadableWalReplayIfNecessary();
try {
final IndexedConsensusRequest walEntry =
subscriptionWALIterator.next();
+ if (hasLocalSearchIndex(walEntry) && walEntry.getSearchIndex() >=
endSearchIndexExclusive) {
+ // A sparse WAL can jump past the pending boundary. Preserve its
next request while
+ // handing the admitted entry back to the pending path, and report
only the missing gap.
+ subscriptionWALIterator.retry(walEntry);
+ reportWalGapBefore(endSearchIndexExclusive);
+ nextExpectedSearchIndex.set(endSearchIndexExclusive);
+ break;
+ }
entriesRead++;
if (isBeforeLocalCursor(walEntry)) {
@@ -2495,6 +2506,11 @@ public class ConsensusPrefetchingQueue {
}
final long actualSearchIndex = request.getSearchIndex();
+ reportWalGapBefore(actualSearchIndex);
+ nextExpectedSearchIndex.set(actualSearchIndex + 1);
+ }
+
+ private void reportWalGapBefore(final long actualSearchIndex) {
final long expectedSearchIndex = nextExpectedSearchIndex.get();
if (actualSearchIndex > expectedSearchIndex) {
final long skippedEntries = actualSearchIndex - expectedSearchIndex;
@@ -2508,7 +2524,6 @@ public class ConsensusPrefetchingQueue {
skippedEntries,
totalSkippedEntries);
}
- nextExpectedSearchIndex.set(actualSearchIndex + 1);
}
private void ensureSubscriptionWalReadable(final
ProgressWALIterator.ScanBudget walScanBudget) {
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueDeliveryTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueDeliveryTest.java
index e553a509766..1f1a14535ca 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueDeliveryTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueDeliveryTest.java
@@ -257,7 +257,9 @@ public class ConsensusPrefetchingQueueDeliveryTest {
config.setSubscriptionConsensusBatchMaxTabletCount(2);
writeWal(1, true);
currentSearchIndex.set(2);
- final ConsensusPrefetchingQueue queue = newQueue();
+ // Keep the mixed batch deterministic when first-use class loading
exhausts a scan slice.
+ // Bounded scan yielding is covered by
ConsensusPrefetchingQueueWalBackpressureTest.
+ final ConsensusPrefetchingQueue queue = newQueue(true);
assertNull(queue.poll(CONSUMER));
assertTrue(pendingEntries(queue).offer(request(2, 2, WRITER)));
queue.drivePrefetchOnce();
@@ -591,6 +593,10 @@ public class ConsensusPrefetchingQueueDeliveryTest {
}
private ConsensusPrefetchingQueue newQueue() {
+ return newQueue(false);
+ }
+
+ private ConsensusPrefetchingQueue newQueue(final boolean
unboundedWalFixture) {
final IoTConsensusServerImpl server = mock(IoTConsensusServerImpl.class);
when(server.getConsensusReqReader()).thenReturn(walNode);
when(server.getWriterSafeFrontierTracker()).thenReturn(new
WriterSafeFrontierTracker());
@@ -614,6 +620,14 @@ public class ConsensusPrefetchingQueueDeliveryTest {
@Override
protected ProgressWALIterator createSubscriptionWALIterator(final
long startSearchIndex) {
walIteratorCreations.incrementAndGet();
+ if (unboundedWalFixture) {
+ return new ProgressWALIterator(walNode, startSearchIndex) {
+ @Override
+ boolean hasNext(final ScanBudget scanBudget) {
+ return super.hasNext(null);
+ }
+ };
+ }
return super.createSubscriptionWALIterator(startSearchIndex);
}
};
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 204a557367e..adcf18a50d8 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
@@ -22,11 +22,16 @@ 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.memory.AtomicLongMemoryBlock;
+import org.apache.iotdb.commons.memory.IMemoryBlock;
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.consensus.iot.logdispatcher.IoTConsensusMemoryManager;
+import
org.apache.iotdb.consensus.iot.subscription.SubscriptionQueueRejectionReason;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryWeightUtil;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode;
import org.apache.iotdb.db.queryengine.plan.statement.StatementTestUtils;
@@ -62,6 +67,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
@@ -247,6 +253,181 @@ public class ConsensusPrefetchingQueueWalBackpressureTest
{
}
}
+ @Test
+ public void testWalGapRecoveryResumesRealtimeAfterConsensusMemoryRejection()
throws Exception {
+ assertWalGapRecoveryResumesRealtime(false, false, false);
+ }
+
+ @Test
+ public void testWalGapRecoveryResumesRealtimeAfterAck() throws Exception {
+ assertWalGapRecoveryResumesRealtime(true, false, false);
+ }
+
+ @Test
+ public void testWalGapRecoveryResumesPendingWhenWalStartsAfterGap() throws
Exception {
+ assertWalGapRecoveryResumesRealtime(false, true, false);
+ }
+
+ @Test
+ public void testWalGapRecoveryPreservesFollowerBeforePendingBoundary()
throws Exception {
+ assertWalGapRecoveryResumesRealtime(false, false, true);
+ }
+
+ private void assertWalGapRecoveryResumesRealtime(
+ final boolean memoryBackpressure,
+ final boolean missingWalGap,
+ final boolean followerBeforePending)
+ throws Exception {
+ final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
+ final CommonConfig config = CommonDescriptor.getInstance().getConfig();
+ final int originalBatchMaxWalEntries =
config.getSubscriptionConsensusBatchMaxWalEntries();
+ final int originalBatchMaxTabletCount =
config.getSubscriptionConsensusBatchMaxTabletCount();
+ final int originalBatchMaxDelay =
config.getSubscriptionConsensusBatchMaxDelayInMs();
+ final long originalWatermarkInterval =
config.getSubscriptionConsensusWatermarkIntervalMs();
+ final IoTConsensusMemoryManager requestMemoryManager =
IoTConsensusMemoryManager.getInstance();
+ final IMemoryBlock originalRequestMemoryBlock =
requestMemoryManager.getMemoryBlock();
+ final double originalQueueRatio =
requestMemoryManager.getMaxMemoryRatioForQueue();
+ final AtomicLongMemoryBlock requestMemoryBlock =
+ new AtomicLongMemoryBlock("subscription-gap-recovery", null,
1_000_000L);
+ final File systemDir = temporaryFolder.newFolder("system-gap-recovery");
+ final File walDirectory = temporaryFolder.newFolder("wal-gap-recovery");
+ final int requestCount = 8;
+ ConsensusPrefetchingQueue queue = null;
+ try {
+ config.setSubscriptionConsensusBatchMaxWalEntries(128);
+ config.setSubscriptionConsensusBatchMaxTabletCount(1);
+ config.setSubscriptionConsensusBatchMaxDelayInMs(0);
+ config.setSubscriptionConsensusWatermarkIntervalMs(0L);
+ writeSealedWal(walDirectory, requestCount, missingWalGap ? 2L : 1L,
followerBeforePending);
+
+ final WALNode walNode = mock(WALNode.class);
+ when(walNode.getLogDirectory()).thenReturn(walDirectory);
+ when(walNode.getCurrentSearchIndex()).thenReturn((long) requestCount);
+ when(walNode.getCurrentWALFileVersion()).thenReturn(1L);
+ when(walNode.getCurrentWALMetaDataSnapshot()).thenReturn(new
WALMetaData());
+ final IoTConsensusServerImpl serverImpl =
mock(IoTConsensusServerImpl.class);
+ when(serverImpl.getConsensusReqReader()).thenReturn(walNode);
+ when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new
WriterSafeFrontierTracker());
+ final List<Long> conversions = new ArrayList<>();
+ final ConsensusLogToTabletConverter converter =
mock(ConsensusLogToTabletConverter.class);
+ when(converter.convert(any()))
+ .thenAnswer(
+ invocation -> {
+ conversions.add(((InsertNode)
invocation.getArgument(0)).getSearchIndex());
+ return Collections.singletonList(createTablet());
+ });
+ when(converter.getDatabaseName()).thenReturn("db");
+ final DataRegionId regionId = new DataRegionId(1);
+ final ConsensusSubscriptionCommitManager commitManager =
newCommitManager(systemDir);
+ queue =
+ new ConsensusPrefetchingQueue(
+ "consumerGroup",
+ "topic",
+ followerBeforePending
+ ? TopicConstant.ORDER_MODE_MULTI_WRITER_VALUE
+ : TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE,
+ regionId,
+ serverImpl,
+ new SubscriptionWalRetentionPolicy(
+ "topic",
+ SubscriptionWalRetentionPolicy.UNBOUNDED,
+ SubscriptionWalRetentionPolicy.UNBOUNDED),
+ converter,
+ commitManager,
+ new RegionProgress(Collections.emptyMap()),
+ 1L,
+ 1L,
+ true);
+ if (followerBeforePending) {
+ queue.setActiveWriterNodeIds(Set.of(WRITER_NODE_ID, WRITER_NODE_ID +
1));
+ }
+ final long oneTabletBytes =
PipeMemoryWeightUtil.calculateTabletSizeInBytes(createTablet());
+ final TrackingSubscriptionMemoryManager memoryManager =
+ new TrackingSubscriptionMemoryManager(
+ memoryBackpressure ? oneTabletBytes * 4L : oneTabletBytes *
requestCount * 4L);
+ queue.setSubscriptionMemoryManager(memoryManager);
+ assertNull(queue.poll("consumer"));
+
+ // A rejected request creates a WAL-only hole. Once consensus memory is
released, subsequent
+ // requests are admitted again and must be delivered from pending after
that hole is filled.
+ requestMemoryManager.init(requestMemoryBlock, 1.0);
+ requestMemoryBlock.forceAllocateWithoutLimitation(1_000_000L);
+ assertFalse(pendingEntries(queue).offer(createRequest(1L)));
+ assertEquals(
+
SubscriptionQueueRejectionReason.CONSENSUS_REQUEST_MEMORY_LIMIT.getCode(),
+ queue.getLastAdmissionRejectionCode());
+ requestMemoryBlock.release(1_000_000L);
+ for (long index = 2L; index <= requestCount; index++) {
+ assertTrue(pendingEntries(queue).offer(createRequest(index)));
+ }
+
+ long deliveredLocalSeq = missingWalGap ? 1L : 0L;
+ for (int round = 0; round < 32 && deliveredLocalSeq < requestCount;
round++) {
+ queue.drivePrefetchOnce();
+ assertMemoryBounded(queue, memoryManager);
+ SubscriptionEvent event;
+ while ((event = queue.poll("consumer")) != null) {
+ assertEquals(WRITER_NODE_ID,
event.getCommitContext().getWriterId().getNodeId());
+ deliveredLocalSeq++;
+ assertEquals(
+ deliveredLocalSeq,
event.getCommitContext().getWriterProgress().getLocalSeq());
+ assertTrue(queue.ack("consumer", event.getCommitContext()));
+ }
+ assertEquals(
+ followerBeforePending ? oneTabletBytes : 0L,
memoryManager.getUsedMemorySizeInBytes());
+ }
+ assertEquals(requestCount, deliveredLocalSeq);
+ if (followerBeforePending) {
+ // 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));
+ queue.drivePrefetchOnce();
+ final SubscriptionEvent follower = queue.poll("consumer");
+ assertNotNull(follower);
+ assertEquals(WRITER_NODE_ID + 1,
follower.getCommitContext().getWriterId().getNodeId());
+ assertEquals(
+ new WriterProgress(2000L, 1L),
follower.getCommitContext().getWriterProgress());
+ assertTrue(queue.ack("consumer", follower.getCommitContext()));
+ assertNull(queue.poll("consumer"));
+ assertEquals(0L, memoryManager.getUsedMemorySizeInBytes());
+ }
+ assertEquals(
+ missingWalGap ? 0L : (followerBeforePending ? 2L : 1L),
+ queue.getWalPathAcceptedEntries());
+ assertEquals(requestCount - 1L, queue.getPendingPathAcceptedEntries());
+ assertEquals(1L, queue.getRealtimeAdmissionRejectionCount());
+ assertEquals(missingWalGap ? 1L : 0L, queue.getWalGapSkippedEntries());
+ assertEquals(0L, queue.getRetainedRequestBytes());
+ assertEquals(0L, requestMemoryBlock.getUsedMemoryInBytes());
+ assertTrue(pendingEntries(queue).isEmpty());
+ assertEquals(
+ new WriterProgress(1000L + requestCount, requestCount),
+ commitManager
+ .getCommittedRegionProgress("consumerGroup", "topic", regionId)
+ .getWriterPositions()
+ .get(new WriterId(regionId.toString(), WRITER_NODE_ID)));
+ if (!memoryBackpressure) {
+ assertEquals(
+ 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)),
+ conversions);
+ }
+ } finally {
+ if (queue != null) {
+ queue.close();
+ }
+ requestMemoryManager.init(originalRequestMemoryBlock,
originalQueueRatio);
+
config.setSubscriptionConsensusBatchMaxWalEntries(originalBatchMaxWalEntries);
+
config.setSubscriptionConsensusBatchMaxTabletCount(originalBatchMaxTabletCount);
+ config.setSubscriptionConsensusBatchMaxDelayInMs(originalBatchMaxDelay);
+
config.setSubscriptionConsensusWatermarkIntervalMs(originalWatermarkInterval);
+
IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
+ }
+ }
+
@Test
public void testAckResumesPendingSuffixWithoutWalReplay() throws Exception {
final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
@@ -690,6 +871,21 @@ public class ConsensusPrefetchingQueueWalBackpressureTest {
private static void writeSealedWal(final File walDirectory, final int
requestCount)
throws Exception {
+ writeSealedWal(walDirectory, requestCount, 1L);
+ }
+
+ private static void writeSealedWal(
+ final File walDirectory, final int requestCount, final long
firstSearchIndex)
+ throws Exception {
+ writeSealedWal(walDirectory, requestCount, firstSearchIndex, false);
+ }
+
+ private static void writeSealedWal(
+ final File walDirectory,
+ final int requestCount,
+ final long firstSearchIndex,
+ final boolean followerBeforePending)
+ throws Exception {
final File historicalWal =
new File(
walDirectory, WALFileUtils.getLogFileName(0L, 0L,
WALFileStatus.CONTAINS_SEARCH_INDEX));
@@ -699,8 +895,11 @@ public class ConsensusPrefetchingQueueWalBackpressureTest {
WALFileUtils.getLogFileName(1L, requestCount,
WALFileStatus.CONTAINS_SEARCH_INDEX));
try (WALWriter writer = new WALWriter(historicalWal, WALFileVersion.V3)) {
- for (long searchIndex = 1L; searchIndex <= requestCount; searchIndex++) {
+ for (long searchIndex = firstSearchIndex; searchIndex <= requestCount;
searchIndex++) {
writeWalEntry(writer, searchIndex);
+ if (searchIndex == 1L && followerBeforePending) {
+ writeWalEntry(writer, -1L, 2000L, WRITER_NODE_ID + 1, 1L);
+ }
}
}
try (WALWriter ignored = new WALWriter(liveWal, WALFileVersion.V3)) {
@@ -710,6 +909,16 @@ public class ConsensusPrefetchingQueueWalBackpressureTest {
private static void writeWalEntry(final WALWriter writer, final long
searchIndex)
throws Exception {
+ writeWalEntry(writer, searchIndex, 1000L + searchIndex, WRITER_NODE_ID,
searchIndex);
+ }
+
+ private static void writeWalEntry(
+ final WALWriter writer,
+ final long searchIndex,
+ final long physicalTime,
+ final int writerNodeId,
+ final long localSeq)
+ throws Exception {
final RelationalInsertRowNode insertNode =
StatementTestUtils.genInsertRowNode(Math.toIntExact(searchIndex));
insertNode.setSearchIndex(searchIndex);
@@ -719,13 +928,7 @@ public class ConsensusPrefetchingQueueWalBackpressureTest {
walEntry.serialize(new WALByteBufferForTest(buffer));
final WALMetaData metaData = new WALMetaData();
- metaData.add(
- walEntry.serializedSize(),
- searchIndex,
- 1L,
- 1000L + searchIndex,
- WRITER_NODE_ID,
- searchIndex);
+ metaData.add(walEntry.serializedSize(), searchIndex, 1L, physicalTime,
writerNodeId, localSeq);
writer.write(buffer, metaData);
}
