[FLINK-40227][core] Use readFully to read Variant value/metadata in VariantSerializer.deserialize

[avalonalex]

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?

• Yes (Claude Code)

Generated-by: Claude Code (Anthropic Claude)

[flinkbot]

CI report:

• 83fcd9f Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[ferenc-csaky]

Thanks for the fix! Logic LGTM, but those LLM comments are way too much.

[ferenc-csaky]

Also, can you pls. open backport PR-s with the same content against the release-2.3 and release-2.2 branches? I think this would worth fixing in those versions too.

You can omit the PR description there, just link this one and prefix it with the backport version, e.g. [BP-2.2] like #28840.

[avalonalex]

Thanks for the review! Trimmed the comments as suggested. Backports are open: #28851 (release-2.3) and #28852 (release-2.2).

[ferenc-csaky]

LGTM, thanks for the quick response!