apache/kafka · error · InvalidRecordException

Invalid wrapper compressionType found in legacy deep record

Error message

Invalid wrapper compressionType found in legacy deep record iterator {}

What it means

Thrown by DeepRecordsIterator (line 331) as an InvalidRecordException when the wrapper record's compressionType equals ZSTD. ZSTD support (KIP-110) was added only for message format v2; legacy v0/v1 message sets may only use NONE, GZIP, SNAPPY, or LZ4. Encountering ZSTD on a legacy wrapper is invalid by spec.

Source

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

    }

    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,
                                    int maxMessageSize,
                                    BufferSupplier bufferSupplier) {
            LegacyRecord wrapperRecord = wrapperEntry.outerRecord();
            this.wrapperMagic = wrapperRecord.magic();
            if (wrapperMagic != RecordBatch.MAGIC_VALUE_V0 && wrapperMagic != RecordBatch.MAGIC_VALUE_V1)
                throw new InvalidRecordException("Invalid wrapper magic found in legacy deep record iterator " + wrapperMagic);

            CompressionType compressionType = wrapperRecord.compressionType();
            if (compressionType == CompressionType.ZSTD)
                throw new InvalidRecordException("Invalid wrapper compressionType found in legacy deep record iterator " + wrapperMagic);
            ByteBuffer wrapperValue = wrapperRecord.value();
            if (wrapperValue == null)
                throw new InvalidRecordException("Found invalid compressed record set with null value (magic = " +
                        wrapperMagic + ")");

            InputStream stream = Compression.of(compressionType).build().wrapForInput(wrapperValue, wrapperRecord.magic(), bufferSupplier);
            LogInputStream<AbstractLegacyRecordBatch> logStream = new DataLogInputStream(stream, maxMessageSize);

            long lastOffsetFromWrapper = wrapperEntry.lastOffset();
            long timestampFromWrapper = wrapperRecord.timestamp();
            this.innerEntries = new ArrayDeque<>();

            // If relative offset is used, we need to decompress the entire message first to compute
            // the absolute offset. For simplicity and because it's a format that is on its way out, we
            // do the same for message format version 0
            try {
                while (true) {
                    AbstractLegacyRecordBatch innerEntry = logStream.nextBatch();

View on GitHub (pinned to c31c9215e1)

Solutions

Produce with magic v2 if you need ZSTD compression (set message.format.version >= 0.11.0 on the broker).
If you must use legacy v0/v1, restrict producer compression.type to gzip, snappy, lz4, or none.
Reproduce with kafka-dump-log --deep-iteration to confirm which segment/offset has the bad attributes byte and re-send those records correctly.
Audit non-Java producers for the attributes-byte encoding of compression (low 3 bits) to ensure they are not emitting 4 (ZSTD) for v0/v1.

Example fix

// before
ProducerConfig: compression.type=zstd, broker message.format.version=0.10.2 (v1)

// after
Either set message.format.version=0.11+ (v2) and keep zstd,
or set compression.type=lz4 while staying on v1.

Defensive patterns

Strategy: try-catch

Try / catch

// Legacy v0/v1 batches must not use ZSTD compression (only v2 supports it).
// The error surfaces during deep iteration of a compressed legacy wrapper.
import org.apache.kafka.common.errors.InvalidRecordException;

try {
    for (Record r : legacyBatch) { /* process */ }
} catch (InvalidRecordException e) {
    // a v0/v1 producer incorrectly produced ZSTD-compressed records.
    log.error("ZSTD compression not permitted in legacy format; batch is malformed", e);
}

Prevention

ZSTD is only valid for message format v2 (KIP-110); never enable ZSTD on producers targeting a v0/v1 topic.
Keep compression.type and the topic's message.format.version consistent; if the topic is v0/v1, use NONE/GZIP/SNAPPY/LZ4 only.
When upgrading to ZSTD, raise message.format.version to v2 on the broker and verify producer client version supports KIP-110 before switching.
Quarantine and re-drive batches that trigger this error rather than retrying; the bytes are structurally invalid for the declared format.

When it happens

Trigger: A v0/v1 message batch whose attributes byte encodes ZSTD. Produced by a buggy custom producer, an interop client that mis-sets the attributes byte, or by corruption that flipped bits into the ZSTD code. The check fires the moment DeepRecordsIterator tries to decompress.

Common situations: Cross-language clients (e.g. older librdkafka forks) that set the wrong attributes byte. Bit-flip corruption in the attributes byte of a legacy record. Test code that constructs LegacyRecord with CompressionType.ZSTD directly. A producer forced to magic=1 but configured for zstd compression.

Related errors

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