From 9a932cfb03d567ed415afbbc3c5c0b637186dd76 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Wed, 5 Aug 2026 23:11:20 +0200 Subject: [PATCH 1/2] AVRO-4324: [Java] Guard ReflectDatumReader.readArray eager allocation ReflectDatumReader.readArray allocated the backing Java array for the declared array block count (Array.newInstance) before reading any element, unlike GenericDatumReader.readArray which already validates the count against the bytes remaining and caps element types whose minimum encoded size is zero. Apply the same guards (ensureAvailableCollectionBytes plus checkMaxCollectionAllocation for zero-byte element types) before the eager allocation, so a malformed or truncated record mapped to a Java array field (e.g. long[]) fails fast with an EOFException instead of over-allocating. Valid arrays continue to read unchanged. --- .../avro/reflect/ReflectDatumReader.java | 12 ++++++++++ .../avro/reflect/TestReflectDatumReader.java | 22 +++++++++++++++++++ 2 files changed, 34 insertions(+) diff --git a/lang/java/avro/src/main/java/org/apache/avro/reflect/ReflectDatumReader.java b/lang/java/avro/src/main/java/org/apache/avro/reflect/ReflectDatumReader.java index 7ba8e4827c6..9433c040600 100644 --- a/lang/java/avro/src/main/java/org/apache/avro/reflect/ReflectDatumReader.java +++ b/lang/java/avro/src/main/java/org/apache/avro/reflect/ReflectDatumReader.java @@ -32,6 +32,7 @@ import org.apache.avro.LogicalType; import org.apache.avro.Schema; import org.apache.avro.Schema.Field; +import org.apache.avro.SystemLimitException; import org.apache.avro.generic.IndexedRecord; import org.apache.avro.io.Decoder; import org.apache.avro.io.ResolvingDecoder; @@ -143,6 +144,17 @@ protected Object readArray(Object old, Schema expected, ResolvingDecoder in) thr if (l <= 0) { return newArray(old, 0, expected); } + // Match GenericDatumReader.readArray: before eagerly allocating the backing + // array for the declared block count, verify the input could plausibly hold + // that many elements (guarding against a malformed or truncated payload), + // and separately cap element types whose minimum encoded size is zero, which + // the bytes-remaining check cannot bound. Without this a small malformed + // record mapped to a Java array field (e.g. long[]) could drive a very large + // eager allocation before any element is read. + ensureAvailableCollectionBytes(in, l, expectedType); + if (isZeroByteSchema(expectedType)) { + SystemLimitException.checkMaxCollectionAllocation(0, l); + } Object array = newArray(old, (int) l, expected); if (array instanceof Collection) { @SuppressWarnings("unchecked") diff --git a/lang/java/avro/src/test/java/org/apache/avro/reflect/TestReflectDatumReader.java b/lang/java/avro/src/test/java/org/apache/avro/reflect/TestReflectDatumReader.java index ecd2cecb677..1dbce82bcc0 100644 --- a/lang/java/avro/src/test/java/org/apache/avro/reflect/TestReflectDatumReader.java +++ b/lang/java/avro/src/test/java/org/apache/avro/reflect/TestReflectDatumReader.java @@ -22,6 +22,7 @@ import static org.junit.jupiter.api.Assertions.assertThrows; import java.io.ByteArrayOutputStream; +import java.io.EOFException; import java.io.IOException; import java.util.Arrays; import java.util.HashSet; @@ -102,6 +103,27 @@ void read_PojoWithArray() throws IOException { assertEquals(pojoWithArray, deserialized); } + /** + * A malformed or truncated record can declare an array block count far larger + * than the data that follows. The reader must reject it before eagerly + * allocating the backing Java array, the same way GenericDatumReader does. + */ + @Test + void read_PojoWithArray_rejectsOversizedArrayCount() throws IOException { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + Encoder encoder = EncoderFactory.get().binaryEncoder(out, null); + encoder.writeInt(42); // record field "id" + encoder.writeLong(2_000_000_000L); // array block count for "relatedIds", with no items following + encoder.flush(); + + byte[] malformed = out.toByteArray(); + + Decoder decoder = DecoderFactory.get().binaryDecoder(malformed, null); + ReflectDatumReader reflectDatumReader = new ReflectDatumReader<>(PojoWithArray.class); + + assertThrows(EOFException.class, () -> reflectDatumReader.read(new PojoWithArray(), decoder)); + } + @Test public void testRead_PojoWithSet() throws IOException { PojoWithSet pojoWithSet = new PojoWithSet(); From fad19cd1b32b5af14d5f8d9ff71fc1bf670b0d02 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Isma=C3=ABl=20Mej=C3=ADa?= Date: Thu, 6 Aug 2026 11:26:40 +0200 Subject: [PATCH 2/2] AVRO-4324: Address review: bound cumulative allocation across array blocks The zero-byte allocation guard was only applied to the first array block, so a large logical array split across multiple blocks could pass the first check while the cumulative count still exceeded the heap-aware limit. Mirror GenericDatumReader by re-validating each continuation block (via a shared nextArrayBlock helper) against the bytes remaining and, for zero-byte element types, the cumulative allocation limit, in both readObjectArray and readCollection. --- .../avro/reflect/ReflectDatumReader.java | 38 +++++++++++++-- .../avro/reflect/TestReflectDatumReader.java | 47 +++++++++++++++++++ 2 files changed, 81 insertions(+), 4 deletions(-) diff --git a/lang/java/avro/src/main/java/org/apache/avro/reflect/ReflectDatumReader.java b/lang/java/avro/src/main/java/org/apache/avro/reflect/ReflectDatumReader.java index 9433c040600..bbd90d96e65 100644 --- a/lang/java/avro/src/main/java/org/apache/avro/reflect/ReflectDatumReader.java +++ b/lang/java/avro/src/main/java/org/apache/avro/reflect/ReflectDatumReader.java @@ -199,6 +199,7 @@ private Object readPrimitiveArray(Object array, Class c, long l, ResolvingDec private Object readObjectArray(Object[] array, Schema expectedType, long l, ResolvingDecoder in) throws IOException { LogicalType logicalType = expectedType.getLogicalType(); Conversion conversion = getData().getConversionFor(logicalType); + boolean zeroByte = isZeroByteSchema(expectedType); int index = 0; if (logicalType != null && conversion != null) { do { @@ -208,7 +209,7 @@ private Object readObjectArray(Object[] array, Schema expectedType, long l, Reso array[index] = element; index++; } - } while ((l = in.arrayNext()) > 0); + } while ((l = nextArrayBlock(in, expectedType, index, zeroByte)) > 0); } else { do { int limit = index + (int) l; @@ -217,7 +218,7 @@ private Object readObjectArray(Object[] array, Schema expectedType, long l, Reso array[index] = element; index++; } - } while ((l = in.arrayNext()) > 0); + } while ((l = nextArrayBlock(in, expectedType, index, zeroByte)) > 0); } return array; } @@ -226,24 +227,53 @@ private Object readCollection(Collection c, Schema expectedType, long l, throws IOException { LogicalType logicalType = expectedType.getLogicalType(); Conversion conversion = getData().getConversionFor(logicalType); + boolean zeroByte = isZeroByteSchema(expectedType); + long count = 0; if (logicalType != null && conversion != null) { do { for (int i = 0; i < l; i++) { Object element = readWithConversion(null, expectedType, logicalType, conversion, in); c.add(element); } - } while ((l = in.arrayNext()) > 0); + count += l; + } while ((l = nextArrayBlock(in, expectedType, count, zeroByte)) > 0); } else { do { for (int i = 0; i < l; i++) { Object element = readWithoutConversion(null, expectedType, in); c.add(element); } - } while ((l = in.arrayNext()) > 0); + count += l; + } while ((l = nextArrayBlock(in, expectedType, count, zeroByte)) > 0); } return c; } + /** + * Read and validate the next array block count, mirroring + * {@link org.apache.avro.generic.GenericDatumReader#readArray}: bound the + * declared count against the bytes remaining, and for element types whose + * minimum encoded size is zero bound the cumulative allocation (which the + * bytes-remaining check cannot). This closes the gap where a large logical + * array split across multiple blocks would otherwise pass only the first + * block's guard. + * + * @param in the decoder + * @param expectedType the array element schema + * @param existing the number of elements already read + * @param zeroByte whether the element type's minimum encoded size is zero + * @return the validated next block count + */ + private long nextArrayBlock(ResolvingDecoder in, Schema expectedType, long existing, boolean zeroByte) + throws IOException { + long l = in.arrayNext(); + ensureAvailableCollectionBytes(in, l, expectedType); + if (zeroByte && l > 0) { + SystemLimitException.checkMaxCollectionAllocation(existing, l); + } + return l; + } + @Override protected Object readString(Object old, Decoder in) throws IOException { return super.readString(null, in).toString(); diff --git a/lang/java/avro/src/test/java/org/apache/avro/reflect/TestReflectDatumReader.java b/lang/java/avro/src/test/java/org/apache/avro/reflect/TestReflectDatumReader.java index 1dbce82bcc0..56670639d0c 100644 --- a/lang/java/avro/src/test/java/org/apache/avro/reflect/TestReflectDatumReader.java +++ b/lang/java/avro/src/test/java/org/apache/avro/reflect/TestReflectDatumReader.java @@ -33,6 +33,7 @@ import java.util.Optional; import org.apache.avro.Schema; +import org.apache.avro.SystemLimitException; import org.apache.avro.io.Decoder; import org.apache.avro.io.DecoderFactory; import org.apache.avro.io.Encoder; @@ -124,6 +125,52 @@ void read_PojoWithArray_rejectsOversizedArrayCount() throws IOException { assertThrows(EOFException.class, () -> reflectDatumReader.read(new PojoWithArray(), decoder)); } + /** + * Elements whose minimum encoded size is zero (here an empty record) consume no + * input, so a large logical array can be split across blocks that each pass the + * first-block guard. The cumulative allocation must still be bounded across + * blocks, mirroring GenericDatumReader. + */ + @Test + void read_PojoWithZeroByteList_rejectsCumulativeCountAcrossBlocks() throws IOException { + System.setProperty(SystemLimitException.MAX_COLLECTION_ALLOCATION_PROPERTY, "1000"); + org.apache.avro.TestSystemLimitException.resetLimits(); + try { + // Two blocks of 600 zero-byte records each (1200 > 1000): the first block + // passes, the cumulative count must be rejected on the second block. + ByteArrayOutputStream out = new ByteArrayOutputStream(); + Encoder encoder = EncoderFactory.get().binaryEncoder(out, null); + encoder.writeLong(600L); + encoder.writeLong(600L); + encoder.writeLong(0L); // array terminator (not reached) + encoder.flush(); + + byte[] malformed = out.toByteArray(); + Decoder decoder = DecoderFactory.get().binaryDecoder(malformed, null); + ReflectDatumReader reader = new ReflectDatumReader<>(PojoWithZeroByteList.class); + + assertThrows(SystemLimitException.class, () -> reader.read(new PojoWithZeroByteList(), decoder)); + } finally { + System.clearProperty(SystemLimitException.MAX_COLLECTION_ALLOCATION_PROPERTY); + org.apache.avro.TestSystemLimitException.resetLimits(); + } + } + + /** + * An empty record encodes to zero bytes, making it a zero-byte element type. + */ + public static class EmptyRecord { + public EmptyRecord() { + } + } + + public static class PojoWithZeroByteList { + public List items; + + public PojoWithZeroByteList() { + } + } + @Test public void testRead_PojoWithSet() throws IOException { PojoWithSet pojoWithSet = new PojoWithSet();