[server] Use logStartOffset when rebuild writer state

Author: LiebingYu

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

@@ -107,17 +107,12 @@ public LoadedLogOffsets load() throws IOException {
         // WriterStateManager used during log recovery may have deleted some files without the
         // LogLoader.writerStateManager instance witnessing the deletion.
         writerStateManager.removeStraySnapshots(logSegments.baseOffsets());
-
-        // TODO, Here, we use 0 as the logStartOffset passed into rebuildWriterState. The reason is
-        // that the current implementation of logStartOffset in Fluss is not yet fully refined, and
-        // there may be cases where logStartOffset is not updated. As a result, logStartOffset is
-        // not yet reliable. Once the issue with correctly updating logStartOffset is resolved in
-        // issue https://github.com/apache/fluss/issues/744, we can use logStartOffset here.
-        // Additionally, using 0 versus using logStartOffset does not affect correctness—they both
-        // can restore the complete WriterState. The only difference is that using logStartOffset
-        // can potentially skip over more segments.
         LogTablet.rebuildWriterState(
-                writerStateManager, logSegments, 0, nextOffset, isCleanShutdown);
+                writerStateManager,
+                logSegments,
+                logStartOffsetCheckpoint,
+                nextOffset,
+                isCleanShutdown);
 
         LogSegment activeSegment = logSegments.lastSegment().get();
         activeSegment.resizeIndexes((int) conf.get(ConfigOptions.LOG_INDEX_FILE_SIZE).getBytes());
@@ -261,16 +256,12 @@ private int recoverSegment(LogSegment segment) throws IOException {
                         logSegments.getTableBucket(),
                         logTabletDir,
                         this.writerStateManager.writerExpirationMs());
-        // TODO, Here, we use 0 as the logStartOffset passed into rebuildWriterState. The reason is
-        // that the current implementation of logStartOffset in Fluss is not yet fully refined, and
-        // there may be cases where logStartOffset is not updated. As a result, logStartOffset is
-        // not yet reliable. Once the issue with correctly updating logStartOffset is resolved in
-        // issue https://github.com/apache/fluss/issues/744, we can use logStartOffset here.
-        // Additionally, using 0 versus using logStartOffset does not affect correctness—they both
-        // can restore the complete WriterState. The only difference is that using logStartOffset
-        // can potentially skip over more segments.
         LogTablet.rebuildWriterState(
-                writerStateManager, logSegments, 0, segment.getBaseOffset(), false);
+                writerStateManager,
+                logSegments,
+                logStartOffsetCheckpoint,
+                segment.getBaseOffset(),
+                false);
         int bytesTruncated = segment.recover();
         // once we have recovered the segment's data, take a snapshot to ensure that we won't
         // need to reload the same segment again while recovering another segment.

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

@@ -1178,16 +1178,8 @@ private void rebuildWriterState(long lastOffset, WriterStateManager writerStateM
             throws IOException {
         synchronized (lock) {
             localLog.checkIfMemoryMappedBufferClosed();
-            // TODO, Here, we use 0 as the logStartOffset passed into rebuildWriterState. The reason
-            // is that the current implementation of logStartOffset in Fluss is not yet fully
-            // refined, and there may be cases where logStartOffset is not updated. As a result,
-            // logStartOffset is not yet reliable. Once the issue with correctly updating
-            // logStartOffset is resolved in issue https://github.com/apache/fluss/issues/744, we
-            // can use logStartOffset here.
-            // Additionally, using 0 versus using logStartOffset does not affect correctness—they
-            // both can restore the complete WriterState. The only difference is that using
-            // logStartOffset can potentially skip over more segments.
-            rebuildWriterState(writerStateManager, localLog.getSegments(), 0, lastOffset, false);
+            rebuildWriterState(
+                    writerStateManager, localLog.getSegments(), logStartOffset, lastOffset, false);
         }
     }
 
@@ -1386,7 +1378,7 @@ private static void loadWritersFromRecords(
         Map<Long, WriterAppendInfo> loadedWriters = new HashMap<>();
         for (LogRecordBatch batch : records.batches()) {
             if (batch.hasWriterId()) {
-                updateWriterAppendInfo(writerStateManager, batch, loadedWriters, true);
+                updateWriterAppendInfo(writerStateManager, batch, loadedWriters, false);
             }
         }
         loadedWriters.values().forEach(writerStateManager::update);

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

@@ -228,6 +228,16 @@ public void deleteSnapshotsBefore(long offset) throws IOException {
         }
     }
 
+    /**
+     * Deletes the writer snapshot files after the given offset (exclusive) in a thread safe manner.
+     */
+    @VisibleForTesting
+    public void deleteSnapshotAfter(long offset) throws IOException {
+        for (SnapshotFile snapshot : snapshots.tailMap(offset, false).values()) {
+            removeAndDeleteSnapshot(snapshot.offset);
+        }
+    }
+
     /** Fetch the snapshot file for the end offset of the log segment. */
     public Optional<File> fetchSnapshot(long offset) {
         return Optional.ofNullable(snapshots.get(offset)).map(SnapshotFile::file);