...

Current state: Under Discussion

Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread]

JIRA: here [Change the link from KAFKA-1 to your own ticket]

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

...

As it’s only addition, it would still be compatible with existing consumer code.

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,
                                         ConsumerRecord<byte[], byte[]> record,
                                         String message,
                                         Throwable cause) {
       super(message, cause);
       this.partition = partition;
       this.consumerRecord = record;
   }

   public TopicPartition topicPartition() {
       return partition;
   }

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

   public ConsumerRecord<byte[], byte[]> getConsumerRecord() {
       return consumerRecord;
   }
}

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.

...

This change is backward compatible and will have no impact on existing application.

Test Plan

• Unit test CompletedFetchTest

Rejected Alternatives

• 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 to user’s code.