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 054f5dedae5 [Subscription] Report retained WAL file gaps and retry
transient reads (#18818)
054f5dedae5 is described below
commit 054f5dedae5e0d12eaacd2c35e1d70dc31fb7d5e
Author: Caideyipi <[email protected]>
AuthorDate: Sat Oct 10 17:15:04 2026 +0800
[Subscription] Report retained WAL file gaps and retry transient reads
(#18818)
* [Subscription] Report retained WAL file gaps and retry transient reads
* [Subscription] Declare runtime Logback dependencies for WAL log assertions
---
.../base/AbstractSubscriptionConsumer.java | 18 +-
.../base/SubscriptionConsumerLifecycleTest.java | 50 ++++
iotdb-core/datanode/pom.xml | 16 ++
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 16 ++
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 16 ++
.../consensus/ConsensusPrefetchingQueue.java | 56 ++++-
.../broker/consensus/ProgressWALIterator.java | 260 ++++++++++++++++-----
...nsensusSubscriptionPrefetchingQueueMetrics.java | 20 ++
.../consensus/ConsensusPrefetchingQueueTest.java | 32 +++
.../broker/consensus/ProgressWALIteratorTest.java | 207 +++++++++++++++-
...susSubscriptionPrefetchingQueueMetricsTest.java | 65 ++++++
.../iotdb/commons/service/metric/enums/Metric.java | 2 +
12 files changed, 690 insertions(+), 68 deletions(-)
diff --git
a/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java
b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java
index 4eecc4d028c..113971ac122 100644
---
a/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java
+++
b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java
@@ -130,6 +130,8 @@ abstract class AbstractSubscriptionConsumer implements
AutoCloseable {
private final AtomicBoolean isClosed = new AtomicBoolean(true);
private final AtomicReference<SubscriptionConsumerFencedException>
fencedException =
new AtomicReference<>();
+ private final AtomicReference<SubscriptionRuntimeCriticalException>
pendingCriticalPollError =
+ new AtomicReference<>();
// This variable indicates whether the consumer has ever been closed.
private final AtomicBoolean isReleased = new AtomicBoolean(false);
@@ -779,6 +781,11 @@ abstract class AbstractSubscriptionConsumer implements
AutoCloseable {
protected List<SubscriptionMessage> multiplePoll(
/* @NotNull */ final Set<String> topicNames, final long timeoutMs) {
checkIfFenced();
+ final SubscriptionRuntimeCriticalException pendingError =
+ pendingCriticalPollError.getAndSet(null);
+ if (pendingError != null) {
+ throw pendingError;
+ }
if (topicNames.isEmpty()) {
return Collections.emptyList();
}
@@ -865,10 +872,13 @@ abstract class AbstractSubscriptionConsumer implements
AutoCloseable {
// case.
checkIfFenced();
- // even if a SubscriptionRuntimeCriticalException is encountered, try to
deliver the message to
- // the client
- if (messages.isEmpty() &&
Objects.nonNull(lastSubscriptionRuntimeCriticalException)) {
- throw lastSubscriptionRuntimeCriticalException;
+ if (Objects.nonNull(lastSubscriptionRuntimeCriticalException)) {
+ if (messages.isEmpty()) {
+ throw lastSubscriptionRuntimeCriticalException;
+ }
+ // Deliver healthy sibling results once, then surface the critical error
on the next poll.
+ // Otherwise a busy topic can indefinitely mask another topic's
permanent WAL replay gap.
+ pendingCriticalPollError.compareAndSet(null,
lastSubscriptionRuntimeCriticalException);
}
return messages;
diff --git
a/iotdb-client/subscription/src/test/java/org/apache/iotdb/session/subscription/consumer/base/SubscriptionConsumerLifecycleTest.java
b/iotdb-client/subscription/src/test/java/org/apache/iotdb/session/subscription/consumer/base/SubscriptionConsumerLifecycleTest.java
index 75f1ed78650..51f35c5d820 100644
---
a/iotdb-client/subscription/src/test/java/org/apache/iotdb/session/subscription/consumer/base/SubscriptionConsumerLifecycleTest.java
+++
b/iotdb-client/subscription/src/test/java/org/apache/iotdb/session/subscription/consumer/base/SubscriptionConsumerLifecycleTest.java
@@ -22,6 +22,7 @@ package org.apache.iotdb.session.subscription.consumer.base;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import
org.apache.iotdb.rpc.subscription.exception.SubscriptionConsumerFencedException;
import org.apache.iotdb.rpc.subscription.exception.SubscriptionException;
+import
org.apache.iotdb.rpc.subscription.exception.SubscriptionRuntimeCriticalException;
import
org.apache.iotdb.rpc.subscription.exception.SubscriptionRuntimeNonCriticalException;
import
org.apache.iotdb.rpc.subscription.payload.poll.SubscriptionCommitContext;
import org.apache.iotdb.rpc.subscription.payload.poll.SubscriptionPollResponse;
@@ -230,6 +231,55 @@ public class SubscriptionConsumerLifecycleTest {
Assert.assertTrue(consumer.isFenced());
}
+ @Test
+ public void
testCriticalParallelPollErrorIsDeliveredAfterHealthySiblingMessages()
+ throws Exception {
+ for (final boolean failedTaskFirst : new boolean[] {false, true}) {
+ final TestPullConsumer consumer = new TestPullConsumer();
+ final SubscriptionRuntimeCriticalException gap =
+ new SubscriptionRuntimeCriticalException("WAL replay gap");
+ final SubscriptionMessage message =
+ new SubscriptionMessage(
+ new SubscriptionCommitContext(0, 0, "healthy_topic",
CONSUMER_GROUP_ID, 1L), 1L);
+ final CompletableFuture<List<SubscriptionMessage>> failed = new
CompletableFuture<>();
+ failed.completeExceptionally(gap);
+ final CompletableFuture<List<SubscriptionMessage>> healthy =
+
CompletableFuture.completedFuture(Collections.singletonList(message));
+ Assert.assertEquals(
+ Collections.singletonList(message),
+ consumer.collectMultiplePollResults(
+ failedTaskFirst ? Arrays.asList(failed, healthy) :
Arrays.asList(healthy, failed),
+ new HashSet<>(Arrays.asList("healthy_topic", "gap_topic"))));
+ try {
+ consumer.multiplePoll(Collections.singleton("healthy_topic"), 100L);
+ Assert.fail("A healthy topic must not hide the pending critical WAL
replay error");
+ } catch (final SubscriptionRuntimeCriticalException expected) {
+ Assert.assertSame(gap, expected);
+ }
+ Assert.assertTrue(consumer.multiplePoll(Collections.emptySet(),
100L).isEmpty());
+ Assert.assertFalse(consumer.isFenced());
+ }
+ }
+
+ @Test
+ public void
testCriticalParallelPollErrorWithoutMessagesIsDeliveredImmediately()
+ throws Exception {
+ final TestPullConsumer consumer = new TestPullConsumer();
+ final SubscriptionRuntimeCriticalException gap =
+ new SubscriptionRuntimeCriticalException("WAL replay gap");
+ final CompletableFuture<List<SubscriptionMessage>> failed = new
CompletableFuture<>();
+ failed.completeExceptionally(gap);
+ try {
+ consumer.collectMultiplePollResults(
+
Arrays.asList(CompletableFuture.completedFuture(Collections.emptyList()),
failed),
+ Collections.singleton("gap_topic"));
+ Assert.fail("A critical WAL replay error must fail an otherwise empty
poll");
+ } catch (final SubscriptionRuntimeCriticalException expected) {
+ Assert.assertSame(gap, expected);
+ }
+ Assert.assertTrue(consumer.multiplePoll(Collections.emptySet(),
100L).isEmpty());
+ }
+
@Test
public void testFencedAsyncCommitFailsBeforeReadingMessages() throws
Exception {
final TestPullConsumer consumer = new TestPullConsumer();
diff --git a/iotdb-core/datanode/pom.xml b/iotdb-core/datanode/pom.xml
index 5cd76ae2ac2..48bc0490611 100644
--- a/iotdb-core/datanode/pom.xml
+++ b/iotdb-core/datanode/pom.xml
@@ -177,6 +177,16 @@
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
</dependency>
+ <dependency>
+ <groupId>ch.qos.logback</groupId>
+ <artifactId>logback-classic</artifactId>
+ <scope>runtime</scope>
+ </dependency>
+ <dependency>
+ <groupId>ch.qos.logback</groupId>
+ <artifactId>logback-core</artifactId>
+ <scope>runtime</scope>
+ </dependency>
<dependency>
<groupId>net.java.dev.jna</groupId>
<artifactId>jna</artifactId>
@@ -462,6 +472,12 @@
<ignoredDependency>org.apache.iotdb:isession</ignoredDependency>
<ignoredDependency>at.yawk.lz4:lz4-java</ignoredDependency>
</ignoredDependencies>
+ <ignoredNonTestScopedDependencies>
+ <!-- Logback supplies the server's runtime SLF4J
backend and LogbackMetrics.
+ Direct references in this module also support
test log assertions. -->
+
<ignoredNonTestScopedDependency>ch.qos.logback:logback-classic</ignoredNonTestScopedDependency>
+
<ignoredNonTestScopedDependency>ch.qos.logback:logback-core</ignoredNonTestScopedDependency>
+ </ignoredNonTestScopedDependencies>
</configuration>
</plugin>
<plugin>
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 a8896d1cf41..e6268fbe01e 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
@@ -2218,6 +2218,22 @@ public final class DataNodePipeMessages {
"ProgressWALIterator: skipped {} unreadable retained WAL files in
directory {}, "
+ "firstFile={}, lastFile={}, firstError={}; historical subscription
data in these "
+ "files cannot be replayed";
+ public static final String
+
LOG_PROGRESSWALITERATOR_SKIPPED_UNREADABLE_RETAINED_WAL_FILE_ARG_HISTORICAL_SUBSCRIPTION_DATA_MAY_BE_LOST_AE0DBAB1
=
+ "ProgressWALIterator: skipped unreadable retained WAL file {};
historical subscription data may be lost";
+ public static final String
+
MESSAGE_FILE_ARG_VERSIONID_ARG_FILESEARCHINDEXRANGE_ARG_ARG_ENTRYRANGE_ARG_ARG_SKIPPEDENTRIES_ARG_ERROR_ARG_0D6D77B0
=
+ "file=%s, versionId=%d, fileSearchIndexRange=(%s, %s],
entryRange=[%d, %s), skippedEntries=%s, error=%s";
+ public static final String
+
MESSAGE_SUBSCRIPTION_WAL_REPLAY_SKIPPED_ARG_UNREADABLE_RETAINED_WAL_FILE_S_IN_QUEUE_ARG_REGION_ARG_HISTORICAL_DATA_MAY_BE_LOST_DETAILS_RECENT_FILES_FULL_DETAILS_IN_SERVER_LOG_ARG_FEDEFF7E
=
+ "Subscription WAL replay skipped %d unreadable retained WAL file(s)
in queue %s, region %s. Historical data may be lost. Details (recent files;
full details in server log): %s";
+ public static final String MESSAGE_UNKNOWN_AD921D60 = "unknown";
+ public static final String
+
MESSAGE_WAL_FILE_ARG_VERSIONID_ARG_ENTRYOFFSET_ARG_IS_TEMPORARILY_UNREADABLE_REPLAY_WILL_RETRY_WITHOUT_SKIPPING_ARG_EA11FBDD
=
+ "WAL file %s (versionId=%d, entryOffset=%d) is temporarily
unreadable; replay will retry without skipping: %s";
+ public static final String
+
MESSAGE_FAILED_TO_REOPEN_WAL_FILE_ARG_AT_ENTRY_OFFSET_ARG_ITERATOR_COULD_NOT_SKIP_TO_THE_REQUESTED_POSITION_332B3AD9
=
+ "failed to reopen WAL file %s at entry offset %s: iterator could not
skip to the requested position";
public static final String
LOG_ARG_WAL_GAP_ARG_ARG_SKIPPED_ARG_TOTAL_SKIPPED_ARG_CAUSE_UNDETERMINED_D4E6491A
=
"{}: WAL gap [{}, {}), skipped {}, total skipped {}; cause
undetermined";
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 f9a4df76fb7..64cd9c7a2d6 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
@@ -2059,6 +2059,22 @@ public final class DataNodePipeMessages {
public static final String
PIPE_LOG_PROGRESSWALITERATOR_SKIPPED_UNREADABLE_RETAINED_WAL_FILES_FFC8455E =
"ProgressWALIterator：跳过了 {} 个无法读取的保留 WAL 文件，directory={}，firstFile={}，"
+ "lastFile={}，firstError={}；这些文件中的历史订阅数据无法重放";
+ public static final String
+
LOG_PROGRESSWALITERATOR_SKIPPED_UNREADABLE_RETAINED_WAL_FILE_ARG_HISTORICAL_SUBSCRIPTION_DATA_MAY_BE_LOST_AE0DBAB1
=
+ "ProgressWALIterator：跳过了无法读取的保留 WAL 文件 {}；历史订阅数据可能丢失";
+ public static final String
+
MESSAGE_FILE_ARG_VERSIONID_ARG_FILESEARCHINDEXRANGE_ARG_ARG_ENTRYRANGE_ARG_ARG_SKIPPEDENTRIES_ARG_ERROR_ARG_0D6D77B0
=
+ "file=%s，versionId=%d，fileSearchIndexRange=(%s, %s]，entryRange=[%d,
%s)，skippedEntries=%s，error=%s";
+ public static final String
+
MESSAGE_SUBSCRIPTION_WAL_REPLAY_SKIPPED_ARG_UNREADABLE_RETAINED_WAL_FILE_S_IN_QUEUE_ARG_REGION_ARG_HISTORICAL_DATA_MAY_BE_LOST_DETAILS_RECENT_FILES_FULL_DETAILS_IN_SERVER_LOG_ARG_FEDEFF7E
=
+ "订阅 WAL 重放累计跳过了 %d 个无法读取的保留 WAL 文件，队列 %s，region
%s。历史数据可能丢失。详情（近期文件；完整信息见服务端日志）：%s";
+ public static final String MESSAGE_UNKNOWN_AD921D60 = "未知";
+ public static final String
+
MESSAGE_WAL_FILE_ARG_VERSIONID_ARG_ENTRYOFFSET_ARG_IS_TEMPORARILY_UNREADABLE_REPLAY_WILL_RETRY_WITHOUT_SKIPPING_ARG_EA11FBDD
=
+ "WAL 文件 %s（versionId=%d，entryOffset=%d）暂不可读；重放将重试而不跳过：%s";
+ public static final String
+
MESSAGE_FAILED_TO_REOPEN_WAL_FILE_ARG_AT_ENTRY_OFFSET_ARG_ITERATOR_COULD_NOT_SKIP_TO_THE_REQUESTED_POSITION_332B3AD9
=
+ "重新打开 WAL 文件 %s（条目偏移 %s）失败：迭代器无法跳到请求的位置";
public static final String
LOG_ARG_WAL_GAP_ARG_ARG_SKIPPED_ARG_TOTAL_SKIPPED_ARG_CAUSE_UNDETERMINED_D4E6491A
=
"{}：WAL 缺口 [{}, {})，跳过 {} 条，累计跳过 {} 条；原因尚未确定";
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 af3bf6344a0..e24748edc0b 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
@@ -213,10 +213,16 @@ public class ConsensusPrefetchingQueue {
private final AtomicLong walGapSkippedEntries = new AtomicLong(0);
+ private final AtomicLong unreadableWalFileCount = new AtomicLong(0);
+
+ private final AtomicLong unreadableWalEntryCount = new AtomicLong(0);
+
private final AtomicReference<String> pendingWalReplayError = new
AtomicReference<>();
private int reportedSkippedBrokenWalFileCount = 0;
+ private long reportedSkippedBrokenWalEntryCount = 0L;
+
/**
* Guards queue state transitions that touch replay positioning, seek state,
and writer buffers.
*/
@@ -853,7 +859,7 @@ public class ConsensusPrefetchingQueue {
}
final String walReplayError = pendingWalReplayError.getAndSet(null);
if (Objects.nonNull(walReplayError)) {
- return generateErrorResponse(walReplayError);
+ return generateErrorResponse(walReplayError, true);
}
if (!prefetchInitialized && !initPrefetch(regionProgress)) {
return null;
@@ -2139,12 +2145,24 @@ public class ConsensusPrefetchingQueue {
if (skippedBrokenWalFileCount <= reportedSkippedBrokenWalFileCount) {
return;
}
+ final long skippedBrokenWalEntryCount =
subscriptionWALIterator.getSkippedBrokenWalEntryCount();
+ final long totalSkippedFiles =
+ unreadableWalFileCount.addAndGet(
+ skippedBrokenWalFileCount - reportedSkippedBrokenWalFileCount);
+ unreadableWalEntryCount.addAndGet(
+ skippedBrokenWalEntryCount - reportedSkippedBrokenWalEntryCount);
+ final String detail =
+
subscriptionWALIterator.getSkippedBrokenWalFileDetails(reportedSkippedBrokenWalFileCount);
reportedSkippedBrokenWalFileCount = skippedBrokenWalFileCount;
+ reportedSkippedBrokenWalEntryCount = skippedBrokenWalEntryCount;
pendingWalReplayError.set(
String.format(
DataNodePipeMessages
-
.SUBSCRIPTION_ERROR_WAL_REPLAY_SKIPPED_UNREADABLE_RETAINED_WAL_FILES_D60C5FB2,
- skippedBrokenWalFileCount));
+
.MESSAGE_SUBSCRIPTION_WAL_REPLAY_SKIPPED_ARG_UNREADABLE_RETAINED_WAL_FILE_S_IN_QUEUE_ARG_REGION_ARG_HISTORICAL_DATA_MAY_BE_LOST_DETAILS_RECENT_FILES_FULL_DETAILS_IN_SERVER_LOG_ARG_FEDEFF7E,
+ totalSkippedFiles,
+ getPrefetchingQueueId(),
+ consensusGroupId,
+ detail));
}
private void advanceWalReplayCursorIfPresent(final IndexedConsensusRequest
request) {
@@ -2187,6 +2205,10 @@ public class ConsensusPrefetchingQueue {
return;
}
reportUnreadableWalReplayIfNecessary();
+ if (subscriptionWALIterator.isWaitingForReadableWal()) {
+ // A retry must preserve the incomplete writer request and its
file/entry offset.
+ return;
+ }
if (!(consensusReqReader instanceof WALNode)) {
return;
}
@@ -2209,6 +2231,7 @@ public class ConsensusPrefetchingQueue {
private void resetSubscriptionWALPosition(final long startSearchIndex) {
closeSubscriptionWALIterator();
reportedSkippedBrokenWalFileCount = 0;
+ reportedSkippedBrokenWalEntryCount = 0L;
subscriptionWALIterator = createSubscriptionWALIterator(startSearchIndex);
}
@@ -2227,7 +2250,12 @@ public class ConsensusPrefetchingQueue {
// applies the pending realignment. Returning true keeps the worker
scheduled for that round.
return true;
}
- return Objects.nonNull(subscriptionWALIterator) &&
subscriptionWALIterator.hasNext();
+ if (Objects.isNull(subscriptionWALIterator)) {
+ return false;
+ }
+ final boolean readable = subscriptionWALIterator.hasNext();
+ reportUnreadableWalReplayIfNecessary();
+ return readable;
}
private void requestSubscriptionWalReset(
@@ -2261,6 +2289,7 @@ public class ConsensusPrefetchingQueue {
if (Objects.isNull(subscriptionWALIterator)) {
return;
}
+ reportUnreadableWalReplayIfNecessary();
try {
subscriptionWALIterator.close();
} catch (final IOException e) {
@@ -4155,9 +4184,14 @@ public class ConsensusPrefetchingQueue {
}
private SubscriptionEvent generateErrorResponse(final String errorMessage) {
+ return generateErrorResponse(errorMessage, false);
+ }
+
+ private SubscriptionEvent generateErrorResponse(
+ final String errorMessage, final boolean critical) {
return new SubscriptionEvent(
SubscriptionPollResponseType.ERROR.getType(),
- new ErrorPayload(errorMessage, false),
+ new ErrorPayload(errorMessage, critical),
createNonCommittableContext(IoTDBDescriptor.getInstance().getConfig().getDataNodeId()));
}
@@ -4214,6 +4248,16 @@ public class ConsensusPrefetchingQueue {
return walGapSkippedEntries.get();
}
+ /** Cumulative file-skip occurrences, including replay after an iterator
reset or seek. */
+ public long getUnreadableWalFileCount() {
+ return unreadableWalFileCount.get();
+ }
+
+ /** Known skipped WAL entries/fragments; unreadable footers leave their
entry counts unknown. */
+ public long getUnreadableWalEntryCount() {
+ return unreadableWalEntryCount.get();
+ }
+
public long getEpochChangeCount() {
return runtimeVersionChangeCount.get();
}
@@ -4718,6 +4762,8 @@ public class ConsensusPrefetchingQueue {
result.put("walPathAcceptedEntries",
String.valueOf(getWalPathAcceptedEntries()));
result.put("seekGeneration", String.valueOf(seekGeneration.get()));
result.put("walGapSkippedEntries",
String.valueOf(walGapSkippedEntries.get()));
+ result.put("unreadableWalFileCount",
String.valueOf(getUnreadableWalFileCount()));
+ result.put("unreadableWalEntryCount",
String.valueOf(getUnreadableWalEntryCount()));
result.put("bufferedRealtimeEntryCount",
String.valueOf(getRealtimeBufferedEntryCount()));
result.put("lag", String.valueOf(getLag()));
result.put("isClosed", String.valueOf(isClosed));
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 6572473cb58..cbf0649f224 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
@@ -23,10 +23,12 @@ import
org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeType;
import org.apache.iotdb.commons.request.IConsensusRequest;
import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest;
import org.apache.iotdb.consensus.common.request.IoTConsensusRequest;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.SearchNode;
import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntryType;
import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALInfoEntry;
+import
org.apache.iotdb.db.storageengine.dataregion.wal.exception.BrokenWALFileException;
import org.apache.iotdb.db.storageengine.dataregion.wal.io.ProgressWALReader;
import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALFileVersion;
import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALMetaData;
@@ -64,6 +66,7 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
}
private static final Logger LOGGER =
LoggerFactory.getLogger(ProgressWALIterator.class);
+ private static final int MAX_SKIPPED_FILE_DETAILS = 8;
private static final int SEARCH_INDEX_OFFSET =
WALInfoEntry.FIXED_SERIALIZED_SIZE + PlanNodeType.BYTES;
@@ -82,10 +85,13 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
private boolean currentReaderUsesLiveSnapshot = false;
private int consumedEntryCountInCurrentFile = 0;
private final Set<Long> skippedBrokenWalVersionIds = new HashSet<>();
- private int unreportedSkippedBrokenWalFileCount = 0;
- private String firstUnreportedSkippedBrokenWalFile;
- private String lastUnreportedSkippedBrokenWalFile;
- private String firstUnreportedSkippedBrokenWalError;
+ private final List<String> skippedBrokenWalFileDetails = new ArrayList<>();
+ private long skippedBrokenWalEntryCount;
+ private boolean retryCurrentReader;
+ private long eofRetryVersionId = -1L;
+ private int eofRetryEntryOffset = -1;
+ private long eofRetryFileLength = -1L;
+ private long eofRetryFileModified = -1L;
private IOException lastError;
private boolean incompleteScan = false;
private String incompleteScanDetail;
@@ -194,7 +200,6 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
DataNodePipeMessages.PIPE_LOG_PROGRESSWALITERATOR_ERROR_READING_WAL_2DB46D41,
e);
}
if (nextReady == null) {
- logSkippedBrokenWalFilesIfNecessary();
return false;
}
}
@@ -366,6 +371,23 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
return skippedBrokenWalVersionIds.size();
}
+ long getSkippedBrokenWalEntryCount() {
+ return skippedBrokenWalEntryCount;
+ }
+
+ boolean isWaitingForReadableWal() {
+ return retryCurrentReader;
+ }
+
+ String getSkippedBrokenWalFileDetails(final int firstFileIndex) {
+ final int firstAvailableIndex =
+ getSkippedBrokenWalFileCount() - skippedBrokenWalFileDetails.size();
+ final int start = Math.max(0, firstFileIndex - firstAvailableIndex);
+ return (firstFileIndex < firstAvailableIndex ? "...; " : "")
+ + String.join(
+ "; ", skippedBrokenWalFileDetails.subList(start,
skippedBrokenWalFileDetails.size()));
+ }
+
public boolean hasIncompleteScan() {
return incompleteScan || hasReadError() || hasSkippedBrokenWalFiles();
}
@@ -398,6 +420,21 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
private IndexedConsensusRequest advance() throws IOException {
while (true) {
+ if (retryCurrentReader) {
+ refresh();
+ final int retryFileIndex =
findFileIndexByVersion(currentReaderVersionId);
+ if (retryFileIndex < 0) {
+ // Do not let a temporarily invisible file make replay silently move
to its successor.
+ return null;
+ }
+ if (!openReaderAtIndex(retryFileIndex,
consumedEntryCountInCurrentFile)) {
+ if (retryCurrentReader) {
+ return null;
+ }
+ continue;
+ }
+ retryCurrentReader = false;
+ }
if (currentReader != null && currentReader.hasNext()) {
try {
final ByteBuffer buffer = currentReader.next();
@@ -430,25 +467,21 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
}
continue;
} catch (final EOFException eofException) {
- if (!currentReaderUsesLiveSnapshot) {
- final IndexedConsensusRequest flushed =
skipUnreadableCurrentWalFile(eofException);
- if (flushed != null && !shouldSkip(flushed)) {
- return flushed;
- }
- continue;
+ if (!isConfirmedRetainedWalTruncation()) {
+ deferCurrentReader(eofException);
+ return null;
}
- // Live snapshot metadata may get ahead of the bytes currently
visible in the file. Treat
- // EOF as "this snapshot is exhausted for now" instead of
terminating the iterator.
- final IndexedConsensusRequest flushed = flushPending();
+ final IndexedConsensusRequest flushed =
skipUnreadableCurrentWalFile(eofException);
if (flushed != null && !shouldSkip(flushed)) {
- closeCurrentReader();
return flushed;
}
- if (reopenLiveSnapshotReader()) {
- continue;
- }
- return null;
+ continue;
} catch (final IOException readException) {
+ if (!(readException instanceof BrokenWALFileException)
+ && !isCurrentEntryAboveSizeLimit()) {
+ deferCurrentReader(readException);
+ return null;
+ }
final IndexedConsensusRequest flushed =
skipUnreadableCurrentWalFile(readException);
if (flushed != null && !shouldSkip(flushed)) {
return flushed;
@@ -479,6 +512,9 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
}
if (!openNextReader()) {
+ if (retryCurrentReader) {
+ return null;
+ }
final IndexedConsensusRequest flushed = flushPending();
if (flushed != null && !shouldSkip(flushed)) {
return flushed;
@@ -493,6 +529,9 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
if (openReaderAtIndex(currentFileIndex, 0)) {
return true;
}
+ if (retryCurrentReader) {
+ return false;
+ }
}
return false;
}
@@ -522,10 +561,14 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
if (previousFileIndex < 0) {
return openFirstReaderAfterVersion(currentReaderVersionId);
}
+ final long versionToReopen = currentReaderVersionId;
if (openReaderAtIndex(previousFileIndex, consumedEntryCountInCurrentFile))
{
return true;
}
- return openFirstReaderAfterVersion(currentReaderVersionId);
+ if (retryCurrentReader) {
+ return false;
+ }
+ return openFirstReaderAfterVersion(versionToReopen);
}
private boolean openReaderAtIndex(final int fileIndex, final int
skipEntries) throws IOException {
@@ -546,11 +589,13 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
throws IOException {
final File walFile = walFiles[fileIndex];
final long versionId = walFileVersionIds[fileIndex];
+ currentFileIndex = fileIndex;
final boolean useLiveSnapshot =
liveWalNode != null && versionId ==
liveWalNode.getCurrentWALFileVersion();
+ ProgressWALReader reader = null;
try {
- final ProgressWALReader reader =
+ reader =
useLiveSnapshot
? new ProgressWALReader(
walFile,
@@ -562,10 +607,14 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
reader.close();
markIncompleteScan(
String.format(
- "failed to reopen WAL file %s at entry offset %s: iterator
could not skip to the requested position",
- walFile.getName(), skipEntries),
+ DataNodePipeMessages
+
.MESSAGE_FAILED_TO_REOPEN_WAL_FILE_ARG_AT_ENTRY_OFFSET_ARG_ITERATOR_COULD_NOT_SKIP_TO_THE_REQUESTED_POSITION_332B3AD9,
+ walFile.getName(),
+ skipEntries),
null);
- resetCurrentFileTracking();
+ currentReaderVersionId = versionId;
+ consumedEntryCountInCurrentFile = skipEntries;
+ retryCurrentReader = true;
return false;
}
currentReader = reader;
@@ -573,9 +622,19 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
currentReaderVersionId = versionId;
currentReaderUsesLiveSnapshot = useLiveSnapshot;
consumedEntryCountInCurrentFile = skipEntries;
+ retryCurrentReader = false;
return true;
} catch (final IOException e) {
- if (isNearLiveWalVersion(versionId)) {
+ if (reader != null) {
+ try {
+ reader.close();
+ } catch (final IOException closeException) {
+ e.addSuppressed(closeException);
+ }
+ }
+ if (isNearLiveWalVersion(versionId)
+ || !(e instanceof BrokenWALFileException)
+ || (liveWalNode == null && fileIndex == walFiles.length - 1)) {
LOGGER.debug(
DataNodePipeMessages
.PIPE_LOG_PROGRESSWALITERATOR_FAILED_TO_OPEN_NEAR_LIVE_WAL_FILE_RETRYING_5AEB94AC,
@@ -592,33 +651,123 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
}
markIncompleteScan(
String.format(
- "failed to open near-live WAL file %s while replay scan was
still in progress",
- walFile.getName()),
+ DataNodePipeMessages
+
.MESSAGE_WAL_FILE_ARG_VERSIONID_ARG_ENTRYOFFSET_ARG_IS_TEMPORARILY_UNREADABLE_REPLAY_WILL_RETRY_WITHOUT_SKIPPING_ARG_EA11FBDD,
+ walFile.getAbsolutePath(),
+ versionId,
+ skipEntries,
+ summarizeException(e)),
e);
+ currentReaderVersionId = versionId;
+ consumedEntryCountInCurrentFile = skipEntries;
+ retryCurrentReader = true;
return false;
}
- recordSkippedBrokenWalFile(versionId, walFile, e);
+ recordSkippedBrokenWalFile(versionId, walFile, e, null, skipEntries);
+ pendingRequests.clear();
+ pendingSearchIndex = Long.MIN_VALUE;
+ pendingLocalSeq = Long.MIN_VALUE;
+ resetCurrentFileTracking();
return false;
}
}
private void recordSkippedBrokenWalFile(
- final long versionId, final File walFile, final IOException error) {
+ final long versionId,
+ final File walFile,
+ final IOException error,
+ final WALMetaData metadata,
+ final int firstSkippedEntryOffset) {
if (!skippedBrokenWalVersionIds.add(versionId)) {
return;
}
- if (unreportedSkippedBrokenWalFileCount == 0) {
- firstUnreportedSkippedBrokenWalFile = walFile.getName();
- firstUnreportedSkippedBrokenWalError = summarizeException(error);
- }
- lastUnreportedSkippedBrokenWalFile = walFile.getName();
- unreportedSkippedBrokenWalFileCount++;
- LOGGER.debug(
+ final int entryCount = metadata == null ? -1 :
metadata.getBuffersSize().size();
+ final long skippedEntries =
+ entryCount < 0 ? -1L : Math.max(0L, (long) entryCount -
firstSkippedEntryOffset);
+ if (skippedEntries > 0L) {
+ skippedBrokenWalEntryCount += skippedEntries;
+ }
+ // Filename boundaries are file-level search-index bounds, not an entry
count. Footer entry
+ // ordinals also include request fragments and entries from other writers.
+ final long firstSearchIndex =
WALFileUtils.parseStartSearchIndex(walFile.getName());
+ final int fileIndex = findFileIndexByVersion(versionId);
+ final long lastSearchIndex =
+ fileIndex >= 0 && fileIndex + 1 < walFiles.length
+ ? WALFileUtils.parseStartSearchIndex(walFiles[fileIndex +
1].getName())
+ : -1L;
+ final String detail =
+ String.format(
+ DataNodePipeMessages
+
.MESSAGE_FILE_ARG_VERSIONID_ARG_FILESEARCHINDEXRANGE_ARG_ARG_ENTRYRANGE_ARG_ARG_SKIPPEDENTRIES_ARG_ERROR_ARG_0D6D77B0,
+ walFile.getAbsolutePath(),
+ versionId,
+ formatKnownValue(firstSearchIndex),
+ formatKnownValue(lastSearchIndex),
+ firstSkippedEntryOffset,
+ formatKnownValue(entryCount),
+ formatKnownValue(skippedEntries),
+ summarizeException(error));
+ if (skippedBrokenWalFileDetails.size() == MAX_SKIPPED_FILE_DETAILS) {
+ skippedBrokenWalFileDetails.remove(0);
+ }
+ skippedBrokenWalFileDetails.add(detail);
+ LOGGER.warn(
DataNodePipeMessages
-
.PIPE_LOG_PROGRESSWALITERATOR_FAILED_TO_PROCESS_WAL_FILE_SKIPPING_REMAINING_ENTRIES_093F14A9,
- walFile.getName(),
+
.LOG_PROGRESSWALITERATOR_SKIPPED_UNREADABLE_RETAINED_WAL_FILE_ARG_HISTORICAL_SUBSCRIPTION_DATA_MAY_BE_LOST_AE0DBAB1,
+ detail,
+ error);
+ }
+
+ private static String formatKnownValue(final long value) {
+ return value < 0 ? DataNodePipeMessages.MESSAGE_UNKNOWN_AD921D60 :
String.valueOf(value);
+ }
+
+ private boolean isConfirmedRetainedWalTruncation() {
+ final File walFile = walFiles[currentFileIndex];
+ final boolean sealed =
+ liveWalNode != null
+ ? !isNearLiveWalVersion(currentReaderVersionId)
+ : currentFileIndex + 1 < walFiles.length;
+ final boolean confirmed =
+ !currentReaderUsesLiveSnapshot
+ && sealed
+ && eofRetryVersionId == currentReaderVersionId
+ && eofRetryEntryOffset == consumedEntryCountInCurrentFile
+ && eofRetryFileLength == walFile.length()
+ && eofRetryFileModified == walFile.lastModified();
+ eofRetryVersionId = currentReaderVersionId;
+ eofRetryEntryOffset = consumedEntryCountInCurrentFile;
+ eofRetryFileLength = walFile.length();
+ eofRetryFileModified = walFile.lastModified();
+ return confirmed;
+ }
+
+ private void deferCurrentReader(final IOException error) throws IOException {
+ // Keep the incomplete request and the offset of the first unread entry.
Flushing or reopening
+ // at offset zero here would either deliver a partial request or duplicate
its fragments.
+ markIncompleteScan(
+ String.format(
+ DataNodePipeMessages
+
.MESSAGE_WAL_FILE_ARG_VERSIONID_ARG_ENTRYOFFSET_ARG_IS_TEMPORARILY_UNREADABLE_REPLAY_WILL_RETRY_WITHOUT_SKIPPING_ARG_EA11FBDD,
+ walFiles[currentFileIndex].getAbsolutePath(),
+ currentReaderVersionId,
+ consumedEntryCountInCurrentFile,
+ summarizeException(error)),
error);
+ retryCurrentReader = true;
+ closeCurrentReader();
+ }
+
+ private boolean isCurrentEntryAboveSizeLimit() {
+ final int entryIndex = currentReader.getCurrentEntryIndex();
+ final List<Integer> sizes = currentReader.getMetaData().getBuffersSize();
+ // An entry above the supported limit is deterministically unreadable,
even after reopening.
+ // Preserve the existing large-entry escape path, but report the resulting
data gap explicitly.
+ return entryIndex >= 0
+ && entryIndex < sizes.size()
+ && sizes.get(entryIndex)
+ >
IoTDBDescriptor.getInstance().getConfig().getWalEntrySizeLimitInByte();
}
private IndexedConsensusRequest skipUnreadableCurrentWalFile(final
IOException error) {
@@ -631,6 +780,13 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
failedFileIndex >= 0 && failedFileIndex < walFileVersionIds.length
? walFileVersionIds[failedFileIndex]
: currentReaderVersionId;
+ final WALMetaData failedFileMetadata =
+ currentReader == null ? null : currentReader.getMetaData();
+ final int firstSkippedEntryOffset =
+ Math.max(
+ 0,
+ consumedEntryCountInCurrentFile
+ - (isFailedEntryPartOfPendingRequest() ?
pendingRequests.size() : 0));
final IndexedConsensusRequest flushed =
isFailedEntryPartOfPendingRequest() ? null : flushPending();
@@ -651,7 +807,8 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
error);
return flushed;
}
- recordSkippedBrokenWalFile(versionId, walFile, error);
+ recordSkippedBrokenWalFile(
+ versionId, walFile, error, failedFileMetadata,
firstSkippedEntryOffset);
return flushed;
}
@@ -664,25 +821,6 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
currentReader.getCurrentEntryLocalSeq());
}
- 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()
@@ -707,6 +845,9 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
if (openReaderAtIndex(i, 0)) {
return true;
}
+ if (retryCurrentReader) {
+ return false;
+ }
}
resetCurrentFileTracking();
return false;
@@ -783,6 +924,9 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
currentReaderVersionId = -1L;
currentReaderUsesLiveSnapshot = false;
consumedEntryCountInCurrentFile = 0;
+ retryCurrentReader = false;
+ eofRetryVersionId = -1L;
+ eofRetryEntryOffset = -1;
}
private void markIncompleteScan(final String detail, final IOException
cause) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetrics.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetrics.java
index 0a549565b06..359da8f9239 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetrics.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetrics.java
@@ -116,6 +116,18 @@ public class ConsensusSubscriptionPrefetchingQueueMetrics
implements IMetricSet
queue,
ConsensusPrefetchingQueue::getWalGapSkippedEntries,
key.getTags());
+ metricService.createAutoGauge(
+ Metric.SUBSCRIPTION_CONSENSUS_WAL_FILE_GAP.toString(),
+ MetricLevel.IMPORTANT,
+ queue,
+ ConsensusPrefetchingQueue::getUnreadableWalFileCount,
+ key.getTags());
+ metricService.createAutoGauge(
+ Metric.SUBSCRIPTION_CONSENSUS_WAL_FILE_GAP_ENTRIES.toString(),
+ MetricLevel.IMPORTANT,
+ queue,
+ ConsensusPrefetchingQueue::getUnreadableWalEntryCount,
+ key.getTags());
metricService.createAutoGauge(
Metric.SUBSCRIPTION_CONSENSUS_ROUTING_EPOCH_CHANGE.toString(),
MetricLevel.IMPORTANT,
@@ -309,6 +321,14 @@ public class ConsensusSubscriptionPrefetchingQueueMetrics
implements IMetricSet
MetricType.AUTO_GAUGE, Metric.SUBSCRIPTION_CONSENSUS_LAG.toString(),
key.getTags());
metricService.remove(
MetricType.AUTO_GAUGE,
Metric.SUBSCRIPTION_CONSENSUS_WAL_GAP.toString(), key.getTags());
+ metricService.remove(
+ MetricType.AUTO_GAUGE,
+ Metric.SUBSCRIPTION_CONSENSUS_WAL_FILE_GAP.toString(),
+ key.getTags());
+ metricService.remove(
+ MetricType.AUTO_GAUGE,
+ Metric.SUBSCRIPTION_CONSENSUS_WAL_FILE_GAP_ENTRIES.toString(),
+ key.getTags());
metricService.remove(
MetricType.AUTO_GAUGE,
Metric.SUBSCRIPTION_CONSENSUS_ROUTING_EPOCH_CHANGE.toString(),
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 371bb149c4c..600073d7e3d 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
@@ -391,8 +391,15 @@ public class ConsensusPrefetchingQueueTest {
final ProgressWALIterator iterator = mock(ProgressWALIterator.class);
when(iterator.getSkippedBrokenWalFileCount()).thenReturn(1);
+ when(iterator.getSkippedBrokenWalEntryCount()).thenReturn(5L);
+ when(iterator.getSkippedBrokenWalFileDetails(0))
+ .thenReturn("file=_0-0-1.wal, versionId=0, error=EOFException");
setSubscriptionWalIterator(queue, iterator);
invokeReportUnreadableWalReplayIfNecessary(queue);
+ invokeReportUnreadableWalReplayIfNecessary(queue);
+ assertEquals(1L, queue.getUnreadableWalFileCount());
+ assertEquals(5L, queue.getUnreadableWalEntryCount());
+ assertEquals(0L, queue.getWalGapSkippedEntries());
final SubscriptionEvent event = queue.poll("consumer");
assertNotNull(event);
@@ -400,7 +407,32 @@ public class ConsensusPrefetchingQueueTest {
SubscriptionPollResponseType.ERROR.getType(),
event.getCurrentResponse().getResponseType());
assertTrue(event.getCurrentResponse().getPayload() instanceof
ErrorPayload);
+ final ErrorPayload error = (ErrorPayload)
event.getCurrentResponse().getPayload();
+ assertTrue(error.isCritical());
+ assertTrue(error.getErrorMessage().contains("_0-0-1.wal"));
+ assertTrue(error.getErrorMessage().contains("EOFException"));
+
assertTrue(error.getErrorMessage().contains(queue.getPrefetchingQueueId()));
+
assertTrue(error.getErrorMessage().contains(queue.getConsensusGroupId().toString()));
assertNull(queue.poll("consumer"));
+
+ final Method reset =
+ ConsensusPrefetchingQueue.class.getDeclaredMethod(
+ "resetSubscriptionWALPosition", long.class);
+ reset.setAccessible(true);
+ reset.invoke(queue, 1L);
+ assertEquals(1L, queue.getUnreadableWalFileCount());
+ assertEquals(5L, queue.getUnreadableWalEntryCount());
+ final ProgressWALIterator replacement = mock(ProgressWALIterator.class);
+ when(replacement.getSkippedBrokenWalFileCount()).thenReturn(2);
+ // No readable footer: file skips remain observable even when entry
counts are unknown.
+ when(replacement.getSkippedBrokenWalFileDetails(0))
+ .thenReturn("file=_1-1-1.wal, skippedEntries=unknown,
error=BrokenWALFileException");
+ setSubscriptionWalIterator(queue, replacement);
+ invokeReportUnreadableWalReplayIfNecessary(queue);
+ assertEquals(3L, queue.getUnreadableWalFileCount());
+ assertEquals(5L, queue.getUnreadableWalEntryCount());
+ assertTrue(
+ ((ErrorPayload)
queue.poll("consumer").getCurrentResponse().getPayload()).isCritical());
} finally {
if (queue != null) {
queue.close();
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 c800a2327e9..8af97b31964 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
@@ -32,13 +32,20 @@ import
org.apache.iotdb.db.storageengine.dataregion.wal.node.WALNode;
import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileStatus;
import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileUtils;
+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.junit.Assume;
import org.junit.Test;
+import org.slf4j.LoggerFactory;
import java.io.File;
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@@ -361,6 +368,13 @@ public class ProgressWALIteratorTest {
final File lastWal =
dir.resolve(WALFileUtils.getLogFileName(2, 2,
WALFileStatus.CONTAINS_SEARCH_INDEX))
.toFile();
+ final Logger logger = (Logger)
LoggerFactory.getLogger(ProgressWALIterator.class);
+ final Level originalLevel = logger.getLevel();
+ logger.setLevel(Level.WARN);
+ final ListAppender<ILoggingEvent> appender = new ListAppender<>();
+ appender.setContext(logger.getLoggerContext());
+ appender.start();
+ logger.addAppender(appender);
try {
Files.write(firstBrokenWal.toPath(), new byte[128]);
@@ -372,6 +386,18 @@ public class ProgressWALIteratorTest {
try (ProgressWALIterator iterator = new
ProgressWALIterator(dir.toFile(), Long.MIN_VALUE)) {
assertTrue(iterator.hasNext());
+ // A continuously readable iterator must log every skip before
reaching exhaustion.
+ assertEquals(2, appender.list.size());
+
assertTrue(appender.list.get(0).getFormattedMessage().contains(firstBrokenWal.getName()));
+
assertTrue(appender.list.get(1).getFormattedMessage().contains(secondBrokenWal.getName()));
+
assertTrue(appender.list.get(0).getFormattedMessage().contains("versionId=0"));
+ assertTrue(
+
appender.list.get(0).getFormattedMessage().contains("error=BrokenWALFileException"));
+ assertEquals(0L, iterator.getSkippedBrokenWalEntryCount());
+ final String detail = iterator.getSkippedBrokenWalFileDetails(0);
+ assertTrue(detail.contains(firstBrokenWal.getName()));
+ assertTrue(detail.contains(secondBrokenWal.getName()));
+ assertTrue(detail.contains("skippedEntries=unknown"));
assertEquals(2L, iterator.next().getSearchIndex());
assertFalse(iterator.hasNext());
assertTrue(iterator.hasSkippedBrokenWalFiles());
@@ -380,6 +406,9 @@ public class ProgressWALIteratorTest {
assertFalse(iterator.hasReadError());
}
} finally {
+ logger.detachAppender(appender);
+ logger.setLevel(originalLevel);
+ appender.stop();
Files.deleteIfExists(firstBrokenWal.toPath());
Files.deleteIfExists(secondBrokenWal.toPath());
Files.deleteIfExists(
@@ -421,6 +450,8 @@ public class ProgressWALIteratorTest {
assertEquals(3L, iterator.next().getSearchIndex());
assertFalse(iterator.hasNext());
assertEquals(1, iterator.getSkippedBrokenWalFileCount());
+ assertEquals(1L, iterator.getSkippedBrokenWalEntryCount());
+
assertTrue(iterator.getSkippedBrokenWalFileDetails(0).contains("entryRange=[1,
2)"));
assertTrue(iterator.hasIncompleteScan());
}
} finally {
@@ -459,6 +490,7 @@ public class ProgressWALIteratorTest {
assertEquals(3L, iterator.next().getSearchIndex());
assertFalse(iterator.hasNext());
assertEquals(1, iterator.getSkippedBrokenWalFileCount());
+ assertEquals(2L, iterator.getSkippedBrokenWalEntryCount());
}
} finally {
IoTDBDescriptor.getInstance().getConfig().setWalEntrySizeLimitInByte(originalEntrySizeLimit);
@@ -468,6 +500,178 @@ public class ProgressWALIteratorTest {
}
}
+ @Test
+ public void testRetainedEofRetriesAndPreservesPendingFragments() throws
Exception {
+ verifyEofRetryPreservesPendingFragments(false);
+ }
+
+ @Test
+ public void testNearLiveEofDoesNotBlacklistFileOnRepeatedReads() throws
Exception {
+ verifyEofRetryPreservesPendingFragments(true);
+ }
+
+ private void verifyEofRetryPreservesPendingFragments(final boolean nearLive)
throws Exception {
+ final Path dir =
Files.createTempDirectory("progress-wal-iterator-eof-retry");
+ final File dataWal =
+ dir.resolve(WALFileUtils.getLogFileName(0, 0,
WALFileStatus.CONTAINS_SEARCH_INDEX))
+ .toFile();
+ final File successorWal =
+ dir.resolve(WALFileUtils.getLogFileName(1, 1,
WALFileStatus.CONTAINS_SEARCH_INDEX))
+ .toFile();
+ try {
+ writeFragmentedWal(dataWal, 38);
+ final WALMetaData successorMetadata = singleEntryMeta(19, 2L, 1L, 200L,
7, 2L);
+ try (WALWriter writer = new WALWriter(successorWal, WALFileVersion.V3)) {
+ writer.write(searchableEntry(2L), successorMetadata);
+ }
+ final WALNode walNode = mock(WALNode.class);
+ when(walNode.getLogDirectory()).thenReturn(dir.toFile());
+ when(walNode.getCurrentWALFileVersion()).thenReturn(1L);
+
when(walNode.getCurrentWALMetaDataSnapshot()).thenReturn(successorMetadata);
+ try (ProgressWALIterator iterator =
+ nearLive
+ ? new ProgressWALIterator(walNode, 1L)
+ : new ProgressWALIterator(dir.toFile(), 1L)) {
+ assertFalse(iterator.hasNext());
+ assertTrue(iterator.isWaitingForReadableWal());
+ assertEquals(0, iterator.getSkippedBrokenWalFileCount());
+ if (nearLive) {
+ assertFalse(iterator.hasNext());
+ assertEquals(0, iterator.getSkippedBrokenWalFileCount());
+ }
+ // Simulate bytes/footer becoming readable after the first EOF. The
retry must resume at
+ // entry offset one, retaining the first fragment exactly once.
+ Files.delete(dataWal.toPath());
+ writeFragmentedWal(dataWal, 19);
+ assertTrue(iterator.hasNext());
+ final IndexedConsensusRequest recovered = iterator.next();
+ assertEquals(1L, recovered.getSearchIndex());
+ assertEquals(2, recovered.getRequests().size());
+ assertTrue(iterator.hasNext());
+ assertEquals(2L, iterator.next().getSearchIndex());
+ assertFalse(iterator.hasNext());
+ assertEquals(0, iterator.getSkippedBrokenWalFileCount());
+ }
+ } finally {
+ Files.deleteIfExists(dataWal.toPath());
+ Files.deleteIfExists(successorWal.toPath());
+ Files.deleteIfExists(dir);
+ }
+ }
+
+ @Test
+ public void testRepeatedEofInStableRetainedFileReportsTruncatedRequest()
throws Exception {
+ final Path dir =
Files.createTempDirectory("progress-wal-iterator-confirmed-truncation");
+ final File dataWal =
+ dir.resolve(WALFileUtils.getLogFileName(0, 0,
WALFileStatus.CONTAINS_SEARCH_INDEX))
+ .toFile();
+ final File successorWal =
+ dir.resolve(WALFileUtils.getLogFileName(1, 1,
WALFileStatus.CONTAINS_SEARCH_INDEX))
+ .toFile();
+ try {
+ writeFragmentedWal(dataWal, 38);
+ try (WALWriter writer = new WALWriter(successorWal, WALFileVersion.V3)) {
+ writer.write(searchableEntry(2L), singleEntryMeta(19, 2L, 1L, 200L, 7,
2L));
+ }
+ try (ProgressWALIterator iterator = new
ProgressWALIterator(dir.toFile(), 1L)) {
+ assertFalse(iterator.hasNext());
+ assertEquals(0, iterator.getSkippedBrokenWalFileCount());
+ assertTrue(iterator.hasNext());
+ assertEquals(2L, iterator.next().getSearchIndex());
+ assertEquals(1, iterator.getSkippedBrokenWalFileCount());
+ assertEquals(2L, iterator.getSkippedBrokenWalEntryCount());
+ final String detail = iterator.getSkippedBrokenWalFileDetails(0);
+ assertTrue(detail.contains(dataWal.getName()));
+ assertTrue(detail.contains("fileSearchIndexRange=(0, 1]"));
+ assertTrue(detail.contains("entryRange=[0, 2)"));
+ assertTrue(detail.contains("error=EOFException"));
+ }
+ } finally {
+ Files.deleteIfExists(dataWal.toPath());
+ Files.deleteIfExists(successorWal.toPath());
+ Files.deleteIfExists(dir);
+ }
+ }
+
+ @Test
+ public void
testLiveSnapshotEofPreservesIncompleteRequestUntilBytesAreVisible() throws
Exception {
+ final Path dir =
Files.createTempDirectory("progress-wal-iterator-live-eof-retry");
+ final File liveWal =
+ dir.resolve(WALFileUtils.getLogFileName(0, 0,
WALFileStatus.CONTAINS_SEARCH_INDEX))
+ .toFile();
+ try {
+ final WALMetaData metadata = singleEntryMeta(19, 1L, 1L, 100L, 7, 1L);
+ metadata.add(19, 1L, 1L, 100L, 7, 1L);
+ final WALNode walNode = mock(WALNode.class);
+ when(walNode.getLogDirectory()).thenReturn(dir.toFile());
+ when(walNode.getCurrentWALFileVersion()).thenReturn(0L);
+ when(walNode.getCurrentWALMetaDataSnapshot()).thenReturn(metadata);
+ try (WALWriter writer = new WALWriter(liveWal, WALFileVersion.V3)) {
+ writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 100L, 7,
1L));
+ try (ProgressWALIterator iterator = new ProgressWALIterator(walNode,
1L)) {
+ assertFalse(iterator.hasNext());
+ assertEquals(0, iterator.getSkippedBrokenWalFileCount());
+ writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 100L,
7, 1L));
+ assertTrue(iterator.hasNext());
+ final IndexedConsensusRequest recovered = iterator.next();
+ assertEquals(1L, recovered.getSearchIndex());
+ assertEquals(2, recovered.getRequests().size());
+ assertEquals(0, iterator.getSkippedBrokenWalFileCount());
+ }
+ }
+ } finally {
+ Files.deleteIfExists(liveWal.toPath());
+ Files.deleteIfExists(dir);
+ }
+ }
+
+ private static void writeFragmentedWal(final File walFile, final int
secondEntrySize)
+ throws Exception {
+ try (WALWriter writer = new WALWriter(walFile, WALFileVersion.V3)) {
+ writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 100L, 7,
1L));
+ writer.write(searchableEntry(1L), singleEntryMeta(secondEntrySize, 1L,
1L, 100L, 7, 1L));
+ }
+ }
+
+ @Test
+ public void testSkippedFileDetailsRemainBoundedDuringContinuousReplay()
throws Exception {
+ final Path dir =
Files.createTempDirectory("progress-wal-iterator-bounded-errors");
+ final List<File> files = new ArrayList<>();
+ try {
+ for (int version = 0; version < 10; version++) {
+ final File file =
+ dir.resolve(
+ WALFileUtils.getLogFileName(
+ version, version, WALFileStatus.CONTAINS_SEARCH_INDEX))
+ .toFile();
+ files.add(file);
+ Files.write(file.toPath(), new byte[128]);
+ }
+ final File successor =
+ dir.resolve(WALFileUtils.getLogFileName(10, 10,
WALFileStatus.CONTAINS_SEARCH_INDEX))
+ .toFile();
+ files.add(successor);
+ try (WALWriter writer = new WALWriter(successor, WALFileVersion.V3)) {
+ writer.write(searchableEntry(11L), singleEntryMeta(19, 11L, 1L, 200L,
7, 11L));
+ }
+ try (ProgressWALIterator iterator = new
ProgressWALIterator(dir.toFile(), 1L)) {
+ assertTrue(iterator.hasNext());
+ assertEquals(10, iterator.getSkippedBrokenWalFileCount());
+ final String detail = iterator.getSkippedBrokenWalFileDetails(0);
+ assertEquals(8, detail.split("versionId=").length - 1);
+ assertFalse(detail.contains(files.get(0).getAbsolutePath()));
+ assertTrue(detail.contains(files.get(9).getAbsolutePath()));
+ assertEquals(2,
iterator.getSkippedBrokenWalFileDetails(8).split("versionId=").length - 1);
+ }
+ } finally {
+ for (final File file : files) {
+ Files.deleteIfExists(file.toPath());
+ Files.deleteIfExists(file.toPath().resolveSibling(file.getName() +
".broken"));
+ }
+ Files.deleteIfExists(dir);
+ }
+ }
+
@Test
public void testLiveWalReopenReusesMetadataSnapshot() throws Exception {
final Path dir =
Files.createTempDirectory("progress-wal-iterator-live-snapshot");
@@ -522,7 +726,8 @@ public class ProgressWALIteratorTest {
assertFalse(iterator.hasNext());
assertTrue(iterator.hasIncompleteScan());
assertTrue(iterator.hasReadError());
- assertTrue(iterator.getIncompleteScanDetail().contains("near-live WAL
file"));
+
assertTrue(iterator.getIncompleteScanDetail().contains(brokenLiveWal.getName()));
+ assertEquals(0, iterator.getSkippedBrokenWalFileCount());
}
} finally {
Files.deleteIfExists(brokenLiveWal.toPath());
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetricsTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetricsTest.java
index 782cebe30cd..8038d9cb0ad 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetricsTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetricsTest.java
@@ -30,9 +30,11 @@ import org.apache.iotdb.metrics.utils.MetricLevel;
import org.apache.iotdb.metrics.utils.MetricType;
import org.junit.Test;
+import org.mockito.ArgumentCaptor;
import java.lang.reflect.Field;
import java.util.Map;
+import java.util.function.ToDoubleFunction;
import static org.junit.Assert.assertEquals;
import static org.mockito.ArgumentMatchers.any;
@@ -60,6 +62,10 @@ public class
ConsensusSubscriptionPrefetchingQueueMetricsTest {
when(firstQueue.getConsensusGroupId()).thenReturn(firstRegionId);
when(secondQueue.getPrefetchingQueueId()).thenReturn(queueId);
when(secondQueue.getConsensusGroupId()).thenReturn(secondRegionId);
+ when(firstQueue.getUnreadableWalFileCount()).thenReturn(4L);
+ when(firstQueue.getUnreadableWalEntryCount()).thenReturn(7L);
+ when(secondQueue.getUnreadableWalFileCount()).thenReturn(5L);
+ when(secondQueue.getUnreadableWalEntryCount()).thenReturn(9L);
when(staleFirstQueue.getPrefetchingQueueId()).thenReturn(queueId);
when(staleFirstQueue.getConsensusGroupId()).thenReturn(firstRegionId);
when(metricService.getOrCreateRate(
@@ -103,6 +109,43 @@ public class
ConsensusSubscriptionPrefetchingQueueMetricsTest {
assertEquals(2, queueMap.size());
assertEquals(2, rateMap.size());
+ for (final Metric metric :
+ new Metric[] {
+ Metric.SUBSCRIPTION_CONSENSUS_WAL_FILE_GAP,
+ Metric.SUBSCRIPTION_CONSENSUS_WAL_FILE_GAP_ENTRIES
+ }) {
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ final ArgumentCaptor<ToDoubleFunction<ConsensusPrefetchingQueue>>
mapper =
+ ArgumentCaptor.forClass((Class) ToDoubleFunction.class);
+ verify(metricService)
+ .createAutoGauge(
+ eq(metric.toString()),
+ eq(MetricLevel.IMPORTANT),
+ eq(firstQueue),
+ mapper.capture(),
+ eq(Tag.NAME.toString()),
+ eq(queueId),
+ eq(Tag.REGION.toString()),
+ eq(firstRegionId.toString()));
+ verify(metricService)
+ .createAutoGauge(
+ eq(metric.toString()),
+ eq(MetricLevel.IMPORTANT),
+ eq(secondQueue),
+ mapper.capture(),
+ eq(Tag.NAME.toString()),
+ eq(queueId),
+ eq(Tag.REGION.toString()),
+ eq(secondRegionId.toString()));
+ assertEquals(
+ metric == Metric.SUBSCRIPTION_CONSENSUS_WAL_FILE_GAP ? 4.0 : 7.0,
+ mapper.getAllValues().get(0).applyAsDouble(firstQueue),
+ 0.0);
+ assertEquals(
+ metric == Metric.SUBSCRIPTION_CONSENSUS_WAL_FILE_GAP ? 5.0 : 9.0,
+ mapper.getAllValues().get(1).applyAsDouble(secondQueue),
+ 0.0);
+ }
verify(metricService)
.createAutoGauge(
eq(Metric.SUBSCRIPTION_CONSENSUS_LAG.toString()),
@@ -169,6 +212,28 @@ public class
ConsensusSubscriptionPrefetchingQueueMetricsTest {
metrics.deregister(firstQueue);
assertEquals(1, queueMap.size());
assertEquals(1, rateMap.size());
+ for (final Metric metric :
+ new Metric[] {
+ Metric.SUBSCRIPTION_CONSENSUS_WAL_FILE_GAP,
+ Metric.SUBSCRIPTION_CONSENSUS_WAL_FILE_GAP_ENTRIES
+ }) {
+ verify(metricService)
+ .remove(
+ MetricType.AUTO_GAUGE,
+ metric.toString(),
+ Tag.NAME.toString(),
+ queueId,
+ Tag.REGION.toString(),
+ firstRegionId.toString());
+ verify(metricService, never())
+ .remove(
+ MetricType.AUTO_GAUGE,
+ metric.toString(),
+ Tag.NAME.toString(),
+ queueId,
+ Tag.REGION.toString(),
+ secondRegionId.toString());
+ }
verify(metricService)
.remove(
MetricType.AUTO_GAUGE,
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java
index d412d74d0cc..e20919801e4 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java
@@ -216,6 +216,8 @@ public enum Metric {
SUBSCRIPTION_EVENT_TRANSFER("subscription_event_transfer"),
SUBSCRIPTION_CONSENSUS_LAG("subscription_consensus_lag"),
SUBSCRIPTION_CONSENSUS_WAL_GAP("subscription_consensus_wal_gap"),
+ SUBSCRIPTION_CONSENSUS_WAL_FILE_GAP("subscription_consensus_wal_file_gap"),
+
SUBSCRIPTION_CONSENSUS_WAL_FILE_GAP_ENTRIES("subscription_consensus_wal_file_gap_entries"),
SUBSCRIPTION_CONSENSUS_ROUTING_EPOCH_CHANGE("subscription_consensus_routing_epoch_change"),
SUBSCRIPTION_CONSENSUS_WATERMARK("subscription_consensus_watermark"),
SUBSCRIPTION_CONSENSUS_SEEK_GENERATION("subscription_consensus_seek_generation"),
