diff --git a/lang/perl/Makefile.PL b/lang/perl/Makefile.PL
index 4c07baac00e..b3b13042432 100644
--- a/lang/perl/Makefile.PL
+++ b/lang/perl/Makefile.PL
@@ -48,7 +48,7 @@ test_requires 'Test::Exception';
 test_requires 'Test::More', 0.88;
 test_requires 'Test::Pod';
 requires 'Compress::Zlib';
-requires 'Compress::Zstd';
+requires 'Compress::Zstd', '0.10'; # 0.10 first provides the streaming Compress::Zstd::Decompressor
 requires 'Encode';
 requires 'Error::Simple';
 requires 'JSON::MaybeXS';
diff --git a/lang/perl/lib/Avro/DataFileReader.pm b/lang/perl/lib/Avro/DataFileReader.pm
index 0ef5f02cff5..9d799d73caf 100644
--- a/lang/perl/lib/Avro/DataFileReader.pm
+++ b/lang/perl/lib/Avro/DataFileReader.pm
@@ -28,17 +28,22 @@ use Object::Tiny qw{
 
 use constant MARKER_SIZE => 16;
 
-# TODO: refuse to read a block more than block_max_size, instead
-# do partial reads
+# A data-file block is decompressed according to the file's codec. A block with
+# a very high compression ratio (or a malformed block) can expand to far more
+# memory than its compressed size. To guard against unbounded allocation, the
+# decompressed size of a single block is capped. This mirrors the Java SDK's
+# decompression limit (AVRO-4247). The default can be overridden with the
+# AVRO_MAX_DECOMPRESS_LENGTH environment variable.
+use constant DEFAULT_MAX_DECOMPRESS_LENGTH => 200 * 1024 * 1024; # 200 MiB
 
 use Avro::DataFile;
 use Avro::BinaryDecoder;
 use Avro::Schema;
 use Carp;
-use Compress::Zstd;
-use IO::Uncompress::Bunzip2 qw(bunzip2);
+use IO::Uncompress::Bunzip2 ();
 use IO::Uncompress::RawInflate ;
 use Fcntl();
+use bytes ();
 
 our $VERSION = '++MODULE_VERSION++';
 
@@ -203,36 +208,193 @@ sub read_block_header {
     $datafile->{block_size} = Avro::BinaryDecoder->decode_long(
         undef, undef, $fh,
     );
+    ## Both are Avro long (zigzag) values, so a malformed/truncated file can
+    ## yield negatives. A negative block_size would flow into $want and a
+    ## negative-length read; a negative object_count is equally nonsensical.
+    ## Reject both before first use.
+    if ($datafile->{object_count} < 0) {
+        croak "Invalid negative object count: $datafile->{object_count}";
+    }
+    if ($datafile->{block_size} < 0) {
+        croak "Invalid negative block size: $datafile->{block_size}";
+    }
     $datafile->{block_start} = tell $fh;
 
     return if $codec eq 'null';
 
-    ## we need to read the entire block into memory, to inflate it
-    my $nread = read $fh, my $block, $datafile->{block_size} + MARKER_SIZE
-        or croak "Error reading from file: $!";
+    ## Guard against an attacker-controlled block_size triggering a huge
+    ## allocation for the compressed block itself, before any decompression
+    ## happens. When the reader is configured with block_max_size, reject a
+    ## block whose declared compressed size exceeds that bound up front.
+    my $block_max = $datafile->{block_max_size};
+    my $block_size = $datafile->{block_size};
+    if (defined $block_max && $block_size > $block_max) {
+        Avro::DataFile::Error::CompressedBlockSize->throw(
+            "Compressed block size $block_size exceeds the configured block_max_size of $block_max bytes"
+        );
+    }
+
+    ## we need to read the entire block into memory, to inflate it. Read in
+    ## bounded chunks (rather than a single read of $want bytes, which would
+    ## pre-extend the buffer to the attacker-controlled block_size before a short
+    ## read is detected) and verify the exact byte count afterward: a short read
+    ## (truncated/malformed file) would otherwise slip through and surface later
+    ## as a confusing marker/decompressor error.
+    my $want = $datafile->{block_size} + MARKER_SIZE;
+    my $block = '';
+    my $chunk_size = 64 * 1024;
+    while (bytes::length($block) < $want) {
+        my $need = $want - bytes::length($block);
+        $need = $chunk_size if $need > $chunk_size;
+        my $nread = read $fh, my $buf, $need;
+        if (!defined $nread) {
+            croak "Error reading from file: $!";
+        }
+        last if $nread == 0;    # EOF
+        $block .= $buf;
+    }
+    if (bytes::length($block) != $want) {
+        croak "Short read: expected $want bytes for the block, got "
+            . bytes::length($block) . " (truncated file?)";
+    }
 
     ## remove the marker
     my $marker = substr $block, -(MARKER_SIZE), MARKER_SIZE, '';
     $datafile->{block_marker} = $marker;
 
+    ## The decompressed size of a block is capped to guard against a block with
+    ## a very high compression ratio expanding to far more memory than its
+    ## compressed size. The limit is the AVRO_MAX_DECOMPRESS_LENGTH environment
+    ## variable, or DEFAULT_MAX_DECOMPRESS_LENGTH. (Note: block_max_size is a
+    ## writer-side flush threshold measured in compressed bytes; it is used above
+    ## as a compressed-size pre-read guard, but is not reused here for the
+    ## decompressed-size cap, to avoid conflating the two units.)
+    my $limit = _max_decompress_length();
+
     ## this is our new reader
     $datafile->{reader} = do {
         if ($codec eq 'deflate') {
-            IO::Uncompress::RawInflate->new(\$block);
+            my $z = IO::Uncompress::RawInflate->new(\$block)
+                or croak "Error inflating block: $IO::Uncompress::RawInflate::RawInflateError";
+            my $uncompressed = _inflate_bounded($z, $limit);
+            _open_decompressed(\$uncompressed);
         }
         elsif ($codec eq 'bzip2') {
-            my $uncompressed;
-            bunzip2 \$block => \$uncompressed;
-            do { open $fh, '<', \$uncompressed; $fh };
+            my $z = IO::Uncompress::Bunzip2->new(\$block)
+                or croak "Error decompressing bzip2 block: $IO::Uncompress::Bunzip2::Bunzip2Error";
+            my $uncompressed = _inflate_bounded($z, $limit);
+            _open_decompressed(\$uncompressed);
         }
         elsif ($codec eq 'zstandard') {
-            do { open $fh, '<', \(decompress(\$block)); $fh };
+            my $uncompressed = _zstd_decompress_bounded(\$block, $limit);
+            _open_decompressed(\$uncompressed);
         }
     };
 
     return;
 }
 
+## Open an in-memory read handle over the decompressed block, surfacing any
+## failure via croak rather than leaving $datafile->{reader} undefined (which
+## would fail later with a less clear error). The handle keeps a reference to
+## the scalar, so the caller's buffer stays alive for the lifetime of the read.
+sub _open_decompressed {
+    my ($uncompressed_ref) = @_;
+    open my $fh, '<', $uncompressed_ref
+        or croak "Error opening decompressed block for reading: $!";
+    return $fh;
+}
+
+## Read from a streaming decompressor in chunks, rejecting the block as soon as
+## its decompressed size would exceed $limit so an over-large (or malicious)
+## block is not fully materialized in memory. Each read is sized to the
+## remaining budget (capped at 64 KiB) so the buffer overshoots $limit by at
+## most one byte before the check fires.
+sub _inflate_bounded {
+    my ($z, $limit) = @_;
+    my $uncompressed = '';
+    my $chunk;
+    my $status;
+    while (1) {
+        my $budget = $limit - bytes::length($uncompressed) + 1;
+        my $to_read = $budget < 65536 ? $budget : 65536;
+        $status = $z->read($chunk, $to_read);
+        last unless defined $status && $status > 0;
+        $uncompressed .= $chunk;
+        _check_decompress_length(bytes::length($uncompressed), $limit);
+    }
+    if (!defined $status || $status < 0) {
+        croak "Error decompressing block: " . $z->error;
+    }
+    return $uncompressed;
+}
+
+## Streaming zstandard decompression, bounded the same way as _inflate_bounded so
+## a high-ratio block is rejected before its full form is materialized.
+sub _zstd_decompress_bounded {
+    my ($block_ref, $limit) = @_;
+    # Load the zstandard decompressor lazily so the reader still loads and works
+    # for other codecs when Compress::Zstd::Decompressor is unavailable (e.g. an
+    # older Compress::Zstd distribution that lacks the Decompressor submodule).
+    # Localize $@ so probing for the module cannot clobber a caller's $@, and
+    # capture the require error before building the message.
+    my $require_error;
+    unless (do { local $@; my $ok = eval { require Compress::Zstd::Decompressor; 1 }; $require_error = $@; $ok }) {
+        Avro::DataFile::Error::UnsupportedCodec->throw(
+            "Cannot read zstandard-compressed block: Compress::Zstd::Decompressor is not available: $require_error"
+        );
+    }
+    my $decompressor = Compress::Zstd::Decompressor->new;
+    my $uncompressed = '';
+    my $length = bytes::length($$block_ref);
+    my $offset = 0;
+    # Feed the compressed input in small pieces. Compress::Zstd's streaming
+    # decompress() has no max-output parameter, so a single call materializes all
+    # output produced from the input it is handed; keeping each fed piece small
+    # bounds the transient $out (roughly one zstd internal block, ~128 KiB) so a
+    # highly-compressible frame cannot balloon a single call's output to
+    # gigabytes before the size check below runs.
+    my $piece_size = 16 * 1024;
+    while ($offset < $length) {
+        my $piece = substr($$block_ref, $offset, $piece_size);
+        $offset += $piece_size;
+        my $out = $decompressor->decompress($piece);
+        # The streaming decompressor croaks on a corrupt frame and otherwise
+        # emits all output produced while consuming the input it is given (there
+        # is no separate flush step in this API). Treat an undefined return as a
+        # failure and fail closed, rather than silently skipping it, so a
+        # malformed block cannot masquerade as a short, within-limit result.
+        unless (defined $out) {
+            croak "Error decompressing zstandard block";
+        }
+        # Check the prospective total before growing $uncompressed so a single
+        # large decompressed chunk cannot transiently balloon memory past the
+        # limit (or double peak memory from string reallocation).
+        _check_decompress_length(
+            bytes::length($uncompressed) + bytes::length($out), $limit);
+        $uncompressed .= $out;
+    }
+    return $uncompressed;
+}
+
+sub _check_decompress_length {
+    my ($length, $limit) = @_;
+    if ($length > $limit) {
+        Avro::DataFile::Error::DecompressionSize->throw(
+            "Decompressed block size $length exceeds the maximum allowed of $limit bytes"
+        );
+    }
+    return;
+}
+
+sub _max_decompress_length {
+    my $value = $ENV{AVRO_MAX_DECOMPRESS_LENGTH};
+    if (defined $value && $value =~ /\A[0-9]+\z/ && $value > 0) {
+        return $value + 0;
+    }
+    return DEFAULT_MAX_DECOMPRESS_LENGTH;
+}
+
 sub verify_marker {
     my $datafile = shift;
 
@@ -308,4 +470,14 @@ sub eof {
 package Avro::DataFile::Error::UnsupportedCodec;
 use parent 'Error::Simple';
 
+package Avro::DataFile::Error::DecompressionSize;
+use parent 'Error::Simple';
+
+## Raised when a block's declared *compressed* size exceeds the reader's
+## configured block_max_size, before the block is read into memory. Kept
+## distinct from DecompressionSize (which is about the *decompressed* size cap)
+## so callers can tell the two conditions apart.
+package Avro::DataFile::Error::CompressedBlockSize;
+use parent 'Error::Simple';
+
 1;
diff --git a/lang/perl/t/04_datafile.t b/lang/perl/t/04_datafile.t
index df7861b191e..e03d2ff7f94 100644
--- a/lang/perl/t/04_datafile.t
+++ b/lang/perl/t/04_datafile.t
@@ -148,31 +148,36 @@ is_deeply $all[0], $data, "Our data is intact!";
 
 
     ## zstandard!
-    $zfh = File::Temp->new(UNLINK => 0);
-    $write_file = Avro::DataFileWriter->new(
-        fh            => $zfh,
-        writer_schema => $schema,
-        codec         => 'zstandard',
-        metadata      => {
-            some => 'metadata',
-        },
-    );
-    $write_file->print($data);
-    $write_file->flush;
+    SKIP: {
+        eval { require Compress::Zstd::Decompressor; 1 }
+            or skip 'Compress::Zstd::Decompressor not available', 4;
 
-    ## rewind
-    seek $zfh, 0, 0;
+        $zfh = File::Temp->new(UNLINK => 0);
+        $write_file = Avro::DataFileWriter->new(
+            fh            => $zfh,
+            writer_schema => $schema,
+            codec         => 'zstandard',
+            metadata      => {
+                some => 'metadata',
+            },
+        );
+        $write_file->print($data);
+        $write_file->flush;
 
-    $read_file = Avro::DataFileReader->new(
-        fh            => $zfh,
-        reader_schema => $schema,
-    );
-    is $read_file->metadata->{'avro.codec'}, 'zstandard', 'avro.codec';
-    is $read_file->metadata->{'some'}, 'metadata', 'custom meta';
+        ## rewind
+        seek $zfh, 0, 0;
 
-    @all = $read_file->all;
-    is scalar @all, 1, "one object back";
-    is_deeply $all[0], $data, "Our data is intact!";
+        $read_file = Avro::DataFileReader->new(
+            fh            => $zfh,
+            reader_schema => $schema,
+        );
+        is $read_file->metadata->{'avro.codec'}, 'zstandard', 'avro.codec';
+        is $read_file->metadata->{'some'}, 'metadata', 'custom meta';
+
+        @all = $read_file->all;
+        is scalar @all, 1, "one object back";
+        is_deeply $all[0], $data, "Our data is intact!";
+    }
 }
 
 ## Test on a slightly larger file with 100 records
diff --git a/lang/perl/t/07_datafile_decompress_limit.t b/lang/perl/t/07_datafile_decompress_limit.t
new file mode 100644
index 00000000000..6d11d8a8819
--- /dev/null
+++ b/lang/perl/t/07_datafile_decompress_limit.t
@@ -0,0 +1,133 @@
+# 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
+
+# A data-file block is decompressed according to the file's codec. A block with
+# a very high compression ratio can expand to far more memory than its
+# compressed size. These tests ensure that reading such a block is rejected
+# instead of allocating without bound.
+
+use strict;
+use warnings;
+use File::Temp;
+use Avro::Schema;
+use Avro::DataFileWriter;
+use Test::More;
+use Test::Exception;
+
+use_ok 'Avro::DataFileReader';
+
+my $schema = Avro::Schema->parse('"string"');
+
+sub codec_file {
+    my ($codec, $payload) = @_;
+    my $fh = File::Temp->new(UNLINK => 1);
+    my $writer = Avro::DataFileWriter->new(
+        fh            => $fh,
+        writer_schema => $schema,
+        codec         => $codec,
+    );
+    $writer->print($payload);
+    $writer->flush;
+    seek $fh, 0, 0;
+    return $fh;
+}
+
+## A large, highly compressible value compresses to a tiny block but would
+## decompress to far more than the configured limit; reading it must be rejected
+## for every codec. The limit is set via AVRO_MAX_DECOMPRESS_LENGTH.
+sub assert_codec_rejects_oversized {
+    my ($codec) = @_;
+    my $big = "a" x (64 * 1024); # 64 KiB
+    my $fh = codec_file($codec, $big);
+    local $ENV{AVRO_MAX_DECOMPRESS_LENGTH} = 1024;
+    my $reader = Avro::DataFileReader->new(
+        fh            => $fh,
+        reader_schema => $schema,
+    );
+    throws_ok { $reader->all }
+        'Avro::DataFile::Error::DecompressionSize',
+        "$codec block exceeding the limit is rejected";
+}
+
+assert_codec_rejects_oversized('deflate');
+
+SKIP: {
+    eval { require IO::Compress::Bzip2; 1 }
+        or skip 'IO::Compress::Bzip2 not available', 1;
+    assert_codec_rejects_oversized('bzip2');
+}
+
+SKIP: {
+    eval { require Compress::Zstd::Decompressor; 1 }
+        or skip 'Compress::Zstd::Decompressor not available', 1;
+    assert_codec_rejects_oversized('zstandard');
+}
+
+## A block within the limit still decodes correctly.
+sub assert_codec_within_limit_decodes {
+    my ($codec) = @_;
+    my $payload = "hello world";
+    my $fh = codec_file($codec, $payload);
+    local $ENV{AVRO_MAX_DECOMPRESS_LENGTH} = 1024 * 1024;
+    my $reader = Avro::DataFileReader->new(
+        fh            => $fh,
+        reader_schema => $schema,
+    );
+    my @all = $reader->all;
+    is_deeply \@all, [$payload], "$codec block within the limit decodes";
+}
+
+assert_codec_within_limit_decodes('deflate');
+
+SKIP: {
+    eval { require IO::Compress::Bzip2; 1 }
+        or skip 'IO::Compress::Bzip2 not available', 1;
+    assert_codec_within_limit_decodes('bzip2');
+}
+
+SKIP: {
+    eval { require Compress::Zstd::Decompressor; 1 }
+        or skip 'Compress::Zstd::Decompressor not available', 1;
+    assert_codec_within_limit_decodes('zstandard');
+}
+
+## When block_max_size is configured on the reader, a block whose declared
+## compressed size exceeds it is rejected before the compressed block is read
+## into memory (guarding against an attacker-controlled block_size allocation).
+{
+    my $payload = "a" x (32 * 1024); # 32 KiB, compresses to a few dozen bytes
+    my $fh = codec_file('deflate', $payload);
+    # Set a large decompressed-size cap so the rejection can only come from the
+    # compressed-size guard, not the decompression limit, regardless of any
+    # AVRO_MAX_DECOMPRESS_LENGTH the runner may have set.
+    local $ENV{AVRO_MAX_DECOMPRESS_LENGTH} = 1024 * 1024 * 1024;
+    my $reader = Avro::DataFileReader->new(
+        fh             => $fh,
+        reader_schema  => $schema,
+        block_max_size => 8, # smaller than any real compressed block
+    );
+    eval { $reader->all };
+    my $err = $@;
+    isa_ok $err, 'Avro::DataFile::Error::CompressedBlockSize',
+        'compressed block exceeding block_max_size is rejected before reading';
+    like "$err", qr/block_max_size/,
+        'rejection is due to the compressed-size guard, not the decompression cap';
+}
+
+done_testing;