add writer state log

Author: LiebingYu

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

@@ -113,8 +113,10 @@ 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, before rebuild: {}", writerStateManager.toJsonString());
         LogTablet.rebuildWriterState(
                 writerStateManager, logSegments, 0, nextOffset, isCleanShutdown);
+        LOG.info("In load, after rebuild: {}", writerStateManager.toJsonString());
 
         LogSegment activeSegment = logSegments.lastSegment().get();
         activeSegment.resizeIndexes((int) conf.get(ConfigOptions.LOG_INDEX_FILE_SIZE).getBytes());
@@ -264,12 +266,14 @@ 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, before rebuild: {}", 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, after rebuild: {}", writerStateManager.toJsonString());
         return bytesTruncated;
     }
 

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> {