[Spark][#36841] Register Spark's streaming internals with Kryo for the Structured Streaming runner - #39939
Merged
Merged
Conversation
A Structured Streaming query serializes two Spark internals behind the runner's back, so streaming pipelines fail spark.kryo.registrationRequired without these registrations: - StateSchemaMetadata is broadcast for every transformWithState query, hit on the very first micro-batch of any pipeline using Beam state or timers. - MemoryWriterCommitMessage is the memory sink's commit message, nested inside the already registered DataWritingSparkTaskResult. Both are registered by name because the shared base also compiles against Spark 3, where neither class exists, and with a JavaSerializer so their whole Scala object graph is covered without tracking Spark's internal field layout across versions. Neither is on a hot path. The registration call sits at the end of the registrator on purpose: Kryo auto assigns ids sequentially, so appending these conditional, by-name registrations keeps the auto assigned ids of everything above identical on Spark 3 and Spark 4 classpaths. SparkKryoRegistratorStreamingTest locks down both registrations and the id parity.
Contributor
|
Assigning reviewers: R: @kennknowles added as fallback since no labels match configuration Note: If you would like to opt out of this review, comment Available commands:
The PR bot will only process comments in the main thread (not review comments). |
Abacn
reviewed
Aug 31, 2026
Abacn
left a comment
Abacn
left a comment
Contributor
There was a problem hiding this comment.
Thanks! Had a few comments
Also have you checked a follow up from the previous PR? #39906 (comment)
| kryo.register(TupleTagList.class); | ||
|
|
||
| // Spark internals only present when running streaming pipelines on Spark 4. These are | ||
| // registered by name because the shared runner base also compiles against Spark 3, where |
Contributor
There was a problem hiding this comment.
because the shared runner base also compiles against Spark 3
this isn't accurate. The sources are compiled on either Spark 3 or 4, producing different Beam artifact.
In any case, please simplify the comment strings. It's unnecessarily verbose as AI tends to write down all its try path into comments
Contributor
Author
tkaymak
changed the title
Corrects the compile model description, the shared sources are compiled once per Spark version into separate artifacts, and trims the javadoc to the essentials per review.
Contributor
Author
|
Run Java PreCommit |
Contributor
Author
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
2 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.
Second slice of the Spark 4 Structured Streaming work split out of #39576, following the dispatch seam in #39906. Addresses #36841.
A Structured Streaming query serializes two Spark internals behind the runner's back, so streaming pipelines fail under spark.kryo.registrationRequired=true without these registrations:
Design notes:
SparkKryoRegistratorStreamingTest locks down both registrations and the id parity.
Full :runners:spark:3:test and :runners:spark:4:test suites green locally on JDK 17 plus spotless, checkstyle and a live ErrorProne compile.
Next slices in order: the DataSourceV2 unbounded source, the state and timer bridge on transformWithState, and the translators with the end to end tests. End to end evidence remains in draft #39576.
R: @Abacn