More ConversionException Improvements

- add consumer records for batch conversion errors.

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/support/converter/ConversionException.java

@@ -16,6 +16,10 @@
 
 package org.springframework.kafka.support.converter;
 
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 
 import org.springframework.kafka.KafkaException;
@@ -33,6 +37,8 @@ public class ConversionException extends KafkaException {
 
 	private final ConsumerRecord<?, ?> record;
 
+	private final List<ConsumerRecord<?, ?>> records = new ArrayList<>();
+
 	private final Message<?> message;
 
 	/**
@@ -59,6 +65,20 @@ public ConversionException(String message, ConsumerRecord<?, ?> record, Throwabl
 		this.message = null;
 	}
 
+	/**
+	 * Construct an instance with the provided properties.
+	 * @param message A text message describing the reason.
+	 * @param records the consumer records.
+	 * @param cause the cause.
+	 * @since 2.7.2
+	 */
+	public ConversionException(String message, List<ConsumerRecord<?, ?>> records, Throwable cause) {
+		super(message, cause);
+		this.record = record;
+		this.records.addAll(records);
+		this.message = null;
+	}
+
 	/**
 	 * Construct an instance with the provided properties.
 	 * @param message A text message describing the reason.
@@ -82,6 +102,16 @@ public ConversionException(String message, Message<?> msg, Throwable cause) {
 		return this.record;
 	}
 
+	/**
+	 * Return the consumer record, if available.
+	 * @return the record.
+	 * @since 2.7.2
+	 */
+	@Nullable
+	public List<ConsumerRecord<?, ?>> getRecords() {
+		return Collections.unmodifiableList(this.records);
+	}
+
 	/**
 	 * Return the {@link Message}, if available.
 	 * @return the message.