apache/kafka · critical · KafkaException

Encountered corrupt message when fetching topic-partition ${

Error message

Encountered corrupt message when fetching topic-partition ${tp.topicPartition()}

What it means

Thrown by ShareFetchCollector.handleInitializeErrors when a fetch response returns Errors.CORRUPT_MESSAGE for a topic-partition. The broker reports that a fetched record could not be checksum-verified (CRC mismatch), indicating on-disk corruption or a transmission error; the share collector converts this into a KafkaException because individual record corruption is not retriable at the fetch layer without operator intervention.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareFetchCollector.java:182

            log.warn("Received unknown topic or partition error in fetch for partition {}.", tp);
            requestMetadataUpdate(metadata, subscriptions, tp.topicPartition());
        } else if (error == Errors.UNKNOWN_TOPIC_ID) {
            log.warn("Received unknown topic ID error in fetch for partition {}.", tp);
            requestMetadataUpdate(metadata, subscriptions, tp.topicPartition());
        } else if (error == Errors.INCONSISTENT_TOPIC_ID) {
            log.warn("Received inconsistent topic ID error in fetch for partition {}.", tp);
            requestMetadataUpdate(metadata, subscriptions, tp.topicPartition());
        } else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
            // 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.topicPartition());
            throw new TopicAuthorizationException(Set.of(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 topic-partition {}.",
                    tp.topicPartition());
        } else if (error == Errors.CORRUPT_MESSAGE) {
            throw new KafkaException("Encountered corrupt message when fetching topic-partition "
                    + tp.topicPartition());
        } else {
            throw new IllegalStateException("Unexpected error code " + error.code()
                    + " while fetching from topic-partition " + tp.topicPartition());
        }
    }
}

View on GitHub (pinned to c31c9215e1)

Solutions

1. Investigate broker logs and disk health (dmesg, smartctl, broker log segments for the reported partition).
2. If corruption is isolated, use kafka-delete-records.sh or kafkacat to truncate past the bad offset, or restore the segment from a healthy replica.
3. Ensure replication factor >= 3 so a corrupt replica can be replaced from a healthy one.
4. Reproduce with the consumer's isolation.level and check producer client for retry/backoff configuration; verify the topic was produced with acks=all.

Example fix

# before: corrupt offset in partition 'orders-3'

# after: truncate past the corrupt offset
bin/kafka-delete-records.sh --bootstrap-server broker:9092 \
  --offset-json-file delete-records.json
# {"partitions":[{"topic":"orders","partition":3,"offset":12345}],"version":1}

try {
    records = consumer.poll(Duration.ofSeconds(5));
} catch (org.apache.kafka.common.errors.KafkaException e) {
    if (e.getMessage() != null && e.getMessage().contains("corrupt message")) {
        // A fetch returned a record that failed checksum / log validation on the broker side.
        // This usually means disk corruption on a specific broker or a bug in a producer's serializer.
        log.error("Corrupt message detected; will skip and continue", e);
        // Share consumers: RELEASE the affected partition's current batch on next poll so the
        // broker can retire the records; quarantine the topic-partition for investigation.
        quarantine(topicPartition);
    } else {
        throw e;
    }
}

When it happens

Trigger: Disk corruption on a broker log segment; bit-flip during network transfer that survives into the broker; hardware fault producing a record whose CRC no longer matches; truncated/partial segment after a hard broker crash.

Common situations: Failing disk on a broker; memory or NIC errors causing silent corruption; running on storage without checksumming (e.g. some ephemeral cloud volumes); topic produced with acks=0 so producers never detected corruption before it landed.

Related errors

• Record batch for partition {} at offset {} is invalid, cause
• Record for partition ${partition.topicPartition()} at offset
• Connection to {node} failed.
• Unknown acknowledge type id: {id}
• Failed to make progress reading messages at {}={}. Received

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