wip

Author: LiebingYu

fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java

@@ -631,6 +631,22 @@ public class ConfigOptions {
                                     + "we would fsync after every message; if it were 5 we would fsync after every "
                                     + "five messages.");
 
+    public static final ConfigOption<Duration> LOG_FLUSH_RECOVERY_OFFSET_CHECKPOINT_INTERVAL =
+            key("log.flush.recovery-offset.checkpoint-interval")
+                    .durationType()
+                    .defaultValue(Duration.ofMinutes(1))
+                    .withDescription(
+                            "The frequency with which the log start offset is saved out to disk. "
+                                    + "The default setting is 1 minute.");
+
+    public static final ConfigOption<Duration> LOG_FLUSH_START_OFFSET_CHECKPOINT_INTERVAL =
+            key("log.flush.start-offset.checkpoint-interval")
+                    .durationType()
+                    .defaultValue(Duration.ofMinutes(1))
+                    .withDescription(
+                            "The frequency with which the log start offset is saved out to disk. "
+                                    + "The default setting is 1 minute.");
+
     public static final ConfigOption<Duration> LOG_REPLICA_HIGH_WATERMARK_CHECKPOINT_INTERVAL =
             key("log.replica.high-watermark.checkpoint-interval")
                     .durationType()

fluss-rpc/src/main/java/org/apache/fluss/rpc/entity/FetchLogResultForBucket.java

