Yuhan Hao created FLINK-40227:
---------------------------------
Summary: 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
{\{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)
