[
https://issues.apache.org/jira/browse/FLINK-40227?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-40227:
-----------------------------------
Labels: pull-request-available (was: )
> VariantSerializer.deserialize ignores DataInputView#read return value,
> corrupting Variant state restore
> -------------------------------------------------------------------------------------------------------
>
> Key: FLINK-40227
> URL: https://issues.apache.org/jira/browse/FLINK-40227
> Project: Flink
> Issue Type: Bug
> Components: API / Type Serialization System
> Affects Versions: 2.2.0
> Reporter: Yuhan Hao
> Priority: Major
> Labels: pull-request-available
> Original Estimate: 168h
> Remaining Estimate: 168h
>
> {\{VariantSerializer#deserialize}} fills the value and metadata byte arrays
> using \{{DataInputView#read(byte\[\])}} and ignores the returned count
> (\{{VariantSerializer.java:84-85}} on master). Per the
> \{{InputStream}}/\{{DataInputView}} contract, \{{read}} may return fewer
> bytes than requested when the array straddles an internal buffer or
> compression-frame boundary. When that happens the tail of the array is left
> unfilled and the stream is desynchronized.
> In practice this surfaces when restoring keyed state containing large Variant
> values from a checkpoint key-group stream after a parallelism change
> (rescale): the misaligned read lands on a bad metadata header
> (\{{MALFORMED_VARIANT}}) or on a length prefix, causing a huge allocation and
> \{{OutOfMemoryError}}. It is intermittent because whether an array straddles
> a boundary varies between restore attempts — jobs can crash-loop on restore
> until manual intervention. Observed in production with a windowed operator
> keyed by Variant
> (\{{KeyGroupPartitioner$PartitioningResultKeyGroupReader.readMappingsInKeyGroup}}
> path): a different subtask failed each restore attempt and the same
> checkpoint later restored cleanly, consistent with a non-deterministic short
> read rather than a format mismatch.
> Fix: use \{{readFully}}, matching every other array-reading serializer in
> flink-core (e.g. \{{BytePrimitiveArraySerializer}}). The wire format is
> unchanged, so the fix is fully snapshot-compatible and needs no state
> migration. A patch with a deterministic regression test (a \{{DataInputView}}
> that legally returns short reads) is ready; PR to follow.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
