This is an automated email from the ASF dual-hosted git repository.
wuchong pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 56dc6dedd [server] Fix TabletServer fails to restart after unclean
shutdown due to EOFException in log recovery (#2942)
56dc6dedd is described below
commit 56dc6dedd55f99dc72268a560330b5962a8e6fee
Author: Liebing <[email protected]>
AuthorDate: Mon May 11 10:44:14 2026 +0800
[server] Fix TabletServer fails to restart after unclean shutdown due to
EOFException in log recovery (#2942)
---
.../org/apache/fluss/server/log/LogLoader.java | 87 +++++----
.../org/apache/fluss/server/log/LogSegment.java | 43 +++--
.../org/apache/fluss/server/log/LogLoaderTest.java | 197 +++++++++++----------
3 files changed, 179 insertions(+), 148 deletions(-)
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/log/LogLoader.java
b/fluss-server/src/main/java/org/apache/fluss/server/log/LogLoader.java
index 572fbd304..b5d5b9f45 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/log/LogLoader.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/log/LogLoader.java
@@ -23,7 +23,6 @@ import org.apache.fluss.exception.InvalidOffsetException;
import org.apache.fluss.exception.LogSegmentOffsetOverflowException;
import org.apache.fluss.exception.LogStorageException;
import org.apache.fluss.metadata.LogFormat;
-import org.apache.fluss.server.exception.CorruptIndexException;
import org.apache.fluss.utils.FlussPaths;
import org.apache.fluss.utils.types.Tuple2;
@@ -137,6 +136,7 @@ final class LogLoader {
*/
private Tuple2<Long, Long> recoverLog() throws IOException {
if (!isCleanShutdown) {
+ long recoverLogStart = System.currentTimeMillis();
List<LogSegment> unflushed =
logSegments.values(recoveryPointCheckpoint,
Long.MAX_VALUE);
int numUnflushed = unflushed.size();
@@ -153,50 +153,57 @@ final class LogLoader {
numUnflushed,
logSegments.getTableBucket());
+ int truncatedBytes = -1;
try {
- segment.sanityCheck();
- } catch (NoSuchFileException | CorruptIndexException e) {
+ truncatedBytes = recoverSegment(segment);
+ } catch (InvalidOffsetException e) {
+ long startOffset = segment.getBaseOffset();
LOG.warn(
- "Found invalid index file corresponding log file
{} for bucket {}, "
- + "recovering segment and rebuilding index
files...",
-
segment.getFileLogRecords().file().getAbsoluteFile(),
+ "Found invalid offset during recovery for bucket
{}. Deleting the corrupt segment "
+ + "and creating an empty one with starting
offset {}",
logSegments.getTableBucket(),
- e);
-
- int truncatedBytes = -1;
- try {
- truncatedBytes = recoverSegment(segment);
- } catch (InvalidOffsetException invalidOffsetException) {
- long startOffset = segment.getBaseOffset();
- LOG.warn(
- "Found invalid offset during recovery for
bucket {}. Deleting the corrupt segment "
- + "and creating an empty one with
starting offset {}",
- logSegments.getTableBucket(),
- startOffset);
- truncatedBytes = segment.truncateTo(startOffset);
- }
+ startOffset);
+ truncatedBytes = segment.truncateTo(startOffset);
+ }
- if (truncatedBytes > 0) {
- // we had an invalid message, delete all remaining log
- LOG.warn(
- "Corruption found in segment {} for bucket {},
truncating to offset {}",
- segment.getBaseOffset(),
- logSegments.getTableBucket(),
- segment.readNextOffset());
- removeAndDeleteSegments(unflushedIter);
- truncated = true;
- }
+ if (truncatedBytes > 0) {
+ // we had an invalid message, delete all remaining log
+ LOG.warn(
+ "Corruption found in segment {} for bucket {},
truncating to offset {}",
+ segment.getBaseOffset(),
+ logSegments.getTableBucket(),
+ segment.readNextOffset());
+ removeAndDeleteSegments(unflushedIter);
+ truncated = true;
+ } else {
+ numFlushed += 1;
}
- numFlushed += 1;
}
+ long recoverLogEnd = System.currentTimeMillis();
+ LOG.info(
+ "Log recovery completed for bucket {} in {} ms",
+ logSegments.getTableBucket(),
+ recoverLogEnd - recoverLogStart);
}
- // TODO truncate log to recover maybe unflush segments.
if (logSegments.isEmpty()) {
+ // TODO: use logStartOffset if issue
https://github.com/apache/fluss/issues/744 ready
logSegments.add(LogSegment.open(logTabletDir, 0L, conf,
logFormat));
}
+
long logEndOffset = logSegments.lastSegment().get().readNextOffset();
- return Tuple2.of(recoveryPointCheckpoint, logEndOffset);
+
+ // Update the recovery point if there was a clean shutdown and did not
perform any changes
+ // to the segment. Otherwise, we just ensure that the recovery point
is not ahead of the log
+ // end offset. To ensure correctness and to make it easier to reason
about, it's best to
+ // only advance the recovery point when the log is flushed. If we
advanced the recovery
+ // point here, we could skip recovery for unflushed segments if the
server crashed after we
+ // checkpointed the recovery point and before we flush the segment.
+ if (isCleanShutdown) {
+ return Tuple2.of(logEndOffset, logEndOffset);
+ } else {
+ return Tuple2.of(Math.min(recoveryPointCheckpoint, logEndOffset),
logEndOffset);
+ }
}
/**
@@ -294,6 +301,20 @@ final class LogLoader {
long baseOffset = FlussPaths.offsetFromFile(file);
LogSegment segment =
LogSegment.open(logTabletDir, baseOffset,
conf, true, 0, logFormat);
+
+ try {
+ segment.sanityCheck();
+ } catch (NoSuchFileException e) {
+ if (isCleanShutdown
+ || segment.getBaseOffset() <
recoveryPointCheckpoint) {
+ LOG.error(
+ "Could not find offset index file
corresponding to log file {} "
+ + "for bucket {}, recovering
segment and rebuilding index files...",
+
segment.getFileLogRecords().file().getAbsoluteFile(),
+ logSegments.getTableBucket());
+ }
+ recoverSegment(segment);
+ }
logSegments.add(segment);
}
}
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/log/LogSegment.java
b/fluss-server/src/main/java/org/apache/fluss/server/log/LogSegment.java
index a88084277..a0d7c6a12 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/log/LogSegment.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/log/LogSegment.java
@@ -20,6 +20,7 @@ package org.apache.fluss.server.log;
import org.apache.fluss.annotation.Internal;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.CorruptMessageException;
import org.apache.fluss.exception.CorruptRecordException;
import org.apache.fluss.exception.InvalidColumnProjectionException;
import org.apache.fluss.exception.InvalidRecordException;
@@ -188,7 +189,6 @@ public final class LogSegment {
+ lazyOffsetIndex.file().getAbsolutePath()
+ " does not exist.");
}
- lazyOffsetIndex.get().sanityCheck();
if (!lazyTimeIndex.file().exists()) {
throw new NoSuchFileException(
@@ -196,7 +196,10 @@ public final class LogSegment {
+ lazyTimeIndex.file().getAbsolutePath()
+ " does not exist.");
}
- lazyTimeIndex.get().sanityCheck();
+
+ // Sanity checks for time index and offset index are skipped because
+ // we will recover the segments above the recovery point in
recoverLog()
+ // in any case so sanity checking them here is redundant.
}
/**
@@ -324,14 +327,17 @@ public final class LogSegment {
// The max timestamp is exposed at the batch level, so no need
to iterate the
// records
- if (batch.commitTimestamp() > maxTimestampSoFar()) {
+ if (batch.commitTimestamp() >
maxTimestampAndStartOffsetSoFar.timestamp) {
maxTimestampAndStartOffsetSoFar =
new TimestampOffset(batch.commitTimestamp(),
batch.baseLogOffset());
}
if (validBytes - lastIndexEntry > indexIntervalBytes) {
offsetIndex().append(batch.lastLogOffset(), validBytes);
- timeIndex().maybeAppend(maxTimestampSoFar(),
startOffsetOfMaxTimestampSoFar());
+ timeIndex()
+ .maybeAppend(
+ maxTimestampAndStartOffsetSoFar.timestamp,
+ maxTimestampAndStartOffsetSoFar.offset);
lastIndexEntry = validBytes;
}
@@ -339,26 +345,27 @@ public final class LogSegment {
validBytes += batch.sizeInBytes();
}
- } catch (CorruptRecordException | InvalidRecordException e) {
+ } catch (CorruptRecordException
+ | InvalidRecordException
+ | CorruptMessageException
+ | IllegalArgumentException
+ | IndexOutOfBoundsException e) {
+ // Data corruption detected during recovery: CRC mismatch, invalid
record format,
+ // or truncated batch from an interrupted write. Truncate from
this point onward.
LOG.warn(
- "Found invalid messages in log segment "
- + fileLogRecords.file().getAbsolutePath()
- + " at byte offset "
- + validBytes
- + ": "
- + e.getMessage()
- + ". "
- + e.getCause());
+ "Found invalid messages in log segment {} at byte offset
{}: {}. {}",
+ fileLogRecords.file().getAbsolutePath(),
+ validBytes,
+ e.getMessage(),
+ e.getCause());
}
int truncated = fileLogRecords.sizeInBytes() - validBytes;
if (truncated > 0) {
LOG.debug(
- "Truncated "
- + truncated
- + " invalid bytes at the end of segment "
- + fileLogRecords.file().getAbsolutePath()
- + " during recovery");
+ "Truncated {} invalid bytes at the end of segment {}
during recovery",
+ truncated,
+ fileLogRecords.file().getAbsolutePath());
}
fileLogRecords.truncateTo(validBytes);
offsetIndex().trimToValidSize();
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/log/LogLoaderTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/log/LogLoaderTest.java
index 582b2934d..075af47c4 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/log/LogLoaderTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/log/LogLoaderTest.java
@@ -45,6 +45,7 @@ import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
@@ -268,9 +269,6 @@ final class LogLoaderTest extends LogTestBase {
corruptSegment.getFileLogRecords().append(memoryLogRecords);
logTablet.close();
- // delete the index file to trigger the recovery
- corruptSegment.offsetIndex().deleteIfExists();
-
logTablet = createLogTablet(false);
// the corrupt segment should be truncated to base offset
assertThat(logTablet.localLogEndOffset()).isEqualTo(corruptSegment.getBaseOffset());
@@ -308,9 +306,6 @@ final class LogLoaderTest extends LogTestBase {
.sorted())
.containsExactly(2L, 4L);
- // Delete all offset index files to trigger segment recover
- deleteAllOffsetIndexFile(log);
-
log = createLogTablet(false);
assertThat(
WriterStateManager.listSnapshotFiles(logDir).stream()
@@ -408,7 +403,7 @@ final class LogLoaderTest extends LogTestBase {
// after this step, the segments should be
// [0-0], [1-1], [2-2], [3-3], [4-4], [5-5], [6-]
// writer id 1 2 1 2 1 2
- // se1 id 0 0 1 1 2 2
+ // seq id 0 0 1 1 2 2
// snapshot 1 2 3 4 5 6
for (int i = 0; i <= 5; i++) {
if (i % 2 == 0) {
@@ -473,22 +468,19 @@ final class LogLoaderTest extends LogTestBase {
long recoveryPoint = segmentOffsets.get(segmentOffsets.size() - 4);
assertThat(recoveryPoint).isLessThan(offsetForSegmentAfterRecoveryPoint);
- // 1. Test unclean shut without any recovery
+ // 1. Test unclean shutdown
// Retain snapshots for the last 2 segments (delete snapshots before
that)
long snapshotRetentionOffset =
segmentOffsets.get(segmentOffsets.size() - 2);
log.writerStateManager().deleteSnapshotsBefore(snapshotRetentionOffset);
log.close();
- // Reopen the log with recovery point. Although we use unclean
shutdown here,
- // all the index files are correctly close, so Fluss will not trigger
recover for any
- // segment.
+ // Reopen the log with recovery point. Since we use unclean shutdown
here,
+ // all the log segments after recovery point will be recovered.
log = createLogTablet(false, recoveryPoint);
- // Expected snapshot offsets: last 2 segment base offsets + log end
offset
- List<Long> lastTowSegmentOffsets =
- segmentOffsets.subList(
- Math.max(0, segmentOffsets.size() - 2),
segmentOffsets.size());
- expectedSnapshotOffsets = new HashSet<>(lastTowSegmentOffsets);
+ // Expected snapshot offsets: segments base offset after recovery point
+ expectedSnapshotOffsets =
+ new HashSet<>(segmentOffsets.subList((int) recoveryPoint,
segmentOffsets.size()));
expectedSnapshotOffsets.add(log.localLogEndOffset());
// Verify that expected snapshot offsets exist
@@ -509,71 +501,6 @@ final class LogLoaderTest extends LogTestBase {
expectedWritersLastBatchSequence.put(wid2, seq2 - 1);
assertThat(actualWritersLastBatchSequence).isEqualTo(expectedWritersLastBatchSequence);
log.close();
-
- // 2. Test unclean shut down without recover segment will rebuild
writer state
- // Delete all snapshot files
- deleteAllWriterSnapshot(logDir);
-
- // Reopen the log with recovery point. Although Fluss will not trigger
recover for any
- // segment, but the writer state should be rebuilt.
- log = createLogTablet(false, recoveryPoint);
-
- actualSnapshotOffsets =
- WriterStateManager.listSnapshotFiles(logDir).stream()
- .map(snapshotFile -> snapshotFile.offset)
- .collect(Collectors.toSet());
- // Will rebuild writer state for all segments, but only take snapshot
for the last 2
- // segments and the last offset
- expectedSnapshotOffsets = new HashSet<>();
- expectedSnapshotOffsets.add(segmentOffsets.get(segmentOffsets.size() -
2));
- expectedSnapshotOffsets.add(segmentOffsets.get(segmentOffsets.size() -
1));
- expectedSnapshotOffsets.add(log.localLogEndOffset());
- assertThat(actualSnapshotOffsets).isEqualTo(expectedSnapshotOffsets);
-
- // Verify that expected writers last batch sequence
- actualWritersLastBatchSequence =
- log.writerStateManager().activeWriters().entrySet().stream()
- .collect(
- Collectors.toMap(
- Map.Entry::getKey, e ->
e.getValue().lastBatchSequence()));
- expectedWritersLastBatchSequence = new HashMap<>();
- expectedWritersLastBatchSequence.put(wid1, seq1 - 1);
- expectedWritersLastBatchSequence.put(wid2, seq2 - 1);
-
assertThat(actualWritersLastBatchSequence).isEqualTo(expectedWritersLastBatchSequence);
- log.close();
-
- // 3. Test unclean shut down with recover segment will rebuild writer
state for each segment
- // after recovery point.
- // Delete all snapshot files and index files (to trigger segment
recover)
- deleteAllWriterSnapshot(logDir);
- deleteAllOffsetIndexFile(log);
-
- // Reopen the log with recovery point. All segments will be recovered
since we delete all
- // the index files
- log = createLogTablet(false, recoveryPoint);
-
- // Writer snapshot files should be rebuilt for each segment after
recovery point
- actualSnapshotOffsets =
- WriterStateManager.listSnapshotFiles(logDir).stream()
- .map(snapshotFile -> snapshotFile.offset)
- .collect(Collectors.toSet());
- expectedSnapshotOffsets =
- log.logSegments(recoveryPoint,
log.localLogEndOffset()).stream()
- .map(LogSegment::getBaseOffset)
- .collect(Collectors.toSet());
- expectedSnapshotOffsets.add(log.localLogEndOffset());
- assertThat(actualSnapshotOffsets).isEqualTo(expectedSnapshotOffsets);
-
- actualWritersLastBatchSequence =
- log.writerStateManager().activeWriters().entrySet().stream()
- .collect(
- Collectors.toMap(
- Map.Entry::getKey, e ->
e.getValue().lastBatchSequence()));
- expectedWritersLastBatchSequence = new HashMap<>();
- expectedWritersLastBatchSequence.put(wid1, seq1 - 1);
- expectedWritersLastBatchSequence.put(wid2, seq2 - 1);
-
assertThat(actualWritersLastBatchSequence).isEqualTo(expectedWritersLastBatchSequence);
- log.close();
}
@Test
@@ -627,6 +554,91 @@ final class LogLoaderTest extends LogTestBase {
.containsExactly(1L, 2L, 4L);
}
+ /**
+ * Tests that log recovery is correctly triggered after an unclean
shutdown (TabletServer
+ * crash).
+ *
+ * <p>The test simulates an unclean shutdown by:
+ *
+ * <ol>
+ * <li>Writing records to a log to create multiple segments
+ * <li>Recording the logEndOffset before corruption
+ * <li>Corrupting the index files (appending garbage bytes, simulating
incomplete flush)
+ * <li>Appending garbage bytes to the active .log file tail (simulating
partial batch write)
+ * <li>Reloading the log with {@code isCleanShutdown=false}
+ * <li>Verifying that recovery truncates the corrupt tail, rebuilds
indexes, and restores the
+ * correct logEndOffset
+ * </ol>
+ */
+ @Test
+ void testLogRecoveryIsCalledUponBrokerCrash() throws Exception {
+ // Step 1: Create log and write enough records to produce multiple
segments
+ int numRecords = 200;
+ LogTablet logTablet = createLogTablet(true);
+ appendRecords(logTablet, numRecords);
+
+ // Step 2: Record the expected logEndOffset after clean writes
+ long expectedLogEndOffset = logTablet.localLogEndOffset();
+ assertThat(expectedLogEndOffset).isEqualTo(numRecords);
+
+ // Step 3: Collect index files and the active segment's .log file
before "crash"
+ List<File> indexFiles = collectIndexFiles(logTablet.logSegments());
+ File activeLogFile =
logTablet.activeLogSegment().getFileLogRecords().file();
+
+ // Step 4: Close the log (simulating the on-disk state at the moment
of crash)
+ logTablet.close();
+
+ // Step 5: Corrupt the index files (simulate incomplete index flush
during crash)
+ for (File indexFile : indexFiles) {
+ try (FileChannel fileChannel =
+ FileChannel.open(indexFile.toPath(),
StandardOpenOption.APPEND)) {
+ // Append 12 bytes of garbage – enough to make the index size
non-multiple of
+ // entrySize and trigger CorruptIndexException during
sanityCheck
+ for (int i = 0; i < 12; i++) {
+ fileChannel.write(ByteBuffer.wrap(new byte[] {(byte) i}));
+ }
+ }
+ }
+
+ // Step 6: Append garbage bytes to the active segment's .log file tail
+ // (simulate a partial/incomplete batch that was being written at
crash time)
+ try (FileChannel logFileChannel =
+ FileChannel.open(activeLogFile.toPath(),
StandardOpenOption.APPEND)) {
+ byte[] garbage = new byte[40];
+ // Corrupt the last log file in the active segment by appending
zero bytes,
+ // simulating a typical interrupted write during an unclean
shutdown where
+ // the filesystem zero-fills pre-allocated but unwritten blocks.
+ Arrays.fill(garbage, (byte) 0);
+ logFileChannel.write(ByteBuffer.wrap(garbage));
+ }
+
+ // Step 7: Reload with isCleanShutdown=false – this must trigger log
recovery
+ logTablet = createLogTablet(false);
+
+ // Step 8: Verify that recovery succeeded and the logEndOffset is
restored correctly.
+ // Recovery should truncate the garbage bytes appended to the .log
file and rebuild
+ // indexes, so the final offset must equal the number of valid records
written.
+
assertThat(logTablet.localLogEndOffset()).isEqualTo(expectedLogEndOffset);
+
+ // Step 9: Verify all index files pass sanity check after recovery
+ for (LogSegment segment : logTablet.logSegments()) {
+ segment.offsetIndex().sanityCheck();
+ segment.timeIndex().sanityCheck();
+ }
+
+ // Step 10: Verify that all previously written records are still
readable after recovery
+ for (int i = 0; i < numRecords; i++) {
+ assertThat(logTablet.lookupOffsetForTimestamp(clock.milliseconds()
+ i * 10L))
+ .isEqualTo(i);
+ }
+
+ // Step 11: Verify that the log is still writable after recovery
+ appendRecords(logTablet, 1, (int) expectedLogEndOffset);
+
assertThat(logTablet.localLogEndOffset()).isEqualTo(expectedLogEndOffset + 1);
+
+ logTablet.close();
+ }
+
private LogTablet createLogTablet(boolean isCleanShutdown) throws
Exception {
return createLogTablet(isCleanShutdown, 0);
}
@@ -649,8 +661,13 @@ final class LogLoaderTest extends LogTestBase {
}
private void appendRecords(LogTablet logTablet, int numRecords) throws
Exception {
- int baseOffset = 0;
- int batchSequence = 0;
+ appendRecords(logTablet, numRecords, 0);
+ }
+
+ private void appendRecords(LogTablet logTablet, int numRecords, int
startOffset)
+ throws Exception {
+ int baseOffset = startOffset;
+ int batchSequence = startOffset;
for (int i = 0; i < numRecords; i++) {
List<Object[]> objects = Collections.singletonList(new Object[]
{1, "a"});
List<ChangeType> changeTypes =
@@ -684,18 +701,4 @@ final class LogLoaderTest extends LogTestBase {
}
return indexFiles;
}
-
- private void deleteAllOffsetIndexFile(LogTablet logTablet) throws
IOException {
- List<LogSegment> logSegments = logTablet.logSegments();
- for (LogSegment segment : logSegments) {
- segment.offsetIndex().deleteIfExists();
- }
- }
-
- private void deleteAllWriterSnapshot(File logDir) throws Exception {
- List<SnapshotFile> snapshotFiles =
WriterStateManager.listSnapshotFiles(logDir);
- for (SnapshotFile snapshotFile : snapshotFiles) {
- snapshotFile.deleteIfExists();
- }
- }
}
