AVRO-4324: [Java] Align ReflectDatumReader.readArray with GenericDatumReader eager-allocation guards

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")

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<PojoWithArray> reflectDatumReader = new ReflectDatumReader<>(PojoWithArray.class);
+
+    assertThrows(EOFException.class, () -> reflectDatumReader.read(new PojoWithArray(), decoder));
+  }
+
   @Test
   public void testRead_PojoWithSet() throws IOException {
     PojoWithSet pojoWithSet = new PojoWithSet();