diff --git a/lang/java/avro/src/main/java/org/apache/avro/SystemLimitException.java b/lang/java/avro/src/main/java/org/apache/avro/SystemLimitException.java
index 003f2e406fd..ef126280c8d 100644
--- a/lang/java/avro/src/main/java/org/apache/avro/SystemLimitException.java
+++ b/lang/java/avro/src/main/java/org/apache/avro/SystemLimitException.java
@@ -42,6 +42,13 @@
  * 0 minimum) that may be allocated at once. Unlike other element types, these
  * cannot be bounded by the number of bytes remaining in the stream, so the
  * limit defaults to a fraction of the maximum heap.</li>
+ * <li><tt>org.apache.avro.limits.decode.maxDepth</tt> limits how deeply nested
+ * a value may be while decoding. A recursive schema (e.g. a linked list or a
+ * tree) lets a small, hostile payload drive arbitrarily deep nesting,
+ * exhausting the call stack ({@link StackOverflowError}) before any allocation
+ * limit is reached. The limit is enforced by counting structural descents (into
+ * a record, array, map or union) and rejecting input that nests deeper than the
+ * configured maximum.</li>
  * </ul>
  *
  * The default is to permit sizes up to {@link #MAX_ARRAY_VM_LIMIT}.
@@ -62,9 +69,24 @@ public class SystemLimitException extends AvroRuntimeException {
   public static final String MAX_COLLECTION_LENGTH_PROPERTY = "org.apache.avro.limits.collectionItems.maxLength";
   public static final String MAX_STRING_LENGTH_PROPERTY = "org.apache.avro.limits.string.maxLength";
 
+  /**
+   * System property bounding how deeply nested a value may be while decoding:
+   * {@value}. See {@link #incrementDecodeDepth()}.
+   */
+  public static final String MAX_DECODE_DEPTH_PROPERTY = "org.apache.avro.limits.decode.maxDepth";
+
+  /**
+   * Default maximum decode nesting depth. Comfortably exceeds any realistic
+   * schema nesting while remaining far below where a recursive decode would
+   * exhaust the call stack. Aligns with the well-known default used by Protocol
+   * Buffers.
+   */
+  static final int DEFAULT_MAX_DECODE_DEPTH = 100;
+
   private static int maxBytesLength = MAX_ARRAY_VM_LIMIT;
   private static int maxCollectionLength = MAX_ARRAY_VM_LIMIT;
   private static int maxStringLength = MAX_ARRAY_VM_LIMIT;
+  private static int maxDecodeDepth = DEFAULT_MAX_DECODE_DEPTH;
 
   private static final Logger LOG = LoggerFactory.getLogger(SystemLimitException.class);
 
@@ -132,6 +154,14 @@ public class SystemLimitException extends AvroRuntimeException {
   private static final class CollectionAllocationScope {
     private int depth;
     private long allocated;
+    /**
+     * Current decode nesting depth (structural descents into records, arrays, maps
+     * and unions). Tracked per thread rather than per reader instance so a reader
+     * reused concurrently cannot corrupt another thread's counter, and so the depth
+     * is threaded implicitly through the recursive decode without changing the
+     * reader method signatures. See {@link #incrementDecodeDepth()}.
+     */
+    private int decodeDepth;
   }
 
   private static final ThreadLocal<CollectionAllocationScope> COLLECTION_ALLOCATION_SCOPE = ThreadLocal
@@ -364,6 +394,11 @@ public static void beginCollectionAllocationScope() {
     CollectionAllocationScope scope = COLLECTION_ALLOCATION_SCOPE.get();
     if (scope.depth == 0) {
       scope.allocated = 0;
+      // Defensively clear the decode depth at the outermost datum boundary. The
+      // counter is already kept balanced by the try/finally around every
+      // increment, but resetting here guarantees a stale value from an abnormally
+      // terminated earlier decode on this thread cannot leak into this one.
+      scope.decodeDepth = 0;
     }
     scope.depth++;
   }
@@ -414,6 +449,47 @@ public static long checkMaxCollectionAllocation(long items) {
     return total;
   }
 
+  /**
+   * Record a structural descent (into a record, array, map or union) while
+   * decoding and verify the nesting has not grown past
+   * {@link #MAX_DECODE_DEPTH_PROPERTY the configured maximum}.
+   * <p>
+   * Avro's binary decoders decode nested values with a recursive call chain, so
+   * the call stack grows in lockstep with the nesting of the data. A recursive
+   * schema (e.g. a linked list or tree) lets a tiny, hostile payload declare
+   * arbitrarily deep nesting, overflowing the stack ({@link StackOverflowError})
+   * long before any allocation limit is reached. Bounding the depth turns such
+   * input into a clean, catchable failure instead of a crash.
+   * <p>
+   * Every call that succeeds must be paired with a matching
+   * {@link #decrementDecodeDepth()} in a {@code finally} block. When the limit
+   * would be exceeded this method throws <em>without</em> incrementing, so the
+   * counter stays balanced as the exception unwinds the enclosing
+   * (already-incremented) frames.
+   *
+   * @throws SystemLimitException if the decode nesting would exceed the maximum.
+   */
+  public static void incrementDecodeDepth() {
+    CollectionAllocationScope scope = COLLECTION_ALLOCATION_SCOPE.get();
+    if (scope.decodeDepth >= maxDecodeDepth) {
+      throw new SystemLimitException("Decode nesting depth exceeds the maximum allowed of " + maxDecodeDepth
+          + " (configure with the system property " + MAX_DECODE_DEPTH_PROPERTY + ")");
+    }
+    scope.decodeDepth++;
+  }
+
+  /**
+   * Record leaving a structural value opened by {@link #incrementDecodeDepth()}.
+   * Must be called from a {@code finally} block so the depth is restored even
+   * when decoding the nested value fails.
+   */
+  public static void decrementDecodeDepth() {
+    CollectionAllocationScope scope = COLLECTION_ALLOCATION_SCOPE.get();
+    if (scope.decodeDepth > 0) {
+      scope.decodeDepth--;
+    }
+  }
+
   /**
    * Check to ensure that reading the string size is within the specified limits.
    *
@@ -468,5 +544,6 @@ static void resetLimits() {
     // zero-byte allocation cap consistent with the other collection limits even
     // when it is configured (or derived from a very large heap) above that.
     maxCollectionAllocation = Math.min(maxCollectionAllocation, MAX_ARRAY_VM_LIMIT);
+    maxDecodeDepth = getLimitFromProperty(MAX_DECODE_DEPTH_PROPERTY, DEFAULT_MAX_DECODE_DEPTH);
   }
 }
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..2749110f0f3 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
@@ -214,15 +214,21 @@ protected Object readWithConversion(Object old, Schema expected, LogicalType log
   protected Object readWithoutConversion(Object old, Schema expected, ResolvingDecoder in) throws IOException {
     switch (expected.getType()) {
     case RECORD:
-      return readRecord(old, expected, in);
-    case ENUM:
-      return readEnum(expected, in);
     case ARRAY:
-      return readArray(old, expected, in);
     case MAP:
-      return readMap(old, expected, in);
     case UNION:
-      return read(old, expected.getTypes().get(in.readIndex()), in);
+      // Descending into a structural value grows the decode call stack. Bound the
+      // nesting depth so a recursive schema fed a deeply nested payload fails with
+      // a SystemLimitException instead of a StackOverflowError. The counter is
+      // decremented on exit via the finally so it stays balanced even on error.
+      SystemLimitException.incrementDecodeDepth();
+      try {
+        return readStructural(old, expected, in);
+      } finally {
+        SystemLimitException.decrementDecodeDepth();
+      }
+    case ENUM:
+      return readEnum(expected, in);
     case FIXED:
       return readFixed(old, expected, in);
     case STRING:
@@ -247,6 +253,26 @@ protected Object readWithoutConversion(Object old, Schema expected, ResolvingDec
     }
   }
 
+  /**
+   * Dispatches the structural (nesting) value types. Split out of
+   * {@link #readWithoutConversion} so the decode-depth guard wraps exactly the
+   * types that grow the recursive call stack.
+   */
+  private Object readStructural(Object old, Schema expected, ResolvingDecoder in) throws IOException {
+    switch (expected.getType()) {
+    case RECORD:
+      return readRecord(old, expected, in);
+    case ARRAY:
+      return readArray(old, expected, in);
+    case MAP:
+      return readMap(old, expected, in);
+    case UNION:
+      return read(old, expected.getTypes().get(in.readIndex()), in);
+    default:
+      throw new AvroRuntimeException("Not a structural type: " + expected);
+    }
+  }
+
   /**
    * Convert an underlying representation of a logical type (such as a ByteBuffer)
    * to a higher level object (such as a BigDecimal).
@@ -819,12 +845,67 @@ public static void skip(Schema schema, Decoder in) throws IOException {
   private static void skipInternal(Schema schema, Decoder in) throws IOException {
     switch (schema.getType()) {
     case RECORD:
-      for (Field field : schema.getFields())
-        skipInternal(field.schema(), in);
+    case ARRAY:
+    case MAP:
+    case UNION:
+      // Skipping descends into nested structural values with the same recursive
+      // call chain as reading, so bound the nesting depth here too. Otherwise a
+      // recursive schema whose deeply nested value is skipped (a writer-only field
+      // during resolution, the fast reader's skip steps, or BinaryData.compare)
+      // would still overflow the stack.
+      SystemLimitException.incrementDecodeDepth();
+      try {
+        skipStructural(schema, in);
+      } finally {
+        SystemLimitException.decrementDecodeDepth();
+      }
       break;
     case ENUM:
       in.readEnum();
       break;
+    case FIXED:
+      in.skipFixed(schema.getFixedSize());
+      break;
+    case STRING:
+      in.skipString();
+      break;
+    case BYTES:
+      in.skipBytes();
+      break;
+    case INT:
+      in.readInt();
+      break;
+    case LONG:
+      in.readLong();
+      break;
+    case FLOAT:
+      in.readFloat();
+      break;
+    case DOUBLE:
+      in.readDouble();
+      break;
+    case BOOLEAN:
+      in.readBoolean();
+      break;
+    case NULL:
+      in.readNull();
+      break;
+    default:
+      throw new RuntimeException("Unknown type: " + schema);
+    }
+  }
+
+  /**
+   * Skips the structural (nesting) value types. Split out of
+   * {@link #skipInternal} so the decode-depth guard wraps exactly the types that
+   * grow the recursive call stack, mirroring {@link #readStructural}.
+   */
+  private static void skipStructural(Schema schema, Decoder in) throws IOException {
+    switch (schema.getType()) {
+    case RECORD:
+      for (Field field : schema.getFields())
+        skipInternal(field.schema(), in);
+      break;
     case ARRAY:
       Schema elementType = schema.getElementType();
       // Bound the cumulative element count: skipping a huge block of elements
@@ -865,35 +946,8 @@ private static void skipInternal(Schema schema, Decoder in) throws IOException {
     case UNION:
       skipInternal(schema.getTypes().get(in.readIndex()), in);
       break;
-    case FIXED:
-      in.skipFixed(schema.getFixedSize());
-      break;
-    case STRING:
-      in.skipString();
-      break;
-    case BYTES:
-      in.skipBytes();
-      break;
-    case INT:
-      in.readInt();
-      break;
-    case LONG:
-      in.readLong();
-      break;
-    case FLOAT:
-      in.readFloat();
-      break;
-    case DOUBLE:
-      in.readDouble();
-      break;
-    case BOOLEAN:
-      in.readBoolean();
-      break;
-    case NULL:
-      in.readNull();
-      break;
     default:
-      throw new RuntimeException("Unknown type: " + schema);
+      throw new RuntimeException("Not a structural type: " + schema);
     }
   }
 
diff --git a/lang/java/avro/src/main/java/org/apache/avro/io/BinaryData.java b/lang/java/avro/src/main/java/org/apache/avro/io/BinaryData.java
index 0df469d4957..9011772bda7 100644
--- a/lang/java/avro/src/main/java/org/apache/avro/io/BinaryData.java
+++ b/lang/java/avro/src/main/java/org/apache/avro/io/BinaryData.java
@@ -23,6 +23,7 @@
 import org.apache.avro.Schema;
 import org.apache.avro.Schema.Field;
 import org.apache.avro.AvroRuntimeException;
+import org.apache.avro.SystemLimitException;
 import org.apache.avro.generic.GenericDatumReader;
 import org.apache.avro.util.internal.ThreadLocalWithInitial;
 
@@ -70,11 +71,15 @@ public static int compare(byte[] b1, int s1, byte[] b2, int s2, Schema schema) {
   public static int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2, Schema schema) {
     Decoders decoders = DECODERS.get();
     decoders.set(b1, s1, l1, b2, s2, l2);
+    // Delimit a decode scope so the recursion-depth counter starts (and is reset)
+    // at this top-level comparison, mirroring GenericDatumReader.
+    SystemLimitException.beginCollectionAllocationScope();
     try {
       return compare(decoders, schema);
     } catch (IOException e) {
       throw new AvroRuntimeException(e);
     } finally {
+      SystemLimitException.endCollectionAllocationScope();
       decoders.clear();
     }
   }
@@ -84,6 +89,25 @@ public static int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2,
    * less than, return LT.
    */
   private static int compare(Decoders d, Schema schema) throws IOException {
+    switch (schema.getType()) {
+    case RECORD:
+    case ARRAY:
+    case UNION:
+      // These descend recursively into nested values, so bound the nesting depth
+      // to prevent a recursive schema comparing deeply nested data from
+      // overflowing the stack (mirrors GenericDatumReader's decode-depth guard).
+      SystemLimitException.incrementDecodeDepth();
+      try {
+        return compareStructural(d, schema);
+      } finally {
+        SystemLimitException.decrementDecodeDepth();
+      }
+    default:
+      return compareScalar(d, schema);
+    }
+  }
+
+  private static int compareStructural(Decoders d, Schema schema) throws IOException {
     Decoder d1 = d.d1;
     Decoder d2 = d.d2;
     switch (schema.getType()) {
@@ -101,17 +125,6 @@ private static int compare(Decoders d, Schema schema) throws IOException {
       }
       return 0;
     }
-    case ENUM:
-    case INT:
-      return Integer.compare(d1.readInt(), d2.readInt());
-    case LONG:
-      return Long.compare(d1.readLong(), d2.readLong());
-    case FLOAT:
-      return Float.compare(d1.readFloat(), d2.readFloat());
-    case DOUBLE:
-      return Double.compare(d1.readDouble(), d2.readDouble());
-    case BOOLEAN:
-      return Boolean.compare(d1.readBoolean(), d2.readBoolean());
     case ARRAY: {
       long i = 0; // position in array
       long r1 = 0, r2 = 0; // remaining in current block
@@ -146,14 +159,34 @@ private static int compare(Decoders d, Schema schema) throws IOException {
         }
       }
     }
-    case MAP:
-      throw new AvroRuntimeException("Can't compare maps!");
     case UNION: {
       int i1 = d1.readInt();
       int i2 = d2.readInt();
       int c = Integer.compare(i1, i2);
       return c == 0 ? compare(d, schema.getTypes().get(i1)) : c;
     }
+    default:
+      throw new AvroRuntimeException("Not a structural type to compare: " + schema);
+    }
+  }
+
+  private static int compareScalar(Decoders d, Schema schema) throws IOException {
+    Decoder d1 = d.d1;
+    Decoder d2 = d.d2;
+    switch (schema.getType()) {
+    case ENUM:
+    case INT:
+      return Integer.compare(d1.readInt(), d2.readInt());
+    case LONG:
+      return Long.compare(d1.readLong(), d2.readLong());
+    case FLOAT:
+      return Float.compare(d1.readFloat(), d2.readFloat());
+    case DOUBLE:
+      return Double.compare(d1.readDouble(), d2.readDouble());
+    case BOOLEAN:
+      return Boolean.compare(d1.readBoolean(), d2.readBoolean());
+    case MAP:
+      throw new AvroRuntimeException("Can't compare maps!");
     case FIXED: {
       int size = schema.getFixedSize();
       int c = compareBytes(d.d1.getBuf(), d.d1.getPos(), size, d.d2.getBuf(), d.d2.getPos(), size);
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..bd8a32c6a7d 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
@@ -419,12 +419,17 @@ private FieldReader createUnionReader(WriterUnion action) throws IOException {
 
   private FieldReader createUnionReader(FieldReader[] unionReaders) {
     return reusingReader((reuse, decoder) -> {
-      final int selection = decoder.readIndex();
-      if (selection < 0 || selection >= unionReaders.length) {
-        throw new AvroTypeException(
-            "Union branch index out of range: must be in [0, " + unionReaders.length + "), but received " + selection);
+      SystemLimitException.incrementDecodeDepth();
+      try {
+        final int selection = decoder.readIndex();
+        if (selection < 0 || selection >= unionReaders.length) {
+          throw new AvroTypeException("Union branch index out of range: must be in [0, " + unionReaders.length
+              + "), but received " + selection);
+        }
+        return unionReaders[selection].read(null, decoder);
+      } finally {
+        SystemLimitException.decrementDecodeDepth();
       }
-      return unionReaders[selection].read(null, decoder);
     });
 
   }
@@ -478,45 +483,54 @@ private FieldReader createArrayReader(Schema readerSchema, Container action) thr
     boolean zeroByteElements = GenericDatumReader.isZeroByteSchema(elementType);
 
     return reusingReader((reuse, decoder) -> {
-      // Open a decode scope so the zero-byte element allocation cap is cumulative
-      // across every block of this array even when the fast reader is used
-      // standalone (i.e. without GenericDatumReader.read opening the outer datum
-      // scope); otherwise a huge array split into many small blocks would bypass
-      // the cap. The scope nests: when a datum scope is already open this simply
-      // accumulates into it, and only the outermost scope resets the running
-      // total (see SystemLimitException). The try/finally guarantees the scope is
-      // always closed so ThreadLocal state cannot leak into later decodes on the
-      // same thread.
+      // Open the collection-allocation scope first: when the fast reader is used
+      // standalone with a top-level array this is the outermost datum boundary,
+      // where the scope clears any stale decode depth. Only after that do we count
+      // this array's nesting level, so the reset cannot wipe the increment and a
+      // stale depth cannot trip the limit before being cleared. Both the depth
+      // decrement and the scope end run in finally blocks, and because the depth
+      // increment sits inside the scope's try, a throw from the depth check still
+      // closes the scope.
       SystemLimitException.beginCollectionAllocationScope();
       try {
-        if (reuse instanceof GenericArray) {
-          GenericArray<Object> reuseArray = (GenericArray<Object>) reuse;
-          long l = decoder.readArrayStart();
-          checkArrayBlock(decoder, elementType, zeroByteElements, l);
-          reuseArray.clear();
-
-          while (l > 0) {
-            for (long i = 0; i < l; i++) {
-              reuseArray.add(elementReader.read(reuseArray.peek(), decoder));
-            }
-            l = decoder.arrayNext();
+        // The scope also makes the zero-byte element allocation cap cumulative
+        // across every block of this array (a huge array split into many small
+        // blocks would otherwise bypass the cap). The scope nests: when a datum
+        // scope is already open this accumulates into it, and only the outermost
+        // scope resets the running total (see SystemLimitException).
+        SystemLimitException.incrementDecodeDepth();
+        try {
+          if (reuse instanceof GenericArray) {
+            GenericArray<Object> reuseArray = (GenericArray<Object>) reuse;
+            long l = decoder.readArrayStart();
             checkArrayBlock(decoder, elementType, zeroByteElements, l);
-          }
-          return reuseArray;
-        } else {
-          long l = decoder.readArrayStart();
-          checkArrayBlock(decoder, elementType, zeroByteElements, l);
-          List<Object> array = (reuse instanceof List) ? (List<Object>) reuse
-              : new GenericData.Array<>(GenericDatumReader.initialCollectionCapacity(l), readerSchema);
-          array.clear();
-          while (l > 0) {
-            for (long i = 0; i < l; i++) {
-              array.add(elementReader.read(null, decoder));
+            reuseArray.clear();
+
+            while (l > 0) {
+              for (long i = 0; i < l; i++) {
+                reuseArray.add(elementReader.read(reuseArray.peek(), decoder));
+              }
+              l = decoder.arrayNext();
+              checkArrayBlock(decoder, elementType, zeroByteElements, l);
             }
-            l = decoder.arrayNext();
+            return reuseArray;
+          } else {
+            long l = decoder.readArrayStart();
             checkArrayBlock(decoder, elementType, zeroByteElements, l);
+            List<Object> array = (reuse instanceof List) ? (List<Object>) reuse
+                : new GenericData.Array<>(GenericDatumReader.initialCollectionCapacity(l), readerSchema);
+            array.clear();
+            while (l > 0) {
+              for (long i = 0; i < l; i++) {
+                array.add(elementReader.read(null, decoder));
+              }
+              l = decoder.arrayNext();
+              checkArrayBlock(decoder, elementType, zeroByteElements, l);
+            }
+            return array;
           }
-          return array;
+        } finally {
+          SystemLimitException.decrementDecodeDepth();
         }
       } finally {
         SystemLimitException.endCollectionAllocationScope();
@@ -637,11 +651,18 @@ public boolean canReuse() {
 
     @Override
     public Object read(Object reuse, Decoder decoder) throws IOException {
-      Object object = supplier.newInstance(reuse, schema);
-      for (ExecutionStep thisStep : readSteps) {
-        thisStep.execute(object, decoder);
+      // Bound decode nesting depth: a recursive schema fed deeply nested data
+      // would otherwise overflow the stack via this recursive descent.
+      SystemLimitException.incrementDecodeDepth();
+      try {
+        Object object = supplier.newInstance(reuse, schema);
+        for (ExecutionStep thisStep : readSteps) {
+          thisStep.execute(object, decoder);
+        }
+        return object;
+      } finally {
+        SystemLimitException.decrementDecodeDepth();
       }
-      return object;
     }
   }
 
@@ -657,19 +678,24 @@ public MapReader(FieldReader keyReader, FieldReader valueReader) {
 
     @Override
     public Object read(Object reuse, Decoder decoder) throws IOException {
-      long l = decoder.readMapStart();
-      Map<Object, Object> targetMap = new HashMap<>();
-
-      while (l > 0) {
-        for (int i = 0; i < l; i++) {
-          Object key = keyReader.read(null, decoder);
-          Object value = valueReader.read(null, decoder);
-          targetMap.put(key, value);
+      SystemLimitException.incrementDecodeDepth();
+      try {
+        long l = decoder.readMapStart();
+        Map<Object, Object> targetMap = new HashMap<>();
+
+        while (l > 0) {
+          for (long i = 0; i < l; i++) {
+            Object key = keyReader.read(null, decoder);
+            Object value = valueReader.read(null, decoder);
+            targetMap.put(key, value);
+          }
+          l = decoder.mapNext();
         }
-        l = decoder.mapNext();
-      }
 
-      return targetMap;
+        return targetMap;
+      } finally {
+        SystemLimitException.decrementDecodeDepth();
+      }
     }
   }
 
diff --git a/lang/java/avro/src/test/java/org/apache/avro/TestSystemLimitException.java b/lang/java/avro/src/test/java/org/apache/avro/TestSystemLimitException.java
index a0b59f55e27..82d5dbfaf62 100644
--- a/lang/java/avro/src/test/java/org/apache/avro/TestSystemLimitException.java
+++ b/lang/java/avro/src/test/java/org/apache/avro/TestSystemLimitException.java
@@ -47,6 +47,7 @@ void reset() {
     System.clearProperty(MAX_COLLECTION_LENGTH_PROPERTY);
     System.clearProperty(MAX_STRING_LENGTH_PROPERTY);
     System.clearProperty(MAX_COLLECTION_ALLOCATION_PROPERTY);
+    System.clearProperty(MAX_DECODE_DEPTH_PROPERTY);
     resetLimits();
   }
 
@@ -155,8 +156,7 @@ void testCheckMaxCollectionAllocationDefaultsToHeapFraction() {
   }
 
   @Test
-  void testCheckMaxCollectionLengthFromNonZero() {
-    // Correct values pass through
+  void testCheckMaxCollectionLengthFromNonZero() { // Correct values pass through
     assertEquals(10, checkMaxCollectionLength(10L, 0L));
     assertEquals(MAX_ARRAY_VM_LIMIT, checkMaxCollectionLength(10L, MAX_ARRAY_VM_LIMIT - 10L));
     assertEquals(MAX_ARRAY_VM_LIMIT, checkMaxCollectionLength(MAX_ARRAY_VM_LIMIT - 10L, 10L));
@@ -205,4 +205,69 @@ void testCheckMaxCollectionLengthFromNonZero() {
     ex = assertThrows(SystemLimitException.class, () -> checkMaxCollectionLength(25, 999));
     assertEquals("Collection length 1024 exceeds maximum allowed", ex.getMessage());
   }
+
+  @Test
+  void testDecodeDepthDefaultAllowsModerateNestingAndRejectsBeyondLimit() {
+    resetLimits();
+    // Descend exactly to the default limit: all increments must succeed.
+    for (int i = 0; i < DEFAULT_MAX_DECODE_DEPTH; i++) {
+      incrementDecodeDepth();
+    }
+    // One level too deep is rejected with a clear, bounded error (not a crash).
+    SystemLimitException ex = assertThrows(SystemLimitException.class, SystemLimitException::incrementDecodeDepth);
+    assertTrue(
+        ex.getMessage().contains("Decode nesting depth exceeds the maximum allowed of " + DEFAULT_MAX_DECODE_DEPTH),
+        ex.getMessage());
+    // The rejected increment must not have advanced the counter: after unwinding
+    // all successful descents the depth returns to zero.
+    for (int i = 0; i < DEFAULT_MAX_DECODE_DEPTH; i++) {
+      decrementDecodeDepth();
+    }
+    // Now at zero again; a fresh descent is permitted.
+    incrementDecodeDepth();
+    decrementDecodeDepth();
+  }
+
+  @Test
+  void testDecodeDepthHonoursCustomLimit() {
+    System.setProperty(MAX_DECODE_DEPTH_PROPERTY, "3");
+    resetLimits();
+    incrementDecodeDepth();
+    incrementDecodeDepth();
+    incrementDecodeDepth();
+    SystemLimitException ex = assertThrows(SystemLimitException.class, SystemLimitException::incrementDecodeDepth);
+    assertTrue(ex.getMessage().contains("maximum allowed of 3"), ex.getMessage());
+    assertTrue(ex.getMessage().contains(MAX_DECODE_DEPTH_PROPERTY), ex.getMessage());
+    decrementDecodeDepth();
+    decrementDecodeDepth();
+    decrementDecodeDepth();
+  }
+
+  @Test
+  void testDecodeDepthResetAtOutermostScope() {
+    System.setProperty(MAX_DECODE_DEPTH_PROPERTY, "5");
+    resetLimits();
+    // Leave a stale depth behind, as an abnormally terminated earlier decode on
+    // this thread might. (In normal operation the try/finally around every
+    // increment keeps it balanced; the reset is a defensive backstop.)
+    incrementDecodeDepth();
+    incrementDecodeDepth();
+    // Opening a fresh outermost datum scope clears the stale depth, so a new
+    // decode starts from zero and gets the full depth budget.
+    beginCollectionAllocationScope();
+    try {
+      for (int i = 0; i < 5; i++) {
+        incrementDecodeDepth();
+      }
+      // Exactly at the limit now: one more level is rejected. This proves the two
+      // stale increments were reset, otherwise the limit would have been reached
+      // two levels early.
+      assertThrows(SystemLimitException.class, SystemLimitException::incrementDecodeDepth);
+      for (int i = 0; i < 5; i++) {
+        decrementDecodeDepth();
+      }
+    } finally {
+      endCollectionAllocationScope();
+    }
+  }
 }
diff --git a/lang/java/avro/src/test/java/org/apache/avro/generic/TestDecodeRecursionDepth.java b/lang/java/avro/src/test/java/org/apache/avro/generic/TestDecodeRecursionDepth.java
new file mode 100644
index 00000000000..9aa29614108
--- /dev/null
+++ b/lang/java/avro/src/test/java/org/apache/avro/generic/TestDecodeRecursionDepth.java
@@ -0,0 +1,134 @@
+/*
+ * 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
+ *
+ *     https://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.avro.generic;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+
+import org.apache.avro.Schema;
+import org.apache.avro.SystemLimitException;
+import org.apache.avro.io.BinaryData;
+import org.apache.avro.io.BinaryEncoder;
+import org.apache.avro.io.Decoder;
+import org.apache.avro.io.DecoderFactory;
+import org.apache.avro.io.EncoderFactory;
+import org.apache.avro.io.FastReaderBuilder;
+import org.apache.avro.io.DatumReader;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+/**
+ * Regression tests for AVRO-4302: decoding a deeply nested payload for a
+ * recursive schema must fail with a bounded {@link SystemLimitException} rather
+ * than crashing the thread with a {@link StackOverflowError}.
+ */
+public class TestDecodeRecursionDepth {
+
+  /** A self-referencing linked-list schema: the classic recursion-bomb shape. */
+  private static final Schema NODE = new Schema.Parser().parse(
+      "{\"type\":\"record\",\"name\":\"Node\",\"fields\":[" + "{\"name\":\"next\",\"type\":[\"null\",\"Node\"]}]}");
+
+  /**
+   * Builds a binary-encoded {@code Node} linked list nested {@code depth} levels
+   * deep. Each level selects the {@code Node} union branch (index 1); the final
+   * level selects {@code null} (index 0) to terminate. Roughly one byte per
+   * level, so a tiny payload encodes enormous nesting.
+   */
+  private static byte[] linkedList(int depth) throws IOException {
+    ByteArrayOutputStream out = new ByteArrayOutputStream();
+    BinaryEncoder encoder = EncoderFactory.get().binaryEncoder(out, null);
+    for (int i = 0; i < depth; i++) {
+      encoder.writeIndex(1); // select "Node": one more level of nesting
+    }
+    encoder.writeIndex(0); // select "null": terminate the list
+    encoder.flush();
+    return out.toByteArray();
+  }
+
+  private static Object read(byte[] bytes, boolean fastReader) throws IOException {
+    GenericData data = new GenericData();
+    data.setFastReaderEnabled(fastReader);
+    GenericDatumReader<Object> reader = new GenericDatumReader<>(NODE, NODE, data);
+    Decoder decoder = DecoderFactory.get().binaryDecoder(bytes, null);
+    return reader.read(null, decoder);
+  }
+
+  @ParameterizedTest
+  @ValueSource(booleans = { false, true })
+  void deeplyNestedInputIsRejectedWithBoundedError(boolean fastReader) throws IOException {
+    // ~100k levels: far beyond the default depth limit and enough to overflow the
+    // stack if it were left unbounded, yet only ~100kB of input.
+    byte[] bomb = linkedList(100_000);
+    assertThrows(SystemLimitException.class, () -> read(bomb, fastReader));
+  }
+
+  @ParameterizedTest
+  @ValueSource(booleans = { false, true })
+  void moderatelyNestedInputWithinLimitStillDecodes(boolean fastReader) throws IOException {
+    // A legitimately nested value comfortably within the limit must still decode.
+    // Two structural descents (union + record) are counted per list level, so
+    // keep the level count well under half the default limit.
+    Object result = read(linkedList(20), fastReader);
+    assertNotNull(result);
+  }
+
+  @Test
+  void deeplyNestedInputIsRejectedWhenSkipped() throws IOException {
+    // The skip path (a writer-only field during resolution, the fast reader's skip
+    // steps) descends recursively too, so it must be bounded as well.
+    byte[] bomb = linkedList(100_000);
+    Decoder decoder = DecoderFactory.get().binaryDecoder(bomb, null);
+    assertThrows(SystemLimitException.class, () -> GenericDatumReader.skip(NODE, decoder));
+  }
+
+  @Test
+  void deeplyNestedInputIsRejectedWhenCompared() throws IOException {
+    // BinaryData.compare descends recursively over the schema and must not
+    // overflow the stack on a deeply nested recursive value either.
+    byte[] bomb = linkedList(100_000);
+    assertThrows(SystemLimitException.class, () -> BinaryData.compare(bomb, 0, bomb, 0, NODE));
+  }
+
+  @Test
+  void standaloneFastReaderTopLevelArrayBoundsDepth() throws IOException {
+    // When the fast reader is used standalone with a top-level array, the array
+    // reader is the outermost scope: it must open the collection scope (which
+    // resets stale depth) before counting its own level, so depth accounting stays
+    // consistent and a deeply nested element is still rejected.
+    Schema arrayOfNode = Schema.createArray(NODE);
+    ByteArrayOutputStream out = new ByteArrayOutputStream();
+    BinaryEncoder encoder = EncoderFactory.get().binaryEncoder(out, null);
+    encoder.writeArrayStart();
+    encoder.setItemCount(1);
+    encoder.startItem();
+    for (int i = 0; i < 100_000; i++) {
+      encoder.writeIndex(1); // one more nested Node
+    }
+    encoder.writeIndex(0); // terminate the linked list
+    encoder.writeArrayEnd();
+    encoder.flush();
+
+    DatumReader<Object> reader = FastReaderBuilder.get().createDatumReader(arrayOfNode);
+    Decoder decoder = DecoderFactory.get().binaryDecoder(out.toByteArray(), null);
+    assertThrows(SystemLimitException.class, () -> reader.read(null, decoder));
+  }
+}