GH-853: Type Safe ErrorHandlingDeserializer

Resolves https://github.com/spring-projects/spring-kafka/issues/853

Since a `null` key is common, we only check for the exception header
if we detect that the error handling deserializer is configured.

* Polishing; add failedDeserializationFunction

spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

@@ -16,6 +16,9 @@
 
 package org.springframework.kafka.listener;
 
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.ObjectInputStream;
 import java.lang.reflect.Type;
 import java.time.Duration;
 import java.util.ArrayList;
@@ -50,6 +53,9 @@ import org.apache.kafka.common.Metric;
 import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.WakeupException;
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.header.internals.RecordHeaders;
 
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.kafka.KafkaException;
@@ -70,6 +76,7 @@ import org.springframework.kafka.support.TopicPartitionInitialOffset;
 import org.springframework.kafka.support.TopicPartitionInitialOffset.SeekPosition;
 import org.springframework.kafka.support.TransactionSupport;
 import org.springframework.kafka.support.serializer.DeserializationException;
+import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer2;
 import org.springframework.kafka.transaction.KafkaAwareTransactionManager;
 import org.springframework.scheduling.SchedulingAwareRunnable;
 import org.springframework.scheduling.TaskScheduler;
@@ -417,6 +424,10 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 
 		private final Duration pollTimeout = Duration.ofMillis(this.containerProperties.getPollTimeout());
 
+		private final boolean checkNullKeyForExceptions;
+
+		private final boolean checkNullValueForExceptions;
+
 		private volatile Map<TopicPartition, OffsetMetadata> definedPartitions;
 
 		private volatile Collection<TopicPartition> assignedPartitions;
@@ -521,6 +532,18 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 			if (this.containerProperties.isLogContainerConfig()) {
 				this.logger.info(this);
 			}
