From e61dce80fd8c1e59ace3470acb32e51bf4b8fdfb Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 31 Jul 2026 17:52:56 +0800 Subject: [PATCH] Subscription: improve WAL replay gap observability --- .../iotdb/db/i18n/DataNodePipeMessages.java | 16 +++-- .../iotdb/db/i18n/DataNodePipeMessages.java | 13 +++- .../consensus/ConsensusPrefetchingQueue.java | 31 ++++++++-- .../broker/consensus/ProgressWALIterator.java | 62 ++++++++++++++++--- .../ConsensusPrefetchingQueueTest.java | 62 +++++++++++++++++++ .../consensus/ProgressWALIteratorTest.java | 38 ++++++++++++ 6 files changed, 204 insertions(+), 18 deletions(-) diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index 8e61bcda388c3..c2b3ea422353f 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java @@ -2099,11 +2099,11 @@ private DataNodePipeMessages() {} public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_PREFETCH_INITIALIZED_STARTSEARCHINDEX_69B53EE6 = "ConsensusPrefetchingQueue {}: prefetch initialized, startSearchIndex={}, progressSource={}, " + "recoveryWriterCount={}"; - public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_PERIODIC_STATS_LAG_PENDINGDELTA_D75375D0 = + public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_PERIODIC_STATS_LAG_PENDINGDELTA_WALGAPSKIPPEDENTRIES_9A4E6608 = "ConsensusPrefetchingQueue {}: periodic stats, lag={}, pendingDelta={}, walDelta={}, " - + "pendingTotal={}, walTotal={}, pendingQueueSize={}, prefetchingQueueSize={}, " - + "inFlightEventsSize={}, realtimeWriterCount={}, walHasNext={}, isActive={}, " - + "subtaskScheduled={}"; + + "pendingTotal={}, walTotal={}, walGapSkippedEntries={}, pendingQueueSize={}, " + + "prefetchingQueueSize={}, inFlightEventsSize={}, realtimeWriterCount={}, " + + "walHasNext={}, isActive={}, subtaskScheduled={}"; public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAITING_MS_FOR_WAL_GAP_TO_BECOME_7D91C6C5 = "ConsensusPrefetchingQueue {}: waiting {}ms for WAL gap [{}, {}) to become visible, " + "currentNextExpected={}, currentWalIndex={}, seekGeneration={}"; @@ -2183,6 +2183,14 @@ private DataNodePipeMessages() {} "ProgressWALIterator: error reading WAL"; public static final String PIPE_LOG_PROGRESSWALITERATOR_FAILED_TO_OPEN_WAL_FILE_SKIPPING_29CA1092 = "ProgressWALIterator: failed to open WAL file {}, skipping"; + public static final String PIPE_LOG_PROGRESSWALITERATOR_SKIPPED_UNREADABLE_RETAINED_WAL_FILES_FFC8455E = + "ProgressWALIterator: skipped {} unreadable retained WAL files in directory {}, " + + "firstFile={}, lastFile={}, firstError={}; historical subscription data in these " + + "files cannot be replayed"; + public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_SKIPPED_UNAVAILABLE_SEARCH_INDEXES_B8023B64 = + "ConsensusPrefetchingQueue {}: WAL replay skipped unavailable search indexes [{}, {}), " + + "skippedEntries={}, totalWalGapSkippedEntries={}; the missing WAL data may have been " + + "reclaimed before subscription consumption"; public static final String PIPE_LOG_PIPE_TERMINATE_EVENT_COMMITTED_FOR_HISTORICAL_TRANSFER_CREATIONTIME_9B807B28 = "Pipe {}@{}: terminate event committed for historical transfer. creationTime: {}, " + "shouldMark: {}. {}"; diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index 2389f97c8ec01..a8c60b2521637 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java @@ -1948,10 +1948,11 @@ private DataNodePipeMessages() {} public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_PREFETCH_INITIALIZED_STARTSEARCHINDEX_69B53EE6 = "ConsensusPrefetchingQueue {}：prefetch 已初始化，startSearchIndex={}，progressSource={}，" + "recoveryWriterCount={}"; - public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_PERIODIC_STATS_LAG_PENDINGDELTA_D75375D0 = + public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_PERIODIC_STATS_LAG_PENDINGDELTA_WALGAPSKIPPEDENTRIES_9A4E6608 = "ConsensusPrefetchingQueue {}：周期统计，lag={}，pendingDelta={}，walDelta={}，pendingTotal={}，" - + "walTotal={}，pendingQueueSize={}，prefetchingQueueSize={}，inFlightEventsSize={}，" - + "realtimeWriterCount={}，walHasNext={}，isActive={}，subtaskScheduled={}"; + + "walTotal={}，walGapSkippedEntries={}，pendingQueueSize={}，prefetchingQueueSize={}，" + + "inFlightEventsSize={}，realtimeWriterCount={}，walHasNext={}，isActive={}，" + + "subtaskScheduled={}"; public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAITING_MS_FOR_WAL_GAP_TO_BECOME_7D91C6C5 = "ConsensusPrefetchingQueue {}：等待 {}ms，使 WAL 缺口 [{}, {}) 可见，currentNextExpected={}，" + "currentWalIndex={}，seekGeneration={}"; @@ -2027,6 +2028,12 @@ private DataNodePipeMessages() {} "ProgressWALIterator：读取 WAL 出错"; public static final String PIPE_LOG_PROGRESSWALITERATOR_FAILED_TO_OPEN_WAL_FILE_SKIPPING_29CA1092 = "ProgressWALIterator：打开 WAL 文件 {} 失败，跳过该文件"; + public static final String PIPE_LOG_PROGRESSWALITERATOR_SKIPPED_UNREADABLE_RETAINED_WAL_FILES_FFC8455E = + "ProgressWALIterator：跳过了 {} 个无法读取的保留 WAL 文件，directory={}，firstFile={}，" + + "lastFile={}，firstError={}；这些文件中的历史订阅数据无法重放"; + public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_SKIPPED_UNAVAILABLE_SEARCH_INDEXES_B8023B64 = + "ConsensusPrefetchingQueue {}：WAL 重放跳过了不可用的 searchIndex 区间 [{}, {})，" + + "skippedEntries={}，totalWalGapSkippedEntries={}；缺失的 WAL 数据可能已在订阅消费前被回收"; public static final String PIPE_LOG_PIPE_TERMINATE_EVENT_COMMITTED_FOR_HISTORICAL_TRANSFER_CREATIONTIME_9B807B28 = "Pipe {}@{}：历史传输的终止事件已提交。creationTime：{}，shouldMark：{}。{}"; public static final String PIPE_LOG_PIPE_HISTORICAL_SOURCE_HAS_SUPPLIED_ALL_EVENTS_EMITTING_8B58DE19 = 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 dd02836cac949..aa7dad56cc4c0 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 @@ -1480,13 +1480,14 @@ private void logPeriodicStatsIfNecessary() { final long currentWalAcceptedEntries = walPathAcceptedEntries.get(); LOGGER.info( DataNodePipeMessages - .PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_PERIODIC_STATS_LAG_PENDINGDELTA_D75375D0, + .PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_PERIODIC_STATS_LAG_PENDINGDELTA_WALGAPSKIPPEDENTRIES_9A4E6608, this, getLag(), currentPendingAcceptedEntries - lastPendingAcceptedEntries, currentWalAcceptedEntries - lastWalAcceptedEntries, currentPendingAcceptedEntries, currentWalAcceptedEntries, + walGapSkippedEntries.get(), pendingEntries.size(), prefetchingQueue.size(), inFlightEvents.size(), @@ -1835,11 +1836,11 @@ private MaterializationResult pumpFromSubscriptionWAL( continue; } if (shouldSkipForRecoveryProgress(walEntry)) { - advanceLocalCursorIfPresent(walEntry); + advanceWalReplayCursorIfPresent(walEntry); continue; } if (shouldSkipForMaterializedProgress(walEntry)) { - advanceLocalCursorIfPresent(walEntry); + advanceWalReplayCursorIfPresent(walEntry); continue; } @@ -1853,7 +1854,7 @@ private MaterializationResult pumpFromSubscriptionWAL( return appendResult; } markMaterializedProgress(walEntry); - advanceLocalCursorIfPresent(walEntry); + advanceWalReplayCursorIfPresent(walEntry); } catch (final Exception e) { LOGGER.warn( DataNodePipeMessages @@ -1875,6 +1876,28 @@ private MaterializationResult pumpFromSubscriptionWAL( return MaterializationResult.SUCCESS; } + private void advanceWalReplayCursorIfPresent(final IndexedConsensusRequest request) { + if (!hasLocalSearchIndex(request)) { + return; + } + + final long actualSearchIndex = request.getSearchIndex(); + final long expectedSearchIndex = nextExpectedSearchIndex.get(); + if (actualSearchIndex > expectedSearchIndex) { + final long skippedEntries = actualSearchIndex - expectedSearchIndex; + final long totalSkippedEntries = walGapSkippedEntries.addAndGet(skippedEntries); + LOGGER.warn( + DataNodePipeMessages + .PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_SKIPPED_UNAVAILABLE_SEARCH_INDEXES_B8023B64, + this, + expectedSearchIndex, + actualSearchIndex, + skippedEntries, + totalSkippedEntries); + } + nextExpectedSearchIndex.set(actualSearchIndex + 1); + } + private void ensureSubscriptionWalReadable() { if (Objects.isNull(subscriptionWALIterator) || subscriptionWALIterator.hasNext() 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 56847c8903c33..5f2bcf41de7e2 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 @@ -74,6 +74,10 @@ public class ProgressWALIterator implements Closeable, Iterator skippedBrokenWalVersionIds = new HashSet<>(); + private int unreportedSkippedBrokenWalFileCount = 0; + private String firstUnreportedSkippedBrokenWalFile; + private String lastUnreportedSkippedBrokenWalFile; + private String firstUnreportedSkippedBrokenWalError; private IOException lastError; private boolean incompleteScan = false; private String incompleteScanDetail; @@ -172,7 +176,9 @@ public boolean hasNext() { } catch (IOException e) { lastError = e; LOGGER.warn(DataNodePipeMessages.PIPE_LOG_PROGRESSWALITERATOR_ERROR_READING_WAL_2DB46D41, e); - return false; + } + if (nextReady == null) { + logSkippedBrokenWalFilesIfNecessary(); } return nextReady != null; } @@ -198,6 +204,10 @@ public boolean hasSkippedBrokenWalFiles() { return !skippedBrokenWalVersionIds.isEmpty(); } + int getSkippedBrokenWalFileCount() { + return skippedBrokenWalVersionIds.size(); + } + public boolean hasIncompleteScan() { return incompleteScan || hasReadError() || hasSkippedBrokenWalFiles(); } @@ -406,16 +416,54 @@ private boolean openReaderAtIndex( e); return false; } - skippedBrokenWalVersionIds.add(versionId); - LOGGER.warn( - DataNodePipeMessages - .PIPE_LOG_PROGRESSWALITERATOR_FAILED_TO_OPEN_WAL_FILE_SKIPPING_29CA1092, - walFile.getName(), - e); + recordSkippedBrokenWalFile(versionId, walFile, e); return false; } } + private void recordSkippedBrokenWalFile( + final long versionId, final File walFile, final IOException error) { + if (!skippedBrokenWalVersionIds.add(versionId)) { + return; + } + + if (unreportedSkippedBrokenWalFileCount == 0) { + firstUnreportedSkippedBrokenWalFile = walFile.getName(); + firstUnreportedSkippedBrokenWalError = summarizeException(error); + } + lastUnreportedSkippedBrokenWalFile = walFile.getName(); + unreportedSkippedBrokenWalFileCount++; + LOGGER.debug( + DataNodePipeMessages.PIPE_LOG_PROGRESSWALITERATOR_FAILED_TO_OPEN_WAL_FILE_SKIPPING_29CA1092, + walFile.getName(), + error); + } + + private void logSkippedBrokenWalFilesIfNecessary() { + if (unreportedSkippedBrokenWalFileCount == 0) { + return; + } + + LOGGER.warn( + DataNodePipeMessages + .PIPE_LOG_PROGRESSWALITERATOR_SKIPPED_UNREADABLE_RETAINED_WAL_FILES_FFC8455E, + unreportedSkippedBrokenWalFileCount, + logDirectory, + firstUnreportedSkippedBrokenWalFile, + lastUnreportedSkippedBrokenWalFile, + firstUnreportedSkippedBrokenWalError); + unreportedSkippedBrokenWalFileCount = 0; + firstUnreportedSkippedBrokenWalFile = null; + lastUnreportedSkippedBrokenWalFile = null; + firstUnreportedSkippedBrokenWalError = null; + } + + private static String summarizeException(final IOException error) { + return error.getMessage() == null + ? error.getClass().getSimpleName() + : error.getClass().getSimpleName() + ": " + error.getMessage(); + } + private boolean skipEntries(final ProgressWALReader reader, final int skipEntries) throws IOException { return reader.skipToEntryIndex(skipEntries); 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 2656c662e96eb..fbd9422e20168 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 @@ -58,6 +58,7 @@ import java.lang.reflect.Modifier; import java.util.Arrays; import java.util.Collections; +import java.util.Iterator; import java.util.List; import java.util.concurrent.BlockingQueue; import java.util.concurrent.CountDownLatch; @@ -742,6 +743,67 @@ ProgressWALReader openProgressWALReader(final File walFile) throws IOException { } } + @Test + public void testWalReplayCountsOnlyUnavailableSearchIndexes() throws Exception { + final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + final File systemDir = temporaryFolder.newFolder("wal-replay-gap-counter"); + ConsensusPrefetchingQueue queue = null; + try { + final DataRegionId regionId = new DataRegionId(9); + final FakeConsensusReqReader reader = new FakeConsensusReqReader(); + reader.currentSearchIndex = 4L; + final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class); + when(serverImpl.getConsensusReqReader()).thenReturn(reader); + when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new WriterSafeFrontierTracker()); + + final ConsensusLogToTabletConverter converter = mock(ConsensusLogToTabletConverter.class); + when(converter.convert(any())).thenReturn(Collections.singletonList(createTablet())); + when(converter.getDatabaseName()).thenReturn("db"); + + final Iterator retainedWalEntries = + Arrays.asList(createRequest(1L), createRequest(4L)).iterator(); + final ProgressWALIterator walIterator = mock(ProgressWALIterator.class); + when(walIterator.hasNext()).thenAnswer(ignored -> retainedWalEntries.hasNext()); + when(walIterator.next()).thenAnswer(ignored -> retainedWalEntries.next()); + + queue = + new ConsensusPrefetchingQueue( + "consumerGroup", + "topic", + TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE, + regionId, + serverImpl, + new SubscriptionWalRetentionPolicy( + "topic", + SubscriptionWalRetentionPolicy.UNBOUNDED, + SubscriptionWalRetentionPolicy.UNBOUNDED), + converter, + newCommitManager(systemDir), + new RegionProgress(Collections.emptyMap()), + 1L, + 1L, + true) { + @Override + protected ProgressWALIterator createSubscriptionWALIterator( + final long startSearchIndex) { + return walIterator; + } + }; + + assertNull(queue.poll("consumer")); + queue.drivePrefetchOnce(); + + assertEquals(2L, queue.getWalPathAcceptedEntries()); + assertEquals(2L, queue.getWalGapSkippedEntries()); + assertEquals(5L, queue.getCurrentReadSearchIndex()); + } finally { + if (queue != null) { + queue.close(); + } + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + @Test public void testActivationRetriesUntilConfigNodeProgressIsExplicitlyAvailable() 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 b37542e936d22..1bb52ab4ebb83 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 @@ -272,6 +272,44 @@ public void testFollowerEntryDoesNotSynthesizeSearchIndexFromProgressLocalSeq() } } + @Test + public void testIteratorAggregatesUnreadableRetainedWalFiles() throws Exception { + final Path dir = Files.createTempDirectory("progress-wal-iterator-unreadable-files"); + final File firstBrokenWal = + dir.resolve(WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + final File secondBrokenWal = + dir.resolve(WALFileUtils.getLogFileName(1, 1, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + final File lastWal = + dir.resolve(WALFileUtils.getLogFileName(2, 2, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + + try { + Files.write(firstBrokenWal.toPath(), new byte[128]); + Files.write(secondBrokenWal.toPath(), new byte[128]); + try (WALWriter writer = new WALWriter(lastWal, WALFileVersion.V3)) { + // Create a readable successor so both malformed WAL files are treated as retained history. + writer.write(searchableEntry(2L), singleEntryMeta(19, 2L, 1L, 200L, 7, 2L)); + } + + try (ProgressWALIterator iterator = new ProgressWALIterator(dir.toFile(), Long.MIN_VALUE)) { + assertTrue(iterator.hasNext()); + assertEquals(2L, iterator.next().getSearchIndex()); + assertFalse(iterator.hasNext()); + assertTrue(iterator.hasSkippedBrokenWalFiles()); + assertEquals(2, iterator.getSkippedBrokenWalFileCount()); + assertTrue(iterator.hasIncompleteScan()); + assertFalse(iterator.hasReadError()); + } + } finally { + Files.deleteIfExists(firstBrokenWal.toPath()); + Files.deleteIfExists(secondBrokenWal.toPath()); + Files.deleteIfExists(lastWal.toPath()); + Files.deleteIfExists(dir); + } + } + @Test public void testIteratorMarksIncompleteScanWhenNearLiveWalCannotBeOpened() throws Exception { final Path dir = Files.createTempDirectory("progress-wal-iterator-incomplete-scan");