From 83fcd9fef699daa05b3a99d57d375fea66ff7a8a Mon Sep 17 00:00:00 2001 From: Yuhan Hao Date: Wed, 1 Jul 2026 14:56:00 -0700 Subject: [PATCH] [FLINK-40227][core] Use readFully to read Variant value/metadata in VariantSerializer.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. --- .../typeutils/base/VariantSerializer.java | 6 +- .../base/VariantSerializerShortReadTest.java | 97 +++++++++++++++++++ 2 files changed, 101 insertions(+), 2 deletions(-) create mode 100644 flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/VariantSerializerShortReadTest.java diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/VariantSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/VariantSerializer.java index 294c30e548bc24..6ad27c54e06635 100644 --- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/VariantSerializer.java +++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/VariantSerializer.java @@ -81,8 +81,10 @@ public Variant deserialize(DataInputView source) throws IOException { byte[] value = new byte[valueLength]; byte[] metaData = new byte[metadataLength]; - source.read(value); - source.read(metaData); + // readFully, not read: read(byte[]) may legally return a short read, leaving the array + // partially filled and the stream desynchronized. + source.readFully(value); + source.readFully(metaData); return new BinaryVariant(value, metaData); } diff --git a/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/VariantSerializerShortReadTest.java b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/VariantSerializerShortReadTest.java new file mode 100644 index 00000000000000..cf81c4d6d12942 --- /dev/null +++ b/flink-core/src/test/java/org/apache/flink/api/common/typeutils/base/VariantSerializerShortReadTest.java @@ -0,0 +1,97 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.api.common.typeutils.base; + +import org.apache.flink.core.memory.DataInputViewStreamWrapper; +import org.apache.flink.core.memory.DataOutputViewStreamWrapper; +import org.apache.flink.types.variant.Variant; +import org.apache.flink.types.variant.VariantBuilder; + +import org.junit.jupiter.api.Test; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.EOFException; +import java.io.FilterInputStream; +import java.io.IOException; +import java.util.Arrays; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Tests that {@link VariantSerializer#deserialize} reads the value/metadata arrays to completion + * even when the source returns short reads. + */ +class VariantSerializerShortReadTest { + + private static final VariantSerializer SERIALIZER = VariantSerializer.INSTANCE; + + /** A Variant large enough that its value and metadata each span many read chunks. */ + private static Variant largeVariant() { + VariantBuilder builder = Variant.newBuilder(); + VariantBuilder.VariantObjectBuilder object = builder.object(); + for (int i = 0; i < 2048; i++) { + object.add("resource-key-" + i, builder.of("resource-value-payload-" + i)); + } + return object.build(); + } + + private static byte[] serialize(Variant variant) throws IOException { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + SERIALIZER.serialize(variant, new DataOutputViewStreamWrapper(out)); + return out.toByteArray(); + } + + @Test + void deserializeRecoversFromShortReads() throws IOException { + Variant original = largeVariant(); + byte[] bytes = serialize(original); + + DataInputViewStreamWrapper source = + new DataInputViewStreamWrapper(new OneByteAtATimeInputStream(bytes)); + + Variant restored = SERIALIZER.deserialize(source); + + assertThat(restored).isEqualTo(original); + } + + @Test + void deserializeThrowsEofOnTruncatedStream() throws IOException { + byte[] bytes = serialize(largeVariant()); + byte[] truncated = Arrays.copyOf(bytes, bytes.length - 1); + + DataInputViewStreamWrapper source = + new DataInputViewStreamWrapper(new ByteArrayInputStream(truncated)); + + assertThatThrownBy(() -> SERIALIZER.deserialize(source)).isInstanceOf(EOFException.class); + } + + /** Returns at most one byte per {@code read(byte[], int, int)} call. */ + private static final class OneByteAtATimeInputStream extends FilterInputStream { + OneByteAtATimeInputStream(byte[] data) { + super(new ByteArrayInputStream(data)); + } + + @Override + public int read(byte[] b, int off, int len) throws IOException { + return super.read(b, off, Math.min(len, 1)); + } + } +}