@@ -35,28 +35,34 @@ public class FetchLogResultForBucket extends ResultForBucket {
     private final @Nullable RemoteLogFetchInfo remoteLogFetchInfo;
     private final @Nullable LogRecords records;
     private final long highWatermark;
+    private final long logStartOffset;
 
     public FetchLogResultForBucket(
-            TableBucket tableBucket, LogRecords records, long highWatermark) {
+            TableBucket tableBucket, LogRecords records, long highWatermark, long logStartOffset) {
         this(
                 tableBucket,
                 null,
                 checkNotNull(records, "records can not be null"),
                 highWatermark,
+                logStartOffset,
                 ApiError.NONE);
     }
 
     public FetchLogResultForBucket(TableBucket tableBucket, ApiError error) {
-        this(tableBucket, null, null, -1L, error);
+        this(tableBucket, null, null, -1L, 0L, error);
     }
 
     public FetchLogResultForBucket(
-            TableBucket tableBucket, RemoteLogFetchInfo remoteLogFetchInfo, long highWatermark) {
+            TableBucket tableBucket,
+            RemoteLogFetchInfo remoteLogFetchInfo,
+            long highWatermark,
+            long logStartOffset) {
         this(
                 tableBucket,
                 checkNotNull(remoteLogFetchInfo, "remote log fetch info can not be null"),
                 null,
                 highWatermark,
+                logStartOffset,
                 ApiError.NONE);
     }
 
@@ -65,11 +71,13 @@ private FetchLogResultForBucket(
             @Nullable RemoteLogFetchInfo remoteLogFetchInfo,
             @Nullable LogRecords records,
             long highWatermark,
+            long logStartOffset,
             ApiError error) {
         super(tableBucket, error);
         this.remoteLogFetchInfo = remoteLogFetchInfo;
         this.records = records;
         this.highWatermark = highWatermark;
+        this.logStartOffset = logStartOffset;
     }
 
     /**
@@ -102,4 +110,8 @@ public LogRecords recordsOrEmpty() {
     public long getHighWatermark() {
         return highWatermark;
     }
+
+    public long getLogStartOffset() {
+        return logStartOffset;
+    }
 }

fluss-rpc/src/main/java/org/apache/fluss/rpc/util/CommonRpcMessageUtils.java

@@ -195,15 +195,22 @@ public static FetchLogResultForBucket getFetchLogResultForBucket(
                                 pbRlfInfo.getFirstStartPos());
                 fetchLogResultForBucket =
                         new FetchLogResultForBucket(
-                                tb, rlFetchInfo, respForBucket.getHighWatermark());
+                                tb,
+                                rlFetchInfo,
+                                respForBucket.getHighWatermark(),
+                                respForBucket.getLogStartOffset());
             } else {
                 ByteBuffer recordsBuffer = toByteBuffer(respForBucket.getRecordsSlice());
                 LogRecords records =
                         respForBucket.hasRecords()
                                 ? MemoryLogRecords.pointToByteBuffer(recordsBuffer)
                                 : MemoryLogRecords.EMPTY;
                 fetchLogResultForBucket =
-                        new FetchLogResultForBucket(tb, records, respForBucket.getHighWatermark());
+                        new FetchLogResultForBucket(
+                                tb,
+                                records,
+                                respForBucket.getHighWatermark(),
+                                respForBucket.getLogStartOffset());
             }
         }
 

fluss-rpc/src/main/proto/FlussApi.proto

@@ -674,7 +674,7 @@ message PbFetchLogRespForBucket {
   optional int32 error_code = 3;
   optional string error_message = 4;
   optional int64 high_watermark = 5;
-  optional int64 log_start_offset = 6; // TODO now we don't introduce log start offset, but remain it in protobuf
+  optional int64 log_start_offset = 6;
   optional PbRemoteLogFetchInfo remote_log_fetch_info = 7;
   optional bytes records = 8;
 }

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

@@ -24,15 +24,24 @@
 /** A class to represent the loaded log offsets. */
 @Internal
 final class LoadedLogOffsets {
+    private final long logStartOffset;
     private final long recoveryPoint;
     private final LogOffsetMetadata nextOffsetMetadata;
 
-    public LoadedLogOffsets(final long recoveryPoint, final LogOffsetMetadata nextOffsetMetadata) {
+    public LoadedLogOffsets(
+            final long logStartOffset,
+            final long recoveryPoint,
+            final LogOffsetMetadata nextOffsetMetadata) {
+        this.logStartOffset = logStartOffset;
         this.recoveryPoint = recoveryPoint;
         this.nextOffsetMetadata =
                 Objects.requireNonNull(nextOffsetMetadata, "nextOffsetMetadata should not be null");
     }
 
+    public long getLogStartOffset() {
+        return logStartOffset;
+    }
+
     public long getRecoveryPoint() {
         return recoveryPoint;
     }
@@ -50,20 +59,24 @@ public boolean equals(Object o) {
             return false;
         }
         final LoadedLogOffsets that = (LoadedLogOffsets) o;
-        return recoveryPoint == that.recoveryPoint
+        return logStartOffset == that.logStartOffset
+                && recoveryPoint == that.recoveryPoint
                 && nextOffsetMetadata.equals(that.nextOffsetMetadata);
     }
 
     @Override
     public int hashCode() {
-        int result = Long.hashCode(recoveryPoint);
+        int result = Long.hashCode(logStartOffset);
+        result = 31 * result + Long.hashCode(recoveryPoint);
         result = 31 * result + nextOffsetMetadata.hashCode();
         return result;
     }
 
     @Override
     public String toString() {
         return "LoadedLogOffsets("
+                + "logStartOffset="
+                + logStartOffset
                 + ", recoveryPoint="
                 + recoveryPoint
                 + ", nextOffsetMetadata="

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

@@ -169,7 +169,7 @@ void checkIfMemoryMappedBufferClosed() {
         }
     }
 
-    private void updateRecoveryPoint(long newRecoveryPoint) {
+    public void updateRecoveryPoint(long newRecoveryPoint) {
         recoveryPoint = newRecoveryPoint;
     }
 

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

@@ -53,6 +53,7 @@ final class LogLoader {
     private final Configuration conf;
     private final LogSegments logSegments;
     private final long recoveryPointCheckpoint;
+    private final long logStartOffsetCheckpoint;
     private final LogFormat logFormat;
     private final WriterStateManager writerStateManager;
     private final boolean isCleanShutdown;
@@ -62,13 +63,15 @@ public LogLoader(
             Configuration conf,
             LogSegments logSegments,
             long recoveryPointCheckpoint,
+            long logStartOffsetCheckpoint,
             LogFormat logFormat,
             WriterStateManager writerStateManager,
             boolean isCleanShutdown) {
         this.logTabletDir = logTabletDir;
         this.conf = conf;
         this.logSegments = logSegments;
         this.recoveryPointCheckpoint = recoveryPointCheckpoint;
+        this.logStartOffsetCheckpoint = logStartOffsetCheckpoint;
         this.logFormat = logFormat;
         this.writerStateManager = writerStateManager;
         this.isCleanShutdown = isCleanShutdown;
@@ -94,6 +97,8 @@ public LoadedLogOffsets load() throws IOException {
         newRecoveryPoint = result.f0;
         nextOffset = result.f1;
 
+        // TODO: revisit here if we need to update logStartOffset if remote log disable
+
         // Any segment loading or recovery code must not use writerStateManager, so that we can
         // build the full state here from scratch.
         if (!writerStateManager.isEmpty()) {
@@ -119,6 +124,7 @@ public LoadedLogOffsets load() throws IOException {
         LogSegment activeSegment = logSegments.lastSegment().get();
         activeSegment.resizeIndexes((int) conf.get(ConfigOptions.LOG_INDEX_FILE_SIZE).getBytes());
         return new LoadedLogOffsets(
+                logStartOffsetCheckpoint,
                 newRecoveryPoint,
                 new LogOffsetMetadata(
                         nextOffset, activeSegment.getBaseOffset(), activeSegment.getSizeInBytes()));