This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch fix/subscription-wal-roll-retention-20261028 in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit efc531082ef8bd2398cb6f4738816542afe670d7 Author: Caideyipi <[email protected]> AuthorDate: Wed Oct 28 18:58:31 2026 +0800 Fix consensus subscription WAL retention and replay cursor --- .../consensus/ConsensusPrefetchingQueue.java | 9 +-- .../broker/consensus/ProgressWALIterator.java | 10 ++- .../consensus/ConsensusPrefetchingQueueTest.java | 25 ++++++-- .../broker/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 df756dc2d22..3668f7dbd09 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 @@ public class ConsensusPrefetchingQueue { 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 cbb0d7099ec..f6daecdb78a 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 class ProgressWALIterator implements Closeable, Iterator<IndexedConsensus 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 @@ public class ProgressWALIterator implements Closeable, Iterator<IndexedConsensus } 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 8d7edd979c5..84ec91d34bc 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 @@ public class ConsensusPrefetchingQueueTest { 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 @@ public class ConsensusPrefetchingQueueTest { 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 3c2080fe673..e2d807614e8 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.i18n.DataNodePipeMessages; 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 class ProgressWALIteratorTest { 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 class ProgressWALIteratorTest { } } + @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");
