apache/kafka · error · CorruptRecordException

Record size exceeds the largest allowable message size (%d).

Error message

Record size exceeds the largest allowable message size (%d).

What it means

Thrown by DataLogInputStream.nextBatch() (line 303) as a CorruptRecordException when the size field of a legacy record frame exceeds the maxMessageSize cap passed to the stream. This guards against unbounded allocations and DoS when materializing a record from an input stream. The limit is the configured maximum record/message size for the reading context.

Source

Thrown at clients/src/main/java/org/apache/kafka/common/record/internal/AbstractLegacyRecordBatch.java:303

        DataLogInputStream(InputStream stream, int maxMessageSize) {
            this.stream = stream;
            this.maxMessageSize = maxMessageSize;
            this.offsetAndSizeBuffer = ByteBuffer.allocate(Records.LOG_OVERHEAD);
        }

        public AbstractLegacyRecordBatch nextBatch() throws IOException {
            offsetAndSizeBuffer.clear();
            Utils.readFully(stream, offsetAndSizeBuffer);
            if (offsetAndSizeBuffer.hasRemaining())
                return null;

            long offset = offsetAndSizeBuffer.getLong(Records.OFFSET_OFFSET);
            int size = offsetAndSizeBuffer.getInt(Records.SIZE_OFFSET);
            if (size < LegacyRecord.RECORD_OVERHEAD_V0)
                throw new CorruptRecordException(String.format("Record size is less than the minimum record overhead (%d)", LegacyRecord.RECORD_OVERHEAD_V0));
            if (size > maxMessageSize)
                throw new CorruptRecordException(String.format("Record size exceeds the largest allowable message size (%d).", maxMessageSize));

            ByteBuffer batchBuffer = ByteBuffer.allocate(size);
            Utils.readFully(stream, batchBuffer);
            if (batchBuffer.hasRemaining())
                return null;
            batchBuffer.flip();

            return new BasicLegacyRecordBatch(offset, new LegacyRecord(batchBuffer));
        }
    }

    private static class DeepRecordsIterator extends AbstractIterator<Record> implements CloseableIterator<Record> {
        private final ArrayDeque<AbstractLegacyRecordBatch> innerEntries;
        private final long absoluteBaseOffset;
        private final byte wrapperMagic;

        private DeepRecordsIterator(AbstractLegacyRecordBatch wrapperEntry,
                                    boolean ensureMatchingMagic,

View on GitHub (pinned to c31c9215e1)

Solutions

Raise the reader-side limit (fetch.max.bytes / max.partition.fetch.bytes for consumers, message.max.bytes for brokers) to at least the producer's max.request.size.
Reduce the payload size on the producer, or enable compression so the on-wire frame is smaller.
Keep producer max.request.size, broker message.max.bytes, and consumer fetch.max.bytes in a consistent hierarchy (consumer >= broker >= producer).
Verify no custom serializer is emitting unexpectedly large values (e.g. unbounded collections).

Example fix

// before
props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 1024 * 1024);

// after
props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 10 * 1024 * 1024);
props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 50 * 1024 * 1024);

Defensive patterns

Strategy: try-catch

Try / catch

// Thrown when a legacy record's declared size exceeds maxMessageSize used to
// construct the iterator (e.g. MemoryRecords.Builder or LogInputStream).
import org.apache.kafka.common.errors.CorruptRecordException;

try {
    for (Record r : records.records()) { /* process */ }
} catch (CorruptRecordException e) {
    // declared size > maxMessageSize: either a genuinely oversized/corrupt record
    // or the limit you passed is too small. Log the configured limit for diagnosis.
    log.error("Legacy record exceeds maxMessageSize={}; possible corruption or misconfigured limit", maxMessageSize, e);
}

Prevention

Configure max.message.bytes on the broker and max.request.size / message.max.bytes on the client consistently so producers cannot send records that exceed the size limit used to read them.
Keep producers on the v2 message format; the legacy-size-check path only runs for v0/v1 batches.
When reading legacy data, size maxMessageSize to at least the broker's absolute max to avoid false positives from legitimately large (but valid) records.
Treat this as data corruption first (skip/advance) and only revisit the limit after confirming the bytes are well-formed.
Log maxMessageSize alongside the failure so you can distinguish 'corrupt record' from 'limit too small'.

When it happens

Trigger: A producer sent a message larger than the reader's max.message.size / message.max.bytes / max.message.bytes, and the reader hits it while iterating compressed inner records (DeepRecordsIterator uses Integer.MAX_VALUE, but a direct DataLogInputStream over a fetch response is bounded). Also when broker message.max.bytes is larger than what the consumer or client is willing to buffer.

Common situations: Producer compression.disabled sending a single large payload, or a compressed wrapper that decompresses to something whose declared size is huge. Broker config message.max.bytes raised without raising replica.fetch.max.bytes / fetch.max.bytes on consumers. Cross-cluster mirroring where the target has a smaller size cap than source.

Related errors

AI-assisted analysis of apache/kafka@c31c9215e1 (2026-08-03). Data as JSON: /data/errors/6ceafa54eb6fcfd3.json. Report an issue: GitHub.