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 4f86b5681ea [Subscription] Diagnose stalled replay and WAL footer
recovery (#18970)
4f86b5681ea is described below
commit 4f86b5681ea075ed4c79287cb572198224cf4ecd
Author: Caideyipi <[email protected]>
AuthorDate: Thu Oct 22 12:32:46 2026 +0800
[Subscription] Diagnose stalled replay and WAL footer recovery (#18970)
---
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 4 +-
.../iotdb/db/i18n/StorageEngineMessages.java | 3 +
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 4 +-
.../iotdb/db/i18n/StorageEngineMessages.java | 3 +
.../dataregion/wal/io/WALMetaData.java | 36 +++++
.../dataregion/wal/recover/WALRepairWriter.java | 15 +++
.../consensus/ConsensusPrefetchingQueue.java | 145 ++++++++++++++++++++-
.../broker/consensus/ProgressWALIterator.java | 22 ++++
.../wal/recover/WALRepairWriterTest.java | 65 +++++++++
.../consensus/ConsensusPrefetchingQueueTest.java | 24 +++-
...onsensusPrefetchingQueueWriterOrderingTest.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 bb04beac563..8eb2b8a6116 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 1adfd8ac3cc..294af1dbb28 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 @@ package org.apache.iotdb.db.i18n;
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 d319d0e3772..ee86a424c29 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 0a0d3b266a4..49f0e760e0a 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 @@ package org.apache.iotdb.db.i18n;
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 50ae0961ce1..5d316a7508f 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.nio.charset.StandardCharsets;
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 class WALMetaData implements SerializedSize {
return localSeqs;
}
+ /** Diagnostic summary for an abnormal footer recovery, never for the normal
write path. */
+ public Map<String, Object> getWriterProgressDiagnostics() {
+ final Map<String, Object> 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<String> 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 94129d9e220..7572eac68cb 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 @@ import java.nio.file.StandardOpenOption;
/** 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 class WALRepairWriter {
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 class WALRepairWriter {
} 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 24366687030..df756dc2d22 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 class ConsensusPrefetchingQueue {
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 @@ public class ConsensusPrefetchingQueue {
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 @@ public class ConsensusPrefetchingQueue {
}
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 @@ public class ConsensusPrefetchingQueue {
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<String, Object> 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<String, Object> 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<Integer, SafeFrontier> 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<String> committedSamples = new ArrayList<>();
+ for (final Map.Entry<WriterId, WriterProgress> 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<Map<String, Object>> sampleWriterDiagnostics(
+ final Set<Integer> writerNodeIds, final Map<Integer, SafeFrontier>
safeFrontiers) {
+ final List<Map<String, Object>> samples = new ArrayList<>();
+ final Set<Integer> 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<String, Object> 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 @@ public class ConsensusPrefetchingQueue {
}
final WriterFrontier frontier = frontiers.peek();
if (Objects.isNull(frontier) || frontier.isBarrier) {
+ lastBlockedWriterFrontier = frontier;
+ lastWriterBarrierInOrderedReplay = false;
return true;
}
final T writerHead = headSupplier.apply(frontier.writerNodeId);
@@ -6243,6 +6371,17 @@ public class ConsensusPrefetchingQueue {
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 c048677a0af..cbb0d7099ec 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
@@ -527,6 +527,28 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
+ "]";
}
+ 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 ea0a615d984..db91a92fece 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 @@ package
org.apache.iotdb.db.storageengine.dataregion.wal.recover;
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.WALFileStatus;
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.After;
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.Path;
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 class WALRepairWriterTest {
}
}
+ @Test
+ public void testFooterRepairReportsUnattributedProgressOnce() throws
Exception {
+ final Logger logger = (Logger)
LoggerFactory.getLogger(WALRepairWriter.class);
+ final Level originalLevel = logger.getLevel();
+ final ListAppender<ILoggingEvent> 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 ba3c2a972ce..8d7edd979c5 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.Arrays;
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 class ConsensusPrefetchingQueueTest {
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 class ConsensusPrefetchingQueueTest {
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 class ConsensusPrefetchingQueueTest {
.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 class ConsensusPrefetchingQueueTest {
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 d943ca41241..972a3a5fdc0 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.SubscriptionWalRetentionPolicy;
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.WALFileVersion;
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.RegionProgress;
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.Before;
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 class ConsensusPrefetchingQueueWriterOrderingTest {
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<ILoggingEvent> 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);
