This is an automated email from the ASF dual-hosted git repository.
ferenc-csaky pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 43bb871e5d1 [FLINK-40227][core] Use `readFully` to read `Variant`
value/metadata in `VariantSerializer.deserialize`
43bb871e5d1 is described below
commit 43bb871e5d1a3edb6122561dc64c22d8d51a0919
Author: Yuhan Hao <[email protected]>
AuthorDate: Fri Jul 31 04:46:28 2026 -0700
[FLINK-40227][core] Use `readFully` to read `Variant` value/metadata in
`VariantSerializer.deserialize`
---
.../common/typeutils/base/VariantSerializer.java | 6 +-
.../base/VariantSerializerShortReadTest.java | 97 ++++++++++++++++++++++
2 files changed, 101 insertions(+), 2 deletions(-)
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 294c30e548b..6ad27c54e06 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 class VariantSerializer extends
TypeSerializerSingleton<Variant> {
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 00000000000..cf81c4d6d12
--- /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));
+ }
+ }
+}
