ConversionException Improvements

- add consumer record or `Message<?>` if available.

Author: garyrussell

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2019-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -47,7 +47,7 @@ protected Object convertPayload(Message<?> message) {
 			return getObjectMapper().writeValueAsBytes(message.getPayload());
 		}
 		catch (JsonProcessingException e) {
-			throw new ConversionException("Failed to convert to JSON", e);
+			throw new ConversionException("Failed to convert to JSON", message, e);
 		}
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -49,7 +49,7 @@ protected Object convertPayload(Message<?> message) {
 			return Bytes.wrap(getObjectMapper().writeValueAsBytes(message.getPayload()));
 		}
 		catch (JsonProcessingException e) {
-			throw new ConversionException("Failed to convert to JSON", e);
+			throw new ConversionException("Failed to convert to JSON", message, e);
 		}
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,7 +16,11 @@
 
 package org.springframework.kafka.support.converter;
 
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+
 import org.springframework.kafka.KafkaException;
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.Message;
 
 /**
  * Exception for conversions.
@@ -27,8 +31,65 @@
 @SuppressWarnings("serial")
 public class ConversionException extends KafkaException {
 
+	private final ConsumerRecord<?, ?> record;
+
+	private final Message<?> message;
+
+	/**
+	 * Construct an instance with the provided properties.
+	 * @param message A text message describing the reason.
+	 * @param cause the cause.
+	 */
 	public ConversionException(String message, Throwable cause) {
 		super(message, cause);
+		this.record = null;
+		this.message = null;
+	}
+
+	/**
+	 * Construct an instance with the provided properties.
+	 * @param message A text message describing the reason.
+	 * @param record the consumer record.
+	 * @param cause the cause.
+	 * @since 2.7.2
+	 */
+	public ConversionException(String message, ConsumerRecord<?, ?> record, Throwable cause) {
+		super(message, cause);
+		this.record = record;
+		this.message = null;
+	}
+
+	/**
+	 * Construct an instance with the provided properties.
+	 * @param message A text message describing the reason.
+	 * @param msg a {@link Message} converted from a consumer record.
+	 * @param cause the cause.
+	 * @since 2.7.2
+	 */
+	public ConversionException(String message, Message<?> msg, Throwable cause) {
+		super(message, cause);
+		this.record = null;
+		this.message = msg;
+	}
+
+	/**
+	 * Return the consumer record, if available.
+	 * @return the record.
+	 * @since 2.7.2
+	 */
+	@Nullable
+	public ConsumerRecord<?, ?> getRecord() {
+		return this.record;
+	}
+
+	/**
+	 * Return the {@link Message}, if available.
+	 * @return the message.
+	 * @since 2.7.2
+	 */
+	@Nullable
+	public Message<?> getMsg() {
+		return this.message;
 	}
 
 }

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

@@ -111,15 +111,15 @@ protected Object extractAndConvertValue(ConsumerRecord<?, ?> record, Type type)
 				return this.objectMapper.readValue((String) value, javaType);
 			}
 			catch (IOException e) {
-				throw new ConversionException("Failed to convert from JSON", e);
+				throw new ConversionException("Failed to convert from JSON", record, e);
 			}
 		}
 		else if (value instanceof byte[]) {
 			try {
 				return this.objectMapper.readValue((byte[]) value, javaType);
 			}
 			catch (IOException e) {
-				throw new ConversionException("Failed to convert from JSON", e);
+				throw new ConversionException("Failed to convert from JSON", record, e);
 			}
 		}
 		else {

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

@@ -93,15 +93,15 @@ protected Object convertFromInternal(Message<?> message, Class<?> targetClass, @
 				return getObjectMapper().readValue((String) value, javaType);
 			}
 			catch (IOException e) {
-				throw new ConversionException("Failed to convert from JSON", e);
+				throw new ConversionException("Failed to convert from JSON", message, e);
 			}
 		}
 		else if (value instanceof byte[]) {
 			try {
 				return getObjectMapper().readValue((byte[]) value, javaType);
 			}
 			catch (IOException e) {
-				throw new ConversionException("Failed to convert from JSON", e);
+				throw new ConversionException("Failed to convert from JSON", message, e);
 			}
 		}
 		else {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -48,7 +48,7 @@ protected Object convertPayload(Message<?> message) {
 					.writeValueAsString(message.getPayload());
 		}
 		catch (JsonProcessingException e) {
-			throw new ConversionException("Failed to convert to JSON", e);
+			throw new ConversionException("Failed to convert to JSON", message, e);
 		}
 	}