GH-2306: Improve DeserEx Message For Improper Ex.

Resolves #2306

Include the original exception message in the `DeserializationException.cause`
when the original exception could not be serialized.

**cherry-pick to 2.9.x, 2.8.x**

Author: garyrussell

Diff for: spring-kafka/src/main/java/org/springframework/kafka/support/serializer/SerializationUtils.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2021 the original author or authors.
+ * Copyright 2020-2022 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.
@@ -156,9 +156,9 @@ public static void deserializationException(Headers headers, byte[] data, Except
 			stream = new ByteArrayOutputStream();
 			try (ObjectOutputStream oos = new ObjectOutputStream(stream)) {
 				exception = new DeserializationException("failed to deserialize",
-						data, isForKeyArg, new RuntimeException("Could not deserialize type "
-						+ ioex.getClass().getName() + " with message " + ioex.getMessage()
-						+ " failure: " + ioex.getMessage()));
+						data, isForKeyArg, new RuntimeException("Could not serialize type "
+						+ ex.getClass().getName() + " with message " + ioex.getMessage()
+						+ ". Original exception message: " + ex.getMessage()));
 				oos.writeObject(exception);
 			}
 			catch (IOException ex2) {

Diff for: spring-kafka/src/test/java/org/springframework/kafka/listener/ErrorHandlingDeserializerTests.java

@@ -111,6 +111,31 @@ public void close() {
 		ehd.close();
 	}
 
+	@Test
+	void notSerializable() {
+		class MyDes implements Deserializer<String> {
+
+			@Override
+			public String deserialize(String topic, byte[] data) {
+				return null;
+			}
+
+			@Override
+			public String deserialize(String topic, Headers headers, byte[] data) {
+				throw new CannotSerializeException("original exception message");
+			}
+
+		}
+		ErrorHandlingDeserializer<String> ehd = new ErrorHandlingDeserializer<>(new MyDes());
+		Headers headers = new RecordHeaders();
+		ehd.deserialize("foo", headers, new byte[1]);
+		DeserializationException dex = ListenerUtils.byteArrayToDeserializationException(null,
+				headers.lastHeader(SerializationUtils.VALUE_DESERIALIZER_EXCEPTION_HEADER).value());
+		assertThat(dex.getMessage())
+				.contains("Could not serialize")
+				.contains("original exception message");
+	}
+
 	@Configuration
 	@EnableKafka
 	public static class Config {
@@ -245,4 +270,19 @@ public static class ExtendedEHD<T> extends ErrorHandlingDeserializer<T> {
 
 	}
 
+	@SuppressWarnings("serial")
+	public static class CannotSerializeException extends RuntimeException {
+
+		private final Foo foo = new Foo();
+
+		public CannotSerializeException(String message) {
+			super(message);
+		}
+
+	}
+
+	public static class Foo {
+
+	}
+
 }