adjust the recovery way

Author: swuferhong

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

@@ -19,10 +19,10 @@
 
 import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
-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;
 
@@ -33,12 +33,8 @@
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.NoSuchFileException;
-import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Comparator;
-import java.util.Iterator;
-import java.util.List;
-import java.util.stream.Collectors;
 
 /* This file is based on source code of Apache Kafka Project (https://kafka.apache.org/), licensed by the Apache
  * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
@@ -172,61 +168,6 @@ private int recoverSegment(LogSegment segment) throws IOException {
      *     overflow
      */
     private Tuple2<Long, Long> recoverLog() throws IOException {
-        if (!isCleanShutdown) {
-            List<LogSegment> unflushed =
-                    logSegments.values(recoveryPointCheckpoint, Long.MAX_VALUE);
-            int numUnflushed = unflushed.size();
-            Iterator<LogSegment> unflushedIter = unflushed.iterator();
-            boolean truncated = false;
-            int numFlushed = 1;
-
-            long startTimeMs = System.currentTimeMillis();
-            while (unflushedIter.hasNext() && !truncated) {
-                LogSegment segment = unflushedIter.next();
-                LOG.info(
-                        "Recovering unflushed segment {}. {}/{} recovered for bucket {}",
-                        segment.getBaseOffset(),
-                        numFlushed,
-                        numUnflushed,
-                        logSegments.getTableBucket());
-
-                int truncatedBytes = -1;
-                try {
-                    truncatedBytes = recoverSegment(segment);
-                } catch (Exception e) {
-                    if (e instanceof 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);
-                    } else {
-                        throw e;
-                    }
-                }
-
-                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;
-                }
-            }
-
-            LOG.info(
-                    "Recovery--11 for bucket {} completed in {} ms",
-                    logSegments.getTableBucket(),
-                    System.currentTimeMillis() - startTimeMs);
-        }
-
         if (logSegments.isEmpty()) {
             // TODO: use logStartOffset if issue https://github.com/apache/fluss/issues/744 ready
             logSegments.add(LogSegment.open(logTabletDir, 0L, conf, logFormat));
@@ -235,49 +176,6 @@ private Tuple2<Long, Long> recoverLog() throws IOException {
         return Tuple2.of(recoveryPointCheckpoint, logEndOffset);
     }
 
-    /**
-     * This method deletes the given log segments and the associated writer snapshots.
-     *
-     * <p>This method does not need to convert IOException to {@link LogStorageException} because it
-     * is either called before all logs are loaded or the immediate caller will catch and handle
-     * IOException
-     *
-     * @param segmentsToDelete The log segments to schedule for deletion
-     */
-    private void removeAndDeleteSegments(Iterator<LogSegment> segmentsToDelete) {
-        if (segmentsToDelete.hasNext()) {
-            List<LogSegment> toDelete = new ArrayList<>();
-            segmentsToDelete.forEachRemaining(toDelete::add);
-
-            LOG.info(
-                    "Deleting segments for bucket {} as part of log recovery: {}",
-                    logSegments.getTableBucket(),
-                    toDelete.stream().map(LogSegment::toString).collect(Collectors.joining(",")));
-            toDelete.forEach(segment -> logSegments.remove(segment.getBaseOffset()));
-
-            try {
-                LocalLog.deleteSegmentFiles(
-                        toDelete, LocalLog.SegmentDeletionReason.LOG_TRUNCATION);
-            } catch (IOException e) {
-                LOG.error(
-                        "Failed to delete truncated segments {} for bucket {}",
-                        toDelete,
-                        logSegments.getTableBucket(),
-                        e);
-            }
-
-            try {
-                LogTablet.deleteWriterSnapshots(toDelete, writerStateManager);
-            } catch (IOException e) {
-                LOG.error(
-                        "Failed to delete truncated writer snapshots {} for bucket {}",
-                        toDelete,
-                        logSegments.getTableBucket(),
-                        e);
-            }
-        }
-    }
-
     /** Loads segments from disk into the provided segments. */
     private void loadSegmentFiles() throws IOException {
         long startTimeMs = System.currentTimeMillis();
@@ -305,17 +203,25 @@ private void loadSegmentFiles() throws IOException {
 
                         try {
                             segment.sanityCheck(timeIndexFileNewlyCreated);
-                        } catch (IOException e) {
+                        } catch (Exception e) {
                             if (e instanceof NoSuchFileException) {
                                 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...",
-                                            logSegments.getTableBucket(),
-                                            segment.getFileLogRecords().file().getAbsoluteFile());
+                                            segment.getFileLogRecords().file().getAbsoluteFile(),
+                                            logSegments.getTableBucket());
                                 }
                                 recoverSegment(segment);
+                            } else if (e instanceof CorruptIndexException) {
+                                LOG.warn(
+                                        "Found a corrupt index file corresponding to log file {} for bucket {}, recovering "
+                                                + "segment and rebuilding index files...",
+                                        segment.getFileLogRecords().file().getAbsoluteFile(),
+                                        logSegments.getTableBucket(),
+                                        e);
+                                recoverSegment(segment);
                             } else {
                                 throw e;
                             }

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

@@ -174,14 +174,14 @@ public void resizeIndexes(int size) throws IOException {
     }
 
     public void sanityCheck(boolean timeIndexFileNewlyCreated) throws IOException {
-        if (lazyOffsetIndex.file().exists()) {
+        if (lazyOffsetIndex.file().exists() && lazyTimeIndex.file().exists()) {
             // Resize the time index file to 0 if it is newly created.
             if (timeIndexFileNewlyCreated) {
                 timeIndex().resize(0);
             }
-            // 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.
+
+            lazyOffsetIndex.get().sanityCheck();
+            lazyTimeIndex.get().sanityCheck();
         } else {
             throw new NoSuchFileException(
                     "Offset index file "
@@ -310,6 +310,7 @@ public int recover() throws IOException {
         maxTimestampAndStartOffsetSoFar = TimestampOffset.UNKNOWN;
         try {
             for (LogRecordBatch batch : fileLogRecords.batches()) {
+                long startTimestampMs = System.currentTimeMillis();
                 batch.ensureValid();
                 ensureOffsetInRange(batch.lastLogOffset());
 
@@ -329,6 +330,10 @@ public int recover() throws IOException {
                 // TODO Adding assign partition leader epoch follow KIP-101
 
                 validBytes += batch.sizeInBytes();
+                LOG.info(
+                        "Recovered one batch of {} bytes costs {} ms",
+                        batch.sizeInBytes(),
+                        System.currentTimeMillis() - startTimestampMs);
             }
         } catch (CorruptRecordException | InvalidRecordException e) {
             LOG.warn(