Repository navigation
fix(spark): serialize connector records under Kryo without a registrator - #25
Merged
Merged
Conversation
Jobs that run with spark.serializer=KryoSerializer, which Uniffle remote
shuffle requires, cannot send the connector's records. Kryo's default
FieldSerializer fails to build a serializer for VortexWriterCommitMessage,
VortexFilePartition and ResolvedTransform ("can't get field offset on a
record class"), so a write job aborts when it returns its commit
messages to the driver. Kryo also corrupts PartitionedWriterCommitMessage,
because it rebuilds the immutable List.of() it holds by calling add().
Until now this was worked around with an out-of-tree KryoRegistrator jar
that registered the records with JavaSerializer. Registering classes
gives them numeric Kryo IDs. Uniffle creates the driver's Kryo instance
early, along a different path from the executors, so the IDs could
differ between the two sides. That would corrupt unrelated objects sent
over the wire, and is the likely cause of the HighlyCompressedMapStatus
deserialization failures ("Index 48 out of bounds for length 3").
Annotate the four types with @DefaultSerializer(JavaSerializer.class).
Kryo now picks Java serialization for them without any registration,
so they take no ID slot. Users no longer need spark.kryo.registrator or
a driver extraClassPath entry. Jobs with
spark.kryo.registrationRequired=true can list the four classes in
spark.kryo.classesToRegister, and Kryo still uses the annotation.
KryoSerializationTest round-trips each type through Spark's
KryoSerializer with spark.kryo.unsafe both off and on, and with the
types registered by name under registrationRequired. VortexKryoWriteTest
runs partitioned and unpartitioned writes and reads in a Kryo session.
Without the annotations every case fails, and the write jobs abort with
the "can't get field offset on a record class" error. The test JVM opens
java.lang.invoke and java.util, which Spark's KryoSerializer needs on
Java 17 and which Spark's launcher normally adds.
Signed-off-by: Anirudh Narasimman <[email protected]>
Co-Authored-By: Claude Opus 5.5 <[email protected]>
| * @param bucketCount -1 if not a bucket transform | ||
| */ | ||
| // Kryo's FieldSerializer cannot rebuild records, so under KryoSerializer this uses Java serialization. | ||
| @DefaultSerializer(JavaSerializer.class) |
Collaborator
There was a problem hiding this comment.
Always alarm bells whenever I see JavaSerializer - but probably fine.
Collaborator
|
This is really a Spark ↔ Vortex bug rather than a Uniffle one: upstream
Also, the PR body says it relies on |
ChristopherWirt
approved these changes
Oct 29, 2026
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
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.
Jobs that run with spark.serializer=KryoSerializer, which Uniffle remote shuffle requires, cannot send the connector's records. Kryo's default FieldSerializer fails to build a serializer for VortexWriterCommitMessage, VortexFilePartition and ResolvedTransform ("can't get field offset on a record class" with spark.kryo.unsafe=true). It also corrupts PartitionedWriterCommitMessage, because it rebuilds the immutable List.of() it holds by calling add().
Until now this was worked around with an out-of-tree KryoRegistrator jar that registered the records with JavaSerializer. Registering classes gives them numeric Kryo IDs. Uniffle creates the driver's Kryo instance early, along a different path from the executors, so the IDs could differ between the two sides. That would corrupt unrelated objects sent over the wire, and is the likely cause of the HighlyCompressedMapStatus deserialization failures ("Index 48 out of bounds for length 3").
Annotate the four types with @DefaultSerializer(JavaSerializer.class). Kryo now picks Java serialization for them without any registration, so they take no ID slot. Users no longer need spark.kryo.registrator or a driver extraClassPath entry. This relies on spark.kryo.registrationRequired being false, which is Spark's default.
KryoSerializationTest round-trips each type through Spark's KryoSerializer with spark.kryo.unsafe both off and on; all cases fail without the annotations. The test JVM opens java.lang.invoke and java.util, which Spark's KryoSerializer needs on Java 17 and which Spark's launcher normally adds.