DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
KIP-334 introduced into the Consumer the RecordDeserializationException with offsets information. That is useful to skip a poison pill but as you do not have access to the Record, it still prevents easy implementation of dead letter queue or simply logging the faulty data.
Public Interfaces
The change changes in the RecordDeserializationException would be modification of the constructor and a new method to get the ConsumerRecordare a new protected constructor with added argument timestampType, timestamp, keyBuffer,valueBuffer and Headers. New accessor methods for the added fields and extra key/value() methods with byte array allocation.
KeyDeserializationException and ValueDeserializationException child methods have similar public constructors, allowing to differentiate if needed the origin of the error.
As it’s only addition, it would still be compatible with existing consumer code.
| Code Block | ||||
|---|---|---|---|---|
| ||||
public class RecordDeserializationException extends SerializationException {
@Deprecated
p
ublicpublic RecordDeserializationException(TopicPartition partition,
long offset,
String message,
Throwable cause);
// New constructor
publicprotected RecordDeserializationException(TopicPartition partition,
long offset,
long timestamp,
TimestampType timestampType,
ByteBuffer keyBuffer,
ByteBuffer valueBuffer,
Record Headers recordheaders,
String message,
Throwable cause);
// New methods
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(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 { public ValueDeserializationException(TopicPartition partition, long offset, long timestamp, TimestampType timestampType, ByteBuffer keyBuffer, ByteBuffer valueBuffer, Headers headers, String message, Throwable cause method public Record record(); } |
Proposed Changes
We propose to include the Record object record content and metadata to the RecordDeserializationException and provide a getConsumerRecord method that , to create 2 childs classes KeyDeserializationException and ValueDeserializationException to differentiate key and value deserialization error. The key() and value() methods will lazily create the needed wrapper object if neededbyte array to access content of the record. This would still allow us 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.
Usage example
Here is an example of basic usage to implement a DLQ feature
| 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.record());
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, RecordRecordDeserializationException recordre) {
var dlqRecord = new ProducerRecord<>(DLQ_TOPIC, Utilsre.toNullableArray(record.key()), Utilsre.toNullableArray(record.value()));
try {
dlqProducer.send(dlqRecord).get();
LOGGER.info("Record sent to DLQ");
} catch (Exception e) {
LOGGER.error("Failed to send corrupted record to DLQ", e);
}
} |
...
- Using a raw byte array consumer and doing the deserialization in the user code. This permits dead letter queue implementation but is moving all the complexity of deseliarization deserialization to user’s code.