Skip to content

Fix WAL roll recovery after closed channel #18653

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jt2594838 merged 1 commit into from
Sep 17, 2026
Merged

Fix WAL roll recovery after closed channel #18653

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 31 additions & 8 deletions .../main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/AbstractWALBuffer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@

import java.io.File;
import java.io.IOException;
import java.nio.file.FileAlreadyExistsException;
import java.nio.file.Files;
import java.nio.file.StandardCopyOption;
import java.util.Arrays;
Expand All @@ -56,6 +57,12 @@ public abstract class AbstractWALBuffer implements IWALBuffer {
@SuppressWarnings("squid:S3077")
protected volatile WALWriter currentWALFileWriter;

// Only the sync thread accesses this state. Once sealed, the old WAL must never be written or
// closed again, even if renaming it or creating its successor fails because the disk is full.
private File pendingRollFile;
private WALFileStatus pendingRollStatus;
private long pendingRollSearchIndex;

protected AbstractWALBuffer(
String identifier, String logDirectory, long startFileVersion, long startSearchIndex)
throws IOException {
Expand Down Expand Up @@ -99,39 +106,55 @@ public long getCurrentWALOriginalFileSize() {
* @throws IOException If failing to close or open the log writer
*/
protected File rollLogWriter(long searchIndex, WALFileStatus fileStatus) throws IOException {
// close file
currentWALFileWriter.close();
addDiskUsage(currentWALFileWriter.size());
addFileNum(1);
File lastFile = currentWALFileWriter.getLogFile();
if (!hasPendingRoll()) {
// Record the boundary only after sealing and forcing the old WAL have both succeeded.

@jt2594838 jt2594838 Sep 16, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Preserves the sealed WAL and current rotation stage across retries, so a later sync task can continue after disk space is restored without closing or counting the old file twice.

currentWALFileWriter.close();
pendingRollFile = currentWALFileWriter.getLogFile();
pendingRollStatus = fileStatus;
pendingRollSearchIndex = searchIndex;
addDiskUsage(currentWALFileWriter.size());
}
File lastFile = pendingRollFile;
String lastName = lastFile.getName();
if (WALFileUtils.parseStatusCode(lastName) != fileStatus) {
if (WALFileUtils.parseStatusCode(lastName) != pendingRollStatus) {
String targetName =
WALFileUtils.getLogFileName(
WALFileUtils.parseVersionId(lastName),
WALFileUtils.parseStartSearchIndex(lastName),
fileStatus);
pendingRollStatus);
File targetFile = SystemFileFactory.INSTANCE.getFile(logDirectory, targetName);
Files.move(
lastFile.toPath(),
targetFile.toPath(),
StandardCopyOption.REPLACE_EXISTING,
StandardCopyOption.ATOMIC_MOVE);
lastFile = targetFile;
pendingRollFile = targetFile;
}
// roll file
long nextFileVersion = currentWALFileVersion + 1;
File nextLogFile =
SystemFileFactory.INSTANCE.getFile(
logDirectory,
WALFileUtils.getLogFileName(
nextFileVersion, searchIndex, WALFileStatus.CONTAINS_SEARCH_INDEX));
nextFileVersion, pendingRollSearchIndex, WALFileStatus.CONTAINS_SEARCH_INDEX));
// A failed header write may leave a partial successor. Do not append to it on retry, or
// overwrite an unexpected existing WAL; either requires recovery rather than online rotation.
if (nextLogFile.length() > 0) {
throw new FileAlreadyExistsException(nextLogFile.toString());
}
currentWALFileWriter = new WALWriter(nextLogFile);
currentWALFileVersion = nextFileVersion;
addFileNum(1);
pendingRollFile = null;
logger.debug(StorageEngineMessages.OPEN_NEW_WAL_FILE_FOR_BUFFER, nextLogFile, identifier);
return lastFile;
}

protected boolean hasPendingRoll() {
return pendingRollFile != null;
}

public long getDiskUsage() {
return diskUsage;
}
Expand Down
81 changes: 71 additions & 10 deletions ...node/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -126,6 +126,12 @@ public class WALBuffer extends AbstractWALBuffer {
private final Map<Long, Set<Long>> memTableIdsOfWal = new ConcurrentHashMap<>();
private final BiConsumer<File, File> walFileRolledListener;

// An entry may span several sync tasks. Never acknowledge its final chunk after an earlier
// chunk failed. Failures before writing a pending successor can be cleared at the batch boundary;
// failures while writing the active WAL require recovery because its record boundary is unknown.
private Exception syncFailure;
private boolean retryAfterFailedBatch;

public WALBuffer(String identifier, String logDirectory) throws IOException {
this(
identifier,
Expand Down Expand Up @@ -585,6 +591,45 @@ public SyncBufferTask(
public void run() {
final long startTime = System.nanoTime();

if (syncFailure != null) {
failListeners(syncFailure);

@jt2594838 jt2594838 Sep 16, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Propagates a failed sync batch to all listeners and keeps the buffer recoverable. SET SYSTEM TO RUNNING alone cannot repair an incomplete WAL record, so retry state is cleared only after the pending roll completes.

// SET SYSTEM TO RUNNING does not repair a failed batch or an unknown record boundary.
if (CommonDescriptor.getInstance().getConfig().isRunning()) {
CommonDescriptor.getInstance().getConfig().handleUnrecoverableError();
}
if (forceFlag && retryAfterFailedBatch) {
syncFailure = null;
}
switchSyncingBufferToIdle();
return;
}

boolean resumedRoll = false;
final boolean hasData = syncingBuffer.position() > 0;
try {
if (hasPendingRoll()) {
// The previous task sealed the old file. Finish opening its successor before touching
// this buffer, so no bytes or metadata can be appended after the old WAL's end marker.
rollLogWriter(searchIndex, fileStatus);
resumedRoll = true;
}
} catch (IOException e) {
logger.error(
StorageEngineMessages
.STORAGE_LOG_FAIL_TO_ROLL_WAL_NODE_S_LOG_WRITER_CHANGE_SYSTEM_MODE_TO_A384AA54,
identifier,
e);
if (!forceFlag) {
syncFailure = e;
retryAfterFailedBatch = true;
}
failListeners(e);
DataNodeExceptionMetrics.getInstance().recordSuspiciousDiskException(e);
CommonDescriptor.getInstance().getConfig().handleUnrecoverableError();
switchSyncingBufferToIdle();
return;
}

makeMemTableCheckpoints();

long walFileVersionId = currentWALFileVersion;
Expand All @@ -610,7 +655,11 @@ public void run() {
.STORAGE_LOG_FAIL_TO_SYNC_WAL_NODE_S_BUFFER_CHANGE_SYSTEM_MODE_TO_ERROR_8C379D57,
identifier,
e);
syncFailure = e instanceof Exception exception ? exception : new IOException(e);
retryAfterFailedBatch = false;
failListeners(syncFailure);
CommonDescriptor.getInstance().getConfig().handleUnrecoverableError();
return;
} finally {
switchSyncingBufferToIdle();
}
Expand All @@ -623,24 +672,24 @@ public void run() {

boolean forceSuccess = false;
// try to roll log writer
if (info.rollWALFileWriterListener != null
if ((info.rollWALFileWriterListener != null && (!resumedRoll || hasData))
// TODO: Control the wal file by the number of WALEntry
|| (forceFlag
&& currentWALFileWriter.originalSize() >= config.getWalFileSizeThresholdInByte())) {
try {
rollLogWriter(searchIndex, currentWALFileWriter.getWalFileStatus());
forceSuccess = true;
if (info.rollWALFileWriterListener != null) {
info.rollWALFileWriterListener.succeed();
}
} catch (IOException e) {
logger.error(
StorageEngineMessages
.STORAGE_LOG_FAIL_TO_ROLL_WAL_NODE_S_LOG_WRITER_CHANGE_SYSTEM_MODE_TO_A384AA54,
identifier,
e);
if (info.rollWALFileWriterListener != null) {
info.rollWALFileWriterListener.fail(e);
failListeners(e);
if (!hasPendingRoll()) {
// A failed seal has no known durable boundary from which to resume rotation.
syncFailure = e;
retryAfterFailedBatch = false;
}
DataNodeExceptionMetrics.getInstance().recordSuspiciousDiskException(e);
CommonDescriptor.getInstance().getConfig().handleUnrecoverableError();
Expand All @@ -657,15 +706,18 @@ public void run() {
identifier,
e);
DataNodeExceptionMetrics.getInstance().recordSuspiciousDiskException(e);
for (WALFlushListener fsyncListener : info.fsyncListeners) {
fsyncListener.fail(e);
}
failListeners(e);
syncFailure = e;
retryAfterFailedBatch = false;
CommonDescriptor.getInstance().getConfig().handleUnrecoverableError();
}
}

// notify all waiting listeners
if (forceSuccess) {
if (info.rollWALFileWriterListener != null) {
info.rollWALFileWriterListener.succeed();
}
for (WALFlushListener fsyncListener : info.fsyncListeners) {
fsyncListener.succeed();
}
Expand All @@ -675,6 +727,15 @@ public void run() {
WRITING_METRICS.recordSyncWALBufferCost(System.nanoTime() - startTime, forceFlag);
}

private void failListeners(Exception e) {
if (info.rollWALFileWriterListener != null) {
info.rollWALFileWriterListener.fail(e);
}
for (WALFlushListener fsyncListener : info.fsyncListeners) {
fsyncListener.fail(e);
}
}

private void makeMemTableCheckpoints() {
if (info.checkpoints.isEmpty()) {
return;
Expand Down Expand Up @@ -767,7 +828,7 @@ public void close() {
shutdownThread(syncBufferThread, ThreadName.WAL_SYNC);
}

if (currentWALFileWriter != null) {
if (currentWALFileWriter != null && !hasPendingRoll()) {
try {
currentWALFileWriter.close();
} catch (IOException e) {
Expand Down
39 changes: 23 additions & 16 deletions ...datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/LogWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,20 +20,16 @@
package org.apache.iotdb.db.storageengine.dataregion.wal.io;

import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.i18n.StorageEngineMessages;
import org.apache.iotdb.db.service.metrics.WritingMetrics;
import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntry;
import org.apache.iotdb.db.storageengine.dataregion.wal.checkpoint.Checkpoint;

import org.apache.tsfile.compress.ICompressor;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.FileChannel;
import java.nio.file.StandardOpenOption;

Expand All @@ -42,8 +38,6 @@
* and writing {@link Checkpoint} into .checkpoint file.
*/
public abstract class LogWriter implements ILogWriter {
private static final Logger logger = LoggerFactory.getLogger(LogWriter.class);

protected final File logFile;
protected final FileChannel logChannel;
protected long originalSize = 0;
Expand Down Expand Up @@ -73,9 +67,23 @@ protected LogWriter(File logFile, WALFileVersion version) throws IOException {
StandardOpenOption.CREATE,
StandardOpenOption.WRITE,
StandardOpenOption.APPEND);
if ((!logFile.exists() || logFile.length() == 0)
&& (version == WALFileVersion.V2 || version == WALFileVersion.V3)) {
this.logChannel.write(ByteBuffer.wrap(version.getVersionBytes()));
try {
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) {
// A full disk can fail initialization after open() succeeds. Release the orphan channel
// before the owner retries creating the successor.
try {
logChannel.close();
} catch (IOException closeException) {
e.addSuppressed(closeException);
}
throw e;
}
}

Expand Down Expand Up @@ -124,12 +132,12 @@ public double write(ByteBuffer buffer, boolean allowCompress) throws IOException
WritingMetrics.getInstance().recordCompressWALBufferCost(System.nanoTime() - startTime);
}
startTime = System.nanoTime();
try {
headerBuffer.flip();
headerBuffer.flip();
while (headerBuffer.hasRemaining()) {
logChannel.write(headerBuffer);
}
while (buffer.hasRemaining()) {
logChannel.write(buffer);

@jt2594838 jt2594838 Sep 16, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Consumes the full buffer and lets ClosedChannelException reach the caller. A partial or failed write must trigger WAL recovery instead of being acknowledged as a successful flush.

} catch (ClosedChannelException e) {
logger.warn(StorageEngineMessages.CANNOT_WRITE_TO, logFile, e);
}
WritingMetrics.getInstance()
.recordWroteWALBuffer(uncompressedSize, bufferSize, System.nanoTime() - startTime);
Expand All @@ -149,9 +157,8 @@ public void force() throws IOException {

@Override
public void force(boolean metaData) throws IOException {
if (logChannel != null && logChannel.isOpen()) {
logChannel.force(metaData);
}
// A closed channel is a failed durability operation, not a successful no-op.
logChannel.force(metaData);
}

@Override
Expand Down
5 changes: 4 additions & 1 deletion ...datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -90,7 +90,10 @@ private synchronized void endFile() throws IOException {

private void writeMetadata(ByteBuffer buffer) throws IOException {
buffer.flip();
logChannel.write(buffer);
// A successful seal is the recovery boundary for switching to the next WAL file.
while (buffer.hasRemaining()) {
logChannel.write(buffer);

@jt2594838 jt2594838 Sep 16, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Persists the complete metadata trailer before treating the file as sealed; this trailer is the recovery boundary for switching to the next WAL file.

}
}

@Override
Expand Down
Loading
Loading