Spark: Extract streaming initial offset store - #18092
Open
huan233usc wants to merge 4 commits into
Open
huan233usc wants to merge 4 commits into
huan233usc wants to merge 4 commits into
Conversation
Move initial-offset checkpoint persistence into a package-private helper with a lazy offset supplier. Preserve checkpoint format and Hadoop FileIO behavior, and verify that recovery does not recompute the initial offset. Split from apache#17953. Generated-by: Codex
added 3 commits
September 13, 2026 15:01
Move nested checkpoint creation coverage to the offset store unit tests. Verify restoration of existing checkpoints and empty-table start offsets without recomputing the initial offset. Check the Iceberg source checkpoint in the retained Hadoop FileIO integration test. Generated-by: Codex
Keep the initial offset store refactor focused on extraction and helper unit tests. Restore the existing Hadoop FileIO integration test. Generated-by: Codex
Add Delta-inspired coverage for unknown, invalid, and missing format versions, malformed JSON, and both initial-snapshot flag values. Verify invalid checkpoints fail without reinitializing or overwriting the offset. Generated-by: Codex
uros-b
reviewed
Sep 15, 2026
uros-b
left a comment
uros-b
left a comment
Member
There was a problem hiding this comment.
Looks like a nice, behavior-preserving extraction, thank you @huan233usc for working on this! The checkpoint path, HadoopFileIO, JSON format, and read-before-write guard are byte-identical to the old inner class, so existing checkpoints recover after upgrade.
Member
|
Please ping committers with more expertise in Iceberg Spark, for further review. e.g. @aokolnychyi has context on streaming source / checkpoint path changes |
szehon-ho
approved these changes
Sep 17, 2026
szehon-ho
left a comment
szehon-ho
left a comment
Member
There was a problem hiding this comment.
lgtm , clean refactor, class is package private
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Spark 4.2 currently embeds initial-offset checkpoint persistence inside
SparkMicroBatchStream. Extract it into a package-privateStreamingInitialOffsetStoreso other streaming sources can reuse the same checkpoint handling.The existing stream supplies its starting-offset calculation lazily, after checking for a persisted offset. The checkpoint location (
offsets/0), JSON format, and use of Hadoop FileIO remain unchanged. Unit tests cover creating nested checkpoint directories, restoring an existing checkpoint, and restoring an empty-table start offset without invoking the initializer again.Following the applicable cases in DeltaSourceOffsetSuite, the store tests also cover unknown, non-integer, and missing format versions, malformed JSON, and both values of the initial-snapshot flag. Invalid checkpoints must fail without invoking the initializer or overwriting the persisted file.
This is the first preparatory refactor split from #17953 and can merge independently.
Validation:
TestStreamingInitialOffsetStore(9 cases),TestStreamingOffset(2 cases),TestStreamingCheckpointHadoopIO, and the checkpoint-resume and timestamp-start cases inTestStructuredStreamingRead3; Spark 4.2 Spotless check.AI Disclosure