From efc531082ef8bd2398cb6f4738816542afe670d7 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Wed, 28 Oct 2026 18:58:31 +0800 Subject: [PATCH 1/2] Fix consensus subscription WAL retention and replay cursor --- .../consensus/ConsensusPrefetchingQueue.java | 9 +-- .../broker/consensus/ProgressWALIterator.java | 10 ++- .../ConsensusPrefetchingQueueTest.java | 25 +++++-- .../consensus/ProgressWALIteratorTest.java | 71 +++++++++++++++++++ 4 files changed, 101 insertions(+), 14 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 df756dc2d22cd..3668f7dbd0924 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 @@ -4791,14 +4791,9 @@ private boolean refreshCommittedWalRetentionBound() { commitManager.getCommittedRegionProgress(consumerGroupId, topicName, consensusGroupId); synchronized (committedRetentionLock) { - if (Objects.equals(lastCommittedProgressForRetention, committedRegionProgress) - && Objects.nonNull(lastRetainedWalFileForRetention) - && lastRetainedWalFileForRetention.exists()) { - return false; - } - // 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. + // it is the current live WAL file. Recheck even unchanged progress: a roll may have sealed + // the formerly live file after its last ACK, making that file reclaimable without new writes. if (canReuseCommittedWalRetentionBound(committedRegionProgress)) { lastCommittedProgressForRetention = committedRegionProgress; return false; 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 cbb0d7099ec30..f6daecdb78ae5 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 @@ -321,7 +321,10 @@ public void advanceTo( return; } - final int firstFileToSkip = Math.max(0, currentFileIndex); + // Without a reader (including a deferred one), currentFileIndex is the predecessor already + // passed by replay or a previous fast-forward. It may have been reclaimed in the meantime. + final int firstFileToSkip = + Math.max(0, currentReaderVersionId < 0 ? currentFileIndex + 1 : currentFileIndex); try { for (int fileIndex = firstFileToSkip; fileIndex < targetFileIndex; fileIndex++) { if (!isWalFileCovered(fileIndex, writerProgressCoverage)) { @@ -351,10 +354,11 @@ private void refreshForNewerLiveWalFile() { } private int locateTargetFile(final long targetSearchIndex) { - if (walFiles.length == 0 - || (liveWalNode != null && targetSearchIndex > liveWalNode.getCurrentSearchIndex())) { + if (walFiles.length == 0) { return -1; } + // Pending delivery may have advanced one past the live tail. Still realign to its file after + // verifying coverage of the sealed prefix, before committed history can be reclaimed. return WALFileUtils.binarySearchFileBySearchIndex(walFiles, targetSearchIndex); } 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 8d7edd979c50f..84ec91d34bce6 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 @@ -973,9 +973,8 @@ ProgressWALReader openProgressWALReader(final File walFile) throws IOException { 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. - } + writeWalMetadata(nextLiveWal, 4L, 8L, 1008L, 7); + // 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)))); @@ -990,9 +989,27 @@ ProgressWALReader openProgressWALReader(final File walFile) throws IOException { assertEquals(3L, committedBound.getLong(queue)); committedProgress.set( - new RegionProgress(Collections.singletonMap(writerId, new WriterProgress(1001L, 1L)))); + new RegionProgress(Collections.singletonMap(writerId, new WriterProgress(1008L, 8L)))); + assertFalse((Boolean) refreshRetentionBound.invoke(queue)); + assertEquals(3, progressWalReaderOpenCount.get()); + final File finalLiveWal = + new File( + walDirectory, + WALFileUtils.getLogFileName(4L, 5L, WALFileStatus.CONTAINS_SEARCH_INDEX)); + try (WALWriter ignored = new WALWriter(finalLiveWal, WALFileVersion.V3)) { + // No new ACK follows this roll: the same committed progress must release sealed v3. + } + currentWalVersion.set(4L); assertTrue((Boolean) refreshRetentionBound.invoke(queue)); assertEquals(4, progressWalReaderOpenCount.get()); + assertEquals(4L, committedBound.getLong(queue)); + assertFalse((Boolean) refreshRetentionBound.invoke(queue)); + assertEquals(4, progressWalReaderOpenCount.get()); + + committedProgress.set( + new RegionProgress(Collections.singletonMap(writerId, new WriterProgress(1001L, 1L)))); + assertTrue((Boolean) refreshRetentionBound.invoke(queue)); + assertEquals(5, progressWalReaderOpenCount.get()); assertEquals(0L, committedBound.getLong(queue)); assertEquals(0L, retentionSupplier.get().getAsLong()); } finally { 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 3c2080fe67392..e2d807614e802 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 @@ -26,6 +26,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.SearchNode; import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntryType; import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALInfoEntry; +import org.apache.iotdb.db.storageengine.dataregion.wal.io.ProgressWALReader; import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALFileVersion; import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALMetaData; import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALWriter; @@ -1250,6 +1251,33 @@ public void testAdvanceToSkipsOnlyFullyCoveredWalFiles() throws Exception { assertTrue(iterator.hasNext()); assertEquals(6L, iterator.next().getSearchIndex()); } + + final WALNode walNode = mock(WALNode.class); + when(walNode.getLogDirectory()).thenReturn(dir.toFile()); + when(walNode.getCurrentWALFileVersion()).thenReturn(2L); + when(walNode.getCurrentSearchIndex()).thenReturn(7L); + try (ProgressWALReader reader = new ProgressWALReader(targetWal)) { + when(walNode.getWALMetaDataSnapshot(2L)).thenReturn(reader.getMetaData()); + } + // A pending-path cursor can advance one past the live tail. Local search indexes still + // cannot justify skipping an uncovered follower with no local search index. + try (ProgressWALIterator iterator = new ProgressWALIterator(walNode, 1L)) { + iterator.advanceTo(8L, (physicalTime, nodeId, localSeq) -> nodeId == 7); + assertTrue(iterator.hasNext()); + assertEquals(8, iterator.next().getNodeId()); + } + try (ProgressWALIterator iterator = new ProgressWALIterator(walNode, 1L)) { + iterator.advanceTo(8L, (physicalTime, nodeId, localSeq) -> true); + // Committed/materialized history may now be reclaimed before the next WAL probe. The + // iterator must already be positioned beyond those files, without reporting data loss. + Files.delete(localWal.toPath()); + Files.delete(followerWal.toPath()); + for (int probe = 0; probe < 3; probe++) { + assertFalse(iterator.hasNext()); + } + assertEquals(0, iterator.getSkippedBrokenWalFileCount()); + assertFalse(iterator.isWaitingForReadableWal()); + } } finally { Files.deleteIfExists(localWal.toPath()); Files.deleteIfExists(followerWal.toPath()); @@ -1258,6 +1286,49 @@ public void testAdvanceToSkipsOnlyFullyCoveredWalFiles() throws Exception { } } + @Test + public void testRepeatedAdvanceDoesNotRecheckReclaimedPrefix() throws Exception { + final Path dir = Files.createTempDirectory("progress-wal-iterator-reclaimed-prefix"); + final File historyWal = + dir.resolve(WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + final File middleWal = + dir.resolve(WALFileUtils.getLogFileName(1, 2, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + final File targetWal = + dir.resolve(WALFileUtils.getLogFileName(2, 4, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + try { + try (WALWriter writer = new WALWriter(historyWal, WALFileVersion.V3)) { + writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 100L, 7, 1L)); + writer.write(searchableEntry(2L), singleEntryMeta(19, 2L, 1L, 200L, 7, 2L)); + } + try (WALWriter writer = new WALWriter(middleWal, WALFileVersion.V3)) { + writer.write(searchableEntry(3L), singleEntryMeta(19, 3L, 1L, 300L, 7, 3L)); + writer.write(searchableEntry(4L), singleEntryMeta(19, 4L, 1L, 400L, 7, 4L)); + } + try (WALWriter writer = new WALWriter(targetWal, WALFileVersion.V3)) { + writer.write(searchableEntry(5L), singleEntryMeta(19, 5L, 1L, 500L, 7, 5L)); + } + try (ProgressWALIterator iterator = new ProgressWALIterator(dir.toFile(), 1L)) { + iterator.advanceTo(3L, (physicalTime, nodeId, localSeq) -> true); + Files.delete(historyWal.toPath()); + // No new WAL roll refreshes the file list between these pending-path advances. + iterator.advanceTo(5L, (physicalTime, nodeId, localSeq) -> true); + Files.delete(middleWal.toPath()); + assertTrue(iterator.hasNext()); + assertEquals(5L, iterator.next().getSearchIndex()); + assertFalse(iterator.hasSkippedBrokenWalFiles()); + assertFalse(iterator.isWaitingForReadableWal()); + } + } finally { + Files.deleteIfExists(historyWal.toPath()); + Files.deleteIfExists(middleWal.toPath()); + Files.deleteIfExists(targetWal.toPath()); + Files.deleteIfExists(dir); + } + } + @Test public void testAdvanceToKeepsIncompleteWriterMetadata() throws Exception { final Path dir = Files.createTempDirectory("progress-wal-iterator-incomplete-progress"); From 135535a5f673846243485bf0e1bbe2bb99e5496b Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 30 Oct 2026 10:52:02 +0800 Subject: [PATCH 2/2] Address WAL retention and replay review feedback --- .../broker/consensus/ConsensusPrefetchingQueue.java | 10 ++++++++++ .../broker/consensus/ProgressWALIterator.java | 3 +++ 2 files changed, 13 insertions(+) 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 3668f7dbd0924..b34dc9d2ac93f 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 @@ -4791,6 +4791,16 @@ private boolean refreshCommittedWalRetentionBound() { commitManager.getCommittedRegionProgress(consumerGroupId, topicName, consensusGroupId); synchronized (committedRetentionLock) { + // Unchanged ACK progress can reuse a live boundary, but a roll must recheck the sealed file. + if (Objects.equals(lastCommittedProgressForRetention, committedRegionProgress) + && consensusReqReader instanceof WALNode + && committedRetainedMinVersionId + == ((WALNode) consensusReqReader).getCurrentWALFileVersion() + && Objects.nonNull(lastRetainedWalFileForRetention) + && lastRetainedWalFileForRetention.exists()) { + return false; + } + // The previous boundary still applies while its sealed requirement is uncovered, or while // it is the current live WAL file. Recheck even unchanged progress: a roll may have sealed // the formerly live file after its last ACK, making that file reclaimable without new writes. 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 f6daecdb78ae5..6d450d9f32dc4 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 @@ -359,6 +359,9 @@ private int locateTargetFile(final long targetSearchIndex) { } // Pending delivery may have advanced one past the live tail. Still realign to its file after // verifying coverage of the sealed prefix, before committed history can be reclaimed. + if (liveWalNode != null && targetSearchIndex > liveWalNode.getCurrentSearchIndex()) { + return walFiles.length - 1; + } return WALFileUtils.binarySearchFileBySearchIndex(walFiles, targetSearchIndex); }