...
The idea is to expose CorruptRecordException when the error is already present in the broker's fetch 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
Expose CorruptRecordException when it is present in broker responses, indicating a disk failure.
Enhance CorruptRecordException to contain the topic partition and corrupted offset.
...