What causes "Found invalid number of record headers {}" in apache/kafka?

Use kafka-dump-log --files --print-data-log to locate the record with the bad header count and identify the producing client.

apache/kafka · error · InvalidRecordException

Found invalid number of record headers {}

Error message

Found invalid number of record headers {}

What it means

Thrown on the ByteBuffer deserialization path when the varint read for the per-record headers count is negative (numHeaders < 0). Magic-v2 records encode the header count as a varint and a negative value means the bytes are not a valid count; the library refuses to allocate a negative-sized header array.

Source

Thrown at clients/src/main/java/org/apache/kafka/common/record/internal/DefaultRecord.java:340

                timestamp = logAppendTime;

            int offsetDelta = ByteUtils.readVarint(buffer);
            long offset = baseOffset + offsetDelta;
            int sequence = baseSequence >= 0 ?
                    DefaultRecordBatch.incrementSequence(baseSequence, offsetDelta) :
                    RecordBatch.NO_SEQUENCE;

            // read key
            int keySize = ByteUtils.readVarint(buffer);
            ByteBuffer key = Utils.readBytes(buffer, keySize);

            // read value
            int valueSize = ByteUtils.readVarint(buffer);
            ByteBuffer value = Utils.readBytes(buffer, valueSize);

            int numHeaders = ByteUtils.readVarint(buffer);
            if (numHeaders < 0)
                throw new InvalidRecordException("Found invalid number of record headers " + numHeaders);
            if (numHeaders > buffer.remaining())
                throw new InvalidRecordException("Found invalid number of record headers. " + numHeaders + " is larger than the remaining size of the buffer");

            final Header[] headers;
            if (numHeaders == 0)
                headers = Record.EMPTY_HEADERS;
            else
                headers = readHeaders(buffer, numHeaders);

            // validate whether we have read all header bytes in the current record
            if (buffer.position() - recordStart != sizeOfBodyInBytes)
                throw new InvalidRecordException("Invalid record size: expected to read " + sizeOfBodyInBytes +
                        " bytes in record payload, but instead read " + (buffer.position() - recordStart));

            int totalSizeInBytes = ByteUtils.sizeOfVarint(sizeOfBodyInBytes) + sizeOfBodyInBytes;
            return new DefaultRecord(totalSizeInBytes, attributes, offset, timestamp, sequence, key, value, headers);
        } catch (BufferUnderflowException | IllegalArgumentException e) {
            throw new InvalidRecordException("Found invalid record structure", e);

View on GitHub (pinned to c31c9215e1)

Solutions

Use kafka-dump-log --files <segment> --print-data-log to locate the record with the bad header count and identify the producing client.
Ensure the producer actually uses the Apache Kafka serializer path (ProducerRecord + DefaultRecord.writeTo) rather than writing raw bytes; never hand-format the headers section.
If caused by disk corruption, verify checksums (DefaultRecordBatch validates CRC) and recover the partition from a healthy in-sync replica.
Add a producer-side integration test that round-trips records with the same header set the consumer reads, to catch off-spec serializers before deploy.

Defensive patterns

Strategy: try-catch

Validate before calling

// numHeaders is decoded internally; you can only pre-check that the varint
// stream still has room. Full prevention requires re-reading the varint
// yourself, which duplicates the parser. Rely on try-catch.
// If you do decode numHeaders manually:
if (numHeaders < 0) { /* malformed; skip record */ }

Type guard

// Guard a decoded header count before further parsing.
private static boolean isValidHeaderCount(int numHeaders) {
    return numHeaders >= 0;
}

Try / catch

try {
    DefaultRecord r = DefaultRecord.readFrom(buffer, baseOffset, baseTimestamp, baseSequence, logAppendTime);
} catch (InvalidRecordException e) {
    // negative numHeaders means the varint field was corrupted
    LOG.warn("Malformed record (bad header count) near offset {}", baseOffset, e);
}

Prevention

Use only Kafka-native serializers (ByteArraySerializer, StringSerializer, etc.) so the varint header-count field is always well-formed.
Do not bit-edit produced record bytes; the header-count varint is positional and a one-byte shift produces a huge or negative value.
Validate record headers on the producer side (count, key length) before send to surface corruption early.

When it happens

Trigger: Raised at DefaultRecord.java:339-340 inside the private readFrom(ByteBuffer, ...) after ByteUtils.readVarint(buffer) at line 338 returns a negative number. Triggered by any corrupt or hand-crafted batch where the bytes that should encode numHeaders instead decode as -1 (the null-marker) or another negative varint.

Common situations: A producer wrote a record with a malformed header section (e.g. via a custom Serializer that bypassed DefaultRecord.writeTo), bit-flip on disk or in transit, partial overwrite of a log segment, or an attempt to consume a topic produced by a non-Apache-Kafka client with an off-spec record format.

Related errors

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