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 999721fcafd Fix consensus subscription WAL retention and replay cursor
(#19042)
999721fcafd is described below
commit 999721fcafd5e95ff5f3de931ad34175ea315946
Author: Caideyipi <[email protected]>
AuthorDate: Mon Nov 2 11:34:47 2026 +0800
Fix consensus subscription WAL retention and replay cursor (#19042)
* Fix consensus subscription WAL retention and replay cursor
* Address WAL retention and replay review feedback
---
.../consensus/ConsensusPrefetchingQueue.java | 7 ++-
.../broker/consensus/ProgressWALIterator.java | 13 +++-
.../consensus/ConsensusPrefetchingQueueTest.java | 25 ++++++--
.../broker/consensus/ProgressWALIteratorTest.java | 71 ++++++++++++++++++++++
4 files changed, 108 insertions(+), 8 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..b34dc9d2ac9 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,19 @@ public class ConsensusPrefetchingQueue {
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. 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..6d450d9f32d 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,14 @@ 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.
+ if (liveWalNode != null && targetSearchIndex >
liveWalNode.getCurrentSearchIndex()) {
+ return walFiles.length - 1;
+ }
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");
