[FLINK-40227][core] Use readFully to read Variant value/metadata in VariantSerializer.deserialize#28810
Open
avalonalex wants to merge 1 commit into
Open
[FLINK-40227][core] Use readFully to read Variant value/metadata in VariantSerializer.deserialize#28810avalonalex wants to merge 1 commit into
avalonalex wants to merge 1 commit into
Conversation
…ariantSerializer.deserialize VariantSerializer.deserialize filled the value and metadata byte arrays with DataInputView#read(byte[]) and ignored the returned count. That read is permitted to return a partial (short) read when the array straddles a buffer or compression-frame boundary in the source stream. A short read leaves the tail of the array unfilled and desyncs the stream, and when restoring large Variant keys from a checkpoint key-group stream on rescale it surfaces as MALFORMED_VARIANT (a misaligned read lands on a bad metadata header) or an OutOfMemoryError (a misaligned read lands on a length prefix). It is intermittent because whether an array straddles a boundary varies between restore attempts. Use readFully instead: it loops until each buffer is filled and throws EOFException only on a genuine end-of-stream. The on-wire format is unchanged, so this is fully snapshot-compatible and needs no state migration.
Collaborator
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.
What is the purpose of the change
VariantSerializer#deserializefills the value and metadata byte arrays withDataInputView#read(byte[])and ignores the returned count.read()may legally return fewer bytes than requested when the array straddles an internal buffer or compression-frame boundary; the array tail is then left unfilled and the stream desynchronizes. Restoring keyed state containing large Variant values after a parallelism change (key-group re-partitioning) fails intermittently with MALFORMED_VARIANT (misaligned read lands on a bad metadata header) or OutOfMemoryError (misaligned read lands on a length prefix), and jobs can crash-loop on restore. Observed in production on Flink 2.2.1 with a window operator keyed by Variant: a different subtask failed each restore attempt and the same checkpoint later restored cleanly — consistent with a nondeterministic short read, not a format mismatch.Brief change log
VariantSerializer#deserializeusesreadFullyfor the value and metadata arrays, matching every other array-reading serializer in flink-core (e.g.BytePrimitiveArraySerializer).VariantSerializerShortReadTest: aDataInputViewwrapper that legally returns short reads; deserialize corrupts data before the fix and round-trips correctly after.Verifying this change
This change added tests and can be verified as follows:
VariantSerializerShortReadTestfails on the previous implementation and passes with the fix.VariantSerializerTestandVariantSerializerUpgradeTestpass unchanged.Does this pull request potentially affect one of the following parts:
@Public(Evolving): nodeserializeswitchesreadtoreadFully, the same pattern the other array-reading serializers use; no extra work when the first read fills the bufferDocumentation
Was generative AI tooling used to co-author this PR?
Generated-by: Claude Code (Anthropic Claude)