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 a4e56ce5c2f [Subscription] Fix WAL replay after leader failover
(#18474)
a4e56ce5c2f is described below
commit a4e56ce5c2f27c5dbddee661aa6d051c5c579271
Author: Caideyipi <[email protected]>
AuthorDate: Mon Aug 24 11:45:07 2026 +0800
[Subscription] Fix WAL replay after leader failover (#18474)
---
.../consensus/ConsensusPrefetchingQueue.java | 20 ++---
.../consensus/ConsensusPrefetchingQueueTest.java | 86 ++++++++++++++++++++++
.../broker/consensus/ProgressWALIteratorTest.java | 11 ++-
3 files changed, 100 insertions(+), 17 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 4dd71c3b8f2..6bdbdbc642e 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
@@ -1003,8 +1003,7 @@ public class ConsensusPrefetchingQueue {
final Map<WriterId, WriterProgress> effectiveRecoveryWriterProgress =
new LinkedHashMap<>(requestedWriterProgress);
final Set<WriterId> exactVisibleWriterIds = new LinkedHashSet<>();
- Long firstUncoveredReplayableSearchIndex = null;
- boolean sawBlockingNonReplayableUncovered = false;
+ Long firstUncoveredLocalSearchIndex = null;
while (requests.hasNext()) {
final IndexedConsensusRequest request = requests.next();
@@ -1026,11 +1025,9 @@ public class ConsensusPrefetchingQueue {
}
if (request.getSearchIndex() >= 0) {
- if (Objects.isNull(firstUncoveredReplayableSearchIndex)) {
- firstUncoveredReplayableSearchIndex = request.getSearchIndex();
+ if (Objects.isNull(firstUncoveredLocalSearchIndex)) {
+ firstUncoveredLocalSearchIndex = request.getSearchIndex();
}
- } else if (Objects.isNull(firstUncoveredReplayableSearchIndex)) {
- sawBlockingNonReplayableUncovered = true;
}
}
@@ -1045,14 +1042,11 @@ public class ConsensusPrefetchingQueue {
final RegionProgress effectiveRecoveryRegionProgress =
new RegionProgress(effectiveRecoveryWriterProgress);
- if (sawBlockingNonReplayableUncovered) {
- return ReplayLocateDecision.locateMiss(
- effectiveRecoveryRegionProgress,
- "uncovered non-replayable WAL records appear before the first local
replayable record");
- }
- if (Objects.nonNull(firstUncoveredReplayableSearchIndex)) {
+ // The iterator's lower bound filters only locally indexed requests.
Replicated requests stay
+ // visible and are deduplicated by writer progress, so they do not block
local cursor lookup.
+ if (Objects.nonNull(firstUncoveredLocalSearchIndex)) {
return ReplayLocateDecision.found(
- firstUncoveredReplayableSearchIndex,
+ firstUncoveredLocalSearchIndex,
effectiveRecoveryRegionProgress,
"resolved first uncovered replayable WAL record");
}
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 5b3fd00b928..da73a003808 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
@@ -106,6 +106,78 @@ public class ConsensusPrefetchingQueueTest {
.getModifiers()));
}
+ @Test
+ public void testReplayStartPreservesUncoveredFollowerEntries() throws
Exception {
+ final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
+ final File systemDir =
temporaryFolder.newFolder("replay-start-with-follower-entry");
+ ConsensusPrefetchingQueue queue = null;
+ try {
+ final DataRegionId regionId = new DataRegionId(1);
+ final FakeConsensusReqReader reader = new FakeConsensusReqReader();
+ reader.currentSearchIndex = 1L;
+ final IoTConsensusServerImpl serverImpl =
mock(IoTConsensusServerImpl.class);
+ when(serverImpl.getConsensusReqReader()).thenReturn(reader);
+ when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new
WriterSafeFrontierTracker());
+
+ queue =
+ new ConsensusPrefetchingQueue(
+ "consumerGroup",
+ "topic",
+ TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE,
+ regionId,
+ serverImpl,
+ new SubscriptionWalRetentionPolicy(
+ "topic",
+ SubscriptionWalRetentionPolicy.UNBOUNDED,
+ SubscriptionWalRetentionPolicy.UNBOUNDED),
+ mock(ConsensusLogToTabletConverter.class),
+ newCommitManager(systemDir),
+ new RegionProgress(Collections.emptyMap()),
+ 1L,
+ 1L,
+ true);
+
+ final WriterId formerLeader = new WriterId(regionId.toString(), 8);
+ final WriterProgress committedProgress = new WriterProgress(100L, 10L);
+ final RegionProgress regionProgress =
+ new RegionProgress(Collections.singletonMap(formerLeader,
committedProgress));
+ final List<IndexedConsensusRequest> requests =
+ Arrays.asList(
+ createRequest(-1L, 10L, 100L, 8),
+ createRequest(-1L, 11L, 101L, 8),
+ createRequest(1L, 1L, 200L, 7));
+
+ final ConsensusPrefetchingQueue.ReplayLocateDecision decision =
+ queue.scanReplayStartForRequests(requests.iterator(),
regionProgress, true);
+
+ assertEquals(ConsensusPrefetchingQueue.ReplayLocateStatus.FOUND,
decision.getStatus());
+ assertEquals(1L, decision.getStartSearchIndex());
+ assertEquals(
+ committedProgress,
+
decision.getRecoveryRegionProgress().getWriterPositions().get(formerLeader));
+
+ // With no uncovered local request, keep the local cursor at the tail
without advancing the
+ // recovery progress past the still-uncovered follower request.
+ reader.currentSearchIndex = 5L;
+ final ConsensusPrefetchingQueue.ReplayLocateDecision tailDecision =
+ queue.scanReplayStartForRequests(
+ Collections.singletonList(createRequest(-1L, 11L, 101L,
8)).iterator(),
+ regionProgress,
+ true);
+
+ assertEquals(ConsensusPrefetchingQueue.ReplayLocateStatus.AT_END,
tailDecision.getStatus());
+ assertEquals(5L, tailDecision.getStartSearchIndex());
+ assertEquals(
+ committedProgress,
+
tailDecision.getRecoveryRegionProgress().getWriterPositions().get(formerLeader));
+ } finally {
+ if (queue != null) {
+ queue.close();
+ }
+
IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
+ }
+ }
+
@Test
@SuppressWarnings("unchecked")
public void testAdmissionClearCannotLeaveEntryEnqueuedAfterFence() throws
Exception {
@@ -1920,6 +1992,20 @@ public class ConsensusPrefetchingQueueTest {
.setNodeId(7);
}
+ private static IndexedConsensusRequest createRequest(
+ final long searchIndex,
+ final long localSeq,
+ final long physicalTime,
+ final int writerNodeId) {
+ return new IndexedConsensusRequest(
+ searchIndex,
+ localSeq,
+ Collections.singletonList(
+
StatementTestUtils.genInsertRowNode(Math.toIntExact(localSeq))))
+ .setPhysicalTime(physicalTime)
+ .setNodeId(writerNodeId);
+ }
+
private static IndexedConsensusRequest createSizedRequest(
final long searchIndex, final long rawMemorySize, final int
serializedMemorySize) {
final IndexedConsensusRequest request =
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 92928663126..fed8febea52 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
@@ -242,7 +242,8 @@ public class ProgressWALIteratorTest {
}
@Test
- public void
testFollowerEntryDoesNotSynthesizeSearchIndexFromProgressLocalSeq() throws
Exception {
+ public void
testLocalLowerBoundKeepsFollowerEntryWithoutSynthesizingSearchIndex()
+ throws Exception {
final Path dir =
Files.createTempDirectory("progress-wal-iterator-follower");
final File firstWal =
dir.resolve(WALFileUtils.getLogFileName(0, 0,
WALFileStatus.CONTAINS_SEARCH_INDEX))
@@ -255,17 +256,19 @@ public class ProgressWALIteratorTest {
try (WALWriter writer = new WALWriter(firstWal, WALFileVersion.V3)) {
writer.write(searchableEntry(-1L), singleEntryMeta(19, -1L, 1L, 900L,
5, 1009L));
}
- try (WALWriter ignored = new WALWriter(lastWal, WALFileVersion.V3)) {
- // Create a readable successor for the first WAL file.
+ try (WALWriter writer = new WALWriter(lastWal, WALFileVersion.V3)) {
+ writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 1000L,
6, 1L));
}
- try (ProgressWALIterator iterator = new
ProgressWALIterator(dir.toFile(), Long.MIN_VALUE)) {
+ try (ProgressWALIterator iterator = new
ProgressWALIterator(dir.toFile(), 1L)) {
assertTrue(iterator.hasNext());
final IndexedConsensusRequest request = iterator.next();
assertEquals(-1L, request.getSearchIndex());
assertEquals(1009L, request.getProgressLocalSeq());
assertEquals(900L, request.getPhysicalTime());
assertEquals(5, request.getNodeId());
+ assertTrue(iterator.hasNext());
+ assertEquals(1L, iterator.next().getSearchIndex());
assertFalse(iterator.hasNext());
}
} finally {
