DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
| Table of Contents |
|---|
Status
Current state: Under Discussion Adopted
Discussion thread: here
JIRA: here
...
The changes in the RecordDeserializationException are a new protected constructor with added argument timestampType, timestamp, keyBuffer, valueBuffer and Headers, Headers and a new enum DeserializationExceptionOrigin. New accessor methods for the added fields and extra key/value() methods with byte array allocation..
A new enum DeserializationExceptionOrigin field is KeyDeserializationException and ValueDeserializationException child methods have similar public constructors, allowing to differentiate if needed the origin of the error: KEY or VALUE.
As it’s only addition, it would still be compatible with existing consumer code.
| Code Block | ||||
|---|---|---|---|---|
| ||||
public class RecordDeserializationException extends SerializationException {
@Deprecated
public RecordDeserializationException(TopicPartition partition,// New enum
public enum DeserializationExceptionOrigin {
long offsetKEY,
String message,
Throwable cause);
// New constructor
protectedVALUE
}
@Deprecated
public RecordDeserializationException(TopicPartition partition,
long offset,
long timestamp,
TimestampType timestampType,
ByteBuffer keyBuffer,
ByteBuffer valueBuffer,
Headers headers,
String message,
Throwable cause);
// New methodsconstructor
public TimestampType timestampType();
public long timestamp();
public ByteBuffer keyBuffer();
public ByteBuffer valueBuffer();
public byte[] key();
public byte[] value();
public Headers headers(); } | ||||
| Code Block | ||||
| ||||
public class KeyDeserializationException extends RecordDeserializationException { public KeyDeserializationException(RecordDeserializationException(DeserializationExceptionOrigin origin, TopicPartition partition, long offset, long timestamp, TimestampType timestampType, ByteBuffer keyBuffer, ByteBuffer valueBuffer, Headers headers, String message, Throwable cause); } | ||||
| Code Block | ||||
| ||||
public class ValueDeserializationException extends RecordDeserializationException { // New methods public publicDeserializationExceptionOrigin ValueDeserializationException(TopicPartition partition,origin(); public long offset, long timestamp, TimestampType timestampType, ByteBuffer keyBuffer, ByteBuffer valueBuffer, Headers headers, String message, Throwable causeTimestampType timestampType(); public long timestamp(); public ByteBuffer keyBuffer(); public ByteBuffer valueBuffer(); public Headers headers(); } |
Proposed Changes
We propose to include record content and metadata to the RecordDeserializationException, to create 2 childs classes KeyDeserializationException and ValueDeserializationException to differentiate key and value deserialization error. The keykeyBuffer() and valuevalueBuffer() methods will lazily create the needed byte array allow to access the content of the record. This would still allow us permit to access offsets and then skip the record but also to fetch the concerned data for specific processing by sending it to a dead letter queue or whatever action that makes sense.
...
| Code Block | ||
|---|---|---|
| ||
{
// …
try (KafkaConsumer<String, SimpleValue> consumer = new KafkaConsumer<>(settings())) {
// Subscribe to our topic
LOGGER.info("Subscribing to topic " + KAFKA_TOPIC);
consumer.subscribe(List.of(KAFKA_TOPIC));
LOGGER.info("Subscribed !");
try (KafkaProducer<byte[], byte[]> dlqProducer = new KafkaProducer<>(producerSettings())) {
//noinspection InfiniteLoopStatement
while (true) {
try {
final var records = consumer.poll(POLL_TIMEOUT);
LOGGER.info("poll() returned {} records", records.count());
for (var record : records) {
LOGGER.info("Fetch record key={} value={}", record.key(), record.value());
// Any processing
// ...
}
} catch (RecordDeserializationException re) {
long offset = re.offset();
Throwable t = re.getCause();
LOGGER.error("Failed to consume at partition={} offset={}", re.topicPartition().partition(), offset, t);
sendDlqRecord(dlqProducer, re());
LOGGER.info("Skipping offset={}", offset);
consumer.seek(re.topicPartition(), offset + 1);
} catch (Exception e) {
LOGGER.error("Failed to consume", e);
}
}
}
} finally {
LOGGER.info("Closing consumer");
}
}
void sendDlqRecord(KafkaProducer<byte[], byte[]> dlqProducer, RecordDeserializationException re) {
var dlqRecord = new ProducerRecord<>(DLQ_TOPIC, Utils.toNullableArray(re.keykeyBuffer()), Utils.toNullableArray(re.valuevalueBuffer()));
try try {
dlqProducer.send(dlqRecord).get();
LOGGER.info("Record sent to DLQ");
} catch (Exception e) {
LOGGER.error("Failed to send corrupted record to DLQ", e);
}
} |
...
This change is backward compatible and will have no impact on existing application.
The previous RecordDeserializationException constructor is deprecated as no more used.
Test Plan
- Unit test CompletedFetchTest
...