From 99bbbd85a5089dbc31a797750341b6d853e016a4 Mon Sep 17 00:00:00 2001 From: Robert Yokota Date: Thu, 15 Oct 2026 17:40:46 -0700 Subject: [PATCH] AVRO-4361: [java] Size collection guards from the writer's schema --- .../avro/generic/GenericDatumReader.java | 128 +++++++++++++-- .../org/apache/avro/io/FastReaderBuilder.java | 4 +- .../avro/reflect/ReflectDatumReader.java | 29 ++-- .../avro/generic/TestGenericDatumReader.java | 148 ++++++++++++++++++ .../avro/reflect/TestReflectDatumReader.java | 20 +++ 5 files changed, 299 insertions(+), 30 deletions(-) diff --git a/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader.java b/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader.java index 9d80d774981..fd81fd038a4 100644 --- a/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader.java +++ b/lang/java/avro/src/main/java/org/apache/avro/generic/GenericDatumReader.java @@ -56,6 +56,10 @@ public class GenericDatumReader implements DatumReader { private ResolvingDecoder creatorResolver = null; private final Thread creator; + // Whether the writer's and reader's schemas differ: -1 until computed, then 0 + // or 1. Reset whenever either schema changes. + private int resolving = -1; + public GenericDatumReader() { this(null, null, GenericData.get()); } @@ -99,6 +103,7 @@ public void setSchema(Schema writer) { } creatorResolver = null; fastDatumReader = null; + resolving = -1; } /** Get the reader's schema. */ @@ -110,6 +115,7 @@ public Schema getExpected() { public void setExpected(Schema reader) { this.expected = reader; creatorResolver = null; + resolving = -1; } private static final ThreadLocal>> RESOLVER_CACHE = ThreadLocalWithInitial @@ -327,7 +333,8 @@ protected Object readArray(Object old, Schema expected, ResolvingDecoder in) thr long l = in.readArrayStart(); long base = 0; if (l > 0) { - ensureAvailableCollectionBytes(in, l, expectedType); + int minBytes = minBytesToRead(expectedType); + ensureAvailableCollectionBytes(in, l, minBytes); // Elements whose minimum encoded size is zero (null, a zero-length fixed, a // record whose fields are all zero-byte, or a recursive schema where the // cycle is broken with a 0 minimum) consume no guaranteed input, so @@ -335,7 +342,7 @@ protected Object readArray(Object old, Schema expected, ResolvingDecoder in) thr // remaining. Cap such collections against a heap-aware limit so a tiny // payload cannot declare a huge block count and drive an unbounded // backing-array allocation. - boolean zeroByteElements = isZeroByteSchema(expectedType); + boolean zeroByteElements = minBytes == 0; if (zeroByteElements) { SystemLimitException.checkMaxCollectionAllocation(l); } @@ -354,7 +361,7 @@ protected Object readArray(Object old, Schema expected, ResolvingDecoder in) thr } } base += l; - l = arrayNext(in, expectedType); + l = arrayNext(in, minBytes); if (zeroByteElements && l > 0) { SystemLimitException.checkMaxCollectionAllocation(l); } @@ -369,9 +376,9 @@ protected Object readArray(Object old, Schema expected, ResolvingDecoder in) thr * Reads the next array block count and validates remaining bytes before the * caller allocates storage. */ - private long arrayNext(ResolvingDecoder in, Schema elementType) throws IOException { + private long arrayNext(ResolvingDecoder in, int minBytes) throws IOException { long l = in.arrayNext(); - ensureAvailableCollectionBytes(in, l, elementType); + ensureAvailableCollectionBytes(in, l, minBytes); return l; } @@ -410,7 +417,8 @@ protected Object readMap(Object old, Schema expected, ResolvingDecoder in) throw long l = in.readMapStart(); LogicalType logicalType = eValue.getLogicalType(); Conversion conversion = getData().getConversionFor(logicalType); - ensureAvailableMapBytes(in, l, eValue); + int minValueBytes = minBytesToRead(eValue); + ensureAvailableMapBytes(in, l, minValueBytes); Object map = newMap(old, initialCollectionCapacity(l)); if (l > 0) { do { @@ -424,7 +432,7 @@ protected Object readMap(Object old, Schema expected, ResolvingDecoder in) throw addToMap(map, readMapKey(null, expected, in), readWithoutConversion(null, eValue, in)); } } - } while ((l = mapNext(in, eValue)) > 0); + } while ((l = mapNext(in, minValueBytes)) > 0); } return map; } @@ -433,23 +441,23 @@ protected Object readMap(Object old, Schema expected, ResolvingDecoder in) throw * Reads the next map block count and validates remaining bytes before the * caller allocates storage. */ - private long mapNext(ResolvingDecoder in, Schema valueType) throws IOException { + private long mapNext(ResolvingDecoder in, int minValueBytes) throws IOException { long l = in.mapNext(); - ensureAvailableMapBytes(in, l, valueType); + ensureAvailableMapBytes(in, l, minValueBytes); return l; } /** * Validates remaining bytes for a map block. Each map entry has a string key - * (at least 1 byte for the length varint) plus a value, so the minimum bytes - * per entry is {@code 1 + minBytesPerElement(valueSchema)}. + * (at least 1 byte for the length varint) plus a value of at least + * {@code minValueBytes}. */ - private static void ensureAvailableMapBytes(Decoder decoder, long count, Schema valueSchema) throws EOFException { + private static void ensureAvailableMapBytes(Decoder decoder, long count, int minValueBytes) throws EOFException { if (count <= 0) { return; } // Map keys are always strings: at least 1 byte for the length varint - long minBytesPerEntry = 1L + minBytesPerElement(valueSchema); + long minBytesPerEntry = 1L + minValueBytes; int remaining = decoder.remainingBytes(); if (remaining >= 0 && count * minBytesPerEntry > remaining) { throw new EOFException("Map claims " + count + " entries with at least " + minBytesPerEntry @@ -538,6 +546,86 @@ private static int minBytesPerElement(Schema schema, Set visited) { } } + /** + * Returns the minimum number of bytes required to encode a value of any + * writer's schema that resolves to the given reader's schema. Used when the + * writer's element schema is not at hand, since a value can encode smaller than + * its reader's schema implies: the writer's record may lack a field the reader + * fills from its default, and a {@code float} or {@code double} may be promoted + * from a 1-byte {@code int} or {@code long}. + *

+ * Returns 0 where a resolved value can encode to no bytes, such as + * {@code null}, a zero-length {@code fixed}, or a record whose fields without + * defaults are all zero-byte. + */ + static int minResolvedBytesPerElement(Schema readerSchema) { + return minResolvedBytesPerElement(readerSchema, Collections.newSetFromMap(new IdentityHashMap<>())); + } + + private static int minResolvedBytesPerElement(Schema schema, Set visited) { + switch (schema.getType()) { + case NULL: + return 0; + case FIXED: + // A fixed only resolves from a fixed of the same size + return schema.getFixedSize(); + case RECORD: + if (!visited.add(schema)) { + return 0; // break recursion for self-referencing schemas + } + long sum = 0; + for (Schema.Field f : schema.getFields()) { + // The writer's record may lack a field that has a default + if (!f.hasDefaultValue()) { + sum += minResolvedBytesPerElement(f.schema(), visited); + if (sum >= Integer.MAX_VALUE) { + sum = Integer.MAX_VALUE; + break; + } + } + } + visited.remove(schema); + return (int) sum; + case UNION: + // A writer's union has a branch index of at least 1 byte, and any other + // writer's schema resolves to one of the branches + int min = 1; + for (Schema branch : schema.getTypes()) { + min = Math.min(min, minResolvedBytesPerElement(branch, visited)); + } + return min; + default: + // BOOLEAN, INT, LONG, ENUM, STRING, BYTES, ARRAY, MAP are all >= 1 byte, and + // FLOAT and DOUBLE may be promoted from a 1-byte INT or LONG + return 1; + } + } + + /** + * Returns the minimum number of bytes required to encode a value that this + * reader reads as the given reader's schema. When the writer's and reader's + * schemas are the same, this is {@link #minBytesPerElement}; otherwise it is + * {@link #minResolvedBytesPerElement}, since the value was encoded with the + * writer's schema. + * + * @param readerSchema the element (or map value) schema of the reader + * @return the minimum encoded size of a value read as {@code readerSchema} + */ + protected int minBytesToRead(Schema readerSchema) { + return isResolving() ? minResolvedBytesPerElement(readerSchema) : minBytesPerElement(readerSchema); + } + + private boolean isResolving() { + int r = resolving; + if (r < 0) { + Schema writer = actual; + Schema reader = expected; + r = writer == reader || (writer != null && writer.equals(reader)) ? 0 : 1; + resolving = r; + } + return r == 1; + } + /** * Validates that the decoder has enough remaining bytes to hold {@code count} * elements of the given schema, assuming each element requires at least @@ -547,14 +635,24 @@ private static int minBytesPerElement(Schema schema, Set visited) { * This check prevents out-of-memory errors from pre-allocating huge backing * arrays when the source data is truncated or malicious. It is exposed so the * fast reader ({@code FastReaderBuilder}) can apply the same guard as this - * classic reader. + * classic reader. The elements were encoded with the writer's schema, so + * {@code elementSchema} must be the writer's element schema. */ public static void ensureAvailableCollectionBytes(Decoder decoder, long count, Schema elementSchema) throws EOFException { + ensureAvailableCollectionBytes(decoder, count, minBytesPerElement(elementSchema)); + } + + /** + * Validates that the decoder has enough remaining bytes to hold {@code count} + * elements of at least {@code minBytes} bytes each, such as computed by + * {@link #minBytesToRead}. Throws {@link EOFException} if the decoder reports + * fewer remaining bytes than required. + */ + protected static void ensureAvailableCollectionBytes(Decoder decoder, long count, int minBytes) throws EOFException { if (count <= 0) { return; } - int minBytes = minBytesPerElement(elementSchema); if (minBytes > 0) { int remaining = decoder.remainingBytes(); if (remaining >= 0 && count * (long) minBytes > remaining) { diff --git a/lang/java/avro/src/main/java/org/apache/avro/io/FastReaderBuilder.java b/lang/java/avro/src/main/java/org/apache/avro/io/FastReaderBuilder.java index f8d66c7069b..f18e8092a21 100644 --- a/lang/java/avro/src/main/java/org/apache/avro/io/FastReaderBuilder.java +++ b/lang/java/avro/src/main/java/org/apache/avro/io/FastReaderBuilder.java @@ -473,7 +473,9 @@ private Optional> findClass(String clazz) { @SuppressWarnings("unchecked") private FieldReader createArrayReader(Schema readerSchema, Container action) throws IOException { FieldReader elementReader = getReaderFor(action.elementAction, null); - Schema elementType = readerSchema.getElementType(); + // The elements were encoded with the writer's schema, which may encode them + // smaller than the reader's (e.g. a field the reader fills from its default) + Schema elementType = action.writer.getElementType(); boolean zeroByteElements = GenericDatumReader.isZeroByteSchema(elementType); 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 0a82faf2c08..c51c804758d 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 @@ -151,8 +151,9 @@ protected Object readArray(Object old, Schema expected, ResolvingDecoder in) thr // 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)) { + int minBytes = minBytesToRead(expectedType); + ensureAvailableCollectionBytes(in, l, minBytes); + if (minBytes == 0) { SystemLimitException.checkMaxCollectionAllocation(l); } Object array = newArray(old, (int) l, expected); @@ -199,7 +200,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 minBytes = minBytesToRead(expectedType); int index = 0; if (logicalType != null && conversion != null) { do { @@ -209,7 +210,7 @@ private Object readObjectArray(Object[] array, Schema expectedType, long l, Reso array[index] = element; index++; } - } while ((l = nextArrayBlock(in, expectedType, zeroByte)) > 0); + } while ((l = nextArrayBlock(in, minBytes)) > 0); } else { do { int limit = index + (int) l; @@ -218,7 +219,7 @@ private Object readObjectArray(Object[] array, Schema expectedType, long l, Reso array[index] = element; index++; } - } while ((l = nextArrayBlock(in, expectedType, zeroByte)) > 0); + } while ((l = nextArrayBlock(in, minBytes)) > 0); } return array; } @@ -227,21 +228,21 @@ private Object readCollection(Collection c, Schema expectedType, long l, throws IOException { LogicalType logicalType = expectedType.getLogicalType(); Conversion conversion = getData().getConversionFor(logicalType); - boolean zeroByte = isZeroByteSchema(expectedType); + int minBytes = minBytesToRead(expectedType); 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 = nextArrayBlock(in, expectedType, zeroByte)) > 0); + } while ((l = nextArrayBlock(in, minBytes)) > 0); } else { do { for (int i = 0; i < l; i++) { Object element = readWithoutConversion(null, expectedType, in); c.add(element); } - } while ((l = nextArrayBlock(in, expectedType, zeroByte)) > 0); + } while ((l = nextArrayBlock(in, minBytes)) > 0); } return c; } @@ -257,15 +258,15 @@ private Object readCollection(Collection c, Schema expectedType, long l, * 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 zeroByte whether the element type's minimum encoded size is zero + * @param in the decoder + * @param minBytes the minimum encoded size of an element, from + * {@link #minBytesToRead} * @return the validated next block count */ - private long nextArrayBlock(ResolvingDecoder in, Schema expectedType, boolean zeroByte) throws IOException { + private long nextArrayBlock(ResolvingDecoder in, int minBytes) throws IOException { long l = in.arrayNext(); - ensureAvailableCollectionBytes(in, l, expectedType); - if (zeroByte && l > 0) { + ensureAvailableCollectionBytes(in, l, minBytes); + if (minBytes == 0 && l > 0) { SystemLimitException.checkMaxCollectionAllocation(l); } return l; diff --git a/lang/java/avro/src/test/java/org/apache/avro/generic/TestGenericDatumReader.java b/lang/java/avro/src/test/java/org/apache/avro/generic/TestGenericDatumReader.java index 51884efbb05..548ee22d50b 100644 --- a/lang/java/avro/src/test/java/org/apache/avro/generic/TestGenericDatumReader.java +++ b/lang/java/avro/src/test/java/org/apache/avro/generic/TestGenericDatumReader.java @@ -31,7 +31,9 @@ import java.util.Arrays; import java.util.Collection; import java.util.Collections; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.Random; import java.util.stream.Collectors; import java.util.stream.IntStream; @@ -185,6 +187,29 @@ void testMinBytesPerElementRecord() { assertEquals(12, GenericDatumReader.minBytesPerElement(recWithMultipleFields)); } + @Test + void testMinResolvedBytesPerElement() { + Schema nullType = Schema.create(Schema.Type.NULL); + Schema intType = Schema.create(Schema.Type.INT); + // FLOAT and DOUBLE may be promoted from a 1-byte INT or LONG + assertEquals(1, GenericDatumReader.minResolvedBytesPerElement(Schema.create(Schema.Type.FLOAT))); + assertEquals(1, GenericDatumReader.minResolvedBytesPerElement(Schema.create(Schema.Type.DOUBLE))); + assertEquals(0, GenericDatumReader.minResolvedBytesPerElement(nullType)); + assertEquals(16, GenericDatumReader.minResolvedBytesPerElement(Schema.createFixed("F16", null, "test", 16))); + // A non-union writer's schema may resolve to the null branch + assertEquals(0, GenericDatumReader.minResolvedBytesPerElement(Schema.createUnion(nullType, intType))); + assertEquals(1, + GenericDatumReader.minResolvedBytesPerElement(Schema.createUnion(intType, Schema.create(Schema.Type.DOUBLE)))); + + // A field with a default may be missing from the writer's record + Schema rec = Schema.createRecord("WithDefault", null, "test", false); + rec.setFields(Arrays.asList(new Schema.Field("x", intType), + new Schema.Field("y", Schema.create(Schema.Type.DOUBLE), null, 0.0), + new Schema.Field("z", Schema.createFixed("F8", null, "test", 8)))); + assertEquals(9, GenericDatumReader.minResolvedBytesPerElement(rec)); + assertEquals(17, GenericDatumReader.minBytesPerElement(rec)); + } + @Test void testMinBytesPerElementNestedCollections() { // Array and map types are >= 1 byte (count varint) @@ -751,4 +776,127 @@ void resolvingSkipOfHugeNullArrayFieldIsBounded() throws Exception { org.apache.avro.TestSystemLimitException.resetLimits(); } } + + private static final String ITEM_V1 = "{\"type\":\"record\",\"name\":\"Item\",\"fields\":[" + + "{\"name\":\"id\",\"type\":\"int\"}]}"; + private static final String ITEM_V2 = "{\"type\":\"record\",\"name\":\"Item\",\"fields\":[" + + "{\"name\":\"id\",\"type\":\"int\"},{\"name\":\"qty\",\"type\":\"int\",\"default\":0}]}"; + + private static byte[] encode(Schema schema, Object datum) throws IOException { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + BinaryEncoder enc = EncoderFactory.get().binaryEncoder(out, null); + new GenericDatumWriter<>(schema).write(datum, enc); + enc.flush(); + return out.toByteArray(); + } + + private static Object readWith(Schema writer, Schema reader, boolean fastReader, byte[] bytes) throws IOException { + GenericData data = new GenericData(); + data.setFastReaderEnabled(fastReader); + GenericDatumReader datumReader = new GenericDatumReader<>(writer, reader, data); + return datumReader.read(null, DecoderFactory.get().binaryDecoder(bytes, null)); + } + + /** + * The elements of a collection are encoded with the writer's schema, which can + * be smaller than the reader's: here the reader's element record adds a field + * with a default, so each 1-byte element is read as a 2-field record. + */ + @Test + void arrayOfRecordsWithAddedDefaultFieldDecodes() throws Exception { + Schema writer = Schema.createArray(new Schema.Parser().parse(ITEM_V1)); + Schema reader = Schema.createArray(new Schema.Parser().parse(ITEM_V2)); + List items = new ArrayList<>(); + for (int i = 0; i < 100; i++) { + GenericRecord item = new GenericData.Record(writer.getElementType()); + item.put("id", 0); + items.add(item); + } + byte[] bytes = encode(writer, items); + for (boolean fast : new boolean[] { true, false }) { + Collection result = (Collection) readWith(writer, reader, fast, bytes); + assertEquals(100, result.size(), "fastReader=" + fast); + assertEquals(0, ((GenericRecord) result.iterator().next()).get("qty"), "fastReader=" + fast); + } + } + + /** + * A promoted element type ({@code int} or {@code float} read as {@code double}) + * is encoded smaller than the reader's type. + */ + @Test + void arrayWithPromotedElementTypeDecodes() throws Exception { + Schema doubleArray = Schema.createArray(Schema.create(Schema.Type.DOUBLE)); + Object[][] cases = { { Schema.create(Schema.Type.INT), 0 }, { Schema.create(Schema.Type.LONG), 0L }, + { Schema.create(Schema.Type.FLOAT), 0f } }; + for (Object[] c : cases) { + Schema writer = Schema.createArray((Schema) c[0]); + byte[] bytes = encode(writer, Collections.nCopies(100, c[1])); + for (boolean fast : new boolean[] { true, false }) { + Collection result = (Collection) readWith(writer, doubleArray, fast, bytes); + assertEquals(100, result.size(), () -> writer + " fastReader=" + fast); + } + } + } + + @Test + void mapWithPromotedValueTypeDecodes() throws Exception { + Schema writer = Schema.createMap(Schema.create(Schema.Type.INT)); + Schema reader = Schema.createMap(Schema.create(Schema.Type.DOUBLE)); + Map map = new HashMap<>(); + for (int i = 0; i < 100; i++) { + map.put(Integer.toString(i), 0); + } + byte[] bytes = encode(writer, map); + for (boolean fast : new boolean[] { true, false }) { + Map result = (Map) readWith(writer, reader, fast, bytes); + assertEquals(100, result.size(), "fastReader=" + fast); + } + } + + /** + * Resolving against a reader's schema still bounds a huge declared count by the + * bytes remaining. + */ + @Test + void resolvedArrayRejectsHugeCount() throws Exception { + Schema writer = Schema.createArray(Schema.create(Schema.Type.INT)); + Schema reader = Schema.createArray(Schema.create(Schema.Type.DOUBLE)); + byte[] data = encodeVarints(10_000_000L, 0L); + for (boolean fast : new boolean[] { true, false }) { + assertThrows(EOFException.class, () -> readWith(writer, reader, fast, data), "fastReader=" + fast); + } + } + + /** + * Equal writer's and reader's schemas, even as separate instances, keep the + * exact bound of the element type. + */ + @Test + void equalSchemasKeepExactBound() throws Exception { + String json = "{\"type\":\"array\",\"items\":\"double\"}"; + Schema writer = new Schema.Parser().parse(json); + Schema reader = new Schema.Parser().parse(json); + byte[] data = encodeVarints(100L, 0L); + for (boolean fast : new boolean[] { true, false }) { + EOFException e = assertThrows(EOFException.class, () -> readWith(writer, reader, fast, data)); + assertTrue(e.getMessage().contains("at least 8 bytes"), () -> "fastReader=" + fast + ": " + e.getMessage()); + } + } + + /** + * A non-union writer's element ({@code null}) resolves to a branch of the + * reader's union, so it is encoded without the union's branch index. + */ + @Test + void arrayOfNullsReadAsNullableUnionDecodes() throws Exception { + Schema nullType = Schema.create(Schema.Type.NULL); + Schema writer = Schema.createArray(nullType); + Schema reader = Schema.createArray(Schema.createUnion(nullType, Schema.create(Schema.Type.INT))); + byte[] bytes = encode(writer, Collections.nCopies(100, null)); + for (boolean fast : new boolean[] { true, false }) { + Collection result = (Collection) readWith(writer, reader, fast, bytes); + assertEquals(100, result.size(), "fastReader=" + fast); + } + } } 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 56670639d0c..046d1f24b11 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 @@ -25,6 +25,8 @@ import java.io.EOFException; import java.io.IOException; import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; import java.util.HashSet; import java.util.HashMap; import java.util.List; @@ -125,6 +127,24 @@ void read_PojoWithArray_rejectsOversizedArrayCount() throws IOException { assertThrows(EOFException.class, () -> reflectDatumReader.read(new PojoWithArray(), decoder)); } + /** + * An array whose writer's elements ({@code int}) encode smaller than the + * reader's ({@code double}) must still decode. + */ + @Test + void read_arrayWithPromotedElementType() throws IOException { + Schema writer = Schema.createArray(Schema.create(Schema.Type.INT)); + Schema reader = Schema.createArray(Schema.create(Schema.Type.DOUBLE)); + ByteArrayOutputStream out = new ByteArrayOutputStream(); + Encoder encoder = EncoderFactory.get().binaryEncoder(out, null); + new ReflectDatumWriter>(writer).write(Collections.nCopies(100, 0), encoder); + encoder.flush(); + + Decoder decoder = DecoderFactory.get().binaryDecoder(out.toByteArray(), null); + Object result = new ReflectDatumReader<>(writer, reader).read(null, decoder); + assertEquals(100, ((Collection) result).size()); + } + /** * 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