diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/SerializationUtils.java b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/SerializationUtils.java index bd525469..cbe82faf 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/SerializationUtils.java +++ b/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 final class SerializationUtils { 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 --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ErrorHandlingDeserializerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ErrorHandlingDeserializerTests.java index 63726115..b73e00ed 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/ErrorHandlingDeserializerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/ErrorHandlingDeserializerTests.java @@ -110,6 +110,31 @@ public class ErrorHandlingDeserializerTests { ehd.close(); } + @Test + void notSerializable() { + class MyDes implements Deserializer { + + @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 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 { @@ -237,4 +262,19 @@ public class ErrorHandlingDeserializerTests { } + @SuppressWarnings("serial") + public static class CannotSerializeException extends RuntimeException { + + private final Foo foo = new Foo(); + + public CannotSerializeException(String message) { + super(message); + } + + } + + public static class Foo { + + } + }