[I] [Bug] Lake snapshot completion is not checkpointed before changelog records [fluss]

["luoyuxia (via GitHub)" <gi...@apache.org>]

luoyuxia opened a new issue, #3933:
URL: https://github.com/apache/fluss/issues/3933

   ### Search before asking
   
   - [x] I searched in the [issues](https://github.com/apache/fluss/issues) and found no issue covering this lake EOF state transition.
   
   ### Fluss version
   
   main (development)
   
   ### Please describe the bug 🐞
   
   For a streaming union read of a lake-enabled primary-key table, a
   `LakeSnapshotAndFlussLogSplit` remains active after its bounded lake snapshot is
   consumed because the same source split continues reading the Fluss changelog.
   
   The reader currently persists `isLakeSplitFinished=true` only after emitting
   the first changelog record. Reaching the end of the lake snapshot closes the
   bounded reader but does not update the checkpointed split state.
   
   A failure can be reproduced with the following sequence:
   
   1. Start a streaming full scan of a lake-enabled primary-key table.
   2. Consume all records from the lake snapshot for a hybrid split.
   3. Do not produce any new changelog records for that split.
   4. Complete a checkpoint while the source is waiting for changelog data.
   5. Expire or compact the old lake snapshot so that the serialized `DataSplit`
      or `FileScanTask` references files that are no longer available.
   6. Restore the job from the checkpoint.
   
   The checkpoint still contains `isLakeSplitFinished=false`. During restore,
   `LakeSplitReaderGenerator` therefore queues the bounded lake phase again and
   creates a reader from the serialized lake split. Paimon reopens the old
   `DataSplit`, and Iceberg reopens the old `FileScanTask`. Restore can fail when
   the referenced data or delete files have been removed by snapshot expiration,
   compaction, or retention cleanup.
   
   If the files still exist, restore unnecessarily reopens the final inner lake
   split and skips its already-consumed records. The state remains vulnerable
   until the first changelog record is emitted and advances the split to the log
   phase.
   
   Expected behavior:
   
   - Reaching the end of the complete lake snapshot immediately updates the
     checkpointed state to `isLakeSplitFinished=true`.
   - A restored reader skips the bounded lake phase and subscribes to the Fluss
     changelog from the existing starting offset.
   - The phase transition does not emit a user record or change the log offset.
   
   Actual behavior:
   
   - Lake EOF is not represented in source state.
   - The first changelog record is required to mark the lake phase as finished.
   - Recovery may depend on lake files that were fully consumed before the
     checkpoint and have since been deleted.
   
   ### Solution
   
   When a streaming hybrid bounded reader reaches lake EOF, send an internal
   lake-phase-finished control element through the SourceReader mailbox. The
   record emitter should consume this element by marking
   `LakeSnapshotAndFlussLogSplitState` as lake-finished and returning without
   calling `SourceOutput.collect()` or changing the log starting offset.
   
   The existing `isLakeSplitFinished` field is already serialized, so this does
   not require a split serializer version change.
   
   Add regression coverage that:
   
   - consumes the complete lake phase while the changelog remains empty;
   - checkpoints and restores the hybrid split with
     `isLakeSplitFinished=true`;
   - verifies that restore does not call `LakeSource.createRecordReader()`;
   - verifies that changelog subscription still starts from the original offset;
   - verifies that the control element produces no downstream record.
   
   ### Are you willing to submit a PR?
   
   - [x] I'm willing to submit a PR!
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [I] Lake snapshot completion is not checkpointed before changelog records [fluss]

["loserwang1024 (via GitHub)" <gi...@apache.org>]

loserwang1024 closed issue #3933: Lake snapshot completion is not checkpointed before changelog records
URL: https://github.com/apache/fluss/issues/3933


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org