[flink] Mark lake and KV snapshots finished at EOF before logs

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeRecordRecordEmitter.java

@@ -47,6 +47,11 @@ public void emitRecord(
             LakeSnapshotAndFlussLogSplitState lakeSnapshotAndFlussLogSplitState =
                     (LakeSnapshotAndFlussLogSplitState) splitState;
 
+            if (recordAndPos.isSnapshotPhaseFinished()) {
+                lakeSnapshotAndFlussLogSplitState.markLakeSplitFinished();
+                return;
+            }
+
             // set current split index
             lakeSnapshotAndFlussLogSplitState.setCurrentLakeSplitIndex(
                     recordAndPos.getCurrentSplitIndex());

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/state/LakeSnapshotAndFlussLogSplitState.java

@@ -49,6 +49,11 @@ public void setCurrentLakeSplitIndex(int currentLakeSplitIndex) {
         this.currentLakeSplitIndex = currentLakeSplitIndex;
     }
 
+    /** Marks the lake split as finished. */
+    public void markLakeSplitFinished() {
+        isLakeSplitFinished = true;
+    }
+
     public void setNextLogOffset(long nextOffset) {
         // if set offset, means lake splits is finished
         isLakeSplitFinished = true;

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/emitter/FlinkRecordEmitter.java

@@ -73,6 +73,11 @@ public void emitRecord(
             HybridSnapshotLogSplitState hybridSnapshotLogSplitState =
                     splitState.asHybridSnapshotLogSplitState();
 
+            if (recordAndPosition.isSnapshotPhaseFinished()) {
+                hybridSnapshotLogSplitState.markSnapshotFinished();
+                return;
+            }
+
             ScanRecord scanRecord = recordAndPosition.record();
             if (scanRecord.logOffset() >= 0) {
                 // record is with a valid offset, means it's in incremental phase,

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkRecordsWithSplitIds.java

@@ -119,6 +119,9 @@ public RecordAndPos nextRecordFromSplit() {
                         + "iterate over the records split.");
         if (currentRecordIterator.hasNext()) {
             RecordAndPos recordAndPos = currentRecordIterator.next();
+            if (recordAndPos.isSnapshotPhaseFinished()) {
+                return recordAndPos;
+            }
             long offset = recordAndPos.record().logOffset();
             // the record current offset is not less than the stopping offset,
             // shouldn't emit it

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReader.java

@@ -547,19 +547,28 @@ private long getStoppingOffset(TableBucket tableBucket) {
     }
 
     private FlinkRecordsWithSplitIds finishCurrentBoundedSplit() throws IOException {
-        Set<String> finishedSplits =
-                (currentBoundedSplit instanceof HybridSnapshotLogSplit
-                                        && !((HybridSnapshotLogSplit) currentBoundedSplit)
-                                                .isBatch())
-                                || (currentBoundedSplit instanceof LakeSnapshotAndFlussLogSplit
-                                        && ((LakeSnapshotAndFlussLogSplit) currentBoundedSplit)
-                                                .isStreaming())
-                        // is hybrid split, or is lakeAndFlussLog split in streaming mode,
-                        // not to finish this split
-                        // since it remains log to read
-                        ? Collections.emptySet()
-                        : Collections.singleton(currentBoundedSplit.splitId());
-        final FlinkRecordsWithSplitIds finishRecords = new FlinkRecordsWithSplitIds(finishedSplits);
+        boolean isStreamingHybridSplit =
+                currentBoundedSplit instanceof HybridSnapshotLogSplit
+                        && !((HybridSnapshotLogSplit) currentBoundedSplit).isBatch();
+        boolean isStreamingLakeSplit =
+                currentBoundedSplit instanceof LakeSnapshotAndFlussLogSplit
+                        && ((LakeSnapshotAndFlussLogSplit) currentBoundedSplit).isStreaming();
+        final FlinkRecordsWithSplitIds finishRecords;
+        if (isStreamingHybridSplit || isStreamingLakeSplit) {
+            // is hybrid split, or is lakeAndFlussLog split in streaming mode, send an internal
+            // record
+            // to set snapshot phase finished in split state
+            finishRecords =
+                    forBoundedSplitRecords(
+                            currentBoundedSplit,
+                            CloseableIterator.wrap(
+                                    Collections.singleton(RecordAndPos.snapshotPhaseFinished())
+                                            .iterator()));
+        } else {
+            finishRecords =
+                    new FlinkRecordsWithSplitIds(
+                            Collections.singleton(currentBoundedSplit.splitId()));
+        }
         closeCurrentBoundedSplit();
         return finishRecords;
     }

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/RecordAndPos.java

@@ -54,6 +54,8 @@ public class RecordAndPos {
     // the index for the current split that the record is from
     protected int currentSplitIndex;
 
+    private final boolean isSnapshotPhaseFinished;
+
     public RecordAndPos(ScanRecord scanRecord) {
         this(scanRecord, NO_READ_RECORDS_COUNT, DEFAULT_SPLIT_INDEX);
     }
@@ -63,9 +65,23 @@ public RecordAndPos(ScanRecord scanRecord, long readRecordsCount) {
     }
 
     public RecordAndPos(ScanRecord scanRecord, long readRecordsCount, int currentSplitIndex) {
+        this(scanRecord, readRecordsCount, currentSplitIndex, false);
+    }
+
+    private RecordAndPos(
+            ScanRecord scanRecord,
+            long readRecordsCount,
+            int currentSplitIndex,
+            boolean isSnapshotPhaseFinished) {
         this.scanRecord = scanRecord;
         this.readRecordsCount = readRecordsCount;
         this.currentSplitIndex = currentSplitIndex;
+        this.isSnapshotPhaseFinished = isSnapshotPhaseFinished;
+    }
+
+    /** Creates a marker indicating that the snapshot phase has been fully read. */
+    public static RecordAndPos snapshotPhaseFinished() {
+        return new RecordAndPos(null, NO_READ_RECORDS_COUNT, DEFAULT_SPLIT_INDEX, true);
     }
 
     public long readRecordsCount() {
@@ -80,6 +96,11 @@ public ScanRecord record() {
         return scanRecord;
     }
 
+    /** Returns whether this is a marker indicating that the snapshot phase has been fully read. */
+    public boolean isSnapshotPhaseFinished() {
+        return isSnapshotPhaseFinished;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {
@@ -91,12 +112,14 @@ public boolean equals(Object o) {
         RecordAndPos that = (RecordAndPos) o;
         return readRecordsCount == that.readRecordsCount
                 && currentSplitIndex == that.currentSplitIndex
+                && isSnapshotPhaseFinished == that.isSnapshotPhaseFinished
                 && Objects.equals(scanRecord, that.scanRecord);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(scanRecord, readRecordsCount, currentSplitIndex);
+        return Objects.hash(
+                scanRecord, readRecordsCount, currentSplitIndex, isSnapshotPhaseFinished);
     }
 
     @Override
@@ -108,6 +131,7 @@ public String toString() {
                 + readRecordsCount
                 + ", currentSplitIndex="
                 + currentSplitIndex
+                + (isSnapshotPhaseFinished ? ", isSnapshotPhaseFinished=true" : "")
                 + '}';
     }
 }

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/split/HybridSnapshotLogSplitState.java

@@ -53,6 +53,11 @@ public void setRecordsToSkip(long recordsToSkip) {
         this.recordsToSkip = recordsToSkip;
     }
 
+    /** Marks the snapshot phase as finished. */
+    public void markSnapshotFinished() {
+        snapshotFinished = true;
+    }
+
     public void setNextOffset(long nextOffset) {
         // if set offset, means snapshot is finished
         snapshotFinished = true;