diff --git a/lang/perl/lib/Avro/BinaryDecoder.pm b/lang/perl/lib/Avro/BinaryDecoder.pm
index d2bb65e14f7..79d8828c044 100644
--- a/lang/perl/lib/Avro/BinaryDecoder.pm
+++ b/lang/perl/lib/Avro/BinaryDecoder.pm
@@ -32,6 +32,33 @@ unless ($Config{use64bitint}) {
     $complement = Math::BigInt->new("0b" . ("1" x 57) . ("0" x 7));
 }
 
+## The block count of an array or map is read from the (potentially untrusted or
+## truncated) input and drives allocation of the resulting collection. To guard
+## against unbounded memory allocation from a very large or malformed block
+## count, the number of items in a single decoded array or map is capped. This
+## mirrors the Java SDK's collection item limit. The default can be overridden
+## with the AVRO_MAX_COLLECTION_ITEMS environment variable, or by setting
+## $Avro::BinaryDecoder::MAX_COLLECTION_ITEMS directly.
+our $DEFAULT_MAX_COLLECTION_ITEMS = (2 ** 31) - 8;
+our $MAX_COLLECTION_ITEMS =
+    ( defined $ENV{AVRO_MAX_COLLECTION_ITEMS} && $ENV{AVRO_MAX_COLLECTION_ITEMS} =~ /\A[0-9]+\z/ )
+    ? $ENV{AVRO_MAX_COLLECTION_ITEMS} + 0
+    : $DEFAULT_MAX_COLLECTION_ITEMS;
+
+## Ensure that decoding the next block of $block_count items would not grow the
+## collection beyond $MAX_COLLECTION_ITEMS. Callers pass the normalized
+## (non-negative) block count -- in the Avro encoding a negative count merely
+## signals that a block-size long follows, and its absolute value is the count.
+## The negative check here is a defensive guard against malformed input.
+sub _check_collection_items {
+    my ($existing, $block_count) = @_;
+    if ($block_count < 0 || $existing + $block_count > $MAX_COLLECTION_ITEMS) {
+        throw Avro::BinaryDecoder::Error::CollectionSize(
+            "Cannot read collections larger than $MAX_COLLECTION_ITEMS items");
+    }
+    return;
+}
+
 =head2 decode(%param)
 
 Resolve the given writer and reader_schema to decode the data provided by the
@@ -258,6 +285,7 @@ sub decode_array {
             $block_size = decode_long($class, @_);
             ## XXX we can skip with $reader_schema?
         }
