[server] Fix TabletServer fails to restart after unclean shutdown due to EOFException in log recovery (#2942)

Author: LiebingYu

fluss-server/src/main/java/org/apache/fluss/server/log/LogLoader.java

@@ -23,7 +23,6 @@
 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 @@ public LoadedLogOffsets load() throws IOException {
      */
     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 @@ private Tuple2<Long, Long> recoverLog() throws IOException {
                         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 @@ private void loadSegmentFiles() throws IOException {
                         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);
                     }
                 }

fluss-server/src/main/java/org/apache/fluss/server/log/LogSegment.java

@@ -20,6 +20,7 @@
 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,15 +189,17 @@ public void sanityCheck() throws IOException {
                             + lazyOffsetIndex.file().getAbsolutePath()
                             + " does not exist.");
         }
-        lazyOffsetIndex.get().sanityCheck();
 
         if (!lazyTimeIndex.file().exists()) {
             throw new NoSuchFileException(
                     "Time index file "
                             + 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,41 +327,45 @@ public int recover() throws IOException {
 
                 // 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;
                 }
 
                 // TODO Adding assign partition leader epoch follow KIP-101
 
                 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();