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 25b73c3015274..e669b1a4b8005 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 @@ -20,6 +20,12 @@ package org.apache.iotdb.db.i18n; public final class StorageEngineMessages { + public static final String EXCEPTION_UNSUPPORTED_OR_UNAVAILABLE_WAL_FORMAT_4629CFE4 = + "Unsupported or unavailable WAL format"; + public static final String EXCEPTION_MULTIPLE_WAL_FORMAT_PROVIDERS_ARE_CONFIGURED_DD78877D = + "Multiple WAL format providers are configured"; + public static final String EXCEPTION_INVALID_WAL_LOGICAL_POSITION_8CF34AFE = + "Invalid WAL logical position"; public static final String LOG_REBUILT_UNREADABLE_WAL_FOOTER_FILE_ARG_VERSION_ARG_ORIGINALFILEBYTES_ARG_WRITERPROGRESS_ARG_1DBF836E = "Rebuilt unreadable WAL footer: file={}, version={}, originalFileBytes={}, writerProgress={}."; 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 93f9b29b6f8d9..5f11f94560ddf 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 @@ -20,6 +20,12 @@ package org.apache.iotdb.db.i18n; public final class StorageEngineMessages { + public static final String EXCEPTION_UNSUPPORTED_OR_UNAVAILABLE_WAL_FORMAT_4629CFE4 = + "WAL 格式不支持或对应实现不可用"; + public static final String EXCEPTION_MULTIPLE_WAL_FORMAT_PROVIDERS_ARE_CONFIGURED_DD78877D = + "配置了多个 WAL 格式 Provider"; + public static final String EXCEPTION_INVALID_WAL_LOGICAL_POSITION_8CF34AFE = + "WAL 逻辑位置无效"; public static final String LOG_REBUILT_UNREADABLE_WAL_FOOTER_FILE_ARG_VERSION_ARG_ORIGINALFILEBYTES_ARG_WRITERPROGRESS_ARG_1DBF836E = "已重建不可读的 WAL footer：file={}，version={}，originalFileBytes={}，writerProgress={}。"; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java index f66a76c9f951b..36bb9679f8107 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java @@ -77,6 +77,7 @@ import org.apache.iotdb.db.storageengine.dataregion.flush.TsFileFlushPolicy.DirectFlushPolicy; import org.apache.iotdb.db.storageengine.dataregion.wal.WALManager; import org.apache.iotdb.db.storageengine.dataregion.wal.exception.WALException; +import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALFormatException; import org.apache.iotdb.db.storageengine.dataregion.wal.recover.WALRecoverManager; import org.apache.iotdb.db.storageengine.load.LoadTsFilePieceNodeAssembler; import org.apache.iotdb.db.storageengine.load.active.ActiveLoadAgent; @@ -231,6 +232,13 @@ private void asyncRecoverDataRegion() throws StartupException { WALRecoverManager.getInstance().recover(); } catch (WALException e) { LOGGER.error(StorageEngineMessages.FAIL_TO_RECOVER_WAL, e); + if (WALFormatException.isFatal(e)) { + cachedThreadPool.shutdownNow(); + StartupException failure = + new StartupException(StorageEngineMessages.FAIL_TO_RECOVER_WAL); + failure.initCause(e); + throw failure; + } } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/CheckpointReader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/CheckpointReader.java index 7069781fd8bfe..dfd13cfd571ea 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/CheckpointReader.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/CheckpointReader.java @@ -29,6 +29,7 @@ import java.io.EOFException; import java.io.File; import java.io.IOException; +import java.io.UncheckedIOException; import java.util.ArrayList; import java.util.List; @@ -53,6 +54,8 @@ private void init() { Checkpoint checkpoint = Checkpoint.deserialize(logStream); checkpoints.add(checkpoint); } + } catch (WALFormatException e) { + throw new UncheckedIOException(e); } catch (EOFException e) { logger.debug( StorageEngineMessages @@ -60,6 +63,9 @@ private void init() { logFile, e.getMessage()); } catch (IOException e) { + if (WALFormatException.isFatal(e)) { + throw new UncheckedIOException(e); + } logger.warn( StorageEngineMessages .STORAGE_LOG_MEET_ERROR_WHEN_READING_CHECKPOINT_FILE_SKIP_BROKEN_CHECKPOINTS_DADF9E9D, diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/CheckpointWriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/CheckpointWriter.java index f31f2e9b846df..1f8cd94336062 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/CheckpointWriter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/CheckpointWriter.java @@ -27,6 +27,6 @@ /** CheckpointWriter writes the binary {@link Checkpoint} into .checkpoint file. */ public class CheckpointWriter extends LogWriter { public CheckpointWriter(File logFile) throws IOException { - super(logFile, WALFileVersion.V2); + super(logFile, WALFormats.writeVersion(WALFileVersion.V2)); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/LogWriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/LogWriter.java index 3652c826682dd..abf96cb19c3c4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/LogWriter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/LogWriter.java @@ -41,6 +41,9 @@ public abstract class LogWriter implements ILogWriter { protected final File logFile; protected final FileChannel logChannel; protected long originalSize = 0; + protected WALFileContext fileContext; + protected long blockIndex; + protected long dataStartOffset; /** * 1 byte for whether enable compression, 4 byte for compressedSize, 4 byte for uncompressedSize @@ -68,14 +71,17 @@ protected LogWriter(File logFile, WALFileVersion version) throws IOException { StandardOpenOption.WRITE, StandardOpenOption.APPEND); try { - if (logChannel.size() == 0 + if (version == WALFileVersion.EXTENDED) { + fileContext = WALFormats.create(logFile, logChannel); + } else if (logChannel.size() == 0 && (version == WALFileVersion.V2 || version == WALFileVersion.V3)) { ByteBuffer magic = ByteBuffer.wrap(version.getVersionBytes()); while (magic.hasRemaining()) { logChannel.write(magic); } } - } catch (IOException e) { + dataStartOffset = logChannel.position(); + } catch (IOException | RuntimeException | Error e) { // A full disk can fail initialization after open() succeeds. Release the orphan channel // before the owner retries creating the successor. try { @@ -83,6 +89,13 @@ protected LogWriter(File logFile, WALFileVersion version) throws IOException { } catch (IOException closeException) { e.addSuppressed(closeException); } + if (fileContext != null) { + try { + fileContext.close(); + } catch (RuntimeException | Error closeException) { + e.addSuppressed(closeException); + } + } throw e; } } @@ -132,6 +145,19 @@ public double write(ByteBuffer buffer, boolean allowCompress) throws IOException WritingMetrics.getInstance().recordCompressWALBufferCost(System.nanoTime() - startTime); } startTime = System.nanoTime(); + if (fileContext != null) { + int written = + fileContext.writeData( + logChannel, + blockIndex, + compressed ? compressionType : CompressionType.UNCOMPRESSED, + uncompressedSize, + buffer); + blockIndex++; + WritingMetrics.getInstance() + .recordWroteWALBuffer(uncompressedSize, written, System.nanoTime() - startTime); + return ((double) written / uncompressedSize); + } headerBuffer.flip(); while (headerBuffer.hasRemaining()) { logChannel.write(headerBuffer); @@ -183,7 +209,13 @@ public void close() throws IOException { logChannel.force(true); } } finally { - logChannel.close(); + try { + logChannel.close(); + } finally { + if (fileContext != null) { + fileContext.close(); + } + } } } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFileContext.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFileContext.java new file mode 100644 index 0000000000000..8c925cf9e685e --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFileContext.java @@ -0,0 +1,56 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iotdb.db.storageengine.dataregion.wal.io; + +import org.apache.tsfile.file.metadata.enums.CompressionType; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; + +/** + * One extension-format file. The caller owns the channel; the context owns only format resources. + * Read methods return decoded segments and must distinguish an incomplete tail (EOFException) from + * an unrecoverable format failure (WALFormatException). + */ +public interface WALFileContext extends AutoCloseable { + int writeData( + FileChannel channel, + long index, + CompressionType compression, + int uncompressedSize, + ByteBuffer data) + throws IOException; + + /** Returns an uncompressed segment, or throws EOFException at metadata/the incomplete tail. */ + ByteBuffer readSegment(FileChannel channel, long endOffset, long index) throws IOException; + + /** Writes the format-specific metadata frame and trailer after the final data segment. */ + void seal(FileChannel channel, WALMetaData metadata, long index) throws IOException; + + /** + * Releases a returned segment, including any sensitive scratch state; accepts null. The provider + * owns its allocation and may recycle it. The caller must not separately free its native memory. + */ + void release(ByteBuffer data); + + /** Releases format resources; repeated calls must be safe. The channel belongs to the caller. */ + @Override + void close(); +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFileVersion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFileVersion.java index f643c78e8dddb..c51821b3fb44f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFileVersion.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFileVersion.java @@ -19,6 +19,7 @@ package org.apache.iotdb.db.storageengine.dataregion.wal.io; import org.apache.iotdb.commons.utils.IOUtils; +import org.apache.iotdb.db.i18n.StorageEngineMessages; import java.io.File; import java.io.IOException; @@ -29,19 +30,30 @@ public enum WALFileVersion { V1("WAL"), V2("V2-WAL"), - V3("V3-WAL"); + V3("V3-WAL"), + /** A dispatch marker only; its on-disk magic and layout belong to the installed provider. */ + EXTENDED(null); private final String versionString; private byte[] versionBytes; public String getVersionString() { + requireBuiltInVersion(); return versionString; } public byte[] getVersionBytes() { + requireBuiltInVersion(); return versionBytes; } + private void requireBuiltInVersion() { + if (this == EXTENDED) { + throw new IllegalStateException( + StorageEngineMessages.EXCEPTION_UNSUPPORTED_OR_UNAVAILABLE_WAL_FORMAT_4629CFE4); + } + } + WALFileVersion(String versionString) { this.versionString = versionString; if (versionString != null) { @@ -73,6 +85,9 @@ public static WALFileVersion getVersion(FileChannel channel) throws IOException return version; } } + if (WALFormats.recognizes(channel)) { + return EXTENDED; + } // v1 by default return V1; } finally { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFormatException.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFormatException.java new file mode 100644 index 0000000000000..100a68e1be4d5 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFormatException.java @@ -0,0 +1,54 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iotdb.db.storageengine.dataregion.wal.io; + +import org.apache.iotdb.db.i18n.StorageEngineMessages; + +import java.io.IOException; +import java.util.Collections; +import java.util.IdentityHashMap; +import java.util.Set; + +/** A format/policy failure that recovery must never downgrade to tail truncation or skip. */ +public class WALFormatException extends IOException { + public WALFormatException(String message) { + super(message); + } + + public WALFormatException(String message, Throwable cause) { + super(message, cause); + } + + public static WALFormatException unsupported() { + return new WALFormatException( + StorageEngineMessages.EXCEPTION_UNSUPPORTED_OR_UNAVAILABLE_WAL_FORMAT_4629CFE4); + } + + public static boolean isFatal(Throwable failure) { + Set seen = Collections.newSetFromMap(new IdentityHashMap<>()); + for (Throwable current = failure; + current != null && seen.add(current); + current = current.getCause()) { + if (current instanceof WALFormatException) { + return true; + } + } + return false; + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFormatProvider.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFormatProvider.java new file mode 100644 index 0000000000000..17ada89c5ab81 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFormatProvider.java @@ -0,0 +1,51 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iotdb.db.storageengine.dataregion.wal.io; + +import java.io.File; +import java.io.IOException; +import java.nio.channels.FileChannel; + +/** Optional WAL/checkpoint format implementation, without prescribing algorithms or key storage. */ +public interface WALFormatProvider { + /** + * Stable magic in the reserved {@code V-WAL} namespace, independent of the write policy. + * It must not reuse a built-in version, so missing providers can be rejected without guessing V1. + */ + byte[] getMagic(); + + boolean isEnabled(); + + /** Initializes a new file, including its magic and header; does not take channel ownership. */ + WALFileContext create(File file, FileChannel channel) throws IOException; + + /** Opens an existing extension-format file, leaving the channel at its first segment. */ + WALFileContext open(File file, FileChannel channel) throws IOException; + + WALMetaData readMetadata(File file, FileChannel channel) throws IOException; + + /** Enforces product policy before opening either a legacy or extension-format file. */ + default void validateRead(File file, WALFileVersion version) throws IOException {} + + /** Checks the file before replay or cleanup may have destructive side effects. */ + void validateForRecovery(File file) throws IOException; + + /** Product-specific diagnostics, called once when recovery terminates with a fatal failure. */ + default void onRecoveryFailure(Throwable failure) {} +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFormats.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFormats.java new file mode 100644 index 0000000000000..8691e46709691 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFormats.java @@ -0,0 +1,149 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iotdb.db.storageengine.dataregion.wal.io; + +import org.apache.iotdb.commons.utils.IOUtils; +import org.apache.iotdb.db.i18n.StorageEngineMessages; + +import java.io.File; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.Iterator; +import java.util.ServiceLoader; + +/** Dispatches optional formats without exposing their binary layouts to the standard WAL path. */ +public final class WALFormats { + private WALFormats() {} + + private static class Holder { + private static final WALFormatProvider PROVIDER = + selectProvider(ServiceLoader.load(WALFormatProvider.class).iterator()); + } + + static WALFormatProvider selectProvider(Iterator providers) { + if (!providers.hasNext()) { + return null; + } + WALFormatProvider provider = providers.next(); + if (providers.hasNext()) { + throw new IllegalStateException( + StorageEngineMessages.EXCEPTION_MULTIPLE_WAL_FORMAT_PROVIDERS_ARE_CONFIGURED_DD78877D); + } + byte[] magic = provider.getMagic(); + if (magic == null + || magic.length > 64 + || !new String(magic, StandardCharsets.US_ASCII).matches("V[0-9]+-WAL") + || Arrays.equals(magic, WALFileVersion.V2.getVersionBytes()) + || Arrays.equals(magic, WALFileVersion.V3.getVersionBytes())) { + throw new IllegalStateException( + StorageEngineMessages.EXCEPTION_UNSUPPORTED_OR_UNAVAILABLE_WAL_FORMAT_4629CFE4); + } + return provider; + } + + public static WALFileVersion writeVersion(WALFileVersion requested) { + return Holder.PROVIDER != null && Holder.PROVIDER.isEnabled() + ? WALFileVersion.EXTENDED + : requested; + } + + static boolean recognizes(FileChannel channel) throws IOException { + return recognizes(channel, Holder.PROVIDER); + } + + static boolean recognizes(FileChannel channel, WALFormatProvider provider) throws IOException { + long position = channel.position(); + try { + if (provider != null) { + byte[] magic = provider.getMagic(); + if (channel.size() >= magic.length) { + ByteBuffer bytes = ByteBuffer.allocate(magic.length); + IOUtils.readFully(channel, bytes, 0); + if (Arrays.equals(magic, bytes.array())) { + return true; + } + } + } + // Versioned magic is reserved: an unavailable future format must not fall back to V1. + int size = (int) Math.min(channel.size(), 64); + ByteBuffer prefix = ByteBuffer.allocate(size); + IOUtils.readFully(channel, prefix, 0); + if (new String(prefix.array(), StandardCharsets.US_ASCII).matches("(?s)^V[0-9]+-WAL.*")) { + throw WALFormatException.unsupported(); + } + return false; + } finally { + channel.position(position); + } + } + + public static WALFileContext create(File file, FileChannel channel) throws IOException { + if (Holder.PROVIDER == null || !Holder.PROVIDER.isEnabled()) { + throw WALFormatException.unsupported(); + } + return requireContext(Holder.PROVIDER.create(file, channel)); + } + + public static WALFileContext open(File file, FileChannel channel) throws IOException { + if (Holder.PROVIDER == null) { + throw WALFormatException.unsupported(); + } + return requireContext(Holder.PROVIDER.open(file, channel)); + } + + private static WALFileContext requireContext(WALFileContext context) throws WALFormatException { + if (context == null) { + throw WALFormatException.unsupported(); + } + return context; + } + + public static void validateRead(File file, WALFileVersion version) throws IOException { + if (Holder.PROVIDER != null) { + Holder.PROVIDER.validateRead(file, version); + } + } + + public static WALMetaData readMetadata(File file, FileChannel channel) throws IOException { + if (Holder.PROVIDER == null) { + throw WALFormatException.unsupported(); + } + Holder.PROVIDER.validateRead(file, WALFileVersion.EXTENDED); + return Holder.PROVIDER.readMetadata(file, channel); + } + + public static void validateForRecovery(File file) throws IOException { + try (FileChannel channel = FileChannel.open(file.toPath())) { + WALFileVersion version = WALFileVersion.getVersion(channel); + validateRead(file, version); + if (version == WALFileVersion.EXTENDED) { + Holder.PROVIDER.validateForRecovery(file); + } + } + } + + public static void onRecoveryFailure(Throwable failure) { + if (Holder.PROVIDER != null) { + Holder.PROVIDER.onRecoveryFailure(failure); + } + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java index d4b4a8b78f9e2..7a512ff5a95ab 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java @@ -53,6 +53,7 @@ public class WALInputStream extends InputStream implements AutoCloseable { private ByteBuffer dataBuffer = null; private ByteBuffer compressedBuffer = null; private long currentSegmentOffset = -1; + private long currentSegmentBlockIndex; private long suspendedSegmentOffset = -1; private int suspendedSegmentPosition; private int suspendedSegmentRemaining; @@ -67,6 +68,9 @@ public class WALInputStream extends InputStream implements AutoCloseable { private long endOffset = -1; private long logicalReadOffset; private boolean recoveringEntries; + private WALFileContext fileContext; + private long dataStartOffset; + private long blockIndex; WALFileVersion version; @@ -81,7 +85,12 @@ public WALInputStream(File logFile) throws IOException { try { fileSize = channel.size(); analyzeFileVersion(); - if (ignoreMetadata) { + WALFormats.validateRead(logFile, version); + if (version == WALFileVersion.EXTENDED) { + fileContext = WALFormats.open(logFile, channel); + dataStartOffset = channel.position(); + endOffset = fileSize; + } else if (ignoreMetadata) { // Recovery must not trust even a plausible footer length. Stop at the entry end marker // or the first unreadable entry instead. endOffset = fileSize; @@ -89,8 +98,23 @@ public WALInputStream(File logFile) throws IOException { } else { getEndOffset(); } - } catch (Exception e) { - channel.close(); + } catch (IOException | RuntimeException | Error e) { + try { + channel.close(); + } catch (IOException closeException) { + e.addSuppressed(closeException); + } + if (fileContext != null) { + try { + fileContext.close(); + } catch (RuntimeException | Error closeException) { + e.addSuppressed(closeException); + } + } + if (version == WALFileVersion.EXTENDED && !WALFormatException.isFatal(e)) { + throw new WALFormatException( + StorageEngineMessages.EXCEPTION_UNSUPPORTED_OR_UNAVAILABLE_WAL_FORMAT_4629CFE4, e); + } throw e; } } @@ -212,9 +236,18 @@ public void close() throws IOException { try { channel.close(); } finally { - releaseReadBuffers(); - suspendedSegmentOffset = -1; - suspendedSegmentRemaining = 0; + try { + releaseReadBuffers(); + } finally { + try { + if (fileContext != null) { + fileContext.close(); + } + } finally { + suspendedSegmentOffset = -1; + suspendedSegmentRemaining = 0; + } + } } } @@ -229,7 +262,7 @@ public void releaseReadBuffers() { suspendedSegmentPosition = dataBuffer.position(); suspendedSegmentRemaining = dataBuffer.remaining(); } - MmapUtil.clean(dataBuffer); + releaseSegment(); MmapUtil.clean(compressedBuffer); dataBuffer = null; compressedBuffer = null; @@ -249,13 +282,18 @@ public int available() throws IOException { private void loadNextSegment() throws IOException { if (suspendedSegmentOffset >= 0) { channel.position(suspendedSegmentOffset); + if (version == WALFileVersion.EXTENDED) { + blockIndex = currentSegmentBlockIndex; + } } if (channel.position() >= endOffset) { throw new EOFException(StorageEngineMessages.REACH_END_OFFSET_OF_WAL_FILE); } long startTime = System.nanoTime(); long startPosition = channel.position(); - if (version == WALFileVersion.V2 || version == WALFileVersion.V3) { + if (version == WALFileVersion.EXTENDED) { + loadNextExtendedSegment(); + } else if (version == WALFileVersion.V2 || version == WALFileVersion.V3) { loadNextSegmentV2(); } else if (version == WALFileVersion.V1) { loadNextSegmentV1(); @@ -275,6 +313,30 @@ private void loadNextSegment() throws IOException { .recordWALRead(channel.position() - startPosition, System.nanoTime() - startTime); } + private void loadNextExtendedSegment() throws IOException { + releaseSegment(); + currentSegmentOffset = channel.position(); + currentSegmentBlockIndex = blockIndex; + dataBuffer = fileContext.readSegment(channel, endOffset, blockIndex); + if (dataBuffer == null || !dataBuffer.hasRemaining()) { + throw WALFormatException.unsupported(); + } + blockIndex++; + } + + private void releaseSegment() { + ByteBuffer segment = dataBuffer; + dataBuffer = null; + if (segment == null) { + return; + } + if (fileContext != null) { + fileContext.release(segment); + } else { + MmapUtil.clean(segment); + } + } + private void loadNextSegmentV1() throws IOException { // just read raw data as input if (channel.position() >= fileSize) { @@ -389,6 +451,25 @@ public void skipToGivenLogicalPosition(long pos) throws IOException { releaseReadBuffers(); suspendedSegmentOffset = -1; suspendedSegmentRemaining = 0; + if (version == WALFileVersion.EXTENDED) { + if (pos < 0) { + throw new WALFormatException( + StorageEngineMessages.EXCEPTION_INVALID_WAL_LOGICAL_POSITION_8CF34AFE); + } + channel.position(dataStartOffset); + blockIndex = 0; + logicalReadOffset = 0; + long remaining = pos; + while (true) { + loadNextExtendedSegment(); + if (remaining < dataBuffer.remaining()) { + dataBuffer.position((int) remaining); + logicalReadOffset = pos; + return; + } + remaining -= dataBuffer.remaining(); + } + } if (version == WALFileVersion.V2 || version == WALFileVersion.V3) { channel.position(version.getVersionBytes().length); long posRemain = pos; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALMetaData.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALMetaData.java index 5d316a7508f9a..0cf97ba88095e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALMetaData.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALMetaData.java @@ -412,6 +412,9 @@ public static WALMetaData readFromWALFile(File logFile, FileChannel channel) thr try { return readFromWALFileWithoutRecovery(logFile, channel); } catch (IOException metadataFailure) { + if (WALFormatException.isFatal(metadataFailure)) { + throw metadataFailure; + } logger.warn(StorageEngineMessages.FAIL_TO_READ_WAL_LOGS_SKIP, logFile, metadataFailure); // Keep the original file when some entries are readable. The reconstructed metadata is an // in-memory view of that prefix; rewriting compressed segments while readers hold the file @@ -457,6 +460,9 @@ public static WALMetaData readFromWALFileWithoutRecovery(File logFile, FileChann return new WALMetaData(); } WALFileVersion version = WALFileVersion.getVersion(channel); + if (version == WALFileVersion.EXTENDED) { + return WALFormats.readMetadata(logFile, channel); + } if (channel.size() < version.getVersionBytes().length + Integer.BYTES || !isValidMagicString(channel, version)) { throw new BrokenWALFileException(logFile); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALReader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALReader.java index c54f3e45bc462..a214b25211cc8 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALReader.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALReader.java @@ -32,6 +32,7 @@ import java.io.EOFException; import java.io.File; import java.io.IOException; +import java.io.UncheckedIOException; import java.util.Iterator; import java.util.NoSuchElementException; @@ -85,16 +86,24 @@ public boolean hasNext() { nextEntry = null; return false; } + } catch (WALFormatException e) { + throw new UncheckedIOException(e); } catch (EOFException e) { fileCorrupted = true; return false; } catch (IOException e) { + if (WALFormatException.isFatal(e)) { + throw new UncheckedIOException(e); + } fileCorrupted = true; // log only when file should be complete if (!fileMayCorrupt) { logger.warn(StorageEngineMessages.FAIL_TO_READ_WAL_ENTRY_SKIP_BROKEN, logFile, e); } } catch (Exception e) { + if (WALFormatException.isFatal(e)) { + throw new IllegalStateException(e); + } fileCorrupted = true; logger.warn(StorageEngineMessages.FAIL_TO_READ_WAL_ENTRY_SKIP_BROKEN, logFile, e); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALWriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALWriter.java index d340b7ce0df1c..2c3a1c27acba6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALWriter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALWriter.java @@ -42,8 +42,8 @@ public WALWriter(File logFile) throws IOException { } public WALWriter(File logFile, WALFileVersion version) throws IOException { - super(logFile, version); - this.version = version; + super(logFile, WALFormats.writeVersion(version)); + this.version = WALFormats.writeVersion(version); } /** @@ -68,7 +68,11 @@ public synchronized WALMetaData snapshotMetaData() { private synchronized void endFile() throws IOException { // The writer channel is write-only; its known version and size identify an empty header. - if (version != WALFileVersion.V1 && logChannel.size() == version.getVersionBytes().length) { + if (version != WALFileVersion.V1 + && logChannel.size() + == (version == WALFileVersion.EXTENDED + ? dataStartOffset + : version.getVersionBytes().length)) { // A WAL with no entries is valid and has no marker or metadata trailer to append. super.close(); return; @@ -78,6 +82,10 @@ private synchronized void endFile() throws IOException { // mark info part ends endMarker.serialize(markerBuffer); write(markerBuffer, false); + if (fileContext != null) { + fileContext.seal(logChannel, metaData, blockIndex++); + return; + } int metaDataSize = metaData.serializedSize(version); ByteBuffer buffer = @@ -100,9 +108,12 @@ private void writeMetadata(ByteBuffer buffer) throws IOException { @Override public void close() throws IOException { + if (!logChannel.isOpen()) { + return; + } try { endFile(); - } catch (IOException | RuntimeException e) { + } catch (IOException | RuntimeException | Error e) { // In particular, failed recovery-file sealing must release the handle before its temporary // file can be removed on Windows, while preserving the original failure. try { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALNodeRecoverTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALNodeRecoverTask.java index 73d429e3ccb07..180890e073885 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALNodeRecoverTask.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALNodeRecoverTask.java @@ -33,6 +33,8 @@ import org.apache.iotdb.db.storageengine.dataregion.wal.checkpoint.MemTableInfo; import org.apache.iotdb.db.storageengine.dataregion.wal.exception.BrokenWALFileException; import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALByteBufReader; +import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALFormatException; +import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALFormats; import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALMetaData; import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALReader; import org.apache.iotdb.db.storageengine.dataregion.wal.recover.file.UnsealedTsFileRecoverPerformer; @@ -47,6 +49,7 @@ import java.io.DataInputStream; import java.io.File; import java.io.IOException; +import java.io.UncheckedIOException; import java.nio.ByteBuffer; import java.nio.file.Files; import java.util.ArrayList; @@ -72,7 +75,7 @@ public class WALNodeRecoverTask implements Runnable { // version id of first valid .wal file private long firstValidVersionId = Long.MAX_VALUE; private Map memTableId2Info; - private Map memTableId2RecoverPerformer; + private Map memTableId2RecoverPerformer = new HashMap<>(); public WALNodeRecoverTask(File logDirectory, CountDownLatch allNodesRecoveredLatch) { this.logDirectory = logDirectory; @@ -81,6 +84,32 @@ public WALNodeRecoverTask(File logDirectory, CountDownLatch allNodesRecoveredLat @Override public void run() { + try { + validateFilesBeforeRecovery(); + recoverNode(); + } catch (IOException e) { + throw new UncheckedIOException(e); + } finally { + allNodesRecoveredLatch.countDown(); + } + } + + private void validateFilesBeforeRecovery() throws IOException { + File[] walFiles = WALFileUtils.listAllWALFiles(logDirectory); + File[] checkpoints = CheckpointFileUtils.listAllCheckpointFiles(logDirectory); + if (walFiles != null) { + for (File file : walFiles) { + WALFormats.validateForRecovery(file); + } + } + if (checkpoints != null) { + for (File file : checkpoints) { + WALFormats.validateForRecovery(file); + } + } + } + + private void recoverNode() { logger.info(StorageEngineMessages.START_RECOVERING_WAL_NODE_IN_DIR, logDirectory); // recover version id and search index @@ -95,6 +124,9 @@ public void run() { for (UnsealedTsFileRecoverPerformer recoverPerformer : memTableId2RecoverPerformer.values()) { recoverPerformer.getRecoverListener().fail(e); } + if (WALFormatException.isFatal(e)) { + throw new IllegalStateException(e); + } } finally { for (UnsealedTsFileRecoverPerformer recoverPerformer : memTableId2RecoverPerformer.values()) { try { @@ -107,53 +139,49 @@ public void run() { } } - try { - if (config.getDataRegionConsensusProtocolClass().equals(ConsensusFactory.IOT_CONSENSUS)) { - // delete checkpoint info to avoid repeated recover - File[] checkpointFiles = CheckpointFileUtils.listAllCheckpointFiles(logDirectory); - for (File checkpointFile : checkpointFiles) { - try { - Files.delete(checkpointFile.toPath()); - } catch (IOException e) { - logger.error(StorageEngineMessages.ERROR_DELETE_CHECKPOINT_FILE, checkpointFile, e); - } - } - // register wal node - WALManager.getInstance() - .registerWALNode( - logDirectory.getName(), - logDirectory.getAbsolutePath(), - lastVersionId + 1, - lastSearchIndex); - logger.info( - StorageEngineMessages - .STORAGE_LOG_SUCCESSFULLY_RECOVER_WAL_NODE_IN_THE_DIRECTORY_ADD_THIS_FA6ADE22, - logDirectory); - } else { - if (cleanupRecoveredDirectory(logDirectory)) { - logger.info( - StorageEngineMessages - .STORAGE_LOG_SUCCESSFULLY_RECOVER_WAL_NODE_IN_THE_DIRECTORY_SO_DELETE_A17892D9, - logDirectory); + if (config.getDataRegionConsensusProtocolClass().equals(ConsensusFactory.IOT_CONSENSUS)) { + // delete checkpoint info to avoid repeated recover + File[] checkpointFiles = CheckpointFileUtils.listAllCheckpointFiles(logDirectory); + for (File checkpointFile : checkpointFiles) { + try { + Files.delete(checkpointFile.toPath()); + } catch (IOException e) { + logger.error(StorageEngineMessages.ERROR_DELETE_CHECKPOINT_FILE, checkpointFile, e); } } - - // IoTConsensusV2 will not only delete WAL node folder, but also register WAL node. - if (config.getDataRegionConsensusProtocolClass().equals(ConsensusFactory.IOT_CONSENSUS_V2)) { - // register wal node - WALManager.getInstance() - .registerWALNode( - logDirectory.getName(), - logDirectory.getAbsolutePath(), - lastVersionId + 1, - lastSearchIndex); + // register wal node + WALManager.getInstance() + .registerWALNode( + logDirectory.getName(), + logDirectory.getAbsolutePath(), + lastVersionId + 1, + lastSearchIndex); + logger.info( + StorageEngineMessages + .STORAGE_LOG_SUCCESSFULLY_RECOVER_WAL_NODE_IN_THE_DIRECTORY_ADD_THIS_FA6ADE22, + logDirectory); + } else { + if (cleanupRecoveredDirectory(logDirectory)) { logger.info( StorageEngineMessages - .STORAGE_LOG_SUCCESSFULLY_RECOVER_WAL_NODE_IN_THE_DIRECTORY_ADD_THIS_FA6ADE22, + .STORAGE_LOG_SUCCESSFULLY_RECOVER_WAL_NODE_IN_THE_DIRECTORY_SO_DELETE_A17892D9, logDirectory); } - } finally { - allNodesRecoveredLatch.countDown(); + } + + // IoTConsensusV2 will not only delete WAL node folder, but also register WAL node. + if (config.getDataRegionConsensusProtocolClass().equals(ConsensusFactory.IOT_CONSENSUS_V2)) { + // register wal node + WALManager.getInstance() + .registerWALNode( + logDirectory.getName(), + logDirectory.getAbsolutePath(), + lastVersionId + 1, + lastSearchIndex); + logger.info( + StorageEngineMessages + .STORAGE_LOG_SUCCESSFULLY_RECOVER_WAL_NODE_IN_THE_DIRECTORY_ADD_THIS_FA6ADE22, + logDirectory); } } @@ -192,6 +220,9 @@ private long[] readLastFileInfoAndRepairIt() { previousLogicalOffset = logicalOffset; } } catch (Exception e) { + if (WALFormatException.isFatal(e)) { + throw new IllegalStateException(e); + } logger.warn(StorageEngineMessages.FAIL_TO_READ_WAL_LOGS_SKIP, lastWALFile, e); } // make sure last wal file is correct @@ -241,6 +272,9 @@ private static boolean repairWalFileIfBroken(File walFile, WALMetaData metaData) try { return walRepairWriter.repair(metaData); } catch (IOException e) { + if (WALFormatException.isFatal(e)) { + throw new IllegalStateException(e); + } logger.error(StorageEngineMessages.FAIL_TO_RECOVER_WAL_METADATA, walFile, e); DataNodeExceptionMetrics.getInstance().recordSuspiciousDiskException(e); return false; @@ -284,6 +318,9 @@ private void recoverTsFiles() { recoverPerformer.startRecovery(); } catch (Exception e) { recoverPerformer.getRecoverListener().fail(e); + if (WALFormatException.isFatal(e)) { + throw new IllegalStateException(e); + } } } // find all valid .wal files @@ -335,12 +372,18 @@ private void recoverTsFiles() { logDirectory.getName(), e); } catch (IOException e) { + if (WALFormatException.isFatal(e)) { + throw new IllegalStateException(e); + } logger.warn( StorageEngineMessages .STORAGE_LOG_FAIL_TO_READ_MEMTABLE_IDS_FROM_THE_WAL_FILE_OF_WAL_NODE_FBCE8D93, walFile.getAbsoluteFile(), e); } catch (Exception e) { + if (WALFormatException.isFatal(e)) { + throw new IllegalStateException(e); + } logger.warn(StorageEngineMessages.FAIL_TO_READ_WAL_LOGS_SKIP, walFile, e); } } @@ -355,6 +398,9 @@ private void endRecovery() { recoverPerformer.getRecoverListener().succeed(); } catch (Exception e) { recoverPerformer.getRecoverListener().fail(e); + if (WALFormatException.isFatal(e)) { + throw new IllegalStateException(e); + } } } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRecoverManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRecoverManager.java index 54a45626a3d67..c6ae45b497c7a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRecoverManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRecoverManager.java @@ -31,6 +31,8 @@ import org.apache.iotdb.db.exception.runtime.StorageEngineFailureException; import org.apache.iotdb.db.i18n.StorageEngineMessages; import org.apache.iotdb.db.storageengine.dataregion.wal.exception.WALRecoverException; +import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALFormatException; +import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALFormats; import org.apache.iotdb.db.storageengine.dataregion.wal.recover.file.UnsealedTsFileRecoverPerformer; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.listener.WALRecoverListener; @@ -108,12 +110,32 @@ public void recover() throws WALRecoverException, StartupException { recoverThreadPool = IoTDBThreadPoolFactory.newCachedThreadPool(ThreadName.WAL_RECOVER.getName()); CountDownLatch allNodesRecoveredLatch = new CountDownLatch(walNodeDirs.size()); + List> recoveries = new ArrayList<>(); for (File walNodeDir : walNodeDirs) { - recoverThreadPool.submit(new WALNodeRecoverTask(walNodeDir, allNodesRecoveredLatch)); + recoveries.add( + recoverThreadPool.submit(new WALNodeRecoverTask(walNodeDir, allNodesRecoveredLatch))); } try { allNodesRecoveredLatch.await(); + ExecutionException failure = null; + for (Future recovery : recoveries) { + try { + recovery.get(); + } catch (ExecutionException e) { + if (failure == null) { + failure = e; + } else if (WALFormatException.isFatal(e) && !WALFormatException.isFatal(failure)) { + e.addSuppressed(failure); + failure = e; + } else { + failure.addSuppressed(e); + } + } + } + if (failure != null) { + throw failure; + } } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new WALRecoverException(StorageEngineMessages.FAIL_TO_RECOVER_WAL, e); @@ -128,6 +150,10 @@ public void recover() throws WALRecoverException, StartupException { absolutePath2RecoverPerformer.values()) { recoverPerformer.getRecoverListener().fail(e); } + if (WALFormatException.isFatal(e)) { + WALFormats.onRecoveryFailure(e); + throw new WALRecoverException(StorageEngineMessages.FAIL_TO_RECOVER_WAL, e); + } } finally { for (UnsealedTsFileRecoverPerformer recoverPerformer : absolutePath2RecoverPerformer.values()) { @@ -169,25 +195,31 @@ private void asyncRecoverLeftTsFiles() { recoverPerformer.getTsFileAbsolutePath(), e); recoverPerformer.getRecoverListener().fail(e); + if (WALFormatException.isFatal(e)) { + throw e; + } } return null; }; futures.add(recoverTsFilesThreadPool.submit(recoverTsFileTask)); } // wait until all tasks done - for (Future future : futures) { - try { - future.get(); - } catch (ExecutionException e) { - throw new StorageEngineFailureException( - StorageEngineMessages.STORAGE_ENGINE_FAILED_TO_RECOVER, e); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new StorageEngineFailureException( - StorageEngineMessages.STORAGE_ENGINE_FAILED_TO_RECOVER, e); + try { + for (Future future : futures) { + try { + future.get(); + } catch (ExecutionException e) { + throw new StorageEngineFailureException( + StorageEngineMessages.STORAGE_ENGINE_FAILED_TO_RECOVER, e); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new StorageEngineFailureException( + StorageEngineMessages.STORAGE_ENGINE_FAILED_TO_RECOVER, e); + } } + } finally { + recoverTsFilesThreadPool.shutdownNow(); } - recoverTsFilesThreadPool.shutdown(); } public WALRecoverListener addRecoverPerformer(UnsealedTsFileRecoverPerformer recoverPerformer) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRepairWriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRepairWriter.java index 7572eac68cb96..962c4d5077fb9 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRepairWriter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/WALRepairWriter.java @@ -23,6 +23,7 @@ import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntryType; import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALByteBufReader; import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALFileVersion; +import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALFormatException; import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALMetaData; import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALWriter; @@ -146,6 +147,9 @@ private boolean hasReadableMetadata() { WALMetaData.readFromWALFileWithoutRecovery(logFile, channel); return true; } catch (IOException | RuntimeException e) { + if (WALFormatException.isFatal(e)) { + throw new IllegalStateException(e); + } return false; } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALFileUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALFileUtils.java index 5cff6d4b13cba..abb137448b84c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALFileUtils.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALFileUtils.java @@ -339,11 +339,11 @@ private static void forEachSealedSearchableRequest( for (final File walFile : walFiles) { try (final FileChannel channel = FileChannel.open(walFile.toPath())) { - if (WALFileVersion.getVersion(channel) != WALFileVersion.V3) { + WALFileVersion version = WALFileVersion.getVersion(channel); + if (version != WALFileVersion.V3 && version != WALFileVersion.EXTENDED) { continue; } - // V3 WAL stores writer progress in the footer, so these lookups do not deserialize entry - // bodies. + // Formats with writer progress in the footer avoid deserializing entry bodies here. final WALMetaData metadata = WALMetaData.readFromWALFile(walFile, channel); final List physicalTimes = metadata.getPhysicalTimes(); final List nodeIds = metadata.getNodeIds(); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFormatExtensionTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFormatExtensionTest.java new file mode 100644 index 0000000000000..3ead148f06a6d --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFormatExtensionTest.java @@ -0,0 +1,649 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iotdb.db.storageengine.dataregion.wal.io; + +import org.apache.iotdb.commons.concurrent.ExceptionalCountDownLatch; +import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.exception.StartupException; +import org.apache.iotdb.consensus.ConsensusFactory; +import org.apache.iotdb.db.conf.IoTDBConfig; +import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.storageengine.StorageEngine; +import org.apache.iotdb.db.storageengine.dataregion.memtable.IMemTable; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; +import org.apache.iotdb.db.storageengine.dataregion.wal.checkpoint.Checkpoint; +import org.apache.iotdb.db.storageengine.dataregion.wal.checkpoint.CheckpointType; +import org.apache.iotdb.db.storageengine.dataregion.wal.checkpoint.MemTableInfo; +import org.apache.iotdb.db.storageengine.dataregion.wal.exception.WALRecoverException; +import org.apache.iotdb.db.storageengine.dataregion.wal.recover.WALNodeRecoverTask; +import org.apache.iotdb.db.storageengine.dataregion.wal.recover.WALRecoverManager; +import org.apache.iotdb.db.storageengine.dataregion.wal.recover.WALRepairWriter; +import org.apache.iotdb.db.storageengine.dataregion.wal.recover.file.UnsealedTsFileRecoverPerformer; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.listener.WALRecoverListener; +import org.apache.iotdb.db.utils.MmapUtil; + +import io.netty.util.internal.PlatformDependent; +import org.apache.tsfile.file.metadata.enums.CompressionType; +import org.apache.tsfile.fileSystem.FSFactoryProducer; +import org.junit.After; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.mockito.ArgumentCaptor; + +import java.io.EOFException; +import java.io.File; +import java.io.IOException; +import java.io.UncheckedIOException; +import java.lang.reflect.Field; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.net.URL; +import java.net.URLClassLoader; +import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.Enumeration; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.regex.Pattern; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.CALLS_REAL_METHODS; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** Exercises dispatch with a test-only provider; it implements no encryption or product format. */ +public class WALFormatExtensionTest { + @Rule public final TemporaryFolder folder = new TemporaryFolder(); + + @Test + public void testExtensionDispatchInIsolatedRuntime() throws Exception { + String serviceName = "META-INF/services/" + WALFormatProvider.class.getName(); + File serviceFile = new File(folder.getRoot(), serviceName); + Files.createDirectories(serviceFile.toPath().getParent()); + Files.writeString(serviceFile.toPath(), StubProvider.class.getName()); + URL service = serviceFile.toURI().toURL(); + String[] classpath = + System.getProperty("surefire.test.class.path", System.getProperty("java.class.path")) + .split(Pattern.quote(File.pathSeparator)); + URL[] urls = new URL[classpath.length]; + for (int i = 0; i < classpath.length; i++) { + urls[i] = new File(classpath[i]).toURI().toURL(); + } + // Do not register a global test provider: downstream editions install their own provider. + ClassLoader previous = Thread.currentThread().getContextClassLoader(); + try (URLClassLoader isolated = + new URLClassLoader(urls, ClassLoader.getPlatformClassLoader()) { + @Override + public Enumeration getResources(String name) throws IOException { + return serviceName.equals(name) + ? Collections.enumeration(Collections.singleton(service)) + : super.getResources(name); + } + }) { + Thread.currentThread().setContextClassLoader(isolated); + Class cases = isolated.loadClass(Cases.class.getName()); + Object result = + isolated + .loadClass("org.junit.runner.JUnitCore") + .getMethod("runClasses", Class[].class) + .invoke(null, (Object) new Class[] {cases}); + assertTrue( + result.getClass().getMethod("getFailures").invoke(result).toString(), + (boolean) result.getClass().getMethod("wasSuccessful").invoke(result)); + assertEquals(16, result.getClass().getMethod("getRunCount").invoke(result)); + } finally { + Thread.currentThread().setContextClassLoader(previous); + } + } + + public static class Cases { + @Rule public final TemporaryFolder folder = new TemporaryFolder(); + private WALFormatProvider provider; + private WALFileContext context; + + @Before + public void setUp() throws Exception { + provider = mock(WALFormatProvider.class); + context = mock(WALFileContext.class); + when(provider.isEnabled()).thenReturn(true); + when(provider.create(any(), any())).thenReturn(context); + when(provider.open(any(), any())).thenReturn(context); + StubProvider.delegate = provider; + } + + @After + public void tearDown() { + StubProvider.delegate = null; + } + + @Test + public void testInitializationFailureClosesCallerChannels() throws Exception { + File file = folder.newFile(); + WALFormatException failure = WALFormatException.unsupported(); + when(provider.create(any(), any())).thenThrow(failure); + assertSame(failure, assertThrows(IOException.class, () -> new WALWriter(file))); + ArgumentCaptor created = ArgumentCaptor.forClass(FileChannel.class); + verify(provider).create(eq(file), created.capture()); + assertFalse(created.getValue().isOpen()); + when(provider.open(any(), any())).thenThrow(failure); + assertSame(failure, assertThrows(IOException.class, () -> new WALInputStream(file))); + ArgumentCaptor opened = ArgumentCaptor.forClass(FileChannel.class); + verify(provider).open(eq(file), opened.capture()); + assertFalse(opened.getValue().isOpen()); + } + + @Test + public void testWriterDelegatesSegmentsMetadataAndClose() throws Exception { + when(context.writeData(any(), anyLong(), any(), anyInt(), any())) + .thenAnswer( + invocation -> { + FileChannel channel = invocation.getArgument(0); + ByteBuffer bytes = invocation.getArgument(4); + int size = bytes.remaining(); + while (bytes.hasRemaining()) { + channel.write(bytes); + } + return size; + }); + File file = folder.newFile(); + try (WALWriter writer = new WALWriter(file)) { + assertEquals(WALFileVersion.EXTENDED, WALFileVersion.getVersion(file)); + ByteBuffer bytes = ByteBuffer.allocate(Integer.BYTES).putInt(42); + writer.write(bytes, false); + } + ArgumentCaptor indexes = ArgumentCaptor.forClass(Long.class); + verify(context, times(2)) + .writeData(any(), indexes.capture(), eq(CompressionType.UNCOMPRESSED), anyInt(), any()); + assertEquals(Arrays.asList(0L, 1L), indexes.getAllValues()); + verify(context).seal(any(), any(), eq(2L)); + verify(context).close(); + } + + @Test + public void testCheckpointUsesExtensionWithoutWalMetadata() throws Exception { + try (CheckpointWriter writer = new CheckpointWriter(folder.newFile())) { + writer.write(ByteBuffer.allocate(Long.BYTES).putLong(7), false); + } + verify(context).writeData(any(), eq(0L), eq(CompressionType.UNCOMPRESSED), eq(8), any()); + verify(context, never()).seal(any(), any(), anyLong()); + verify(context).close(); + } + + @Test + public void testReaderResumesAndSeeksThroughProvider() throws Exception { + File file = extensionFile(); + when(context.readSegment(any(), anyLong(), anyLong())) + .thenAnswer( + invocation -> { + if ((long) invocation.getArgument(2) != 0) { + throw new EOFException(); + } + FileChannel channel = invocation.getArgument(0); + ByteBuffer bytes = ByteBuffer.allocate(4); + while (bytes.hasRemaining()) { + if (channel.read(bytes) < 0) { + throw new EOFException(); + } + } + return bytes.flip(); + }); + doAnswer( + invocation -> { + ByteBuffer bytes = invocation.getArgument(0); + if (bytes != null) { + Arrays.fill(bytes.array(), (byte) 0); + } + return null; + }) + .when(context) + .release(any()); + try (WALInputStream reader = new WALInputStream(file)) { + assertEquals(1, reader.read()); + reader.releaseReadBuffers(); + assertEquals(2, reader.read()); + reader.skipToGivenLogicalPosition(3); + assertEquals(4, reader.read()); + } + verify(context, times(3)).readSegment(any(), anyLong(), eq(0L)); + verify(context).close(); + } + + @Test + public void testFatalMetadataCannotTriggerRepairOrQuarantine() throws Exception { + File file = extensionFile(); + byte[] original = Files.readAllBytes(file.toPath()); + WALFormatException failure = WALFormatException.unsupported(); + when(provider.readMetadata(any(), any())).thenThrow(failure); + try (FileChannel channel = FileChannel.open(file.toPath())) { + assertSame( + failure, + assertThrows(IOException.class, () -> WALMetaData.readFromWALFile(file, channel))); + } + IllegalStateException repairFailure = + assertThrows( + IllegalStateException.class, + () -> new WALRepairWriter(file).repair(new WALMetaData())); + assertTrue(WALFormatException.isFatal(repairFailure)); + assertArrayEquals(original, Files.readAllBytes(file.toPath())); + assertFalse(new File(file + ".broken").exists()); + verify(provider, never()).open(any(), any()); + } + + @Test + public void testFatalSegmentFailureIsNotSwallowedByReaders() throws Exception { + File file = extensionFile(); + when(context.readSegment(any(), anyLong(), anyLong())) + .thenThrow(new IOException(WALFormatException.unsupported())); + try (WALReader reader = new WALReader(file, true)) { + assertTrue( + WALFormatException.isFatal(assertThrows(UncheckedIOException.class, reader::hasNext))); + } + assertTrue( + WALFormatException.isFatal( + assertThrows(UncheckedIOException.class, () -> new CheckpointReader(file)))); + } + + @Test + public void testRecoveryPreflightFailureReachesCallerAndKeepsFiles() throws Exception { + File walRoot = folder.newFolder("wal"); + File node = new File(walRoot, "node"); + assertTrue(node.mkdir()); + File wal = new File(node, "_0-0-1.wal"); + Files.write(wal.toPath(), StubProvider.MAGIC); + byte[] original = Files.readAllBytes(wal.toPath()); + doThrow(WALFormatException.unsupported()).when(provider).validateForRecovery(any()); + String[] previous = CommonDescriptor.getInstance().getConfig().getWalDirs(); + WALRecoverManager recovery = WALRecoverManager.getInstance(); + try { + recovery.clear(); + CommonDescriptor.getInstance() + .getConfig() + .setWalDirs(new String[] {walRoot.getAbsolutePath()}); + recovery.setAllDataRegionScannedLatch(new ExceptionalCountDownLatch(0)); + assertTrue( + WALFormatException.isFatal(assertThrows(WALRecoverException.class, recovery::recover))); + assertArrayEquals(original, Files.readAllBytes(wal.toPath())); + verify(provider).onRecoveryFailure(any()); + } finally { + recovery.clear(); + CommonDescriptor.getInstance().getConfig().setWalDirs(previous); + } + } + + @Test + public void testOrdinaryFailureDoesNotHideLaterFatalFailure() throws Exception { + assertMultipleNodeFailures(false, true); + } + + @Test + public void testFatalFailureRetainsLaterOrdinaryFailure() throws Exception { + assertMultipleNodeFailures(true, true); + } + + @Test + public void testOrdinaryNodeFailuresKeepExistingRecoveryPolicy() throws Exception { + assertMultipleNodeFailures(false, false); + } + + private void assertMultipleNodeFailures(boolean fatalFirst, boolean hasFatal) throws Exception { + File firstRoot = folder.newFolder("first-wal"); + File secondRoot = folder.newFolder("second-wal"); + File[] roots = {firstRoot, secondRoot}; + IOException ordinary = new IOException("ordinary recovery failure"); + IOException other = hasFatal ? WALFormatException.unsupported() : new IOException("other"); + IOException[] failures = + fatalFirst ? new IOException[] {other, ordinary} : new IOException[] {ordinary, other}; + List files = new ArrayList<>(); + for (int i = 0; i < roots.length; i++) { + File node = new File(roots[i], "node"); + assertTrue(node.mkdir()); + File wal = new File(node, "_0-0-1.wal"); + Files.write(wal.toPath(), StubProvider.MAGIC); + files.add(wal); + doThrow(failures[i]).when(provider).validateForRecovery(wal); + } + String[] previous = CommonDescriptor.getInstance().getConfig().getWalDirs(); + WALRecoverManager recovery = WALRecoverManager.getInstance(); + try { + recovery.clear(); + CommonDescriptor.getInstance() + .getConfig() + .setWalDirs(new String[] {firstRoot.getAbsolutePath(), secondRoot.getAbsolutePath()}); + recovery.setAllDataRegionScannedLatch(new ExceptionalCountDownLatch(0)); + if (hasFatal) { + WALRecoverException thrown = assertThrows(WALRecoverException.class, recovery::recover); + assertTrue(WALFormatException.isFatal(thrown)); + Throwable primary = thrown.getCause(); + assertSame(other, primary.getCause().getCause()); + assertEquals(1, primary.getSuppressed().length); + assertSame(ordinary, primary.getSuppressed()[0].getCause().getCause()); + verify(provider).onRecoveryFailure(primary); + } else { + recovery.recover(); + verify(provider, never()).onRecoveryFailure(any()); + } + for (File wal : files) { + verify(provider).validateForRecovery(wal); + assertArrayEquals(StubProvider.MAGIC, Files.readAllBytes(wal.toPath())); + } + } finally { + recovery.clear(); + CommonDescriptor.getInstance().getConfig().setWalDirs(previous); + } + } + + private File extensionFile() throws IOException { + File file = folder.newFile(); + ByteBuffer bytes = ByteBuffer.allocate(StubProvider.MAGIC.length + 4); + bytes.put(StubProvider.MAGIC).put(new byte[] {1, 2, 3, 4}); + Files.write(file.toPath(), bytes.array()); + return file; + } + + @Test + public void testProviderOwnsPooledDirectBufferAcrossPauseSeekAndClose() throws Exception { + File file = extensionFile(); + ByteBuffer pooled = ByteBuffer.allocateDirect(4); + long address = PlatformDependent.directBufferAddress(pooled); + assertTrue(address != 0); + when(context.readSegment(any(), anyLong(), anyLong())) + .thenAnswer( + call -> { + FileChannel channel = call.getArgument(0); + pooled.clear(); + while (pooled.hasRemaining()) { + if (channel.read(pooled) < 0) { + throw new EOFException(); + } + } + return pooled.flip(); + }); + try { + try (WALInputStream reader = new WALInputStream(file)) { + assertEquals(1, reader.read()); + reader.releaseReadBuffers(); + assertEquals(address, PlatformDependent.directBufferAddress(pooled)); + assertEquals(2, reader.read()); + reader.skipToGivenLogicalPosition(3); + assertEquals(address, PlatformDependent.directBufferAddress(pooled)); + assertEquals(4, reader.read()); + } + assertEquals(address, PlatformDependent.directBufferAddress(pooled)); + verify(context, times(3)).release(pooled); + } finally { + if (PlatformDependent.directBufferAddress(pooled) != 0) { + MmapUtil.clean(pooled); + } + } + } + + @Test + public void testFatalRepairWriteFailureKeepsOriginalFile() throws Exception { + assertFatalRepairFailure(false); + } + + @Test + public void testFatalRepairSealFailureKeepsOriginalFile() throws Exception { + assertFatalRepairFailure(true); + } + + private void assertFatalRepairFailure(boolean seal) throws Exception { + File file = extensionFile(); + byte[] original = Files.readAllBytes(file.toPath()); + when(provider.readMetadata(any(), any())).thenThrow(new IOException("missing footer")); + when(context.writeData(any(), anyLong(), any(), anyInt(), any())) + .thenAnswer( + call -> { + FileChannel channel = call.getArgument(0); + ByteBuffer data = call.getArgument(4); + int size = data.remaining(); + while (data.hasRemaining()) { + channel.write(data); + } + return size; + }); + when(context.readSegment(any(), anyLong(), anyLong())) + .thenReturn(ByteBuffer.wrap(new byte[] {1, 2, 3, 4})); + WALFormatException failure = WALFormatException.unsupported(); + if (seal) { + doThrow(failure).when(context).seal(any(), any(), anyLong()); + } else { + doThrow(failure).when(context).writeData(any(), anyLong(), any(), anyInt(), any()); + } + WALMetaData metadata = new WALMetaData(); + metadata.add(4, 0, 1); + Method repair = + WALNodeRecoverTask.class.getDeclaredMethod( + "repairWalFileIfBroken", File.class, WALMetaData.class); + repair.setAccessible(true); + InvocationTargetException thrown = + assertThrows(InvocationTargetException.class, () -> repair.invoke(null, file, metadata)); + assertTrue(WALFormatException.isFatal(thrown.getCause())); + assertArrayEquals(original, Files.readAllBytes(file.toPath())); + assertFalse(new File(file + ".broken").exists()); + assertEquals(1, folder.getRoot().listFiles().length); + } + + @Test(timeout = 10000) + public void testFatalReplayFailureCompletesAllTaskListeners() throws Exception { + assertTaskListenersComplete(false); + } + + @Test(timeout = 10000) + public void testFatalFinalizationFailureCompletesAllTaskListeners() throws Exception { + assertTaskListenersComplete(true); + } + + private void assertTaskListenersComplete(boolean failOnEnd) throws Exception { + WALRecoverManager recovery = WALRecoverManager.getInstance(); + recovery.clear(); + File node = folder.newFolder("node"); + List performers = new ArrayList<>(); + List tables = new ArrayList<>(); + WALFormatException failure = WALFormatException.unsupported(); + try { + for (int i = 1; i <= 2; i++) { + File tsFile = new File(folder.getRoot(), "sequence/root.sg/0/0/" + i + "-1-0-0.tsfile"); + UnsealedTsFileRecoverPerformer performer = mock(UnsealedTsFileRecoverPerformer.class); + WALRecoverListener listener = new WALRecoverListener(tsFile.getAbsolutePath()); + when(performer.getTsFileResource()).thenReturn(new TsFileResource(tsFile)); + when(performer.getTsFileAbsolutePath()).thenReturn(tsFile.getAbsolutePath()); + when(performer.getRecoverListener()).thenReturn(listener); + recovery.addRecoverPerformer(performer); + performers.add(performer); + IMemTable table = mock(IMemTable.class); + when(table.getMemTableId()).thenReturn((long) i); + when(table.getDataRegionId()).thenReturn("0"); + tables.add(new MemTableInfo(table, tsFile.getAbsolutePath(), 0)); + } + Checkpoint checkpoint = new Checkpoint(CheckpointType.GLOBAL_MEMORY_TABLE_INFO, tables); + ByteBuffer bytes = ByteBuffer.allocate(Long.BYTES + checkpoint.serializedSize()); + bytes.putLong(2); + checkpoint.serialize(bytes); + File checkpointFile = new File(node, "_0.checkpoint"); + ByteBuffer fileBytes = ByteBuffer.allocate(StubProvider.MAGIC.length + bytes.position()); + fileBytes.put(StubProvider.MAGIC).put(bytes.array()); + Files.write(checkpointFile.toPath(), fileBytes.array()); + when(context.readSegment(any(), anyLong(), anyLong())) + .thenAnswer( + call -> { + FileChannel channel = call.getArgument(0); + ByteBuffer data = ByteBuffer.allocate(bytes.capacity()); + while (data.hasRemaining()) { + if (channel.read(data) < 0) { + throw new EOFException(); + } + } + return data.flip(); + }); + if (failOnEnd) { + doThrow(new WALRecoverException(failure.getMessage(), failure)) + .when(performers.get(0)) + .endRecovery(); + } else { + doThrow(failure).when(performers.get(0)).startRecovery(); + } + CountDownLatch latch = new CountDownLatch(1); + assertTrue( + WALFormatException.isFatal( + assertThrows( + IllegalStateException.class, () -> new WALNodeRecoverTask(node, latch).run()))); + assertEquals(0, latch.getCount()); + for (UnsealedTsFileRecoverPerformer performer : performers) { + assertEquals( + WALRecoverListener.Status.FAILURE, performer.getRecoverListener().waitForResult()); + verify(performer).close(); + } + assertTrue(checkpointFile.exists()); + } finally { + recovery.clear(); + } + } + + @Test + public void testFatalWalFormatPreventsStorageEngineReadiness() throws Exception { + File walRoot = folder.newFolder("startup-wal"); + File node = new File(walRoot, "node"); + assertTrue(node.mkdir()); + Files.writeString(new File(node, "_0-0-1.wal").toPath(), "V99-WAL"); + String[] previousDirs = CommonDescriptor.getInstance().getConfig().getWalDirs(); + IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig(); + String previousConsensus = config.getDataRegionConsensusProtocolClass(); + WALRecoverManager recovery = WALRecoverManager.getInstance(); + StorageEngine engine = mock(StorageEngine.class, CALLS_REAL_METHODS); + setField(engine, "fsFactory", FSFactoryProducer.getFSFactory()); + setField(engine, "isReadyForReadAndWrite", new AtomicBoolean(true)); + setField(engine, "isReadyForNonReadWriteFunctions", new AtomicBoolean(true)); + doReturn(Collections.emptyMap()).when(engine).getLocalDataRegionInfo(); + try { + recovery.clear(); + CommonDescriptor.getInstance().getConfig().setWalDirs(new String[] {walRoot.getPath()}); + config.setDataRegionConsensusProtocolClass(ConsensusFactory.SIMPLE_CONSENSUS); + Method recover = StorageEngine.class.getDeclaredMethod("asyncRecoverDataRegion"); + recover.setAccessible(true); + InvocationTargetException thrown = + assertThrows(InvocationTargetException.class, () -> recover.invoke(engine)); + assertTrue(thrown.getCause() instanceof StartupException); + assertTrue(WALFormatException.isFatal(thrown.getCause())); + assertFalse(engine.isReadyForReadAndWrite()); + assertFalse(engine.isReadyForNonReadWriteFunctions()); + } finally { + Field pool = StorageEngine.class.getDeclaredField("cachedThreadPool"); + pool.setAccessible(true); + ExecutorService executor = (ExecutorService) pool.get(engine); + if (executor != null) { + executor.shutdownNow(); + } + recovery.clear(); + CommonDescriptor.getInstance().getConfig().setWalDirs(previousDirs); + config.setDataRegionConsensusProtocolClass(previousConsensus); + } + } + + private static void setField(Object target, String name, Object value) throws Exception { + Field field = StorageEngine.class.getDeclaredField(name); + field.setAccessible(true); + field.set(target, value); + } + } + + /** Registered only inside the isolated test runtime, never in a distribution. */ + public static final class StubProvider implements WALFormatProvider { + private static final byte[] MAGIC = "V88-WAL".getBytes(StandardCharsets.US_ASCII); + private static volatile WALFormatProvider delegate; + + @Override + public byte[] getMagic() { + return MAGIC.clone(); + } + + @Override + public boolean isEnabled() { + return delegate != null && delegate.isEnabled(); + } + + private WALFormatProvider implementation() throws IOException { + if (delegate == null) { + throw WALFormatException.unsupported(); + } + return delegate; + } + + @Override + public WALFileContext create(File file, FileChannel channel) throws IOException { + ByteBuffer magic = ByteBuffer.wrap(MAGIC); + while (magic.hasRemaining()) { + channel.write(magic); + } + return implementation().create(file, channel); + } + + @Override + public WALFileContext open(File file, FileChannel channel) throws IOException { + channel.position(MAGIC.length); + return implementation().open(file, channel); + } + + @Override + public WALMetaData readMetadata(File file, FileChannel channel) throws IOException { + return implementation().readMetadata(file, channel); + } + + @Override + public void validateRead(File file, WALFileVersion version) throws IOException { + if (delegate != null) { + delegate.validateRead(file, version); + } + } + + @Override + public void validateForRecovery(File file) throws IOException { + implementation().validateForRecovery(file); + } + + @Override + public void onRecoveryFailure(Throwable failure) { + delegate.onRecoveryFailure(failure); + } + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFormatsTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFormatsTest.java new file mode 100644 index 0000000000000..cd36779ae90b9 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFormatsTest.java @@ -0,0 +1,150 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iotdb.db.storageengine.dataregion.wal.io; + +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.mockito.InOrder; + +import java.io.File; +import java.io.IOException; +import java.lang.reflect.Field; +import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.util.Arrays; +import java.util.Collections; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +public class WALFormatsTest { + @Rule public final TemporaryFolder folder = new TemporaryFolder(); + + @Test + public void testProviderIsOptionalAndUnambiguous() { + assertNull(WALFormats.selectProvider(Collections.emptyIterator())); + WALFormatProvider provider = mock(WALFormatProvider.class); + when(provider.getMagic()).thenReturn("V88-WAL".getBytes(StandardCharsets.US_ASCII)); + assertSame(provider, WALFormats.selectProvider(Collections.singleton(provider).iterator())); + assertThrows( + IllegalStateException.class, + () -> WALFormats.selectProvider(Arrays.asList(provider, provider).iterator())); + } + + @Test + public void testProviderMagicIsBounded() { + for (byte[] magic : + new byte[][] { + null, new byte[0], new byte[65], new byte[] {1, 2, 3}, WALFileVersion.V3.getVersionBytes() + }) { + WALFormatProvider provider = mock(WALFormatProvider.class); + when(provider.getMagic()).thenReturn(magic); + assertThrows( + IllegalStateException.class, + () -> WALFormats.selectProvider(Collections.singleton(provider).iterator())); + } + } + + @Test + public void testUnknownVersionIsNotGuessedAsLegacy() throws Exception { + File file = folder.newFile(); + Files.write(file.toPath(), "V77-WALpayload".getBytes(StandardCharsets.US_ASCII)); + try (FileChannel channel = FileChannel.open(file.toPath())) { + channel.position(2); + assertThrows(WALFormatException.class, () -> WALFileVersion.getVersion(channel)); + assertEquals(2, channel.position()); + assertThrows(WALFormatException.class, () -> WALFormats.recognizes(channel, null)); + assertEquals(2, channel.position()); + } + assertThrows(WALFormatException.class, () -> new WALInputStream(file)); + } + + @Test + public void testRecognitionDoesNotDependOnWritePolicy() throws Exception { + WALFormatProvider provider = mock(WALFormatProvider.class); + byte[] magic = "V88-WAL".getBytes(StandardCharsets.US_ASCII); + when(provider.getMagic()).thenReturn(magic); + when(provider.isEnabled()).thenReturn(false); + File file = folder.newFile(); + Files.write(file.toPath(), magic); + try (FileChannel channel = FileChannel.open(file.toPath())) { + channel.position(1); + assertTrue(WALFormats.recognizes(channel, provider)); + assertEquals(1, channel.position()); + } + } + + @Test + public void testPlainAndEmptyFilesDoNotRequireProvider() throws Exception { + File file = folder.newFile(); + try (FileChannel channel = FileChannel.open(file.toPath())) { + assertFalse(WALFormats.recognizes(channel, null)); + assertEquals(WALFileVersion.V1, WALFileVersion.getVersion(channel)); + } + Files.write(file.toPath(), new byte[] {0, 1, 2}); + try (FileChannel channel = FileChannel.open(file.toPath())) { + assertFalse(WALFormats.recognizes(channel, null)); + assertEquals(WALFileVersion.V1, WALFileVersion.getVersion(channel)); + } + Files.write(file.toPath(), WALFileVersion.V3.getVersionBytes()); + try (FileChannel channel = FileChannel.open(file.toPath())) { + assertEquals(WALFileVersion.V3, WALFileVersion.getVersion(channel)); + assertTrue(WALFileVersion.isEmptyOrHeaderOnly(channel)); + } + } + + @Test + public void testReaderReleasesBufferBeforeClosingFormatContext() throws Exception { + File file = folder.newFile(); + WALInputStream reader = new WALInputStream(file); + WALFileContext context = mock(WALFileContext.class); + ByteBuffer data = ByteBuffer.wrap(new byte[] {1, 2, 3}); + Field contextField = WALInputStream.class.getDeclaredField("fileContext"); + contextField.setAccessible(true); + contextField.set(reader, context); + Field dataField = WALInputStream.class.getDeclaredField("dataBuffer"); + dataField.setAccessible(true); + dataField.set(reader, data); + reader.close(); + InOrder order = inOrder(context); + order.verify(context).release(data); + order.verify(context).close(); + } + + @Test + public void testUnrecoverableFailureSurvivesWrappingWithoutLoopingOnCauses() { + WALFormatException fatal = WALFormatException.unsupported(); + assertTrue(WALFormatException.isFatal(new IllegalStateException(new IOException(fatal)))); + IOException first = new IOException(); + IOException second = new IOException(first); + first.initCause(second); + assertFalse(WALFormatException.isFatal(first)); + assertFalse(WALFormatException.isFatal(new IOException())); + } +}