+        _check_collection_items(scalar(@array), $block_count);
         for (1..$block_count) {
             push @array, $class->decode(
                 writer_schema => $writer_items,
@@ -296,6 +324,10 @@ sub decode_map {
     my $block_count = decode_long($class, @_);
     my $writer_values = $writer_schema->values;
     my $reader_values = $reader_schema->values;
+    ## Track the number of pairs decoded rather than scalar(keys %hash):
+    ## repeated keys collapse in the hash and would otherwise let the cumulative
+    ## check be bypassed by a stream that keeps rewriting the same key.
+    my $pairs_read = 0;
     while ($block_count) {
         my $block_size;
         if ($block_count < 0) {
@@ -303,6 +335,7 @@ sub decode_map {
             $block_size = decode_long($class, @_);
             ## XXX we can skip with $reader_schema?
         }
+        _check_collection_items($pairs_read, $block_count);
         for (1..$block_count) {
             my $key = decode_string($class, @_);
             unless (defined $key && length $key) {
@@ -314,6 +347,7 @@ sub decode_map {
                 reader        => $reader,
             );
         }
+        $pairs_read += $block_count;
         $block_count = decode_long($class, @_);
     }
     return \%hash;
@@ -390,4 +424,7 @@ sub unsigned_varint {
     return $int;
 }
 
+package Avro::BinaryDecoder::Error::CollectionSize;
+use parent -norequire, 'Error::Simple';
+
 1;
diff --git a/lang/perl/t/06_bin_decode_limits.t b/lang/perl/t/06_bin_decode_limits.t
new file mode 100644
index 00000000000..892123f2ed3
--- /dev/null
+++ b/lang/perl/t/06_bin_decode_limits.t
@@ -0,0 +1,95 @@
+# 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.
+
+#!/usr/bin/env perl
+
+# Decoding arrays and maps must bound the block count read from the input. The
+# block count drives allocation of the resulting collection, so a pathological
+# or truncated input declaring a very large block count must raise an error
+# instead of attempting an unbounded allocation.
+
+use strict;
+use warnings;
+
+# Load the decoder with its default limit regardless of the runner environment:
+# $MAX_COLLECTION_ITEMS is initialized from AVRO_MAX_COLLECTION_ITEMS at load
+# time, so clear it before any Avro module is loaded.
+BEGIN { delete $ENV{AVRO_MAX_COLLECTION_ITEMS}; }
+
+use Avro::Schema;
+use Test::More;
+use Test::Exception;
+
+use_ok 'Avro::BinaryDecoder';
+
+my $array_schema = Avro::Schema->parse(q({"type": "array", "items": "null"}));
+my $map_schema   = Avro::Schema->parse(q({"type": "map", "values": "null"}));
+
+sub decode_bytes {
+    my ($schema, $bytes) = @_;
+    open my $reader, '<', \$bytes or die "Can't open memory file: $!";
+    return Avro::BinaryDecoder->decode(
+        writer_schema => $schema,
+        reader_schema => $schema,
+        reader        => $reader,
+    );
+}
+
+my $err = 'Avro::BinaryDecoder::Error::CollectionSize';
+
+{
+    local $Avro::BinaryDecoder::MAX_COLLECTION_ITEMS = 10;
+
+    # zigzag(11) = 0x16: a single block of 11 items exceeds the limit of 10.
+    throws_ok { decode_bytes($array_schema, "\x16\x00") } $err,
+        "array block count above limit is rejected";
+
+    throws_ok { decode_bytes($map_schema, "\x16\x00") } $err,
+        "map block count above limit is rejected";
+
+    # Two blocks of 6 items (zigzag(6) = 0x0c) exceed the limit cumulatively.
+    # Well-formed encoding: two blocks of six null items (0 bytes each) followed
+    # by the terminating 0 block count.
+    throws_ok { decode_bytes($array_schema, "\x0c\x0c\x00") } $err,
+        "array cumulative block count above limit is rejected";
+
+    # Repeated map keys collapse in the hash; the cumulative check must still
+    # count every decoded pair. Two well-formed blocks of 6 pairs all keyed "a"
+    # (zigzag(6)=0x0c, key string = 0x02 0x61, null value = no bytes) then the
+    # terminating 0 block count exceed the limit of 10.
+    my $repeated_key_map = "\x0c" . ("\x02\x61" x 6) . "\x0c" . ("\x02\x61" x 6) . "\x00";
+    throws_ok { decode_bytes($map_schema, $repeated_key_map) } $err,
+        "map cumulative pair count with repeated keys is rejected";
+
+    # Negative count: unsigned varint 0x15 decodes (zigzag) to -11, whose
+    # absolute value (11) is used; a block size long (0x00) follows, then the
+    # terminating 0 block count makes the encoding well-formed.
+    throws_ok { decode_bytes($array_schema, "\x15\x00\x00") } $err,
+        "negative array block count is bounded by its absolute value";
+
+    # zigzag(3) = 0x06: three null items are within the limit and decode fine.
+    my $decoded = decode_bytes($array_schema, "\x06\x00");
+    is_deeply $decoded, [undef, undef, undef],
+        "array within the limit still decodes";
+}
+
+# By default the limit is generous enough not to affect ordinary decoding.
+no warnings 'once'; # $DEFAULT_MAX_COLLECTION_ITEMS is a package global set at load time
+is $Avro::BinaryDecoder::MAX_COLLECTION_ITEMS, $Avro::BinaryDecoder::DEFAULT_MAX_COLLECTION_ITEMS,
+    "default collection item limit restored outside local scope";
+
+done_testing;