avalonalex opened a new pull request, #28810:
URL: https://github.com/apache/flink/pull/28810
## What is the purpose of the change
`VariantSerializer#deserialize` fills the value and metadata byte arrays
with `DataInputView#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#deserialize` uses `readFully` for the value and
metadata arrays, matching every other array-reading serializer in flink-core
(e.g. `BytePrimitiveArraySerializer`).
- Added `VariantSerializerShortReadTest`: a `DataInputView` wrapper 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:
- New unit test `VariantSerializerShortReadTest` fails on the previous
implementation and passes with the fix.
- Existing `VariantSerializerTest` and `VariantSerializerUpgradeTest` pass
unchanged.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with
`@Public(Evolving)`: no
- The serializers: yes — behavior only; the serialized wire format is
unchanged, so the change is fully snapshot-compatible and needs no migration or
serializer-upgrade path
- The runtime per-record code paths (performance sensitive): yes —
`deserialize` switches `read` to `readFully`, the same pattern the other
array-reading serializers use; no extra work when the first read fills the
buffer
- Anything that affects deployment or recovery: JobManager (and its
components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes — fixes restore
correctness for checkpoints/savepoints containing Variant keyed state; no
format change
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? not applicable
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (Claude Code)
Generated-by: Claude Code (Anthropic Claude)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
