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 f58a52fe90c Fix WAL replay stalls on oversized entries (#18728)
f58a52fe90c is described below
commit f58a52fe90c5e40a40bc52a1c3a34b469783d4e3
Author: Caideyipi <[email protected]>
AuthorDate: Tue Sep 29 12:19:41 2026 +0800
Fix WAL replay stalls on oversized entries (#18728)
---
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 12 ++-
.../iotdb/db/i18n/StorageEngineMessages.java | 5 +-
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 11 ++-
.../iotdb/db/i18n/StorageEngineMessages.java | 5 +-
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 11 +++
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 3 +
.../dataregion/wal/buffer/WALBuffer.java | 19 +++++
.../dataregion/wal/io/WALByteBufReader.java | 17 ++--
.../consensus/ConsensusPrefetchingQueue.java | 43 +++++++++-
.../broker/consensus/ProgressWALIterator.java | 61 +++++++++++++-
.../dataregion/wal/buffer/WALBufferCommonTest.java | 27 +++++++
.../dataregion/wal/io/WALFileTest.java | 34 ++++++++
.../consensus/ConsensusPrefetchingQueueTest.java | 63 +++++++++++++++
.../broker/consensus/ProgressWALIteratorTest.java | 92 ++++++++++++++++++++++
.../resources/conf/edge/iotdb-system.properties | 1 +
.../conf/iotdb-system.properties.template | 6 ++
16 files changed, 394 insertions(+), 16 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 3f7330a79ad..59f58407dc4 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
@@ -2220,8 +2220,16 @@ public final class DataNodePipeMessages {
"ProgressWALIterator: failed to open near-live WAL file {}, retrying
without blacklisting";
public static final String
PIPE_LOG_PROGRESSWALITERATOR_ERROR_READING_WAL_2DB46D41 =
"ProgressWALIterator: error reading WAL";
- public static final String
PIPE_LOG_PROGRESSWALITERATOR_FAILED_TO_OPEN_WAL_FILE_SKIPPING_29CA1092 =
- "ProgressWALIterator: failed to open WAL file {}, skipping";
+ public static final String
+
PIPE_LOG_PROGRESSWALITERATOR_FAILED_TO_PROCESS_WAL_FILE_SKIPPING_REMAINING_ENTRIES_093F14A9
=
+ "ProgressWALIterator: failed to process WAL file {}, skipping remaining
entries";
+ public static final String
+
PIPE_LOG_PROGRESSWALITERATOR_FAILED_TO_IDENTIFY_UNREADABLE_WAL_FILE_7BA9F422 =
+ "failed to identify unreadable WAL file during replay";
+ public static final String
+
SUBSCRIPTION_ERROR_WAL_REPLAY_SKIPPED_UNREADABLE_RETAINED_WAL_FILES_D60C5FB2 =
+ "Subscription WAL replay skipped %d unreadable retained WAL file(s).
Historical data in "
+ + "those files cannot be delivered.";
public static final String
PIPE_LOG_PROGRESSWALITERATOR_SKIPPED_UNREADABLE_RETAINED_WAL_FILES_FFC8455E =
"ProgressWALIterator: skipped {} unreadable retained WAL files in
directory {}, "
+ "firstFile={}, lastFile={}, firstError={}; historical subscription
data in these "
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 12ef8456800..032db3e9bb1 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
@@ -349,7 +349,10 @@ public final class StorageEngineMessages {
public static final String FAILED_TO_CREATE_WAL_NODE_AFTER_RETRIES = "Failed
to create WAL node after retries for identifier: ";
public static final String FAIL_TO_CREATE_WAL_NODE = "Fail to create wal
node";
public static final String OVER_SIZED_WAL_ENTRY = "The wal entry size %d
exceeds the limit %d, which may be a result of file corruption or configuration
change."
- + "Please increase wal_buffer_size_in_byte or quarantine the file %s";
+ + "Please increase wal_entry_size_limit_in_byte or quarantine the file
%s";
+ public static final String
+
EXCEPTION_THE_WAL_ENTRY_SIZE_ARG_EXCEEDS_WAL_ENTRY_SIZE_LIMIT_IN_BYTE_ARG_691BB408
=
+ "The WAL entry size %d exceeds wal_entry_size_limit_in_byte %d.";
// ======================== Flush ========================
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 9b124afaffa..48ec49342f3 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
@@ -2062,8 +2062,15 @@ public final class DataNodePipeMessages {
"ProgressWALIterator：打开 near-live WAL 文件 {} 失败，不加入黑名单并重试";
public static final String
PIPE_LOG_PROGRESSWALITERATOR_ERROR_READING_WAL_2DB46D41 =
"ProgressWALIterator：读取 WAL 出错";
- public static final String
PIPE_LOG_PROGRESSWALITERATOR_FAILED_TO_OPEN_WAL_FILE_SKIPPING_29CA1092 =
- "ProgressWALIterator：打开 WAL 文件 {} 失败，跳过该文件";
+ public static final String
+
PIPE_LOG_PROGRESSWALITERATOR_FAILED_TO_PROCESS_WAL_FILE_SKIPPING_REMAINING_ENTRIES_093F14A9
=
+ "ProgressWALIterator：处理 WAL 文件 {} 失败，跳过该文件的剩余条目";
+ public static final String
+
PIPE_LOG_PROGRESSWALITERATOR_FAILED_TO_IDENTIFY_UNREADABLE_WAL_FILE_7BA9F422 =
+ "重放时无法确定不可读的 WAL 文件";
+ public static final String
+
SUBSCRIPTION_ERROR_WAL_REPLAY_SKIPPED_UNREADABLE_RETAINED_WAL_FILES_D60C5FB2 =
+ "订阅 WAL 重放跳过了 %d 个无法读取的保留 WAL 文件。这些文件中的历史数据无法投递。";
public static final String
PIPE_LOG_PROGRESSWALITERATOR_SKIPPED_UNREADABLE_RETAINED_WAL_FILES_FFC8455E =
"ProgressWALIterator：跳过了 {} 个无法读取的保留 WAL 文件，directory={}，firstFile={}，"
+ "lastFile={}，firstError={}；这些文件中的历史订阅数据无法重放";
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 a69ae2a6c8e..43ebfbd35ab 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
@@ -349,7 +349,10 @@ public final class StorageEngineMessages {
public static final String FAILED_TO_CREATE_WAL_NODE_AFTER_RETRIES =
"重试后仍无法创建 WAL 节点，标识符: ";
public static final String FAIL_TO_CREATE_WAL_NODE = "创建 WAL 节点失败";
public static final String OVER_SIZED_WAL_ENTRY = "WAL条目大小 %d 超过阈值 %d,
可能是由于文件损坏或者系统配置变更."
- + "请提高 wal_buffer_size_in_byte 或者隔离该文件 %s";
+ + "请提高 wal_entry_size_limit_in_byte 或者隔离该文件 %s";
+ public static final String
+
EXCEPTION_THE_WAL_ENTRY_SIZE_ARG_EXCEEDS_WAL_ENTRY_SIZE_LIMIT_IN_BYTE_ARG_691BB408
=
+ "WAL 条目大小 %d 超过 wal_entry_size_limit_in_byte %d。";
// ======================== Flush ========================
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index 82a6f397720..fb39bb57a62 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -186,6 +186,9 @@ public class IoTDBConfig implements DefaultEncodingProvider
{
/** Buffer size of each wal node. Unit: byte */
private int walBufferSize = 32 * 1024 * 1024;
+ /** Maximum size of one WAL entry. A non-positive value follows
walBufferSize. Unit: byte */
+ private int walEntrySizeLimitInByte = -1;
+
/** Blocking queue capacity of each delete ahead log buffer */
private int deletionAheadLogBufferQueueCapacity = 500;
@@ -2152,6 +2155,14 @@ public class IoTDBConfig implements
DefaultEncodingProvider {
this.walBufferSize = walBufferSize;
}
+ public int getWalEntrySizeLimitInByte() {
+ return walEntrySizeLimitInByte > 0 ? walEntrySizeLimitInByte :
walBufferSize;
+ }
+
+ public void setWalEntrySizeLimitInByte(int walEntrySizeLimitInByte) {
+ this.walEntrySizeLimitInByte = walEntrySizeLimitInByte;
+ }
+
public int getDeletionAheadLogBufferQueueCapacity() {
return deletionAheadLogBufferQueueCapacity;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
index 098940f8254..a2b8eff3610 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
@@ -1444,6 +1444,9 @@ public class IoTDBDescriptor {
conf.setWalBufferSize(walBufferSize);
}
+ conf.setWalEntrySizeLimitInByte(
+
Integer.parseInt(properties.getProperty("wal_entry_size_limit_in_byte", "-1")));
+
boolean WALInsertNodeCacheShrinkClearEnabled =
Boolean.parseBoolean(
properties.getProperty(
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java
index 27e36f7574c..13930f936d5 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java
@@ -22,6 +22,7 @@ package
org.apache.iotdb.db.storageengine.dataregion.wal.buffer;
import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
import org.apache.iotdb.commons.concurrent.ThreadName;
import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
@@ -68,6 +69,7 @@ import java.util.function.BiConsumer;
import java.util.function.Predicate;
import static
org.apache.iotdb.db.storageengine.dataregion.wal.node.WALNode.DEFAULT_SEARCH_INDEX;
+import static org.apache.iotdb.rpc.TSStatusCode.WAL_ENTRY_TOO_LARGE;
/**
* This buffer guarantees the concurrent safety and uses double buffers
mechanism to accelerate
@@ -247,6 +249,23 @@ public class WALBuffer extends AbstractWALBuffer {
walEntry.getWalFlushListener().fail(new
WALNodeClosedException(identifier));
return;
}
+ if (!walEntry.isSignal() && walEntry.getType() !=
WALEntryType.MEMORY_TABLE_CHECKPOINT) {
+ final int serializedSize = walEntry.serializedSize();
+ final int entrySizeLimit = config.getWalEntrySizeLimitInByte();
+ if (serializedSize > entrySizeLimit) {
+ walEntry
+ .getWalFlushListener()
+ .fail(
+ new IoTDBRuntimeException(
+ String.format(
+ StorageEngineMessages
+
.EXCEPTION_THE_WAL_ENTRY_SIZE_ARG_EXCEEDS_WAL_ENTRY_SIZE_LIMIT_IN_BYTE_ARG_691BB408,
+ serializedSize,
+ entrySizeLimit),
+ WAL_ENTRY_TOO_LARGE.getStatusCode()));
+ return;
+ }
+ }
// just add this WALEntry to queue
try {
walEntries.put(walEntry);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALByteBufReader.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALByteBufReader.java
index 3fa8f23a2ce..f89b1fb5393 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALByteBufReader.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALByteBufReader.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.db.storageengine.dataregion.wal.io;
import org.apache.iotdb.consensus.iot.log.ConsensusReqReader;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.i18n.StorageEngineMessages;
import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntry;
@@ -31,13 +32,13 @@ import java.nio.ByteBuffer;
import java.util.Iterator;
import java.util.List;
-import static
org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALBuffer.ONE_THIRD_WAL_BUFFER_SIZE;
-
/**
* This reader returns {@link WALEntry} as {@link ByteBuffer}, the usage of
WALByteBufReader is like
* {@link Iterator}.
*/
public class WALByteBufReader implements Closeable {
+ private static final int READ_CHUNK_SIZE = 1024 * 1024;
+
private WALMetaData metaData;
private WALInputStream walInputStream;
private DataInputStream logStream;
@@ -86,12 +87,14 @@ public class WALByteBufReader implements Closeable {
public ByteBuffer next() throws IOException {
currentEntryIndex++;
int size = sizeIterator.next();
- if (size > ONE_THIRD_WAL_BUFFER_SIZE) {
+ final int entrySizeLimit =
+ IoTDBDescriptor.getInstance().getConfig().getWalEntrySizeLimitInByte();
+ if (size > entrySizeLimit) {
throw new IOException(
String.format(
StorageEngineMessages.OVER_SIZED_WAL_ENTRY,
size,
- ONE_THIRD_WAL_BUFFER_SIZE,
+ entrySizeLimit,
walInputStream.logFile));
}
// TODO: Reuse this buffer
@@ -101,7 +104,11 @@ public class WALByteBufReader implements Closeable {
logStream.readFully, since this function does not change
the position of the buffer.
*/
- logStream.readFully(buffer.array(), 0, size);
+ for (int offset = 0; offset < size; ) {
+ final int chunkSize = Math.min(READ_CHUNK_SIZE, size - offset);
+ logStream.readFully(buffer.array(), offset, chunkSize);
+ offset += chunkSize;
+ }
return buffer;
}
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 f68d37cc6a2..7f9ba6afd24 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
@@ -89,6 +89,7 @@ import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.function.BiConsumer;
import java.util.function.BooleanSupplier;
@@ -181,6 +182,10 @@ public class ConsensusPrefetchingQueue {
private final AtomicLong walGapSkippedEntries = new AtomicLong(0);
+ private final AtomicReference<String> pendingWalReplayError = new
AtomicReference<>();
+
+ private int reportedSkippedBrokenWalFileCount = 0;
+
/**
* Guards queue state transitions that touch replay positioning, seek state,
and writer buffers.
*/
@@ -749,6 +754,10 @@ public class ConsensusPrefetchingQueue {
if (isClosed || closeRequested || !isActive) {
return null;
}
+ final String walReplayError = pendingWalReplayError.getAndSet(null);
+ if (Objects.nonNull(walReplayError)) {
+ return generateErrorResponse(walReplayError);
+ }
if (!prefetchInitialized && !initPrefetch(regionProgress)) {
return null;
}
@@ -1927,11 +1936,13 @@ public class ConsensusPrefetchingQueue {
}
ensureSubscriptionWalReadable();
+ reportUnreadableWalReplayIfNecessary();
int entriesRead = 0;
while (entriesRead < maxWalEntries
&& subscriptionWALIterator.hasNext()
&& prefetchingQueue.size() < MAX_PREFETCHING_QUEUE_SIZE) {
+ reportUnreadableWalReplayIfNecessary();
try {
final IndexedConsensusRequest walEntry =
subscriptionWALIterator.next();
entriesRead++;
@@ -1968,6 +1979,7 @@ public class ConsensusPrefetchingQueue {
break;
}
}
+ reportUnreadableWalReplayIfNecessary();
if (entriesRead > 0) {
LOGGER.debug(
@@ -1980,6 +1992,22 @@ public class ConsensusPrefetchingQueue {
return MaterializationResult.SUCCESS;
}
+ private void reportUnreadableWalReplayIfNecessary() {
+ if (Objects.isNull(subscriptionWALIterator)) {
+ return;
+ }
+ final int skippedBrokenWalFileCount =
subscriptionWALIterator.getSkippedBrokenWalFileCount();
+ if (skippedBrokenWalFileCount <= reportedSkippedBrokenWalFileCount) {
+ return;
+ }
+ reportedSkippedBrokenWalFileCount = skippedBrokenWalFileCount;
+ pendingWalReplayError.set(
+ String.format(
+ DataNodePipeMessages
+
.SUBSCRIPTION_ERROR_WAL_REPLAY_SKIPPED_UNREADABLE_RETAINED_WAL_FILES_D60C5FB2,
+ skippedBrokenWalFileCount));
+ }
+
private void advanceWalReplayCursorIfPresent(final IndexedConsensusRequest
request) {
if (!hasLocalSearchIndex(request)) {
return;
@@ -2003,14 +2031,24 @@ public class ConsensusPrefetchingQueue {
}
private void ensureSubscriptionWalReadable() {
- if (Objects.isNull(subscriptionWALIterator) ||
subscriptionWALIterator.hasNext()) {
+ if (Objects.isNull(subscriptionWALIterator)) {
+ return;
+ }
+ if (subscriptionWALIterator.hasNext()) {
+ reportUnreadableWalReplayIfNecessary();
return;
}
+ reportUnreadableWalReplayIfNecessary();
// Listing and sorting all retained WAL files is only necessary after the
iterator is
// exhausted. While it still has a readable request, refreshing cannot
affect the next result.
subscriptionWALIterator.refresh();
- if (subscriptionWALIterator.hasNext() || !(consensusReqReader instanceof
WALNode)) {
+ if (subscriptionWALIterator.hasNext()) {
+ reportUnreadableWalReplayIfNecessary();
+ return;
+ }
+ reportUnreadableWalReplayIfNecessary();
+ if (!(consensusReqReader instanceof WALNode)) {
return;
}
@@ -2031,6 +2069,7 @@ public class ConsensusPrefetchingQueue {
private void resetSubscriptionWALPosition(final long startSearchIndex) {
closeSubscriptionWALIterator();
+ reportedSkippedBrokenWalFileCount = 0;
subscriptionWALIterator = createSubscriptionWALIterator(startSearchIndex);
}
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 d9f8ca5e94e..ee61c105c15 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
@@ -431,7 +431,11 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
continue;
} catch (final EOFException eofException) {
if (!currentReaderUsesLiveSnapshot) {
- throw eofException;
+ final IndexedConsensusRequest flushed =
skipUnreadableCurrentWalFile(eofException);
+ if (flushed != null && !shouldSkip(flushed)) {
+ return flushed;
+ }
+ continue;
}
// 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.
@@ -444,6 +448,12 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
continue;
}
return null;
+ } catch (final IOException readException) {
+ final IndexedConsensusRequest flushed =
skipUnreadableCurrentWalFile(readException);
+ if (flushed != null && !shouldSkip(flushed)) {
+ return flushed;
+ }
+ continue;
}
}
@@ -605,11 +615,55 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
lastUnreportedSkippedBrokenWalFile = walFile.getName();
unreportedSkippedBrokenWalFileCount++;
LOGGER.debug(
-
DataNodePipeMessages.PIPE_LOG_PROGRESSWALITERATOR_FAILED_TO_OPEN_WAL_FILE_SKIPPING_29CA1092,
+ DataNodePipeMessages
+
.PIPE_LOG_PROGRESSWALITERATOR_FAILED_TO_PROCESS_WAL_FILE_SKIPPING_REMAINING_ENTRIES_093F14A9,
walFile.getName(),
error);
}
+ private IndexedConsensusRequest skipUnreadableCurrentWalFile(final
IOException error) {
+ final int failedFileIndex = currentFileIndex;
+ final File walFile =
+ failedFileIndex >= 0 && failedFileIndex < walFiles.length
+ ? walFiles[failedFileIndex]
+ : null;
+ final long versionId =
+ failedFileIndex >= 0 && failedFileIndex < walFileVersionIds.length
+ ? walFileVersionIds[failedFileIndex]
+ : currentReaderVersionId;
+ final IndexedConsensusRequest flushed =
+ isFailedEntryPartOfPendingRequest() ? null : flushPending();
+
+ try {
+ closeCurrentReader();
+ } catch (final IOException closeException) {
+ error.addSuppressed(closeException);
+ }
+ pendingRequests.clear();
+ pendingSearchIndex = Long.MIN_VALUE;
+ pendingLocalSeq = Long.MIN_VALUE;
+ resetCurrentFileTracking();
+
+ if (walFile == null || versionId < 0) {
+ markIncompleteScan(
+ DataNodePipeMessages
+
.PIPE_LOG_PROGRESSWALITERATOR_FAILED_TO_IDENTIFY_UNREADABLE_WAL_FILE_7BA9F422,
+ error);
+ return flushed;
+ }
+ recordSkippedBrokenWalFile(versionId, walFile, error);
+ return flushed;
+ }
+
+ private boolean isFailedEntryPartOfPendingRequest() {
+ return currentReader != null
+ && !pendingRequests.isEmpty()
+ && isSamePendingRequest(
+ currentReader.getCurrentEntryPhysicalTime(),
+ currentReader.getCurrentEntryNodeId(),
+ currentReader.getCurrentEntryLocalSeq());
+ }
+
private void logSkippedBrokenWalFilesIfNecessary() {
if (unreportedSkippedBrokenWalFileCount == 0) {
return;
@@ -717,8 +771,9 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
private void closeCurrentReader() throws IOException {
if (currentReader != null) {
- currentReader.close();
+ final ProgressWALReader reader = currentReader;
currentReader = null;
+ reader.close();
}
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBufferCommonTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBufferCommonTest.java
index 253e71e2a02..1d4dcbc8c0e 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBufferCommonTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBufferCommonTest.java
@@ -20,6 +20,7 @@ package
org.apache.iotdb.db.storageengine.dataregion.wal.buffer;
import org.apache.iotdb.calc.exception.QueryProcessException;
import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.commons.exception.IoTDBRuntimeException;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.conf.IoTDBConfig;
@@ -27,6 +28,7 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode;
import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALReader;
import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileUtils;
+import
org.apache.iotdb.db.storageengine.dataregion.wal.utils.listener.WALFlushListener;
import org.apache.iotdb.db.utils.constant.TestConstant;
import org.apache.tsfile.common.conf.TSFileConfig;
@@ -48,7 +50,9 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
+import static org.apache.iotdb.rpc.TSStatusCode.WAL_ENTRY_TOO_LARGE;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
public abstract class WALBufferCommonTest {
@@ -167,11 +171,34 @@ public abstract class WALBufferCommonTest {
public void testHugeWrite() throws Exception {
// use small buffer (only 32 bytes) to simulate huge write request
int prevWalBufferSize = config.getWalBufferSize();
+ int prevWalEntrySizeLimit = config.getWalEntrySizeLimitInByte();
config.setWalBufferSize(32);
+ config.setWalEntrySizeLimitInByte(Integer.MAX_VALUE);
try {
testConcurrentWrite();
} finally {
config.setWalBufferSize(prevWalBufferSize);
+ config.setWalEntrySizeLimitInByte(prevWalEntrySizeLimit);
+ }
+ }
+
+ @Test
+ public void testRejectEntryExceedingConfiguredSizeLimit() throws Exception {
+ final InsertRowNode insertRowNode = getInsertRowNode(devicePath, 1L);
+ final WALEntry walEntry = new WALInfoEntry(1L, insertRowNode, true);
+ final int previousEntrySizeLimit = config.getWalEntrySizeLimitInByte();
+ try {
+ config.setWalEntrySizeLimitInByte(walEntry.serializedSize() - 1);
+ walBuffer.write(walEntry);
+
+ assertEquals(WALFlushListener.Status.FAILURE,
walEntry.getWalFlushListener().waitForResult());
+ assertTrue(walEntry.getWalFlushListener().getCause() instanceof
IoTDBRuntimeException);
+ assertEquals(
+ WAL_ENTRY_TOO_LARGE.getStatusCode(),
+ ((IoTDBRuntimeException)
walEntry.getWalFlushListener().getCause()).getErrorCode());
+ assertTrue(walBuffer.isAllWALEntriesConsumed());
+ } finally {
+ config.setWalEntrySizeLimitInByte(previousEntrySizeLimit);
}
}
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFileTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFileTest.java
index ba34aef1a9e..5c8880686b3 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFileTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFileTest.java
@@ -95,6 +95,40 @@ public class WALFileTest {
Files.deleteIfExists(new File(walFile + ".broken.1").toPath());
}
+ @Test
+ public void testReadEntryLargerThanOneThirdWalBufferWithinIndependentLimit()
throws Exception {
+ final int originalWalBufferSize =
IoTDBDescriptor.getInstance().getConfig().getWalBufferSize();
+ final int originalEntrySizeLimit =
+ IoTDBDescriptor.getInstance().getConfig().getWalEntrySizeLimitInByte();
+ final int entrySize = originalWalBufferSize / 3 + 1;
+ try {
+
IoTDBDescriptor.getInstance().getConfig().setWalEntrySizeLimitInByte(originalWalBufferSize);
+
+ final ByteBuffer entry = ByteBuffer.allocate(entrySize);
+ entry.put(WALEntryType.INSERT_ROW_NODE.getCode());
+ entry.position(entrySize);
+ final WALMetaData metadata = new WALMetaData();
+ metadata.add(entrySize, -1, 42);
+ try (WALWriter writer = new WALWriter(walFile, WALFileVersion.V3)) {
+ writer.write(entry, metadata);
+ }
+
+ try (WALByteBufReader reader = new WALByteBufReader(walFile)) {
+ assertTrue(reader.hasNext());
+ assertArrayEquals(entry.array(), reader.next().array());
+ assertFalse(reader.hasNext());
+ }
+
+
IoTDBDescriptor.getInstance().getConfig().setWalEntrySizeLimitInByte(entrySize
- 1);
+ try (WALByteBufReader reader = new WALByteBufReader(walFile)) {
+ final IOException exception = assertThrows(IOException.class,
reader::next);
+
assertTrue(exception.getMessage().contains("wal_entry_size_limit_in_byte"));
+ }
+ } finally {
+
IoTDBDescriptor.getInstance().getConfig().setWalEntrySizeLimitInByte(originalEntrySizeLimit);
+ }
+ }
+
/** Unexpected channel closure must propagate to the buffer instead of
acknowledging a write. */
@Test
public void testClosedChannelWriteAndForceFail() throws IOException {
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 b2127159672..6dabc8e2a19 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
@@ -44,7 +44,9 @@ import
org.apache.iotdb.db.subscription.agent.SubscriptionConsumerAgent;
import org.apache.iotdb.db.subscription.event.SubscriptionEvent;
import org.apache.iotdb.db.subscription.resource.SubscriptionMemoryManager;
import org.apache.iotdb.rpc.subscription.config.TopicConstant;
+import org.apache.iotdb.rpc.subscription.payload.poll.ErrorPayload;
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.WriterId;
import org.apache.iotdb.rpc.subscription.payload.poll.WriterProgress;
@@ -112,6 +114,59 @@ public class ConsensusPrefetchingQueueTest {
.getModifiers()));
}
+ @Test
+ public void testUnreadableWalReplayIsVisibleAsPollError() throws Exception {
+ final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
+ final File systemDir =
temporaryFolder.newFolder("system-unreadable-wal-poll-error");
+ ConsensusPrefetchingQueue queue = null;
+ try {
+ final FakeConsensusReqReader reader = new FakeConsensusReqReader();
+ final IoTConsensusServerImpl serverImpl =
mock(IoTConsensusServerImpl.class);
+ when(serverImpl.getConsensusReqReader()).thenReturn(reader);
+ when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new
WriterSafeFrontierTracker());
+
+ final ConsensusLogToTabletConverter converter =
mock(ConsensusLogToTabletConverter.class);
+ when(converter.getDatabaseName()).thenReturn("db");
+ when(converter.isTableModel()).thenReturn(true);
+
+ queue =
+ new ConsensusPrefetchingQueue(
+ "consumerGroup",
+ "topic",
+ TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE,
+ new DataRegionId(1),
+ serverImpl,
+ new SubscriptionWalRetentionPolicy(
+ "topic",
+ SubscriptionWalRetentionPolicy.UNBOUNDED,
+ SubscriptionWalRetentionPolicy.UNBOUNDED),
+ converter,
+ newCommitManager(systemDir),
+ new RegionProgress(Collections.emptyMap()),
+ 1L,
+ 1L,
+ true);
+
+ final ProgressWALIterator iterator = mock(ProgressWALIterator.class);
+ when(iterator.getSkippedBrokenWalFileCount()).thenReturn(1);
+ setSubscriptionWalIterator(queue, iterator);
+ invokeReportUnreadableWalReplayIfNecessary(queue);
+
+ final SubscriptionEvent event = queue.poll("consumer");
+ assertNotNull(event);
+ assertEquals(
+ SubscriptionPollResponseType.ERROR.getType(),
+ event.getCurrentResponse().getResponseType());
+ assertTrue(event.getCurrentResponse().getPayload() instanceof
ErrorPayload);
+ assertNull(queue.poll("consumer"));
+ } finally {
+ if (queue != null) {
+ queue.close();
+ }
+
IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
+ }
+ }
+
@Test
public void testPrefetchDoesNotDependOnConsumerMetaReadLock() throws
Exception {
final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
@@ -2272,6 +2327,14 @@ public class ConsensusPrefetchingQueueTest {
method.invoke(queue);
}
+ private static void invokeReportUnreadableWalReplayIfNecessary(
+ final ConsensusPrefetchingQueue queue) throws Exception {
+ final Method method =
+
ConsensusPrefetchingQueue.class.getDeclaredMethod("reportUnreadableWalReplayIfNecessary");
+ method.setAccessible(true);
+ method.invoke(queue);
+ }
+
private static Tablet createTablet() {
final List<String> columnNames = Arrays.asList("device", "temperature");
final List<TSDataType> dataTypes = Arrays.asList(TSDataType.STRING,
TSDataType.DOUBLE);
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 7f816937091..45bff39d6ba 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
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.subscription.broker.consensus;
import
org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeType;
import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
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.io.WALFileVersion;
@@ -311,11 +312,92 @@ public class ProgressWALIteratorTest {
} finally {
Files.deleteIfExists(firstBrokenWal.toPath());
Files.deleteIfExists(secondBrokenWal.toPath());
+ Files.deleteIfExists(
+ firstBrokenWal.toPath().resolveSibling(firstBrokenWal.getName() +
".broken"));
+ Files.deleteIfExists(
+ secondBrokenWal.toPath().resolveSibling(secondBrokenWal.getName() +
".broken"));
Files.deleteIfExists(lastWal.toPath());
Files.deleteIfExists(dir);
}
}
+ @Test
+ public void
testIteratorSkipsUnreadableEntryAndPreservesCompletedPendingRequest()
+ throws Exception {
+ final Path dir =
Files.createTempDirectory("progress-wal-iterator-unreadable-entry");
+ final File dataWal =
+ dir.resolve(WALFileUtils.getLogFileName(0, 0,
WALFileStatus.CONTAINS_SEARCH_INDEX))
+ .toFile();
+ final File successorWal =
+ dir.resolve(WALFileUtils.getLogFileName(1, 3,
WALFileStatus.CONTAINS_SEARCH_INDEX))
+ .toFile();
+ final int originalEntrySizeLimit =
+ IoTDBDescriptor.getInstance().getConfig().getWalEntrySizeLimitInByte();
+
+ try {
+ IoTDBDescriptor.getInstance().getConfig().setWalEntrySizeLimitInByte(64);
+ try (WALWriter writer = new WALWriter(dataWal, WALFileVersion.V3)) {
+ writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 100L, 7,
1L));
+ writer.write(oversizedSearchableEntry(65, 2L), singleEntryMeta(65, 2L,
1L, 200L, 7, 2L));
+ }
+ try (WALWriter writer = new WALWriter(successorWal, WALFileVersion.V3)) {
+ writer.write(searchableEntry(3L), singleEntryMeta(19, 3L, 1L, 300L, 7,
3L));
+ }
+
+ try (ProgressWALIterator iterator = new
ProgressWALIterator(dir.toFile(), Long.MIN_VALUE)) {
+ assertTrue(iterator.hasNext());
+ assertEquals(1L, iterator.next().getSearchIndex());
+ assertTrue(iterator.hasNext());
+ assertEquals(3L, iterator.next().getSearchIndex());
+ assertFalse(iterator.hasNext());
+ assertEquals(1, iterator.getSkippedBrokenWalFileCount());
+ assertTrue(iterator.hasIncompleteScan());
+ }
+ } finally {
+
IoTDBDescriptor.getInstance().getConfig().setWalEntrySizeLimitInByte(originalEntrySizeLimit);
+ Files.deleteIfExists(dataWal.toPath());
+ Files.deleteIfExists(successorWal.toPath());
+ Files.deleteIfExists(dir);
+ }
+ }
+
+ @Test
+ public void
testIteratorDropsPendingRequestWhenUnreadableEntryContinuesSameWriterProgress()
+ throws Exception {
+ final Path dir =
Files.createTempDirectory("progress-wal-iterator-incomplete-request");
+ final File dataWal =
+ dir.resolve(WALFileUtils.getLogFileName(0, 0,
WALFileStatus.CONTAINS_SEARCH_INDEX))
+ .toFile();
+ final File successorWal =
+ dir.resolve(WALFileUtils.getLogFileName(1, 3,
WALFileStatus.CONTAINS_SEARCH_INDEX))
+ .toFile();
+ final int originalEntrySizeLimit =
+ IoTDBDescriptor.getInstance().getConfig().getWalEntrySizeLimitInByte();
+
+ try {
+ IoTDBDescriptor.getInstance().getConfig().setWalEntrySizeLimitInByte(64);
+ try (WALWriter writer = new WALWriter(dataWal, WALFileVersion.V3)) {
+ writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 100L, 7,
1L));
+ writer.write(oversizedSearchableEntry(65, 2L), singleEntryMeta(65, 2L,
1L, 100L, 7, 1L));
+ }
+ try (WALWriter writer = new WALWriter(successorWal, WALFileVersion.V3)) {
+ writer.write(searchableEntry(3L), singleEntryMeta(19, 3L, 1L, 300L, 7,
3L));
+ }
+
+ try (ProgressWALIterator iterator = new
ProgressWALIterator(dir.toFile(), Long.MIN_VALUE)) {
+ assertTrue(iterator.hasNext());
+ assertEquals(3L, iterator.next().getSearchIndex());
+ assertFalse(iterator.hasNext());
+ assertEquals(1, iterator.getSkippedBrokenWalFileCount());
+ }
+ } finally {
+
IoTDBDescriptor.getInstance().getConfig().setWalEntrySizeLimitInByte(originalEntrySizeLimit);
+ Files.deleteIfExists(dataWal.toPath());
+ Files.deleteIfExists(successorWal.toPath());
+ Files.deleteIfExists(dir);
+ }
+ }
+
@Test
public void testLiveWalReopenReusesMetadataSnapshot() throws Exception {
final Path dir =
Files.createTempDirectory("progress-wal-iterator-live-snapshot");
@@ -635,6 +717,16 @@ public class ProgressWALIteratorTest {
return buffer;
}
+ private static ByteBuffer oversizedSearchableEntry(final int size, final
long bodySearchIndex) {
+ final ByteBuffer buffer = ByteBuffer.allocate(size);
+ buffer.put(WALEntryType.INSERT_ROW_NODE.getCode());
+ buffer.putLong(1L);
+ buffer.putShort(PlanNodeType.INSERT_ROW.getNodeType());
+ buffer.putLong(bodySearchIndex);
+ buffer.position(size);
+ return buffer;
+ }
+
private static WALMetaData singleEntryMeta(
final int size,
final long searchIndex,
diff --git
a/iotdb-core/node-commons/src/assembly/resources/conf/edge/iotdb-system.properties
b/iotdb-core/node-commons/src/assembly/resources/conf/edge/iotdb-system.properties
index 82679f99e6a..1ec4ad2fe30 100644
---
a/iotdb-core/node-commons/src/assembly/resources/conf/edge/iotdb-system.properties
+++
b/iotdb-core/node-commons/src/assembly/resources/conf/edge/iotdb-system.properties
@@ -108,6 +108,7 @@ max_allowed_concurrent_queries=100
# ---- memory / file buffers ----
wal_buffer_size_in_byte=1048576
+wal_entry_size_limit_in_byte=-1
group_size_in_byte=4194304
target_chunk_point_num=10000
target_chunk_size=262144
diff --git
a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
index ce806f57541..158a079be6b 100644
---
a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
+++
b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
@@ -1787,6 +1787,12 @@ wal_sync_mode_fsync_delay_in_ms=3
# Datatype: int
wal_buffer_size_in_byte=33554432
+# Maximum size of one WAL entry
+# If it is non-positive, follow wal_buffer_size_in_byte.
+# effectiveMode: restart
+# Datatype: int
+wal_entry_size_limit_in_byte=-1
+
# Whether to cache the sorted WAL file list in each WAL node.
# Enabling this avoids scanning and sorting the WAL directory whenever the
file list is accessed,
# including when a consensus request reader updates its files to search.
