Versions Compared

Old Version 2

changes.mady.by.user Damien Gasparina

Saved on

compared with

New Version 3

changes.mady.by.user Damien Gasparina

Saved on

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.

...

Code Block
languagejava
titleRecordDeserializationException.java
public class RecordDeserializationException extends SerializationException {
   private static final long serialVersionUID = 2L;
   private final TopicPartition partition;
   private final ConsumerRecord<byte[], byte[]> consumerRecord;

   public RecordDeserializationException(TopicPartition partition@Deprecated
  p
ublic RecordDeserializationException(TopicPartition partition,
                                        long offset,
                                        String ConsumerRecord<byte[], byte[]> record,
message,
                                        Throwable cause);

  // New constructor
  public StringRecordDeserializationException(TopicPartition messagepartition,
                                        Record  Throwable cause) {
record,
             super(message, cause);
       this.partition = partition;
       this.consumerRecord = record;
   }

   public TopicPartition topicPartition() {String message,
       return   partition;
   }

   public long offset() {
       return consumerRecord.offset();
   }

   public ConsumerRecord<byte[], byte[]> getConsumerRecord() {
   Throwable cause);
  // returnNew consumerRecord;method
  public Record }record();
}

Proposed Changes

We propose to include the Record object to the RecordDeserializationException and provide a getConsumerRecord method that will lazily create the needed wrapper object if needed. 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
languagejava
{

// …

  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, Record record) {
   var dlqRecord = new ProducerRecord<>(DLQ_TOPIC, Utils.toNullableArray(record.key()), Utils.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);
   }
}

Compatibility, Deprecation, and Migration Plan

...