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 3d57d2507fa [Subscription] Preserve WAL replay progress across pending
gaps (#18831)
3d57d2507fa is described below
commit 3d57d2507faa23e1d32867b913f7772fd380a677
Author: Caideyipi <[email protected]>
AuthorDate: Wed Oct 14 11:15:45 2026 +0800
[Subscription] Preserve WAL replay progress across pending gaps (#18831)
---
.../consensus/ConsensusPrefetchingQueue.java | 13 +-
...nsensusPrefetchingQueueWalBackpressureTest.java | 182 ++++++++++++++++++++-
2 files changed, 191 insertions(+), 4 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 44d9c101443..63c06f06960 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
@@ -2046,7 +2046,7 @@ public class ConsensusPrefetchingQueue {
} finally {
// Advance once for the consumed prefix, including when a gap or memory
block stops the
// batch. Per-entry fast-forward can repeatedly inspect the same
uncovered WAL file footer.
- // Gap recovery creates its own iterator at the current cursor before
reading WAL.
+ // Gap recovery realigns the same iterator at the current cursor before
reading WAL.
final ProgressWALIterator iterator = subscriptionWALIterator;
if (nextExpectedSearchIndex.get() > startSearchIndex
&& Objects.nonNull(iterator)
@@ -2077,10 +2077,16 @@ public class ConsensusPrefetchingQueue {
final int maxTablets,
final long maxBatchBytes) {
pendingWalGapRetryRequested = false;
- resetSubscriptionWALPosition(fromIndex);
if (seekGeneration.get() != expectedSeekGeneration || isClosed) {
return MaterializationResult.STALE;
}
+ // Preserve the reader position and any memory-blocked request across
pending-queue gaps.
+ // Reopening here scans retained WAL prefixes again on every bounded
replay round, and drops
+ // the request cached for a memory retry. advanceTo skips covered work
without losing uncovered
+ // writer progress, including requests without a local search index.
+ if (Objects.nonNull(subscriptionWALIterator)) {
+ 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
@@ -2223,6 +2229,9 @@ public class ConsensusPrefetchingQueue {
.PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_ERROR_READING_SUBSCRIPTION_WAL_A3888AC5,
this,
e);
+ // next() may have consumed a request whose materialization failed.
Reusing this reader
+ // would lose that request, so recover from the unchanged cursor on
the next round.
+ requestSubscriptionWalReset(nextExpectedSearchIndex.get(),
expectedSeekGeneration);
break;
}
}
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 54ea59fbb0a..bb7b2e59371 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
@@ -53,6 +53,7 @@ import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import java.io.File;
+import java.io.IOException;
import java.lang.reflect.Constructor;
import java.lang.reflect.Field;
import java.nio.ByteBuffer;
@@ -61,6 +62,8 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@@ -332,17 +335,192 @@ public class
ConsensusPrefetchingQueueWalBackpressureTest {
}
}
+ @Test
+ public void testPendingGapReplayDoesNotRescanPrefixAcrossBoundedRounds()
throws Exception {
+ assertPendingGapReplay(false, false);
+ }
+
+ @Test
+ public void testPendingGapReplayKeepsMemoryBlockedRequestAcrossRounds()
throws Exception {
+ assertPendingGapReplay(true, false);
+ }
+
+ @Test
+ public void testPendingGapReplayRetriesRequestAfterConversionFailure()
throws Exception {
+ assertPendingGapReplay(false, true);
+ }
+
+ private void assertPendingGapReplay(
+ final boolean memoryBackpressure, final boolean materializationFailure)
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 long originalBatchMaxSize =
config.getSubscriptionConsensusBatchMaxSizeInBytes();
+ final int originalBatchMaxDelay =
config.getSubscriptionConsensusBatchMaxDelayInMs();
+ final File systemDir =
temporaryFolder.newFolder("system-pending-gap-replay");
+ final File walDirectory =
temporaryFolder.newFolder("wal-pending-gap-replay");
+ ConsensusPrefetchingQueue queue = null;
+ try {
+ final long oneTabletBytes = createTablet().ramBytesUsed();
+ config.setSubscriptionConsensusBatchMaxWalEntries(2);
+ config.setSubscriptionConsensusBatchMaxTabletCount(1);
+ config.setSubscriptionConsensusBatchMaxSizeInBytes(oneTabletBytes * 16L);
+ config.setSubscriptionConsensusBatchMaxDelayInMs(60_000);
+
+ final int replayedRequests = materializationFailure ? 7 : 6;
+ final int rounds = memoryBackpressure ? replayedRequests :
(replayedRequests + 1) / 2;
+ final int requestCount = replayedRequests + 2 * rounds;
+ writeSealedWal(walDirectory, requestCount);
+
+ 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 ConsensusLogToTabletConverter converter =
mock(ConsensusLogToTabletConverter.class);
+ final AtomicBoolean failureInjected = new AtomicBoolean();
+ when(converter.convert(any()))
+ .thenAnswer(
+ invocation -> {
+ final InsertNode insertNode = (InsertNode)
invocation.getArgument(0);
+ if (materializationFailure
+ && insertNode.getSearchIndex() == 2L
+ && failureInjected.compareAndSet(false, true)) {
+ throw new IllegalStateException("transient conversion
failure");
+ }
+ return Collections.singletonList(createTablet());
+ });
+ when(converter.getDatabaseName()).thenReturn("db");
+
+ final AtomicLong closedIteratorBodyReads = new AtomicLong();
+ final DataRegionId regionId = new DataRegionId(1);
+ queue =
+ new ConsensusPrefetchingQueue(
+ "consumerGroup",
+ "topic",
+ TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE,
+ regionId,
+ serverImpl,
+ new SubscriptionWalRetentionPolicy(
+ "topic",
+ SubscriptionWalRetentionPolicy.UNBOUNDED,
+ SubscriptionWalRetentionPolicy.UNBOUNDED),
+ converter,
+ newCommitManager(systemDir),
+ new RegionProgress(Collections.emptyMap()),
+ 1L,
+ 1L,
+ true) {
+ @Override
+ protected ProgressWALIterator createSubscriptionWALIterator(
+ final long startSearchIndex) {
+ return new ProgressWALIterator(walNode, startSearchIndex) {
+ @Override
+ public void close() throws IOException {
+ closedIteratorBodyReads.addAndGet(walBodyReadCount(this));
+ super.close();
+ }
+ };
+ }
+ };
+ final TrackingSubscriptionMemoryManager memoryManager =
+ new TrackingSubscriptionMemoryManager(
+ memoryBackpressure
+ ? oneTabletBytes + Math.max(1L, oneTabletBytes / 2L)
+ : oneTabletBytes * 16L);
+ queue.setSubscriptionMemoryManager(memoryManager);
+
+ assertNull(queue.poll("consumer"));
+ // Admission failures leave a hole before newer pending requests. Keep a
suffix queued so
+ // every bounded round enters fillGapFromWAL, including after ACK
releases Tablet memory.
+ for (long index = replayedRequests + 1L; index <= requestCount; index++)
{
+ assertTrue(pendingEntries(queue).offer(createRequest(index)));
+ }
+
+ long deliveredLocalSeq = 0L;
+ for (int round = 0; round < rounds; round++) {
+ queue.drivePrefetchOnce();
+ final long expectedLocalSeq =
+ materializationFailure
+ ? round * 2L + 1L
+ : memoryBackpressure ? round + 1L : (round + 1L) * 2L;
+ assertEquals(expectedLocalSeq + 1L, queue.getCurrentReadSearchIndex());
+ assertMemoryBounded(queue, memoryManager);
+ if (memoryBackpressure) {
+ assertTrue(subscriptionWalIterator(queue).hasBufferedNext());
+ }
+ while (deliveredLocalSeq < expectedLocalSeq) {
+ final SubscriptionEvent event = queue.poll("consumer");
+ assertNotNull(event);
+ deliveredLocalSeq++;
+ assertEquals(
+ deliveredLocalSeq,
event.getCommitContext().getWriterProgress().getLocalSeq());
+ assertEquals(
+ 1000L + deliveredLocalSeq,
+ event.getCommitContext().getWriterProgress().getPhysicalTime());
+ assertTrue(queue.ack("consumer", event.getCommitContext()));
+ }
+ assertEquals(0L, memoryManager.getUsedMemorySizeInBytes());
+ }
+
+ assertEquals(replayedRequests, deliveredLocalSeq);
+ assertEquals(replayedRequests, queue.getWalPathAcceptedEntries());
+ assertEquals(0L, queue.getPendingPathAcceptedEntries());
+ assertEquals(0L, queue.getWalGapSkippedEntries());
+ assertEquals(0L, queue.getSubscriptionUncommittedEventCount());
+ assertEquals(0L, queue.getRetainedTabletBytes());
+ // Grouping reads one request ahead; a blocked request can require one
further lookahead.
+ // Count actual body reads, including discarded iterators, rather than
timing disk caches.
+ // An exceptional conversion failure needs one reset, rereading its
three consumed bodies.
+ assertEquals(
+ replayedRequests + (memoryBackpressure ? 2L : 1L) +
(materializationFailure ? 3L : 0L),
+ closedIteratorBodyReads.get() +
walBodyReadCount(subscriptionWalIterator(queue)));
+ } finally {
+ if (queue != null) {
+ queue.close();
+ }
+
config.setSubscriptionConsensusBatchMaxWalEntries(originalBatchMaxWalEntries);
+
config.setSubscriptionConsensusBatchMaxTabletCount(originalBatchMaxTabletCount);
+ config.setSubscriptionConsensusBatchMaxSizeInBytes(originalBatchMaxSize);
+ config.setSubscriptionConsensusBatchMaxDelayInMs(originalBatchMaxDelay);
+
IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
+ }
+ }
+
+ private static int walBodyReadCount(final ProgressWALIterator iterator) {
+ try {
+ // All requests stay in one historical file, so this offset is the
number of body reads.
+ final Field field =
+
ProgressWALIterator.class.getDeclaredField("consumedEntryCountInCurrentFile");
+ field.setAccessible(true);
+ return field.getInt(iterator);
+ } catch (final ReflectiveOperationException e) {
+ throw new AssertionError(e);
+ }
+ }
+
private static void writeSealedWal(final File walDirectory) throws Exception
{
+ writeSealedWal(walDirectory, REQUEST_COUNT);
+ }
+
+ private static void writeSealedWal(final File walDirectory, final int
requestCount)
+ throws Exception {
final File historicalWal =
new File(
walDirectory, WALFileUtils.getLogFileName(0L, 0L,
WALFileStatus.CONTAINS_SEARCH_INDEX));
final File liveWal =
new File(
walDirectory,
- WALFileUtils.getLogFileName(1L, REQUEST_COUNT,
WALFileStatus.CONTAINS_SEARCH_INDEX));
+ WALFileUtils.getLogFileName(1L, requestCount,
WALFileStatus.CONTAINS_SEARCH_INDEX));
try (WALWriter writer = new WALWriter(historicalWal, WALFileVersion.V3)) {
- for (long searchIndex = 1L; searchIndex <= REQUEST_COUNT; searchIndex++)
{
+ for (long searchIndex = 1L; searchIndex <= requestCount; searchIndex++) {
writeWalEntry(writer, searchIndex);
}
}
