apache/kafka · error · KafkaException

Encountered corrupt message when fetching offset {} for topi

Error message

Encountered corrupt message when fetching offset {} for topic-partition {}

What it means

Thrown as a KafkaException when the broker returns Errors.CORRUPT_MESSAGE for a fetch response in FetchCollector.handleInitializeErrors. It signals that the broker's CRC validation of a record batch at the requested fetch offset failed, so the consumer refuses to hand the malformed data to the application. Unlike transient fetch errors (NOT_LEADER_OR_FOLLOWER, OFFSET_OUT_OF_RANGE) which are handled inline, CORRUPT_MESSAGE is treated as unrecoverable for that position and bubbles up to the caller.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchCollector.java:377

                        throw new OffsetOutOfRangeException(errorMessage,
                                Collections.singletonMap(tp, position.offset));
                    }
                }
            } else {
                log.debug("Unset the preferred read replica {} for partition {} since we got {} when fetching {}",
                        clearedReplicaId.get(), tp, error, fetchOffset);
            }
        } else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
            //we log the actual partition and not just the topic to help with ACL propagation issues in large clusters
            log.warn("Not authorized to read from partition {}.", tp);
            throw new TopicAuthorizationException(Collections.singleton(tp.topic()));
        } else if (error == Errors.UNKNOWN_LEADER_EPOCH) {
            log.debug("Received unknown leader epoch error in fetch for partition {}", tp);
        } else if (error == Errors.UNKNOWN_SERVER_ERROR) {
            log.warn("Unknown server error while fetching offset {} for topic-partition {}",
                    fetchOffset, tp);
        } else if (error == Errors.CORRUPT_MESSAGE) {
            throw new KafkaException("Encountered corrupt message when fetching offset "
                    + fetchOffset
                    + " for topic-partition "
                    + tp);
        } else {
            throw new IllegalStateException("Unexpected error code "
                    + error.code()
                    + " while fetching at offset "
                    + fetchOffset
                    + " from topic-partition " + tp);
        }
    }
}

View on GitHub (pinned to c31c9215e1)

Solutions

Identify the topic-partition and offset from the exception message, then use kafka-console-consumer or kafka-dump-log on the broker segment to confirm the corrupt batch.
Skip the corrupt record by seeking the consumer past the bad offset with consumer.seek(tp, corruptOffset + 1) and resume polling, or use a producer to write a compensating record.
If corruption is widespread, delete and recreate the affected partition's log segment on the broker (it will rebuild from ISR), or restore from a known-good backup/tiered storage.
Replace or diagnose the underlying disk/hardware on the affected broker to prevent recurrence.

Example fix

// before
consumer.subscribe(Collections.singleton("orders"));
while (true) {
    for (ConsumerRecord<String,String> r : consumer.poll(Duration.ofMillis(500))) {
        process(r);
    }
}

// after - skip past the corrupt offset reported by the exception
try {
    consumer.subscribe(Collections.singleton("orders"));
    while (true)
        consumer.poll(Duration.ofMillis(500)).forEach(this::process);
} catch (org.apache.kafka.common.KafkaException e) {
    if (e.getMessage().contains("corrupt message")) {
        TopicPartition tp = parseTp(e);            // from message text
        long bad = parseOffset(e);
        consumer.seek(tp, bad + 1);                 // skip the bad batch
    } else throw e;
}

Defensive patterns

Strategy: try-catch

Try / catch

try {
    ConsumerRecords<K,V> records = consumer.poll(Duration.ofMillis(timeoutMs));
} catch (org.apache.kafka.common.KafkaException e) {
    if (e.getMessage() != null && e.getMessage().contains("corrupt message")) {
        // option 1: seek past the bad offset on the offending partition
        consumer.seek(badPartition, badOffset + 1);
        // option 2: reset to earliest/latest if data is disposable
        consumer.seek(badPartition, consumer.beginningOffsets(Collections.singleton(badPartition)).get(badPartition));
    } else {
        throw e;
    }
}

Prevention

Run with broker log recovery / kafka-dump-log to identify and truncate the corrupt segment on the broker side.
Keep crc.check.enabled at its default true so the broker rejects bad records before they reach consumers.
Monitor broker disk health and filesystem errors; corrupt messages usually stem from disk faults or unclean shutdowns.
Compacted or compacted+deleted topics can surface older corrupt records on rewind; isolate affected partitions and quarantine them.
If corruption is persistent, recreate the partition and re-produce from a validated source rather than skipping offsets blindly.

When it happens

Trigger: Returned by KafkaConsumer.poll() when the leader's log read at the consumer's fetch offset produces a record batch whose checksum does not match. Occurs specifically in the fetch error path that handles Errors.CORRUPT_MESSAGE after a successful FetchRequest; not raised by producer or admin operations.

Common situations: Disk corruption or bit-rot on a broker segment, a partially written segment after a hard broker crash, faulty storage hardware, memory/disk errors during log flush, or a broker version downgrade where record framing differs. Rare under normal operation; often surfaces on one specific partition/offset while others fetch fine.

Related errors

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