Status

Current state: Under Discussion

Discussion thread: here [TBD]

JIRA: here

Motivation

As part of our analysis around KAFKA-19430, we realized it's not possible to handle a CorruptRecordException in Streams because it's not exposed to the client; instead, a generic KafkaException is thrown. This KIP proposes to expose CorruptRecordException directly, allowing the client to decide how to handle this specific error.

Public Interfaces

The idea is to expose CorruptRecordException when the error is already present in the broker's response, which signifies disk corruption. We will also enhance CorruptRecordException to include the TopicPartition and offset.

It's important to highlight that offset here is an offset we're trying to read from, not a offset of corruption!

CorruptRecordException

package org.apache.kafka.common.errors;


import org.apache.kafka.common.TopicPartition;


/**
 * This exception indicates a record has failed its internal CRC check, this generally indicates network or disk corruption.
 */
public class CorruptRecordException extends RetriableException {


    private static final long serialVersionUID = 1L;


    private TopicPartition topicPartition;
    private long corruptOffset;


    public CorruptRecordException() {
        super("This message has failed its CRC checksum, exceeds the valid size, has a null key for a compacted topic, or is otherwise corrupt.");
    }


    public CorruptRecordException(String message) {
        super(message);
    }


    public CorruptRecordException(Throwable cause) {
        super(cause);
    }


    public CorruptRecordException(String message, Throwable cause) {
        super(message, cause);
    }


    public CorruptRecordException(String message, Throwable cause, TopicPartition topicPartition, long corruptOffset) {
        super(message, cause);
        this.topicPartition = topicPartition;
        this.corruptOffset = corruptOffset;
    }


    public TopicPartition topicPartition() {
        return topicPartition;
    }


    public long corruptOffset() {
        return corruptOffset;
    }
}


Proposed Changes

Compatibility, Deprecation, and Migration Plan

Test Plan