+			Map<String, Object> props = KafkaMessageListenerContainer.this.consumerFactory.getConfigurationProperties();
+			this.checkNullKeyForExceptions = checkDeserializer(props.get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG));
+			this.checkNullValueForExceptions = checkDeserializer(
+					props.get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG));
+		}
+
+		private boolean checkDeserializer(Object deser) {
+			return deser instanceof Class
+					? ((Class<?>) deser).equals(ErrorHandlingDeserializer2.class)
+					: deser instanceof String
+						? ((String) deser).equals(ErrorHandlingDeserializer2.class.getName())
+						: false;
 		}
 
 		protected void checkConsumer() {
@@ -1136,6 +1159,12 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 				if (record.key() instanceof DeserializationException) {
 					throw (DeserializationException) record.key();
 				}
+				if (record.value() == null && this.checkNullValueForExceptions) {
+					checkDeser(record, ErrorHandlingDeserializer2.VALUE_DESERIALIZER_EXCEPTION_HEADER);
+				}
+				if (record.key() == null && this.checkNullKeyForExceptions) {
+					checkDeser(record, ErrorHandlingDeserializer2.KEY_DESERIALIZER_EXCEPTION_HEADER);
+				}
 				switch (this.listenerType) {
 					case ACKNOWLEDGING_CONSUMER_AWARE:
 						this.listener.onMessage(record,
@@ -1197,6 +1226,25 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
 			return null;
 		}
 
+		public void checkDeser(final ConsumerRecord<K, V> record, String headerName) {
+			Header header = record.headers().lastHeader(headerName);
+			if (header != null) {
+				try {
+					DeserializationException ex = (DeserializationException) new ObjectInputStream(
+							new ByteArrayInputStream(header.value())).readObject();
+					Headers headers = new RecordHeaders(Arrays.stream(record.headers().toArray())
+							.filter(h -> !h.key()
+									.startsWith(ErrorHandlingDeserializer2.KEY_DESERIALIZER_EXCEPTION_HEADER_PREFIX))
+							.collect(Collectors.toList()));
+					ex.setHeaders(headers);
+					throw ex;
+				}
+				catch (IOException | ClassNotFoundException | ClassCastException e) {
+					this.logger.error("Failed to deserialize a deserialization exception", e);
+				}
+			}
+		}
+
 		public void ackCurrent(final ConsumerRecord<K, V> record, @SuppressWarnings("rawtypes") Producer producer) {
 			if (this.isRecordAck) {
 				Map<TopicPartition, OffsetAndMetadata> offsetsToCommit =

spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DeserializationException.java

@@ -35,14 +35,14 @@ import org.springframework.lang.Nullable;
 public class DeserializationException extends KafkaException {
 
 	@Nullable
-	private final Headers headers;
+	private Headers headers;
 
 	private final byte[] data;
 
 	private final boolean isKey;
 
 	public DeserializationException(String message, byte[] data, boolean isKey, Throwable cause) {
-		this(message, null, data, isKey, cause); // NOSONAR test coverage
+		this(message, null, data, isKey, cause);
 	}
 
 	public DeserializationException(String message, @Nullable Headers headers, byte[] data, // NOSONAR array reference
@@ -59,6 +59,10 @@ public class DeserializationException extends KafkaException {
 		return this.headers;
 	}
 
+	public void setHeaders(@Nullable Headers headers) {
+		this.headers = headers;
+	}
+
 	public byte[] getData() {
 		return this.data; // NOSONAR array reference
 	}

spring-kafka/src/main/java/org/springframework/kafka/support/serializer/ErrorHandlingDeserializer.java

@@ -34,10 +34,12 @@ import org.springframework.util.ClassUtils;
  *
  * @author Gary Russell
  * @author Artem Bilan
+ * @deprecated in favor of {@link ErrorHandlingDeserializer2}.
  *
  * @since 2.2
  *
  */
+@Deprecated
 public class ErrorHandlingDeserializer<T> implements ExtendedDeserializer<T> {
 
 	/**

spring-kafka/src/main/java/org/springframework/kafka/support/serializer/ErrorHandlingDeserializer2.java

@@ -0,0 +1,200 @@
+/*
+ * Copyright 2018 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.support.serializer;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.ObjectOutputStream;
+import java.util.Map;
+import java.util.function.BiFunction;
+
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.header.internals.RecordHeader;
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.ExtendedDeserializer;
+
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+
+/**
+ * Delegating key/value deserializer that catches exceptions, returning them
+ * in the headers as serialized java objects.
+ *
+ * @param <T> class of the entity, representing messages
+ *
+ * @author Gary Russell
+ * @author Artem Bilan
+ *
+ * @since 2.2
+ *
+ */
+public class ErrorHandlingDeserializer2<T> implements ExtendedDeserializer<T> {
+
+	/**
+	 * Header name for deserialization exceptions.
+	 */
+	public static final String KEY_DESERIALIZER_EXCEPTION_HEADER_PREFIX = "springDeserializerException";
+
+	/**
+	 * Header name for deserialization exceptions.
+	 */
+	public static final String KEY_DESERIALIZER_EXCEPTION_HEADER = KEY_DESERIALIZER_EXCEPTION_HEADER_PREFIX + "Key";
+
+	/**
+	 * Heaader name for deserialization exceptions.
+	 */
+	public static final String VALUE_DESERIALIZER_EXCEPTION_HEADER = KEY_DESERIALIZER_EXCEPTION_HEADER_PREFIX + "Value";
+
+	/**
+	 * Supplier for a T when deserialization fails.
+	 */
+	public static final String KEY_FUNCTION = "spring.deserializer.key.function";
+
+	/**
+	 * Supplier for a T when deserialization fails.
+	 */
+	public static final String VALUE_FUNCTION = "spring.deserializer.value.function";
+
+	/**
+	 * Property name for the delegate key deserializer.
+	 */
+	public static final String KEY_DESERIALIZER_CLASS = "spring.deserializer.key.delegate.class";
+
+	/**
+	 * Property name for the delegate value deserializer.
+	 */
+	public static final String VALUE_DESERIALIZER_CLASS = "spring.deserializer.value.delegate.class";
+
+	private ExtendedDeserializer<T> delegate;
+
+	private boolean isKey;
+
+	private BiFunction<byte[], Headers, T> failedDeserializationFunction;
+
+	public ErrorHandlingDeserializer2() {
+		super();
+	}
+
+	public ErrorHandlingDeserializer2(Deserializer<T> delegate) {
+		this.delegate = setupDelegate(delegate);
+	}
+
+	public void setFailedDeserializationFunction(BiFunction<byte[], Headers, T> failedDeserializationFunction) {
+		this.failedDeserializationFunction = failedDeserializationFunction;
+	}
+
+
+	@Override
+	public void configure(Map<String, ?> configs, boolean isKey) {
+		setupDelegate(configs, isKey ? KEY_DESERIALIZER_CLASS : VALUE_DESERIALIZER_CLASS);
+		Assert.state(this.delegate != null, "No delegate deserializer configured");
+		this.delegate.configure(configs, isKey);
+		this.isKey = isKey;
+		setupFunction(configs, isKey ? KEY_FUNCTION : VALUE_FUNCTION);
+	}
+
+	public void setupDelegate(Map<String, ?> configs, String configKey) {
+		if (configs.containsKey(configKey)) {
+			try {
+				Object value = configs.get(configKey);
+				Class<?> clazz = value instanceof Class ? (Class<?>) value : ClassUtils.forName((String) value, null);
+				this.delegate = setupDelegate(clazz.newInstance());
+			}
+			catch (ClassNotFoundException | LinkageError | InstantiationException | IllegalAccessException e) {
+				throw new IllegalStateException(e);
+			}
+		}
+	}
+
+	@SuppressWarnings("unchecked")
+	private ExtendedDeserializer<T> setupDelegate(Object delegate) {
+		Assert.isInstanceOf(Deserializer.class, delegate, "'delegate' must be a 'Deserializer', not a ");
+		return delegate instanceof ExtendedDeserializer
+				? (ExtendedDeserializer<T>) delegate
+				: ExtendedDeserializer.Wrapper.ensureExtended((Deserializer<T>) delegate);
+	}
+
+	@SuppressWarnings("unchecked")
+	private void setupFunction(Map<String, ?> configs, String configKey) {
+		if (configs.containsKey(configKey)) {
+			try {
+				Object value = configs.get(configKey);
+				Class<?> clazz = value instanceof Class ? (Class<?>) value : ClassUtils.forName((String) value, null);
+				Assert.isTrue(BiFunction.class.isAssignableFrom(clazz), "'function' must be a 'BiFunction	', not a "
+						+ clazz.getName());
+				this.failedDeserializationFunction = (BiFunction<byte[], Headers, T>) clazz.newInstance();
+			}
+			catch (ClassNotFoundException | LinkageError | InstantiationException | IllegalAccessException e) {
+				throw new IllegalStateException(e);
+			}
+		}
+	}
+
+	@Override
+	public T deserialize(String topic, byte[] data) {
+		try {
+			return this.delegate.deserialize(topic, data);
+		}
+		catch (Exception e) {
+			return this.failedDeserializationFunction != null
+					? this.failedDeserializationFunction.apply(data, null)
+					: null;
+		}
+	}
+
+	@Override
+	public T deserialize(String topic, Headers headers, byte[] data) {
+		try {
+			return this.delegate.deserialize(topic, headers, data);
+		}
+		catch (Exception e) {
+			deserializationException(headers, data, e);
+			return this.failedDeserializationFunction != null
+					? this.failedDeserializationFunction.apply(data, headers)
+					: null;
+		}
+	}
+
+	@Override
+	public void close() {
+		this.delegate.close();
+	}
+
+	private void deserializationException(Headers headers, byte[] data, Exception e) {
+		ByteArrayOutputStream stream = new ByteArrayOutputStream();
+		DeserializationException exception = new DeserializationException("failed to deserialize", data, this.isKey, e);
+		try {
+			new ObjectOutputStream(stream).writeObject(exception);
+		}
+		catch (IOException ex) {
+			try {
+				exception = new DeserializationException("failed to deserialize",
+						data, this.isKey, new RuntimeException("Could not deserialize type "
+								+ e.getClass().getName() + " with message " + e.getMessage()
+								+ " failure: " + ex.getMessage()));
+				new ObjectOutputStream(stream).writeObject(exception);
+			}
+			catch (IOException ex2) {
+				throw new IllegalStateException("Could not serialize a DeserializationException", ex2);
+			}
+		}
+		headers.add(
+				new RecordHeader(this.isKey ? KEY_DESERIALIZER_EXCEPTION_HEADER : VALUE_DESERIALIZER_EXCEPTION_HEADER,
+						stream.toByteArray()));
+	}
+
+}

spring-kafka/src/test/java/org/springframework/kafka/annotation/BatchListenerConversion2Tests.java

@@ -0,0 +1,221 @@
+/*
+ * Copyright 2018 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.annotation;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BiFunction;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.config.KafkaListenerContainerFactory;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.kafka.support.converter.BytesJsonMessageConverter;
+import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer2;
+import org.springframework.kafka.support.serializer.JsonDeserializer;
+import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ * @author Gary Russell
+ *
+ * @since 2.1.1
+ *
+ */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+@DirtiesContext
+public class BatchListenerConversion2Tests {
+
+	private static final String DEFAULT_TEST_GROUP_ID = "blc2";
+
+	@ClassRule // one topic to preserve order
+	public static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, true, 1, "blc.2.1");
+
+	@Autowired
+	private Config config;
+
+	@Autowired
+	private KafkaTemplate<Integer, String> template;
+
+	@Test
+	public void testBatchOfPojosWithABadOne() throws Exception {
+		Listener listener = this.config.listener1();
+		String topic = "blc.2.1";
+		this.template.send(topic, "{\"bar\":\"baz\"}");
+		this.template.send(topic, "junk");
+		this.template.send(topic, "{\"bar\":\"baz\"}");
+		assertThat(listener.latch1.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(listener.badFoo).isInstanceOf(BadFoo.class);
+		assertThat(listener.receivedFoos).isEqualTo(2);
+	}
+
+	@Configuration
+	@EnableKafka
+	public static class Config {
+
+		@Bean
+		public KafkaListenerContainerFactory<?> kafkaListenerContainerFactory() {
+			ConcurrentKafkaListenerContainerFactory<Integer, Foo> factory =
+					new ConcurrentKafkaListenerContainerFactory<>();
+			factory.setConsumerFactory(consumerFactory());
+			factory.setBatchListener(true);
+			factory.setReplyTemplate(template());
+			return factory;
+		}
+
+		@Bean
+		public DefaultKafkaConsumerFactory<Integer, Foo> consumerFactory() {
+			return new DefaultKafkaConsumerFactory<>(consumerConfigs());
+		}
+
+		@Bean
+		public Map<String, Object> consumerConfigs() {
+			Map<String, Object> consumerProps =
+					KafkaTestUtils.consumerProps(DEFAULT_TEST_GROUP_ID, "false", embeddedKafka.getEmbeddedKafka());
+			consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer2.class);
+			consumerProps.put(ErrorHandlingDeserializer2.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class);
+			consumerProps.put(ErrorHandlingDeserializer2.VALUE_FUNCTION, FailedFooProvider.class);
+			consumerProps.put(JsonDeserializer.VALUE_DEFAULT_TYPE, Foo.class.getName());
+			consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+			return consumerProps;
+		}
+
+		@Bean
+		public KafkaTemplate<Integer, String> template() {
+			KafkaTemplate<Integer, String> kafkaTemplate = new KafkaTemplate<>(producerFactory());
+			return kafkaTemplate;
+		}
+
+		@Bean
+		public BytesJsonMessageConverter converter() {
+			return new BytesJsonMessageConverter();
+		}
+
+		@Bean
+		public ProducerFactory<Integer, String> producerFactory() {
+			return new DefaultKafkaProducerFactory<>(producerConfigs());
+		}
+
+		@Bean
+		public Map<String, Object> producerConfigs() {
+			Map<String, Object> props = KafkaTestUtils.producerProps(embeddedKafka.getEmbeddedKafka());
+			props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+			return props;
+		}
+
+		@Bean
+		public Listener listener1() {
+			return new Listener();
+		}
+
+	}
+
+	public static class Listener {
+
+		private final CountDownLatch latch1 = new CountDownLatch(3);
+
+		private volatile Foo badFoo;
+
+		private volatile int receivedFoos;
+
+		@KafkaListener(id = "deser", topics = "blc.2.1")
+		public void listen1(List<Foo> foos) {
+			foos.forEach(f -> {
+				if (f.getBar() == null) {
+					this.badFoo = f;
+				}
+				else {
+					this.receivedFoos++;
+				}
+				this.latch1.countDown();
+			});
+		}
+
+	}
+
+	public static class Foo {
+
+		private String bar;
+
+		public Foo() {
+			super();
+		}
+
+		public Foo(String bar) {
+			this.bar = bar;
+		}
+
+		public String getBar() {
+			return this.bar;
+		}
+
+		public void setBar(String bar) {
+			this.bar = bar;
+		}
+
+		@Override
+		public String toString() {
+			return "Foo [bar=" + this.bar + "]";
+		}
+
+	}
+
+	public static class BadFoo extends Foo {
+
+		private final byte[] failedDecode;
+
+		public BadFoo(byte[] failedDecode) {
+			this.failedDecode = failedDecode;
+		}
+
+		public byte[] getFailedDecode() {
+			return this.failedDecode;
+		}
+
+	}
+
+	public static class FailedFooProvider implements BiFunction<byte[], Headers, Foo> {
+
+		@Override
+		public Foo apply(byte[] t, Headers u) {
+			return new BadFoo(t);
+		}
+
+	}
+
+}

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

@@ -18,6 +18,8 @@ package org.springframework.kafka.listener;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import java.io.ByteArrayInputStream;
+import java.io.ObjectInputStream;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -25,7 +27,9 @@ import java.util.concurrent.TimeUnit;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.header.Header;
 import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.header.internals.RecordHeaders;
 import org.apache.kafka.common.serialization.Deserializer;
 import org.apache.kafka.common.serialization.ExtendedDeserializer;
 import org.apache.kafka.common.serialization.StringDeserializer;
@@ -44,7 +48,7 @@ import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.core.ProducerFactory;
 import org.springframework.kafka.support.serializer.DeserializationException;
-import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;
+import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer2;
 import org.springframework.kafka.test.EmbeddedKafkaBroker;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
 import org.springframework.test.annotation.DirtiesContext;
@@ -77,11 +81,11 @@ public class ErrorHandlingDeserializerTests {
 	}
 
 	@Test
-	public void unitTests() {
-		ErrorHandlingDeserializer<String> ehd = new ErrorHandlingDeserializer<>(new StringDeserializer());
+	public void unitTests() throws Exception {
+		ErrorHandlingDeserializer2<String> ehd = new ErrorHandlingDeserializer2<>(new StringDeserializer());
 		assertThat(ehd.deserialize("topic", "foo".getBytes())).isEqualTo("foo");
 		ehd.close();
-		ehd = new ErrorHandlingDeserializer<>(new Deserializer<String>() {
+		ehd = new ErrorHandlingDeserializer2<>(new Deserializer<String>() {
 
 			@Override
 			public void configure(Map<String, ?> configs, boolean isKey) {
@@ -97,8 +101,11 @@ public class ErrorHandlingDeserializerTests {
 			}
 
 		});
-		Object result = ehd.deserialize("topic", "foo".getBytes());
-		assertThat(result).isInstanceOf(DeserializationException.class);
+		Headers headers = new RecordHeaders();
+		Object result = ehd.deserialize("topic", headers, "foo".getBytes());
+		assertThat(result).isNull();
+		Header deser = headers.lastHeader(ErrorHandlingDeserializer2.VALUE_DESERIALIZER_EXCEPTION_HEADER);
+		assertThat(new ObjectInputStream(new ByteArrayInputStream(deser.value())).readObject()).isInstanceOf(DeserializationException.class);
 		ehd.close();
 	}
 
@@ -122,7 +129,6 @@ public class ErrorHandlingDeserializerTests {
 			this.latch.countDown();
 		}
 
-
 		@Bean
 		public EmbeddedKafkaBroker embeddedKafka() {
 			return new EmbeddedKafkaBroker(1, true, 1, TOPIC);
@@ -134,11 +140,11 @@ public class ErrorHandlingDeserializerTests {
 					new ConcurrentKafkaListenerContainerFactory<>();
 			factory.setConsumerFactory(cf());
 			factory.setErrorHandler((t, r) -> {
-				if (r.value() instanceof DeserializationException) {
+				if (r.value() == null && t instanceof DeserializationException) {
 					this.valueErrorCount++;
-					this.headers = ((DeserializationException) r.value()).getHeaders();
+					this.headers = ((DeserializationException) t).getHeaders();
 				}
-				else if (r.key() instanceof DeserializationException) {
+				else if (r.key() == null && t instanceof DeserializationException) {
 					this.keyErrorCount++;
 				}
 				this.latch.countDown();
@@ -150,10 +156,10 @@ public class ErrorHandlingDeserializerTests {
 		public ConsumerFactory<String, String> cf() {
 			Map<String, Object> props = KafkaTestUtils.consumerProps(TOPIC, "false", embeddedKafka());
 			props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
-			props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
-			props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
-			props.put(ErrorHandlingDeserializer.KEY_DESERIALIZER_CLASS, FailSometimesDeserializer.class);
-			props.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, FailSometimesDeserializer.class.getName());
+			props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer2.class);
+			props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer2.class);
+			props.put(ErrorHandlingDeserializer2.KEY_DESERIALIZER_CLASS, FailSometimesDeserializer.class);
+			props.put(ErrorHandlingDeserializer2.VALUE_DESERIALIZER_CLASS, FailSometimesDeserializer.class.getName());
 			return new DefaultKafkaConsumerFactory<>(props);
 		}
 

spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java

@@ -84,7 +84,7 @@ import org.springframework.kafka.listener.adapter.FilteringMessageListenerAdapte
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.kafka.support.TopicPartitionInitialOffset;
 import org.springframework.kafka.support.TopicPartitionInitialOffset.SeekPosition;
-import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;
+import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer2;
 import org.springframework.kafka.support.serializer.JsonDeserializer;
 import org.springframework.kafka.support.serializer.JsonSerializer;
 import org.springframework.kafka.test.EmbeddedKafkaBroker;
@@ -1696,8 +1696,8 @@ public class KafkaMessageListenerContainerTests {
 		this.logger.info("Start JSON4");
 		Map<String, Object> props = KafkaTestUtils.consumerProps("testJson", "false", embeddedKafka);
 
-		ErrorHandlingDeserializer<Foo1> errorHandlingDeserializer =
-				new ErrorHandlingDeserializer<>(new JsonDeserializer<>(Foo1.class, false));
+		ErrorHandlingDeserializer2<Foo1> errorHandlingDeserializer =
+				new ErrorHandlingDeserializer2<>(new JsonDeserializer<>(Foo1.class, false));
 
 		DefaultKafkaConsumerFactory<Integer, Foo1> cf = new DefaultKafkaConsumerFactory<>(props,
 				new IntegerDeserializer(), errorHandlingDeserializer);

src/checkstyle/checkstyle-suppressions.xml

@@ -7,4 +7,5 @@
 	<suppress files="[\\/]test[\\/]" checks="RequireThis" />
 	<suppress files="[\\/]test[\\/]" checks="Javadoc*" />
 	<suppress files="KafkaMatchersTests" checks="RegexpSinglelineJava" />
+	<suppress files="DeserializationException" checks="MutableException" />
 </suppressions>

src/reference/asciidoc/kafka.adoc

@@ -1937,22 +1937,28 @@ Generally, the `BytesJsonMessageConverter` is more efficient because it avoids a
 ===== ErrorHandlingDeserializer
 
 When a deserializer fails to deserialize a message, Spring has no way to handle the problem because it occurs before the `poll()` returns.
-To solve this problem, version 2.2 introduced the `ErrorHandlingDeserializer`.
+To solve this problem, version 2.2 introduced the `ErrorHandlingDeserializer2`.
 This deserializer delegates to a real deserializer (key or value).
-If the delegate fails to deserialize the record content, the `ErrorHandlingDeserializer` returns a `DeserializationException` instead, containing the cause and raw bytes.
-When using a record-level `MessageListener`, if either the key or value contains a `DeserializationException`, the container's `ErrorHandler` is called with the failed `ConsumerRecord`.
-When using a `BatchMessageListener`, the failed record is passed to the application along with the remaining records in the batch, so it is the responsibility of the application listener to check whether the key or value in a particular record is a `DeserializationException`.
+If the delegate fails to deserialize the record content, the `ErrorHandlingDeserializer2` returns a `null` value and a `DeserializationException` in a header, containing the cause and raw bytes.
+When using a record-level `MessageListener`, if either the key or value contains a `DeserializationException` header, the container's `ErrorHandler` is called with the failed `ConsumerRecord`; the record is not passed to the listener.
 
-You can use the `DefaultKafkaConsumerFactory` constructor that takes key and value `Deserializer` objects and wire in appropriate `ErrorHandlingDeserializer` configured with the proper delegates.
+Alternatively, you can configure a `failedDeserializationFunction` which is a `BiConsumer<byte[], Headers, T>`.
+This function is invoked to create an instance of `T` which is passed to the listener, as normal.
+The raw record value and headers are provided to the function.
+The `DeserializationException` can be found (as a serialized Java object) in headers; see the javadocs for the `ErrorHandlingDeserializer2` for more information.
+
+When using a `BatchMessageListener`, you **must** provide a `failedDeserializationFunction`, otherwise, the batch of records will not be type safe.
+
+You can use the `DefaultKafkaConsumerFactory` constructor that takes key and value `Deserializer` objects and wire in appropriate `ErrorHandlingDeserializer2` configured with the proper delegates.
 Alternatively, you can use consumer configuration properties which are used by the `ErrorHandlingDeserializer` to instantiate the delegates.
-The property names are `ErrorHandlingDeserializer.KEY_DESERIALIZER_CLASS` and `ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS`; the property value can be a class or class name.
+The property names are `ErrorHandlingDeserializer2.KEY_DESERIALIZER_CLASS` and `ErrorHandlingDeserializer2.VALUE_DESERIALIZER_CLASS`; the property value can be a class or class name.
 For example:
 
 [source, java]
 ----
 ... // other props
-props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
-props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
+props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer2.class);
+props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer2.class);
 props.put(ErrorHandlingDeserializer.KEY_DESERIALIZER_CLASS, JsonDeserializer.class);
 props.put(JsonDeserializer.KEY_DEFAULT_TYPE, "com.example.MyKey")
 props.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class.getName());
@@ -1961,6 +1967,45 @@ props.put(JsonDeserializer.TRUSTED_PACKAGES, "com.example")
 return new DefaultKafkaConsumerFactory<>(props);
 ----
 
+The following is an example of using a `failedDeserializationFunction`.
+
+[source, java]
+----
+public class BadFoo extends Foo {
+
+  private final byte[] failedDecode;
+
+  public BadFoo(byte[] failedDecode) {
+    this.failedDecode = failedDecode;
+  }
+
+  public byte[] getFailedDecode() {
+    return this.failedDecode;
+  }
+
+}
+
+public class FailedFooProvider implements BiFunction<byte[], Headers, Foo> {
+
+  @Override
+  public Foo apply(byte[] t, Headers u) {
+    return new BadFoo(t);
+  }
+
+}
+----
+
+and config
+
+[source, java]
+----
+...
+consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer2.class);
+consumerProps.put(ErrorHandlingDeserializer2.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class);
+consumerProps.put(ErrorHandlingDeserializer2.VALUE_FUNCTION, FailedFooProvider.class);
+...
+----
+
 [[payload-conversion-with-batch]]
 ===== Payload Conversion with Batch Listeners