From ece3c17e1f123818da1079f2409ca8157b6585c0 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Wed, 21 Oct 2026 19:38:45 +0800 Subject: [PATCH] [Subscription] Diagnose stalled replay and WAL footer recovery --- .../iotdb/db/i18n/DataNodePipeMessages.java | 4 +- .../iotdb/db/i18n/StorageEngineMessages.java | 3 + .../iotdb/db/i18n/DataNodePipeMessages.java | 4 +- .../iotdb/db/i18n/StorageEngineMessages.java | 3 + .../dataregion/wal/io/WALMetaData.java | 36 +++++ .../wal/recover/WALRepairWriter.java | 15 ++ .../consensus/ConsensusPrefetchingQueue.java | 145 +++++++++++++++++- .../broker/consensus/ProgressWALIterator.java | 22 +++ .../wal/recover/WALRepairWriterTest.java | 65 ++++++++ .../ConsensusPrefetchingQueueTest.java | 24 ++- ...susPrefetchingQueueWriterOrderingTest.java | 87 +++++++++++ 11 files changed, 399 insertions(+), 9 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 bb04beac563ed..8eb2b8a6116d0 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 @@ -20,8 +20,8 @@ package org.apache.iotdb.db.i18n; public final class DataNodePipeMessages { - public static final String LOG_SUBSCRIPTION_REPLAY_HAS_MADE_NO_CURSOR_PROGRESS_FOR_ARG_MS_ARG_INSPECT_PREFETCH_WORKER_STACKS_AND_WAL_I_O_FAE1F64B = - "Subscription replay has made no cursor progress for {} ms: {}. Inspect prefetch worker stacks and WAL I/O."; + public static final String LOG_SUBSCRIPTION_REPLAY_STALLED_FOR_ARG_MS_ARG_DIAGNOSTICS_ARG_INSPECT_PREFETCH_WORKER_STACKS_AND_WAL_I_O_CB012F98 = + "Subscription replay stalled for {} ms: {}; diagnostics={}. Inspect prefetch worker stacks and WAL I/O."; public static final String LOG_SUBSCRIPTION_PREFETCH_ROUND_STARTING_FOR_ARG_ACTIVE_ARG_INITIALIZED_ARG_CLOSED_ARG_PENDINGENTRIES_ARG_PREFETCHEDEVENTS_ARG_INFLIGHTEVENTS_ARG_NEXTREADSEARCHINDEX_ARG_F75FF2CD = diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java index 1adfd8ac3cc8e..294af1dbb28d9 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java @@ -21,6 +21,9 @@ public final class StorageEngineMessages { + public static final String LOG_REBUILT_UNREADABLE_WAL_FOOTER_FILE_ARG_VERSION_ARG_ORIGINALFILEBYTES_ARG_WRITERPROGRESS_ARG_1DBF836E = + "Rebuilt unreadable WAL footer: file={}, version={}, originalFileBytes={}, writerProgress={}."; + public static final String MESSAGE_LOAD_TSFILE_SLICE_BODY_IS_NULL_OR_EMPTY_2A65366C = "Load TsFile slice body is null or empty"; public static final String MESSAGE_INVALID_LOAD_TSFILE_SLICE_METADATA_SLICECOUNT_ARG_ORIGINBODYSIZE_ARG_379BF1B8 = 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 d319d0e37728b..ee86a424c299f 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 @@ -20,8 +20,8 @@ package org.apache.iotdb.db.i18n; public final class DataNodePipeMessages { - public static final String LOG_SUBSCRIPTION_REPLAY_HAS_MADE_NO_CURSOR_PROGRESS_FOR_ARG_MS_ARG_INSPECT_PREFETCH_WORKER_STACKS_AND_WAL_I_O_FAE1F64B = - "订阅回放游标已连续 {} ms 未推进：{}。请检查预取工作线程栈和 WAL I/O。"; + public static final String LOG_SUBSCRIPTION_REPLAY_STALLED_FOR_ARG_MS_ARG_DIAGNOSTICS_ARG_INSPECT_PREFETCH_WORKER_STACKS_AND_WAL_I_O_CB012F98 = + "订阅回放已停滞 {} ms：{}；diagnostics={}。请检查预取工作线程栈和 WAL I/O。"; public static final String LOG_SUBSCRIPTION_PREFETCH_ROUND_STARTING_FOR_ARG_ACTIVE_ARG_INITIALIZED_ARG_CLOSED_ARG_PENDINGENTRIES_ARG_PREFETCHEDEVENTS_ARG_INFLIGHTEVENTS_ARG_NEXTREADSEARCHINDEX_ARG_F75FF2CD = diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java index 0a0d3b266a4bd..49f0e760e0a36 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java @@ -21,6 +21,9 @@ public final class StorageEngineMessages { + public static final String LOG_REBUILT_UNREADABLE_WAL_FOOTER_FILE_ARG_VERSION_ARG_ORIGINALFILEBYTES_ARG_WRITERPROGRESS_ARG_1DBF836E = + "已重建不可读的 WAL footer：file={}，version={}，originalFileBytes={}，writerProgress={}。"; + public static final String MESSAGE_LOAD_TSFILE_SLICE_BODY_IS_NULL_OR_EMPTY_2A65366C = "Load TsFile 分片内容为 null 或为空"; public static final String MESSAGE_INVALID_LOAD_TSFILE_SLICE_METADATA_SLICECOUNT_ARG_ORIGINBODYSIZE_ARG_379BF1B8 = diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALMetaData.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALMetaData.java index 50ae0961ce1ff..5d316a7508f9a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALMetaData.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALMetaData.java @@ -39,6 +39,7 @@ import java.util.ArrayList; import java.util.HashMap; import java.util.HashSet; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Set; @@ -312,6 +313,41 @@ public List getLocalSeqs() { return localSeqs; } + /** Diagnostic summary for an abnormal footer recovery, never for the normal write path. */ + public Map getWriterProgressDiagnostics() { + final Map result = new LinkedHashMap<>(); + final int entries = buffersSize.size(); + final int progressEntries = + Math.min( + entries, Math.min(physicalTimes.size(), Math.min(nodeIds.size(), localSeqs.size()))); + int attributedEntries = 0; + int defaultLocalProgressEntries = 0; + final List unattributedSamples = new ArrayList<>(); + for (int i = 0; i < progressEntries; i++) { + final long physicalTime = physicalTimes.get(i); + final int nodeId = nodeIds.get(i); + final long localSeq = localSeqs.get(i); + if (nodeId >= 0 && physicalTime >= 0 && localSeq >= 0) { + attributedEntries++; + } else { + if (unattributedSamples.size() < 4) { + unattributedSamples.add(i + ":(" + physicalTime + "," + nodeId + "," + localSeq + ")"); + } + if (nodeId < 0 && physicalTime == 0 && localSeq >= 0) { + defaultLocalProgressEntries++; + } + } + } + result.put("entryCount", entries); + result.put("firstSearchIndex", firstSearchIndex); + result.put("attributedProgressEntries", attributedEntries); + result.put("unattributedProgressEntries", progressEntries - attributedEntries); + result.put("missingProgressEntries", entries - progressEntries); + result.put("defaultLocalProgressEntries", defaultLocalProgressEntries); + result.put("unattributedSamples", unattributedSamples); + return result; + } + private int getNodeIdOverrideCount() { final short defaultNodeId = computeDefaultNodeId(); int count = 0; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRepairWriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRepairWriter.java index 94129d9e22067..7572eac68cb96 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRepairWriter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRepairWriter.java @@ -19,12 +19,16 @@ package org.apache.iotdb.db.storageengine.dataregion.wal.recover; +import org.apache.iotdb.db.i18n.StorageEngineMessages; import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntryType; import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALByteBufReader; 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; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import java.io.File; import java.io.IOException; import java.nio.ByteBuffer; @@ -38,6 +42,7 @@ /** Check whether the wal file is broken and repair it. */ public class WALRepairWriter { + private static final Logger LOGGER = LoggerFactory.getLogger(WALRepairWriter.class); private final File logFile; public WALRepairWriter(File logFile) { @@ -63,6 +68,7 @@ public boolean repair(WALMetaData metaData) throws IOException { quarantine(); return false; } + final long originalFileBytes = logFile.length(); // A channel offset may include a partially read entry in the same compressed segment. Rebuild // complete entries in a temporary file instead of truncating at a read-ahead offset. Publish // only after every entry and the new footer have been written and forced successfully. @@ -111,6 +117,15 @@ public boolean repair(WALMetaData metaData) throws IOException { } finally { Files.deleteIfExists(repaired); } + if (LOGGER.isWarnEnabled()) { + LOGGER.warn( + StorageEngineMessages + .LOG_REBUILT_UNREADABLE_WAL_FOOTER_FILE_ARG_VERSION_ARG_ORIGINALFILEBYTES_ARG_WRITERPROGRESS_ARG_1DBF836E, + logFile, + version, + originalFileBytes, + metaData.getWriterProgressDiagnostics()); + } return true; } 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 1268689749fbe..ad12edb78efc4 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 @@ -351,6 +351,8 @@ public class ConsensusPrefetchingQueue { private long lastReplayProgressNanos; private boolean replayStallWarningLogged; private long lastReplayStallWarningNanos; + private WriterFrontier lastBlockedWriterFrontier; + private boolean lastWriterBarrierInOrderedReplay; private volatile String lastConsumerId = ""; @@ -1687,6 +1689,7 @@ public SubscriptionEvent pollTablets( private static final long PREFETCH_STATS_LOG_INTERVAL_MS = 5_000L; private static final long REPLAY_STALL_WARNING_THRESHOLD_NANOS = TimeUnit.SECONDS.toNanos(60L); private static final long REPLAY_STALL_WARNING_INTERVAL_NANOS = TimeUnit.SECONDS.toNanos(30L); + private static final int MAX_REPLAY_DIAGNOSTIC_WRITERS = 4; private static final long SLOW_PREFETCH_ROUND_THRESHOLD_MS = 10_000L; @@ -1800,6 +1803,8 @@ private PrefetchRoundResult drivePrefetchOnceUnderReadLock() { getPrefetchDurationMs()); } logPeriodicStatsIfNecessary(); + // Preserve the previous round's actual ordering decision until the stall reporter sees it. + lastBlockedWriterFrontier = null; final long currentSeekGeneration = seekGeneration.get(); if (currentSeekGeneration != observedSeekGeneration) { @@ -2067,6 +2072,8 @@ private PrefetchRoundResult driveOrderedWriterReplay( } final WriterFrontier next = frontiers.peek(); if (Objects.isNull(next) || next.isBarrier) { + lastBlockedWriterFrontier = next; + lastWriterBarrierInOrderedReplay = true; break; } final PreparedEntry buffered = peekRealtimeEntry(next.writerNodeId); @@ -2393,19 +2400,138 @@ private void reportReplayStallIfNecessary(final long nowNanos) { final long stalledNanos = nowNanos - lastReplayProgressNanos; if (stalledNanos >= REPLAY_STALL_WARNING_THRESHOLD_NANOS && (!replayStallWarningLogged - || nowNanos - lastReplayStallWarningNanos >= REPLAY_STALL_WARNING_INTERVAL_NANOS)) { + || nowNanos - lastReplayStallWarningNanos >= REPLAY_STALL_WARNING_INTERVAL_NANOS) + && LOGGER.isWarnEnabled()) { LOGGER.warn( DataNodePipeMessages - .LOG_SUBSCRIPTION_REPLAY_HAS_MADE_NO_CURSOR_PROGRESS_FOR_ARG_MS_ARG_INSPECT_PREFETCH_WORKER_STACKS_AND_WAL_I_O_FAE1F64B, + .LOG_SUBSCRIPTION_REPLAY_STALLED_FOR_ARG_MS_ARG_DIAGNOSTICS_ARG_INSPECT_PREFETCH_WORKER_STACKS_AND_WAL_I_O_CB012F98, TimeUnit.NANOSECONDS.toMillis(stalledNanos), - this); + this, + getReplayStallDiagnostics()); lastReplayStallWarningNanos = nowNanos; replayStallWarningLogged = true; } } + private Map getReplayStallDiagnostics() { + // Only called for the throttled stall warning, on the serialized prefetch worker. Do not + // refresh frontiers, probe WAL readers, or dump requests/Tablets while taking this snapshot. + final Map result = new LinkedHashMap<>(); + result.put("runtimeVersion", runtimeVersion); + result.put("admissionBlockReason", realtimeAdmissionBlockReason); + result.put("memoryBlockReason", memoryBlockReason); + result.put("prefetchLimitReached", hasReachedPrefetchLimit()); + result.put("lingerEntryCount", lingerBatch.entryCount); + result.put("lastBlockedFrontier", lastBlockedWriterFrontier); + result.put( + "lastBarrierInOrderedReplay", + Objects.nonNull(lastBlockedWriterFrontier) ? lastWriterBarrierInOrderedReplay : null); + final Map safeFrontiers = + serverImpl.getWriterSafeFrontierTracker().snapshotSafeFrontiers(); + result.put("activeWriterCount", activeWriterNodeIds.size()); + result.put("activeWriters", sampleWriterDiagnostics(activeWriterNodeIds, safeFrontiers)); + result.put("realtimeWriterCount", realtimeEntriesByWriter.size()); + result.put( + "realtimeWriters", + sampleWriterDiagnostics(realtimeEntriesByWriter.keySet(), safeFrontiers)); + result.put("orderedWriterCount", orderedWriterReplay.size()); + result.put( + "orderedWriters", sampleWriterDiagnostics(orderedWriterReplay.keySet(), safeFrontiers)); + final RegionProgress committed = + commitManager.getCommittedRegionProgress(consumerGroupId, topicName, consensusGroupId); + result.put("committedWriterCount", committed.getWriterPositions().size()); + final List committedSamples = new ArrayList<>(); + for (final Map.Entry entry : + committed.getWriterPositions().entrySet()) { + if (committedSamples.size() >= MAX_REPLAY_DIAGNOSTIC_WRITERS) { + break; + } + committedSamples.add(entry.toString()); + } + result.put("committedWriters", committedSamples); + result.put("committedRetainedMinVersionId", committedRetainedMinVersionId); + final WalFileCommitRequirement requirement = + walFileCommitRequirements.get(committedRetainedMinVersionId); + result.put( + "retentionContainsUnsupportedProgress", + Objects.nonNull(requirement) ? requirement.containsUnsupportedProgress : null); + if (consensusReqReader instanceof WALNode) { + result.put("walDirectory", ((WALNode) consensusReqReader).getLogDirectory()); + } + result.put( + "walScanState", + Objects.nonNull(subscriptionWALIterator) + ? subscriptionWALIterator.getDiagnosticState() + : null); + return result; + } + + private List> sampleWriterDiagnostics( + final Set writerNodeIds, final Map safeFrontiers) { + final List> samples = new ArrayList<>(); + final Set sampledWriterNodeIds = new LinkedHashSet<>(); + // Keep unknown progress and the actual blocker visible even when other writers fill the cap. + if (writerNodeIds.contains(-1)) { + sampledWriterNodeIds.add(-1); + } + if (Objects.nonNull(lastBlockedWriterFrontier) + && writerNodeIds.contains(lastBlockedWriterFrontier.writerNodeId)) { + sampledWriterNodeIds.add(lastBlockedWriterFrontier.writerNodeId); + } + for (final int writerNodeId : writerNodeIds) { + if (sampledWriterNodeIds.size() >= MAX_REPLAY_DIAGNOSTIC_WRITERS) { + break; + } + sampledWriterNodeIds.add(writerNodeId); + } + for (final int writerNodeId : sampledWriterNodeIds) { + final Map sample = new LinkedHashMap<>(); + sample.put("writerNodeId", writerNodeId); + final WriterState state = writerStates.get(writerNodeId); + sample.put("effectiveSafePt", Objects.nonNull(state) ? state.effectiveSafePt : null); + sample.put("closed", Objects.nonNull(state) ? state.closed : null); + final SafeFrontier safeFrontier = safeFrontiers.get(writerNodeId); + if (Objects.nonNull(safeFrontier)) { + sample.put("trackerSafePt", safeFrontier.getPhysicalTime()); + sample.put("trackerAppliedLocalSeq", safeFrontier.getLocalSeq()); + } + final PreparedEntry buffered = peekRealtimeEntry(writerNodeId); + if (Objects.nonNull(buffered)) { + sample.put("bufferedHead", buffered.getOrderingKey()); + sample.put("bufferedSearchIndex", buffered.searchIndex); + sample.put("bufferedEntryCount", realtimeEntriesByWriter.get(writerNodeId).size()); + } + sample.put("materializedProgress", materializedProgressByWriter.get(writerNodeId)); + sample.put( + "recoveryProgress", + recoveryWriterProgressByWriter.get( + new WriterId(consensusGroupId.toString(), writerNodeId))); + final WriterReplayState replay = orderedWriterReplay.get(writerNodeId); + if (Objects.nonNull(replay)) { + sample.put("consumedLocalSeq", replay.consumedLocalSeq); + sample.put("caughtUp", replay.caughtUp); + sample.put("fromPending", replay.fromPending); + if (Objects.nonNull(replay.head)) { + sample.put( + "rawHead", + new OrderingKey( + replay.head.getPhysicalTime(), + replay.head.getNodeId(), + replay.head.getProgressLocalSeq())); + sample.put("rawSearchIndex", replay.head.getSearchIndex()); + } + sample.put( + "walScanState", + Objects.nonNull(replay.iterator) ? replay.iterator.getDiagnosticState() : null); + } + samples.add(sample); + } + return samples; + } + private void resetRoundStateForSeek(final long newSeekGeneration) { replayProgressObserved = false; + lastBlockedWriterFrontier = null; restorePendingSubscriptionWalCursor(newSeekGeneration); discardRealtimeEntries(); discardBatch(lingerBatch); @@ -3734,6 +3860,8 @@ private boolean drainWriterEntries( } final WriterFrontier frontier = frontiers.peek(); if (Objects.isNull(frontier) || frontier.isBarrier) { + lastBlockedWriterFrontier = frontier; + lastWriterBarrierInOrderedReplay = false; return true; } final T writerHead = headSupplier.apply(frontier.writerNodeId); @@ -6204,6 +6332,17 @@ private static WriterFrontier forBarrier(final int writerNodeId, final long effe writerNodeId, new OrderingKey(effectiveSafePt, Integer.MIN_VALUE, Long.MIN_VALUE), true); } + @Override + public String toString() { + return "WriterFrontier{writerNodeId=" + + writerNodeId + + ", orderingKey=" + + orderingKey + + ", barrier=" + + isBarrier + + '}'; + } + @Override public int compareTo(final WriterFrontier other) { int cmp = orderingKey.compareTo(other.orderingKey); 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 07438858a35ab..c50498ad3db0a 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 @@ -522,6 +522,28 @@ public String toString() { + "]"; } + String getDiagnosticState() { + // Called only by the throttled queue stall warning; never read entry bodies or advance here. + return this + + "[file=" + + currentReaderFile + + ", liveSnapshot=" + + currentReaderUsesLiveSnapshot + + ", metadataRecoveredFromEntries=" + + (currentReader != null && currentReader.getMetaData().isRecoveredFromEntries()) + + ", pendingProgress=(" + + pendingPhysicalTime + + "," + + pendingNodeId + + "," + + pendingLocalSeq + + "), pendingSearchIndex=" + + pendingSearchIndex + + ", pendingFragments=" + + pendingRequests.size() + + "]"; + } + String getSkippedBrokenWalFileDetails(final int firstFileIndex) { final int firstAvailableIndex = getSkippedBrokenWalFileCount() - skippedBrokenWalFileDetails.size(); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRepairWriterTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRepairWriterTest.java index ea0a615d9846e..db91a92feced8 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRepairWriterTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRepairWriterTest.java @@ -21,6 +21,7 @@ import org.apache.iotdb.commons.exception.IllegalPathException; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; +import org.apache.iotdb.db.i18n.StorageEngineMessages; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntry; @@ -34,6 +35,10 @@ import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileUtils; import org.apache.iotdb.db.utils.constant.TestConstant; +import ch.qos.logback.classic.Level; +import ch.qos.logback.classic.Logger; +import ch.qos.logback.classic.spi.ILoggingEvent; +import ch.qos.logback.core.read.ListAppender; import org.apache.tsfile.common.conf.TSFileConfig; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.utils.Binary; @@ -42,6 +47,7 @@ import org.junit.Assert; import org.junit.Before; import org.junit.Test; +import org.slf4j.LoggerFactory; import java.io.File; import java.io.IOException; @@ -53,6 +59,8 @@ import java.nio.file.StandardOpenOption; import java.util.ArrayList; import java.util.HashSet; +import java.util.List; +import java.util.Map; import java.util.stream.Stream; public class WALRepairWriterTest { @@ -76,6 +84,63 @@ public void tearDown() throws Exception { } } + @Test + public void testFooterRepairReportsUnattributedProgressOnce() throws Exception { + final Logger logger = (Logger) LoggerFactory.getLogger(WALRepairWriter.class); + final Level originalLevel = logger.getLevel(); + final ListAppender appender = new ListAppender<>(); + logger.setLevel(Level.WARN); + appender.start(); + logger.addAppender(appender); + try { + long dataEnd; + try (WALWriter writer = new WALWriter(logFile)) { + for (int i = 1; i <= 8; i++) { + final InsertRowNode node = getInsertRowNode(); + // Include replicated entries whose local search index cannot identify their writer. + node.setSearchIndex(i == 8 ? -1 : i); + final WALInfoEntry entry = new WALInfoEntry(1, node); + final ByteBuffer buffer = ByteBuffer.allocate(entry.serializedSize()); + entry.serialize(new WALByteBufferForTest(buffer)); + final WALMetaData metadata = new WALMetaData(); + metadata.add(entry.serializedSize(), node.getSearchIndex(), 1, 1000, i == 8 ? 8 : 7, i); + writer.write(buffer, metadata); + } + dataEnd = writer.getOffset(); + } + try (FileChannel channel = FileChannel.open(logFile.toPath(), StandardOpenOption.WRITE)) { + channel.truncate(dataEnd); + } + final WALMetaData recovered; + try (FileChannel channel = FileChannel.open(logFile.toPath(), StandardOpenOption.READ)) { + recovered = WALMetaData.readFromWALFile(logFile, channel); + } + Assert.assertTrue(new WALRepairWriter(logFile).repair(recovered)); + // A second check of the intact file must neither rescan diagnostic arrays nor log again. + Assert.assertTrue(new WALRepairWriter(logFile).repair(recovered)); + Assert.assertEquals(1, appender.list.size()); + final ILoggingEvent warning = appender.list.get(0); + Assert.assertEquals( + StorageEngineMessages + .LOG_REBUILT_UNREADABLE_WAL_FOOTER_FILE_ARG_VERSION_ARG_ORIGINALFILEBYTES_ARG_WRITERPROGRESS_ARG_1DBF836E, + warning.getMessage()); + final Map progress = (Map) warning.getArgumentArray()[3]; + Assert.assertEquals(8, progress.get("entryCount")); + Assert.assertEquals(0, progress.get("attributedProgressEntries")); + Assert.assertEquals(8, progress.get("unattributedProgressEntries")); + Assert.assertEquals(7, progress.get("defaultLocalProgressEntries")); + Assert.assertEquals(4, ((List) progress.get("unattributedSamples")).size()); + Assert.assertTrue(warning.getFormattedMessage().contains(logFile.toString())); + try (WALByteBufReader reader = new WALByteBufReader(logFile)) { + Assert.assertEquals(8, reader.getMetaData().getBuffersSize().size()); + } + } finally { + logger.detachAppender(appender); + logger.setLevel(originalLevel); + appender.stop(); + } + } + @Test public void testEmptyFile() throws IOException { // prepare file 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 a985a0f617d8c..494c40bf7dff4 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 @@ -88,6 +88,8 @@ import java.util.Collections; import java.util.Iterator; import java.util.List; +import java.util.Map; +import java.util.Set; import java.util.concurrent.BlockingQueue; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; @@ -146,7 +148,7 @@ public void testReplayStallWarningIsThrottledAndResetsAfterProgress() throws Exc new ConsensusPrefetchingQueue( "consumerGroup", "topic", - TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE, + TopicConstant.ORDER_MODE_MULTI_WRITER_VALUE, new DataRegionId(1), server, new SubscriptionWalRetentionPolicy( @@ -163,6 +165,7 @@ public void testReplayStallWarningIsThrottledAndResetsAfterProgress() throws Exc ConsensusPrefetchingQueue.class.getDeclaredField("prefetchInitialized"); initialized.setAccessible(true); initialized.setBoolean(queue, true); + queue.setActiveWriterNodeIds(Set.of(1, 2, 3, 4, 5, 6, 7, 8)); final Method report = ConsensusPrefetchingQueue.class.getDeclaredMethod( "reportReplayStallIfNecessary", long.class); @@ -176,7 +179,7 @@ public void testReplayStallWarningIsThrottledAndResetsAfterProgress() throws Exc .getMessage() .equals( DataNodePipeMessages - .LOG_SUBSCRIPTION_REPLAY_HAS_MADE_NO_CURSOR_PROGRESS_FOR_ARG_MS_ARG_INSPECT_PREFETCH_WORKER_STACKS_AND_WAL_I_O_FAE1F64B)) + .LOG_SUBSCRIPTION_REPLAY_STALLED_FOR_ARG_MS_ARG_DIAGNOSTICS_ARG_INSPECT_PREFETCH_WORKER_STACKS_AND_WAL_I_O_CB012F98)) .count(); report.invoke(queue, 0L); report.invoke(queue, TimeUnit.SECONDS.toNanos(59L)); @@ -184,6 +187,23 @@ public void testReplayStallWarningIsThrottledAndResetsAfterProgress() throws Exc report.invoke(queue, TimeUnit.SECONDS.toNanos(60L)); report.invoke(queue, TimeUnit.SECONDS.toNanos(89L)); assertEquals(Long.valueOf(1L), warningCount.get()); + final Map diagnostics = + (Map) + appender.list.stream() + .filter( + event -> + event + .getMessage() + .equals( + DataNodePipeMessages + .LOG_SUBSCRIPTION_REPLAY_STALLED_FOR_ARG_MS_ARG_DIAGNOSTICS_ARG_INSPECT_PREFETCH_WORKER_STACKS_AND_WAL_I_O_CB012F98)) + .findFirst() + .get() + .getArgumentArray()[2]; + assertEquals(8, diagnostics.get("activeWriterCount")); + assertEquals(4, ((List) diagnostics.get("activeWriters")).size()); + assertTrue(((List) diagnostics.get("realtimeWriters")).isEmpty()); + assertTrue(((List) diagnostics.get("orderedWriters")).isEmpty()); report.invoke(queue, TimeUnit.SECONDS.toNanos(90L)); assertEquals(Long.valueOf(2L), warningCount.get()); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueWriterOrderingTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueWriterOrderingTest.java index d943ca4124106..972a3a5fdc063 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueWriterOrderingTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueWriterOrderingTest.java @@ -28,6 +28,7 @@ import org.apache.iotdb.consensus.iot.WriterSafeFrontierTracker; import org.apache.iotdb.consensus.iot.log.ConsensusReqReader; import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryWeightUtil; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode; import org.apache.iotdb.db.queryengine.plan.statement.StatementTestUtils; @@ -36,6 +37,7 @@ import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALMetaData; import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALWriter; import org.apache.iotdb.db.storageengine.dataregion.wal.node.WALNode; +import org.apache.iotdb.db.storageengine.dataregion.wal.recover.WALRepairWriter; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALByteBufferForTest; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileStatus; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileUtils; @@ -46,6 +48,10 @@ import org.apache.iotdb.rpc.subscription.payload.poll.SubscriptionPollResponseType; import org.apache.iotdb.rpc.subscription.payload.poll.TabletsPayload; +import ch.qos.logback.classic.Level; +import ch.qos.logback.classic.Logger; +import ch.qos.logback.classic.spi.ILoggingEvent; +import ch.qos.logback.core.read.ListAppender; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.write.record.Tablet; import org.apache.tsfile.write.schema.MeasurementSchema; @@ -54,16 +60,20 @@ import org.junit.Rule; import org.junit.Test; import org.junit.rules.TemporaryFolder; +import org.slf4j.LoggerFactory; import java.io.File; import java.lang.reflect.Field; import java.lang.reflect.Method; import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; import java.nio.file.Files; +import java.nio.file.StandardOpenOption; import java.util.Collections; import java.util.List; import java.util.Set; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import static org.junit.Assert.assertEquals; @@ -116,6 +126,83 @@ public void tearDown() { IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); } + @Test + public void testRecoveredUnknownWriterLogsActualBlockingBarrier() throws Exception { + final Logger logger = (Logger) LoggerFactory.getLogger(ConsensusPrefetchingQueue.class); + final Level originalLevel = logger.getLevel(); + final ListAppender appender = new ListAppender<>(); + logger.setLevel(Level.WARN); + appender.start(); + logger.addAppender(appender); + try { + final File directory = writeWal(8); + final File data = + new File( + directory, WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)); + try (FileChannel channel = FileChannel.open(data.toPath(), StandardOpenOption.WRITE)) { + channel.truncate(channel.size() - 1); + } + final WALMetaData recovered; + try (FileChannel channel = FileChannel.open(data.toPath(), StandardOpenOption.READ)) { + recovered = WALMetaData.readFromWALFile(data, channel); + } + assertTrue(new WALRepairWriter(data).repair(recovered)); + newQueue(walNode(directory, 8), 1024 * 1024); + queue.setActiveWriterNodeIds(Set.of(7)); + for (int round = 0; round < 5; round++) { + queue.drivePrefetchOnce(); + } + assertNull(queue.poll("consumer")); + final int readsBefore = walReads.get(); + final long bytesBefore = queue.getRetainedTabletBytes(); + final Field iteratorField = + ConsensusPrefetchingQueue.class.getDeclaredField("subscriptionWALIterator"); + iteratorField.setAccessible(true); + final ProgressWALIterator iterator = (ProgressWALIterator) iteratorField.get(queue); + assertNotNull(iterator); + final String scanStateBefore = iterator.getDiagnosticState(); + final Method report = + ConsensusPrefetchingQueue.class.getDeclaredMethod( + "reportReplayStallIfNecessary", long.class); + report.setAccessible(true); + final long nowNanos = System.nanoTime(); + report.invoke(queue, nowNanos); + report.invoke(queue, nowNanos + TimeUnit.SECONDS.toNanos(60)); + final ILoggingEvent warning = + appender.list.stream() + .filter( + event -> + event + .getMessage() + .equals( + DataNodePipeMessages + .LOG_SUBSCRIPTION_REPLAY_STALLED_FOR_ARG_MS_ARG_DIAGNOSTICS_ARG_INSPECT_PREFETCH_WORKER_STACKS_AND_WAL_I_O_CB012F98)) + .findFirst() + .get(); + final String diagnostic = warning.getFormattedMessage(); + assertTrue( + diagnostic, diagnostic.contains("lastBlockedFrontier=WriterFrontier{writerNodeId=7")); + assertTrue(diagnostic, diagnostic.contains("orderingKey=(" + Long.MIN_VALUE + ",")); + assertTrue(diagnostic, diagnostic.contains("barrier=true")); + // The replicated entry also loses its progress and sorts before the local defaults. + assertTrue(diagnostic, diagnostic.contains("bufferedHead=(0,-1,-1)")); + assertTrue(diagnostic, diagnostic.contains("bufferedSearchIndex=-1")); + assertTrue(diagnostic, diagnostic.contains("bufferedEntryCount=9")); + assertTrue(diagnostic, diagnostic.contains("admissionBlockReason=WRITER_BACKLOG")); + assertTrue(diagnostic, diagnostic.contains("retentionContainsUnsupportedProgress=true")); + assertTrue(diagnostic, diagnostic.contains(directory.toString())); + // Taking the diagnostic must not consume WAL requests, change ordering, or retain Tablets. + assertEquals(readsBefore, walReads.get()); + assertEquals(scanStateBefore, iterator.getDiagnosticState()); + assertEquals(bytesBefore, queue.getRetainedTabletBytes()); + assertNull(queue.poll("consumer")); + } finally { + logger.detachAppender(appender); + logger.setLevel(originalLevel); + appender.stop(); + } + } + @Test public void testBlockedWriterBarrierRetriesLaterAndResumesAfterIdleBarrier() throws Exception { newQueue(mock(ConsensusReqReader.class), 1024 * 1024);