KAFKA-16507 Add raw record into RecordDeserialisationException

[fred-ro]

Add raw ConsumerRecord data to RecordDeserialisationException to make DLQ implementation easier

Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[kirktrue]

Thanks for the PR @fred-ro!

Since parseRecord catches all RuntimeExceptions, it's very forgiving. No doubt most of the time the errors it catches are related to deserialization issues. However, with this change, the catch block would now also call methods on the incoming Record. Is it possible that an error could be thrown in our efforts to create a ConsumerRecord for the exception?

Perhaps we can rely on the existing upstream code that creates the Record to prevent that?

Anyway, just wondering 🤔

[fvaleri]

What about using a helper method like this to throw the exception?

    private void throwRecordDeserializationException(RecordDeserializationException.DeserializationExceptionOrigin origin,
                                                     TopicPartition partition,
                                                     TimestampType timestampType,
                                                     Record record,
                                                     RuntimeException e) {
        throw new RecordDeserializationException(origin, partition, record.offset(), record.timestamp(), timestampType, record.key(), 
            record.value(), new RecordHeaders(record.headers()), "Error deserializing " + origin.toString() + " for partition " 
                + partition + " at offset " + record.offset() + ". If needed, please seek past the record to continue consumption.", e);
    }