add writer state log

Author: LiebingYu

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

@@ -113,8 +113,18 @@ public LoadedLogOffsets load() throws IOException {
         // 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.
+        LOG.info(
+                "In load for bucket {}, end offset {}, before rebuild: {}",
+                logSegments.getTableBucket(),
+                writerStateManager.mapEndOffset(),
+                writerStateManager.toJsonString());
         LogTablet.rebuildWriterState(
                 writerStateManager, logSegments, 0, nextOffset, isCleanShutdown);
+        LOG.info(
+                "In load for bucket {}, end offset {}, after rebuild: {}",
+                logSegments.getTableBucket(),
+                writerStateManager.mapEndOffset(),
+                writerStateManager.toJsonString());
 
         LogSegment activeSegment = logSegments.lastSegment().get();
         activeSegment.resizeIndexes((int) conf.get(ConfigOptions.LOG_INDEX_FILE_SIZE).getBytes());
@@ -264,12 +274,24 @@ private int recoverSegment(LogSegment segment) throws IOException {
         // 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.
+        LOG.info(
+                "In recoverSegment for bucket {} for segment {}, end offset {}, before rebuild: {}",
+                logSegments.getTableBucket(),
+                segment.getBaseOffset(),
+                writerStateManager.mapEndOffset(),
+                writerStateManager.toJsonString());
         LogTablet.rebuildWriterState(
                 writerStateManager, logSegments, 0, 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.
         writerStateManager.takeSnapshot();
+        LOG.info(
+                "In recoverSegment for bucket {} for segment {}, end offset {}, after rebuild: {}",
+                logSegments.getTableBucket(),
+                segment.getBaseOffset(),
+                writerStateManager.mapEndOffset(),
+                writerStateManager.toJsonString());
         return bytesTruncated;
     }
 

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

@@ -1256,6 +1256,10 @@ static void rebuildWriterState(
                     FetchDataInfo fetchDataInfo =
                             segment.read(startOffset, Integer.MAX_VALUE, maxPosition, false);
                     if (fetchDataInfo != null) {
+                        LOG.info(
+                                "Loading writer state for bucket {} from segment {}",
+                                segments.getTableBucket(),
+                                segment);
                         loadWritersFromRecords(writerStateManager, fetchDataInfo.getRecords());
                     }
                 }

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

@@ -478,6 +478,22 @@ private static void writeSnapshot(File file, Map<Long, WriterStateEntry> entries
         }
     }
 
+    public String toJsonString() {
+        List<WriterSnapshotEntry> snapshotEntries = new ArrayList<>();
+        writers.forEach(
+                (writerId, writerStateEntry) ->
+                        snapshotEntries.add(
+                                new WriterSnapshotEntry(
+                                        writerId,
+                                        writerStateEntry.lastBatchSequence(),
+                                        writerStateEntry.lastDataOffset(),
+                                        writerStateEntry.lastOffsetDelta(),
+                                        writerStateEntry.lastBatchTimestamp())));
+        byte[] jsonBytes = new WriterSnapshotMap(snapshotEntries).toJsonBytes();
+
+        return new String(jsonBytes);
+    }
+
     /** Writer snapshot map json serde. */
     public static class WriterSnapshotMapJsonSerde
             implements JsonSerializer<WriterSnapshotMap>, JsonDeserializer<WriterSnapshotMap> {