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

[Shawn-Hx]

• Yes, generative AI was used: Codex

Purpose

Linked issue: close #3933

When a streaming lake or KV hybrid split reached snapshot EOF, no element was handed to the SourceReader mailbox. The split state therefore remained unfinished until the first log record arrived. A checkpoint taken in that gap restored the snapshot as unfinished, reopening and rescanning it, and also delayed the KV snapshot completion event.

This change records the snapshot-to-log boundary as soon as EOF is consumed, without advancing the saved log offset or emitting a user record.

Brief change log

• Add an internal snapshot-phase-finished marker shared by streaming lake and KV hybrid splits.
• Emit the marker at bounded snapshot EOF while keeping the hybrid split active for log consumption.
• Consume the marker in the mailbox thread and update only the corresponding completion flag.
• Add mailbox, checkpoint, and restore regressions for both lake and KV hybrid splits.
• Keep batch split completion and existing checkpoint serialization formats unchanged.

Tests

• FlinkSourceReaderTest, FlinkSourceSplitReaderTest, RecordAndPosTest, SourceSplitSerializerTest, FlinkRecordEmitterTest, and SourceSplitStateTest
• ./mvnw spotless:check -pl fluss-flink/fluss-flink-common
• Module verification: 593 unit tests and 56 integration tests passed

API and Format

No user-facing API or checkpoint/storage format changes. The completion fields were already serialized, so no serializer version bump is required.

Documentation

No documentation changes are required for this bug fix.

[Copilot]

Pull request overview

Warning

Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.

Fixes a checkpoint/restore gap where streaming hybrid (KV snapshot + log) and streaming lake splits could reach snapshot EOF without marking the snapshot phase finished until the first log record arrived, causing restored checkpoints to rescan finished snapshots.

Changes:

• Introduces an internal RecordAndPos snapshot-phase-finished marker emitted at bounded snapshot EOF for streaming hybrid and streaming lake splits.
• Updates mailbox-thread emitters to consume the marker and only flip the corresponding “snapshot finished” flag without advancing log offsets.
• Adds regression tests covering mailbox delivery, checkpointing, and restore behavior before the first log record.

Reviewed changes

Copilot reviewed 10 out of 10 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/RecordAndPosTest.java	Adds unit test for the new snapshot-finished marker semantics.
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReaderTest.java	Extends split reader tests to validate receiving a single snapshot-finished marker per split.
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/reader/FlinkSourceReaderTest.java	Adds checkpoint/restore regressions ensuring snapshot completion is captured before any log record arrives.
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/split/HybridSnapshotLogSplitState.java	Adds explicit API to mark snapshot finished without moving offsets.
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/RecordAndPos.java	Adds marker flag + factory method; updates equality/hash/toString accordingly.
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkSourceSplitReader.java	Emits marker at bounded snapshot EOF for streaming hybrid and streaming lake splits.
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/reader/FlinkRecordsWithSplitIds.java	Ensures marker bypasses null record offset logic and is forwarded.
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/emitter/FlinkRecordEmitter.java	Consumes marker to flip hybrid snapshot-finished state without emitting user data.
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/state/LakeSnapshotAndFlussLogSplitState.java	Adds explicit API to mark lake snapshot finished without moving offsets.
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeRecordRecordEmitter.java	Consumes marker to flip lake snapshot-finished state without emitting user data.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[luoyuxia]

@Shawn-Hx Thanks for the pr. The change lgtm. @loserwang1024 Could you please have another review?