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 1b9e0713fb1 Fix subscription WAL replay stalls (#18969)
1b9e0713fb1 is described below
commit 1b9e0713fb162a3da4018a1032cd1f3d77892793
Author: Caideyipi <[email protected]>
AuthorDate: Wed Oct 21 19:54:45 2026 +0800
Fix subscription WAL replay stalls (#18969)
Skip non-search WAL separators while fast-forwarding committed progress,
and avoid rescanning covered WAL prefixes on later commits. Move ACK retention
metadata scans outside the in-flight event map lock so event recycling can
proceed during slow scans.
Add regression coverage for separator handling, retention-bound reuse, and
ACK recycling during a blocked retention scan.
---
.../consensus/ConsensusPrefetchingQueue.java | 51 +++++-
.../broker/consensus/ProgressWALIterator.java | 5 +
.../consensus/ConsensusPrefetchingQueueTest.java | 178 ++++++++++++++++++++-
.../broker/consensus/ProgressWALIteratorTest.java | 48 ++++++
4 files changed, 271 insertions(+), 11 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 1268689749f..24366687030 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
@@ -3987,8 +3987,7 @@ public class ConsensusPrefetchingQueue {
return null;
}
- final boolean committed =
- commitAndRefreshWalRetention(commitWriterId,
commitWriterProgress);
+ final boolean committed = commitWriterProgress(commitWriterId,
commitWriterProgress);
if (!committed) {
if (!silent) {
LOGGER.warn(
@@ -4006,6 +4005,12 @@ public class ConsensusPrefetchingQueue {
cleanUpEvent(ev, false);
return null;
});
+ if (acked.get()) {
+ // Retention refresh can read metadata from every newly relevant WAL
file. Keep that I/O out
+ // of ConcurrentHashMap.compute so recycling and requeueing other
in-flight events can
+ // proceed while an ACK advances committed progress.
+ refreshCommittedWalRetentionBoundAndNotify();
+ }
return !found.get()
? InFlightAckResult.MISSING
: acked.get() ? InFlightAckResult.ACKED : InFlightAckResult.REJECTED;
@@ -4526,12 +4531,20 @@ public class ConsensusPrefetchingQueue {
private boolean commitAndRefreshWalRetention(
final WriterId writerId, final WriterProgress writerProgress) {
+ final boolean committed = commitWriterProgress(writerId, writerProgress);
+ if (committed) {
+ refreshCommittedWalRetentionBoundAndNotify();
+ }
+ return committed;
+ }
+
+ private boolean commitWriterProgress(
+ final WriterId writerId, final WriterProgress writerProgress) {
final boolean committed =
commitManager.commit(
consumerGroupId, topicName, consensusGroupId, writerId,
writerProgress);
if (committed) {
lastProgressTimeMs = System.currentTimeMillis();
- refreshCommittedWalRetentionBoundAndNotify();
}
return committed;
}
@@ -4584,7 +4597,7 @@ public class ConsensusPrefetchingQueue {
return true;
}
- private boolean canReuseCommittedWalRetentionBound(final RegionProgress
committedRegionProgress) {
+ private boolean canSkipCoveredWalPrefix(final RegionProgress
committedRegionProgress) {
if (!isProgressAtLeast(committedRegionProgress,
lastCommittedProgressForRetention)
|| Objects.isNull(lastRetainedWalFileForRetention)
|| !lastRetainedWalFileForRetention.exists()
@@ -4593,6 +4606,22 @@ public class ConsensusPrefetchingQueue {
return false;
}
+ return true;
+ }
+
+ private boolean canReuseCommittedWalRetentionBound(final RegionProgress
committedRegionProgress) {
+ if (!canSkipCoveredWalPrefix(committedRegionProgress)) {
+ return false;
+ }
+
+ // The live WAL file must be retained regardless of committed progress.
Its metadata is not
+ // sealed yet, so there is no cached requirement to check until the WAL
rolls.
+ if (consensusReqReader instanceof WALNode
+ && committedRetainedMinVersionId
+ == ((WALNode) consensusReqReader).getCurrentWALFileVersion()) {
+ return true;
+ }
+
final WalFileCommitRequirement requirement =
walFileCommitRequirements.get(committedRetainedMinVersionId);
return Objects.nonNull(requirement) &&
!requirement.isCoveredBy(committedRegionProgress);
@@ -4640,8 +4669,8 @@ public class ConsensusPrefetchingQueue {
return false;
}
- // The previous boundary file remains the first uncommitted file while
its cached
- // requirement is still uncovered by monotonically advancing committed
progress.
+ // The previous boundary still applies while its sealed requirement is
uncovered, or while
+ // it is the current live WAL file. Both cases require monotonically
advancing progress.
if (canReuseCommittedWalRetentionBound(committedRegionProgress)) {
lastCommittedProgressForRetention = committedRegionProgress;
return false;
@@ -4673,8 +4702,18 @@ public class ConsensusPrefetchingQueue {
}
WALFileUtils.ascSortByVersionId(walFiles);
+ // Sealed WAL files before the previous boundary were already covered by
committed progress.
+ // They may remain on disk because of the replay cursor or topic
retention; rescanning their
+ // metadata on every later commit makes sparse topics spend most of their
time in retention.
+ final long firstVersionToCheck =
+ canSkipCoveredWalPrefix(committedRegionProgress)
+ ? committedRetainedMinVersionId
+ : Long.MIN_VALUE;
for (final File walFile : walFiles) {
final long versionId = WALFileUtils.parseVersionId(walFile.getName());
+ if (versionId < firstVersionToCheck) {
+ continue;
+ }
if (versionId >= currentWalVersion) {
return new CommittedWalRetentionBound(Math.max(0L, currentWalVersion),
walFile);
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java
index 07438858a35..c048677a0af 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java
@@ -428,6 +428,11 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
final long physicalTime = physicalTimes.get(entryIndex);
final int nodeId = nodeIds.get(entryIndex);
final long localSeq = localSeqs.get(entryIndex);
+ if (nodeId < 0 && physicalTime == 0L && localSeq < 0L) {
+ // Non-search WAL entries (such as separators) do not participate in
subscription
+ // progress and must not prevent a covered historical file from being
skipped.
+ continue;
+ }
if (nodeId < 0
|| physicalTime < 0
|| localSeq < 0
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 a985a0f617d..a8c880495c9 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
@@ -851,9 +851,8 @@ public class ConsensusPrefetchingQueueTest {
WALFileUtils.getLogFileName(2L, 3L,
WALFileStatus.CONTAINS_SEARCH_INDEX));
writeWalMetadata(firstWal, 1L, 2L, 1002L, 7);
writeWalMetadata(nextWal, 2L, 4L, 1004L, 7);
- try (WALWriter ignored = new WALWriter(liveWal, WALFileVersion.V3)) {
- // Keep a live successor so both data WAL files are eligible for
boundary checks.
- }
+ // The v2 metadata becomes eligible for retention checks only after the
mocked WAL rolls.
+ writeWalMetadata(liveWal, 3L, 7L, 1007L, 7);
assertEquals(3, WALFileUtils.listAllWALFiles(walDirectory).length);
assertFalse(ProgressWALIterator.isHeaderOnlyWalFile(firstWal));
@@ -866,7 +865,8 @@ public class ConsensusPrefetchingQueueTest {
final WALNode walNode = mock(WALNode.class);
when(walNode.getLogDirectory()).thenReturn(walDirectory);
when(walNode.getSortedWalFilesSnapshot()).thenReturn(new File[]
{firstWal, nextWal, liveWal});
- when(walNode.getCurrentWALFileVersion()).thenReturn(2L);
+ final AtomicLong currentWalVersion = new AtomicLong(2L);
+ when(walNode.getCurrentWALFileVersion()).thenAnswer(ignored ->
currentWalVersion.get());
assertEquals(2L, walNode.getCurrentWALFileVersion());
final IoTConsensusServerImpl serverImpl =
mock(IoTConsensusServerImpl.class);
@@ -936,9 +936,43 @@ public class ConsensusPrefetchingQueueTest {
assertEquals(0L, retentionSupplier.get().getAsLong());
committedProgress.set(
- new RegionProgress(Collections.singletonMap(writerId, new
WriterProgress(1001L, 1L))));
+ new RegionProgress(Collections.singletonMap(writerId, new
WriterProgress(1004L, 4L))));
+ assertTrue((Boolean) refreshRetentionBound.invoke(queue));
+ // Advancing past v1 must not reopen v0, which is still retained for
replay.
+ assertEquals(2, progressWalReaderOpenCount.get());
+ assertEquals(2L, committedBound.getLong(queue));
+
+ committedProgress.set(
+ new RegionProgress(Collections.singletonMap(writerId, new
WriterProgress(1005L, 5L))));
+ assertFalse((Boolean) refreshRetentionBound.invoke(queue));
+ // The current WAL file stays retained without repeatedly scanning the
sealed prefix.
+ assertEquals(2, progressWalReaderOpenCount.get());
+ assertEquals(0L, retentionSupplier.get().getAsLong());
+
+ final File nextLiveWal =
+ new File(
+ walDirectory,
+ WALFileUtils.getLogFileName(3L, 4L,
WALFileStatus.CONTAINS_SEARCH_INDEX));
+ try (WALWriter ignored = new WALWriter(nextLiveWal, WALFileVersion.V3)) {
+ // Rolling v2 makes its previously live metadata a sealed retention
requirement.
+ }
+ currentWalVersion.set(3L);
+ committedProgress.set(
+ new RegionProgress(Collections.singletonMap(writerId, new
WriterProgress(1006L, 6L))));
+ assertFalse((Boolean) refreshRetentionBound.invoke(queue));
+ assertEquals(3, progressWalReaderOpenCount.get());
+ assertEquals(2L, committedBound.getLong(queue));
+
+ committedProgress.set(
+ new RegionProgress(Collections.singletonMap(writerId, new
WriterProgress(1007L, 7L))));
assertTrue((Boolean) refreshRetentionBound.invoke(queue));
assertEquals(3, progressWalReaderOpenCount.get());
+ assertEquals(3L, committedBound.getLong(queue));
+
+ committedProgress.set(
+ new RegionProgress(Collections.singletonMap(writerId, new
WriterProgress(1001L, 1L))));
+ assertTrue((Boolean) refreshRetentionBound.invoke(queue));
+ assertEquals(4, progressWalReaderOpenCount.get());
assertEquals(0L, committedBound.getLong(queue));
assertEquals(0L, retentionSupplier.get().getAsLong());
} finally {
@@ -2517,6 +2551,140 @@ public class ConsensusPrefetchingQueueTest {
}
}
+ @Test
+ public void testAckRetentionRefreshDoesNotBlockEventRecycling() throws
Exception {
+ for (final boolean silent : new boolean[] {false, true}) {
+ final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
+ final File systemDir = temporaryFolder.newFolder("ack-retention-system-"
+ silent);
+ final File walDirectory = temporaryFolder.newFolder("ack-retention-wal-"
+ silent);
+ final int originalRecycleInterval =
+ CommonDescriptor.getInstance()
+ .getConfig()
+ .getSubscriptionRecycleUncommittedEventIntervalMs();
+ final CountDownLatch retentionReadStarted = new CountDownLatch(1);
+ final CountDownLatch releaseRetentionRead = new CountDownLatch(1);
+ final AtomicBoolean blockRetentionRead = new AtomicBoolean(false);
+ final ExecutorService executor = Executors.newFixedThreadPool(2);
+ ConsensusPrefetchingQueue queue = null;
+ try {
+
IoTDBDescriptor.getInstance().getConfig().setSystemDir(systemDir.getAbsolutePath());
+ final File firstWal =
+ new File(
+ walDirectory,
+ WALFileUtils.getLogFileName(0L, 0L,
WALFileStatus.CONTAINS_SEARCH_INDEX));
+ final File nextWal =
+ new File(
+ walDirectory,
+ WALFileUtils.getLogFileName(1L, 1L,
WALFileStatus.CONTAINS_SEARCH_INDEX));
+ final File liveWal =
+ new File(
+ walDirectory,
+ WALFileUtils.getLogFileName(2L, 2L,
WALFileStatus.CONTAINS_SEARCH_INDEX));
+ writeWalMetadata(firstWal, 1L, 1L, 1001L, 7);
+ writeWalMetadata(nextWal, 2L, 2L, 1002L, 7);
+ try (WALWriter ignored = new WALWriter(liveWal, WALFileVersion.V3)) {
+ // Keep both data files sealed for retention checks.
+ }
+ final WALNode walNode = mock(WALNode.class);
+ when(walNode.getLogDirectory()).thenReturn(walDirectory);
+ when(walNode.getSortedWalFilesSnapshot())
+ .thenReturn(new File[] {firstWal, nextWal, liveWal});
+ when(walNode.getCurrentWALFileVersion()).thenReturn(2L);
+ when(walNode.getCurrentSearchIndex()).thenReturn(2L);
+ final IoTConsensusServerImpl serverImpl =
mock(IoTConsensusServerImpl.class);
+ when(serverImpl.getConsensusReqReader()).thenReturn(walNode);
+ when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new
WriterSafeFrontierTracker());
+ final ConsensusLogToTabletConverter converter =
mock(ConsensusLogToTabletConverter.class);
+ when(converter.convert(any()))
+ .thenReturn(Collections.singletonList(createTablet()),
Collections.emptyList());
+ when(converter.getDatabaseName()).thenReturn("db");
+ queue =
+ new ConsensusPrefetchingQueue(
+ "consumerGroup",
+ "topic",
+ TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE,
+ new DataRegionId(1),
+ serverImpl,
+ new SubscriptionWalRetentionPolicy(
+ "topic",
+ SubscriptionWalRetentionPolicy.UNBOUNDED,
+ SubscriptionWalRetentionPolicy.UNBOUNDED),
+ converter,
+ newCommitManager(systemDir),
+ new RegionProgress(Collections.emptyMap()),
+ 1L,
+ 1L,
+ true) {
+ @Override
+ ProgressWALReader openProgressWALReader(final File walFile)
throws IOException {
+ if (blockRetentionRead.get()) {
+ retentionReadStarted.countDown();
+ try {
+
assertTrue(releaseRetentionRead.await(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS));
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException(e);
+ }
+ }
+ return super.openProgressWALReader(walFile);
+ }
+ };
+ queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(1024L
* 1024L));
+ assertTrue(pendingEntries(queue).offer(createRequest(1L)));
+ assertTrue(pendingEntries(queue).offer(createRequest(2L)));
+ queue.drivePrefetchOnce();
+ final Method commitAndRetention =
+ ConsensusPrefetchingQueue.class.getDeclaredMethod(
+ "commitAndRefreshWalRetention", WriterId.class,
WriterProgress.class);
+ commitAndRetention.setAccessible(true);
+ commitAndRetention.invoke(
+ queue, new WriterId("DataRegion[1]", 7), new WriterProgress(1000L,
0L));
+ final SubscriptionEvent event = queue.poll("consumer");
+ assertNotNull(event);
+ assertTrue(queue.getRetainedTabletBytes() > 0L);
+ final Field timestampField =
+ SubscriptionEvent.class.getDeclaredField("lastPolledTimestamp");
+ timestampField.setAccessible(true);
+ ((AtomicLong)
timestampField.get(event)).set(System.currentTimeMillis() - 1000L);
+ CommonDescriptor.getInstance()
+ .getConfig()
+ .setSubscriptionRecycleUncommittedEventIntervalMs(1);
+
+ final ConsensusPrefetchingQueue activeQueue = queue;
+ blockRetentionRead.set(true);
+ final Future<Boolean> ack =
+ executor.submit(
+ () ->
+ silent
+ ? activeQueue.ackSilent("consumer",
event.getCommitContext())
+ : activeQueue.ack("consumer",
event.getCommitContext()));
+ assertTrue(retentionReadStarted.await(TEST_TIMEOUT_SECONDS,
TimeUnit.SECONDS));
+ final Method recycle =
+
ConsensusPrefetchingQueue.class.getDeclaredMethod("recycleInFlightEvents");
+ recycle.setAccessible(true);
+ // A slow retention read must not keep the event-map bucket locked by
ACK and stall
+ // a shared prefetch worker trying to recycle that event.
+ executor.submit(() -> recycle.invoke(activeQueue)).get(5,
TimeUnit.SECONDS);
+ assertEquals(0L, activeQueue.getSubscriptionUncommittedEventCount());
+ assertEquals(0L, activeQueue.getRetainedTabletBytes());
+ assertFalse(ack.isDone());
+ releaseRetentionRead.countDown();
+ assertTrue(ack.get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS));
+ } finally {
+ releaseRetentionRead.countDown();
+ executor.shutdownNow();
+ assertTrue(executor.awaitTermination(TEST_TIMEOUT_SECONDS,
TimeUnit.SECONDS));
+ if (queue != null) {
+ queue.close();
+ }
+ CommonDescriptor.getInstance()
+ .getConfig()
+
.setSubscriptionRecycleUncommittedEventIntervalMs(originalRecycleInterval);
+
IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
+ }
+ }
+ }
+
@Test
public void testLateAckDoesNotWaitForPrefetchReadLock() throws Exception {
final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java
index 5202b0ef1b7..3c2080fe673 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java
@@ -1193,6 +1193,11 @@ public class ProgressWALIteratorTest {
try {
try (WALWriter writer = new WALWriter(localWal, WALFileVersion.V3)) {
writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 100L, 7,
1L));
+ final ByteBuffer separator =
ByteBuffer.allocate(WALInfoEntry.FIXED_SERIALIZED_SIZE);
+
separator.put(WALEntryType.CONTINUOUS_SAME_SEARCH_INDEX_SEPARATOR_NODE.getCode());
+ separator.putLong(-1L);
+ writer.write(
+ separator, singleEntryMeta(WALInfoEntry.FIXED_SERIALIZED_SIZE,
-1L, -1L, 0L, -1, -1L));
writer.write(searchableEntry(2L), singleEntryMeta(19, 2L, 1L, 200L, 7,
2L));
}
try (WALWriter writer = new WALWriter(followerWal, WALFileVersion.V3)) {
@@ -1253,6 +1258,49 @@ public class ProgressWALIteratorTest {
}
}
+ @Test
+ public void testAdvanceToKeepsIncompleteWriterMetadata() throws Exception {
+ final Path dir =
Files.createTempDirectory("progress-wal-iterator-incomplete-progress");
+ final File historyWal =
+ dir.resolve(WALFileUtils.getLogFileName(0, 0,
WALFileStatus.CONTAINS_SEARCH_INDEX))
+ .toFile();
+ final File targetWal =
+ dir.resolve(WALFileUtils.getLogFileName(1, 2,
WALFileStatus.CONTAINS_SEARCH_INDEX))
+ .toFile();
+ try {
+ try (WALWriter writer = new WALWriter(targetWal, WALFileVersion.V3)) {
+ writer.write(searchableEntry(6L), singleEntryMeta(19, 6L, 1L, 600L, 7,
6L));
+ writer.write(searchableEntry(7L), singleEntryMeta(19, 7L, 1L, 700L, 7,
7L));
+ }
+ // Only the complete non-search sentinel may be ignored. A missing
writer, time or
+ // sequence on a searchable entry must still force sequential replay.
+ final long[][] incompleteProgress = {
+ {100L, -1L, -1L}, {0L, -1L, 1L}, {0L, 7L, -1L}, {-1L, 7L, 1L}
+ };
+ for (final long[] progress : incompleteProgress) {
+ try (WALWriter writer = new WALWriter(historyWal, WALFileVersion.V3)) {
+ writer.write(
+ searchableEntry(-1L),
+ singleEntryMeta(19, -1L, 1L, progress[0], (int) progress[1],
progress[2]));
+ writer.write(searchableEntry(2L), singleEntryMeta(19, 2L, 1L, 200L,
7, 2L));
+ }
+ try (ProgressWALIterator iterator = new
ProgressWALIterator(dir.toFile(), 6L)) {
+ iterator.advanceTo(6L, (physicalTime, nodeId, localSeq) -> true);
+ final ProgressWALIterator.ScanBudget budget =
+ new ProgressWALIterator.ScanBudget(1, Long.MAX_VALUE);
+ assertFalse(iterator.hasNext(budget));
+ assertTrue(budget.hasYielded());
+ assertEquals(-1L, iterator.next().getSearchIndex());
+ }
+ Files.delete(historyWal.toPath());
+ }
+ } finally {
+ Files.deleteIfExists(historyWal.toPath());
+ Files.deleteIfExists(targetWal.toPath());
+ Files.deleteIfExists(dir);
+ }
+ }
+
@Test
public void testRefreshAfterExhaustionDiscoversNextWalFile() throws
Exception {
final Path dir =
Files.createTempDirectory("progress-wal-iterator-refresh-after-exhaustion");
