Fix comments and notify the timestamp to the tablet server

Author: LiebingYu

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/FlussTableLakeSnapshotCommitter.java

@@ -88,6 +88,10 @@ public void commit(long tableId, CommittedLakeSnapshot committedLakeSnapshot)
             Long partitionId = partitionBucket.f0;
             if (partitionId == null) {
                 tableBucket = new TableBucket(tableId, partitionBucket.f1);
+                // we use -1 since we don't store timestamp in lake snapshot property for
+                // simplicity, it may cause the timestamp to be -1 during constructing lake
+                // snapshot to commit to Fluss.
+                // But it should happen rarely and should be a normal value after next tiering.
                 flussTableLakeSnapshot.addBucketOffsetAndTimestamp(
                         tableBucket, entry.getValue(), -1);
             } else {

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

@@ -474,6 +474,7 @@ message PbNotifyLakeTableOffsetReqForBucket {
   required int64 snapshot_id = 4;
   optional int64 log_start_offset = 5;
   optional int64 log_end_offset = 6;
+  optional int64 max_timestamp = 7;
 }
 
 message NotifyLakeTableOffsetResponse {

fluss-server/src/main/java/org/apache/fluss/server/entity/LakeBucketOffset.java

@@ -25,11 +25,14 @@ public class LakeBucketOffset {
 
     private final Long logStartOffset;
     private final Long logEndOffset;
+    private final Long maxTimestamp;
 
-    public LakeBucketOffset(long snapshotId, Long logStartOffset, Long logEndOffset) {
+    public LakeBucketOffset(
+            long snapshotId, Long logStartOffset, Long logEndOffset, Long maxTimestamp) {
         this.snapshotId = snapshotId;
         this.logStartOffset = logStartOffset;
         this.logEndOffset = logEndOffset;
+        this.maxTimestamp = maxTimestamp;
     }
 
     public long getSnapshotId() {
@@ -43,4 +46,8 @@ public Optional<Long> getLogStartOffset() {
     public Optional<Long> getLogEndOffset() {
         return Optional.ofNullable(logEndOffset);
     }
+
+    public Optional<Long> getMaxTimestamp() {
+        return Optional.ofNullable(maxTimestamp);
+    }
 }

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

@@ -123,6 +123,7 @@ public final class LogTablet {
     // note: currently, for primary key table, the log start offset nerve be updated
     private volatile long lakeLogStartOffset = Long.MAX_VALUE;
     private volatile long lakeLogEndOffset = -1L;
+    private volatile long lakeMaxTimestamp = -1;
 
     private LogTablet(
             PhysicalTablePath physicalPath,
@@ -248,6 +249,10 @@ public long getLakeLogEndOffset() {
         return lakeLogEndOffset;
     }
 
+    public long getLakeMaxTimestamp() {
+        return lakeMaxTimestamp;
+    }
+
     public int getWriterIdCount() {
         return writerStateManager.writerIdCount();
     }
@@ -507,6 +512,12 @@ public void updateLakeLogEndOffset(long lakeLogEndOffset) {
         }
     }
 
+    public void updateLakeMaxTimestamp(long lakeMaxTimestamp) {
+        if (lakeMaxTimestamp > this.lakeMaxTimestamp) {
+            this.lakeMaxTimestamp = lakeMaxTimestamp;
+        }
+    }
+
     public void loadWriterSnapshot(long lastOffset) throws IOException {
         synchronized (lock) {
             rebuildWriterState(lastOffset, writerStateManager);

fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java

@@ -720,6 +720,10 @@ public void notifyLakeTableOffset(
                                 .getLogEndOffset()
                                 .ifPresent(logTablet::updateLakeLogEndOffset);
 
+                        lakeBucketOffset
+                                .getMaxTimestamp()
+                                .ifPresent(logTablet::updateLakeMaxTimestamp);
+
                         responseCallback.accept(new NotifyLakeTableOffsetResponse());
                     }
                 });
@@ -779,6 +783,10 @@ private void updateWithLakeTableSnapshot(Replica replica) throws Exception {
             lakeTableSnapshot
                     .getLogEndOffset(tb)
                     .ifPresent(replica.getLogTablet()::updateLakeLogEndOffset);
+
+            lakeTableSnapshot
+                    .getMaxTimestamp(tb)
+                    .ifPresent(replica.getLogTablet()::updateLakeMaxTimestamp);
         }
     }
 

fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java

@@ -1490,6 +1490,8 @@ public static PbNotifyLakeTableOffsetReqForBucket makeNotifyLakeTableOffsetForBu
 
         lakeTableSnapshot.getLogEndOffset(tableBucket).ifPresent(reqForBucket::setLogEndOffset);
 
+        lakeTableSnapshot.getMaxTimestamp(tableBucket).ifPresent(reqForBucket::setMaxTimestamp);
+
         return reqForBucket;
     }
 
@@ -1509,8 +1511,11 @@ public static NotifyLakeTableOffsetData getNotifyLakeTableOffset(
                     reqForBucket.hasLogStartOffset() ? reqForBucket.getLogStartOffset() : null;
             Long logEndOffset =
                     reqForBucket.hasLogEndOffset() ? reqForBucket.getLogEndOffset() : null;
+            Long maxTimestamp =
+                    reqForBucket.hasMaxTimestamp() ? reqForBucket.getMaxTimestamp() : null;
             lakeBucketOffsetMap.put(
-                    tableBucket, new LakeBucketOffset(snapshotId, logStartOffset, logEndOffset));
+                    tableBucket,
+                    new LakeBucketOffset(snapshotId, logStartOffset, logEndOffset, maxTimestamp));
         }
 
         return new NotifyLakeTableOffsetData(

fluss-server/src/test/java/org/apache/fluss/server/replica/CommitLakeTableSnapshotITCase.java

@@ -131,6 +131,7 @@ void testCommitDataLakeData() throws Exception {
                         assertThat(logTablet.getLakeLogStartOffset())
                                 .isEqualTo(dataLakeLogStartOffset);
                         assertThat(logTablet.getLakeLogEndOffset()).isEqualTo(dataLakeLogEndOffset);
+                        assertThat(logTablet.getLakeMaxTimestamp()).isEqualTo(dataLakeMaxTimestamp);
                     });
         }
 

fluss-server/src/test/java/org/apache/fluss/server/replica/NotifyReplicaLakeTableOffsetTest.java

@@ -43,30 +43,37 @@ void testNotifyWithOutRemoteLog(boolean partitionedTable) throws Exception {
         Replica replica = replicaManager.getReplicaOrException(tb);
 
         // now, notify lake table offset
-        notifyAndVerify(tb, replica, 1, 0L, 20L);
+        notifyAndVerify(tb, replica, 1, 0L, 20L, System.currentTimeMillis());
         // notify again
-        notifyAndVerify(tb, replica, 2, 20L, 30L);
+        notifyAndVerify(tb, replica, 2, 20L, 30L, System.currentTimeMillis());
     }
 
     private void notifyAndVerify(
-            TableBucket tb, Replica replica, long snapshotId, long startOffset, long endOffset)
+            TableBucket tb,
+            Replica replica,
+            long snapshotId,
+            long startOffset,
+            long endOffset,
+            long maxTimestamp)
             throws Exception {
         NotifyLakeTableOffsetData notifyLakeTableOffsetData =
-                getNotifyLakeTableOffset(tb, snapshotId, startOffset, endOffset);
+                getNotifyLakeTableOffset(tb, snapshotId, startOffset, endOffset, maxTimestamp);
         CompletableFuture<NotifyLakeTableOffsetResponse> future = new CompletableFuture<>();
         replicaManager.notifyLakeTableOffset(notifyLakeTableOffsetData, future::complete);
         future.get();
-        verifyLakeTableOffset(replica, snapshotId, startOffset, endOffset);
+        verifyLakeTableOffset(replica, snapshotId, startOffset, endOffset, maxTimestamp);
     }
 
     private void verifyLakeTableOffset(
-            Replica replica, long snapshotId, long startOffset, long endOffset) {
+            Replica replica, long snapshotId, long startOffset, long endOffset, long maxTimestamp) {
         AssertionsForClassTypes.assertThat(replica.getLogTablet().getLakeTableSnapshotId())
                 .isEqualTo(snapshotId);
         AssertionsForClassTypes.assertThat(replica.getLogTablet().getLakeLogStartOffset())
                 .isEqualTo(startOffset);
         AssertionsForClassTypes.assertThat(replica.getLogTablet().getLakeLogEndOffset())
                 .isEqualTo(endOffset);
+        AssertionsForClassTypes.assertThat(replica.getLogTablet().getLakeMaxTimestamp())
+                .isEqualTo(maxTimestamp);
     }
 
     private TableBucket makeTableBucket(boolean partitionTable) {
@@ -82,10 +89,15 @@ private TableBucket makeTableBucket(long tableId, boolean partitionTable) {
     }
 
     private NotifyLakeTableOffsetData getNotifyLakeTableOffset(
-            TableBucket tableBucket, long snapshotId, long startOffset, long endOffset) {
+            TableBucket tableBucket,
+            long snapshotId,
+            long startOffset,
+            long endOffset,
+            long maxTimestamp) {
         return new NotifyLakeTableOffsetData(
                 1,
                 Collections.singletonMap(
-                        tableBucket, new LakeBucketOffset(snapshotId, startOffset, endOffset)));
+                        tableBucket,
+                        new LakeBucketOffset(snapshotId, startOffset, endOffset, maxTimestamp)));
     }
 }