GH-1936: Batch Listener Handle ConversionException

Resolves #1936

Make it easier to detect and recover from conversion exceptions in batch
listeners.

Also add `DelegatingByTypeSerializer`.

* Polishing - improve exception message, add test for DBTSerializer.

* Fix test - supported types order is random.

Author: garyrussell

spring-kafka-docs/src/main/asciidoc/kafka.adoc

@@ -4179,6 +4179,22 @@ It is not necessary to configure serializers or deserializers for these types, t
 
 For another technique to send different types to different topics, see <<routing-template>>.
 
+2.8 introduced the `DelegatingByTypeSerializer`.
+
+====
+[source, java]
+----
+@Bean
+public ProducerFactory<Integer, Object> producerFactory(Map<String, Object> config) {
+	return new DefaultKafkaProducerFactory<>(config,
+			null, new DelegatingByTypeSerializer(Map.of(
+					byte[].class, new ByteArraySerializer(),
+					Bytes.class, new BytesSerializer(),
+					String.class, new StringSerializer())));
+}
+----
+====
+
 [[retrying-deserialization]]
 ===== Retrying Deserializer
 
@@ -4402,6 +4418,28 @@ consumerProps.put(ErrorHandlingDeserializer.VALUE_FUNCTION, FailedFooProvider.cl
 ----
 ====
 
+When using an `ErrorHandlingDeserializer` with a batch listener, you must check for the deserialization exceptions in message headers.
+When used with a `RecoveringBatchErrorHandler`, you can use that header to determine which record the exception failed on and communicate to the error handler via a `BatchListenerFailedException`.
+
+====
+[source, java]
+----
+@KafkaListener(id = "test", topics = "test")
+void listen(List<Thing> in, @Header(KafkaHeaders.BATCH_CONVERTED_HEADERS) List<Map<String, Object>> headers) {
+    for (int i = 0; i < in.size(); i++) {
+        Thing thing = in.get(i);
+        if (thing == null
+                && headers.get(i).get(ErrorHandlingDeserializer.VALUE_DESERIALIZER_EXCEPTION_HEADER) != null) {
+
+            throw new BatchListenerFailedException("deserialization error",
+                    new DeserializationException("Batch listener", null, false, null), i);
+        }
+        process(foo);
+    }
+}
+----
+====
+
 [[payload-conversion-with-batch]]
 ===== Payload Conversion with Batch Listeners
 
@@ -4909,6 +4947,7 @@ The container commits any pending offset commits before calling the error handle
 
 If you are using Spring Boot, you simply need to add the error handler as a `@Bean` and Boot will add it to the auto-configured factory.
 
+[[default-eh]]
 ===== DefaultErrorHandler
 
 This new error handler replaces the `SeekToCurrentErrorHandler` and `RecoveringBatchErrorHandler`, which have been the default error handlers for several releases now.
@@ -5068,6 +5107,31 @@ By default, the exception type is not considered.
 
 Also see <<delivery-header>>.
 
+[[batch-listener-conv-errors]]
+==== Conversion Errors with Batch Error Handlers
+
+Starting with version 2.8, batch listeners can now properly handle conversion errors, when using a `MessageConverter` with a `ByteArrayDeserializer`, a `BytesDeserializer` or a `StringDeserializer`, as well as a `DefaultErrorHandler`.
+When a conversion error occurs, the payload is set to null and a deserialization exception is added to the record headers, similar to the `ErrorHandlingDeserializer`.
+A list of `ConversionException` s is available in the listener so the listener can throw a `BatchListenerFailedException` indicating the first index at which a conversion exception occurred.
+
+Example:
+
+====
+[source, java]
+----
+@KafkaListener(id = "test", topics = "topic")
+void listen(List<Thing> in, @Header(KafkaHeaders.CONVERSION_FAILURES) List<ConversionException> exceptions) {
+    for (int i = 0; i < in.size(); i++) {
+        Foo foo = in.get(i);
+        if (foo == null && exceptions.get(i) != null) {
+            throw new BatchListenerFailedException("Conversion error", exceptions.get(i), i);
+        }
+        process(foo);
+    }
+}
+---
+====
+
 [[retrying-batch-eh]]
 ===== Retrying Batch Error Handler
 

spring-kafka-docs/src/main/asciidoc/whats-new.adoc

@@ -20,8 +20,13 @@ See <<ooo-commits>> for more information.
 
 It is now possible to specify whether the listener method is a batch listener on the method itself.
 This allows the same container factory to be used for both record and batch listeners.
+
 See <<batch-listeners>> for more information.
 
+Batch listeners can now handle conversion exceptions.
+
+See <<batch-listener-conv-errors>> for more information.
+
 [[x28-template]]
 ==== `KafkaTemplate` Changes
 

spring-kafka/src/main/java/org/springframework/kafka/support/KafkaHeaders.java

@@ -314,4 +314,12 @@ public abstract class KafkaHeaders {
 	 * @since 2.2
 	 */
 	public static final String ORIGINAL_TIMESTAMP_TYPE = PREFIX + "original-timestamp-type";
+
+	/**
+	 * The header containing a list of conversion failures (for batch listeners only).
+	 * Type: List<ConversionException>.
+	 * @since 2.8
+	 */
+	public static final String CONVERSION_FAILURES = PREFIX + "conversionFailures";
+
 }

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

@@ -18,6 +18,7 @@
 
 import java.lang.reflect.ParameterizedType;
 import java.lang.reflect.Type;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -28,6 +29,7 @@
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.utils.Bytes;
 
 import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.support.Acknowledgment;
@@ -36,6 +38,7 @@
 import org.springframework.kafka.support.KafkaHeaderMapper;
 import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.kafka.support.KafkaNull;
+import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.support.MessageBuilder;
@@ -153,6 +156,7 @@ public Message<?> toMessage(List<ConsumerRecord<?, ?>> records, @Nullable Acknow
 		List<Map<String, Object>> convertedHeaders = new ArrayList<>();
 		List<Headers> natives = new ArrayList<>();
 		List<ConsumerRecord<?, ?>> raws = new ArrayList<>();
+		List<ConversionException> conversionFailures = new ArrayList<>();
 		if (this.headerMapper != null) {
 			rawHeaders.put(KafkaHeaders.BATCH_CONVERTED_HEADERS, convertedHeaders);
 		}
@@ -164,12 +168,11 @@ public Message<?> toMessage(List<ConsumerRecord<?, ?>> records, @Nullable Acknow
 		}
 		commonHeaders(acknowledgment, consumer, rawHeaders, keys, topics, partitions, offsets, timestampTypes,
 				timestamps);
+		rawHeaders.put(KafkaHeaders.CONVERSION_FAILURES, conversionFailures);
 
 		boolean logged = false;
 		for (ConsumerRecord<?, ?> record : records) {
-			payloads.add(this.recordConverter == null || !containerType(type)
-					? extractAndConvertValue(record, type)
-					: convert(record, type));
+			payloads.add(obtainPayload(type, record, conversionFailures));
 			keys.add(record.key());
 			topics.add(record.topic());
 			partitions.add(record.partition());
@@ -200,6 +203,12 @@ public Message<?> toMessage(List<ConsumerRecord<?, ?>> records, @Nullable Acknow
 		return MessageBuilder.createMessage(payloads, kafkaMessageHeaders);
 	}
 
+	private Object obtainPayload(Type type, ConsumerRecord<?, ?> record, List<ConversionException> conversionFailures) {
+		return this.recordConverter == null || !containerType(type)
+				? extractAndConvertValue(record, type)
+				: convert(record, type, conversionFailures);
+	}
+
 	@Override
 	public List<ProducerRecord<?, ?>> fromMessage(Message<?> message, String defaultTopic) {
 		throw new UnsupportedOperationException();
@@ -222,11 +231,35 @@ protected Object extractAndConvertValue(ConsumerRecord<?, ?> record, Type type)
 	 * @param record the record.
 	 * @param type the type - must be a {@link ParameterizedType} with a single generic
 	 * type parameter.
+	 * @param conversionFailures Conversion failures.
 	 * @return the converted payload.
 	 */
-	protected Object convert(ConsumerRecord<?, ?> record, Type type) {
-		return this.recordConverter
-			.toMessage(record, null, null, ((ParameterizedType) type).getActualTypeArguments()[0]).getPayload();
+	protected Object convert(ConsumerRecord<?, ?> record, Type type, List<ConversionException> conversionFailures) {
+		try {
+			Object payload = this.recordConverter
+				.toMessage(record, null, null, ((ParameterizedType) type).getActualTypeArguments()[0]).getPayload();
+			conversionFailures.add(null);
+			return payload;
+		}
+		catch (ConversionException ex) {
+			byte[] original = null;
+			if (record.value() instanceof byte[]) {
+				original = (byte[]) record.value();
+			}
+			else if (record.value() instanceof Bytes) {
+				original = ((Bytes) record.value()).get();
+			}
+			else if (record.value() instanceof String) {
+				original = ((String) record.value()).getBytes(StandardCharsets.UTF_8);
+			}
+			if (original != null) {
+				ErrorHandlingDeserializer.deserializationException(record.headers(), original, ex, false);
+				conversionFailures.add(ex);
+				return null;
+			}
+			throw new ConversionException("The batch converter can only report conversion failures to the listener "
+					+ "if the record.value() is byte[], Bytes, or String", ex);
+		}
 	}
 
 	/**

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

@@ -35,11 +35,11 @@
 @SuppressWarnings("serial")
 public class ConversionException extends KafkaException {
 
-	private final ConsumerRecord<?, ?> record;
+	private transient ConsumerRecord<?, ?> record;
 
-	private final List<ConsumerRecord<?, ?>> records = new ArrayList<>();
+	private transient List<ConsumerRecord<?, ?>> records = new ArrayList<>();
 
-	private final Message<?> message;
+	private transient Message<?> message;
 
 	/**
 	 * Construct an instance with the provided properties.

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

@@ -0,0 +1,85 @@
+/*
+ * Copyright 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://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.util.HashMap;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import org.apache.kafka.common.errors.SerializationException;
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.serialization.Serializer;
+
+import org.springframework.util.Assert;
+
+/**
+ * Delegates to a serializer based on type.
+ *
+ * @author Gary Russell
+ * @since 2.8
+ *
+ */
+public class DelegatingByTypeSerializer implements Serializer<Object> {
+
+	@SuppressWarnings("rawtypes")
+	private final Map<Class<?>, Serializer> delegates = new HashMap<>();
+
+	/**
+	 * Construct an instance with the map of delegates.
+	 * @param delegates the delegates.
+	 */
+	@SuppressWarnings("rawtypes")
+	public DelegatingByTypeSerializer(Map<Class<?>, Serializer> delegates) {
+		Assert.notNull(delegates, "'delegates' cannot be null");
+		Assert.noNullElements(delegates.values(), "Serializers in delegates map cannot be null");
+		this.delegates.putAll(delegates);
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public void configure(Map<String, ?> configs, boolean isKey) {
+		this.delegates.values().forEach(del -> del.configure(configs, isKey));
+	}
+
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	@Override
+	public byte[] serialize(String topic, Object data) {
+		Serializer delegate = findDelegate(data);
+		return delegate.serialize(topic, data);
+	}
+
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	@Override
+	public byte[] serialize(String topic, Headers headers, Object data) {
+		Serializer delegate = findDelegate(data);
+		return delegate.serialize(topic, headers, data);
+	}
+
+	@SuppressWarnings("rawtypes")
+	private Serializer findDelegate(Object data) {
+		Serializer delegate = this.delegates.get(data.getClass());
+		if (delegate == null) {
+			throw new SerializationException("No matching delegate for type: " + data.getClass().getName()
+					+ "; supported types: " + this.delegates.keySet().stream()
+							.map(clazz -> clazz.getName())
+							.collect(Collectors.toList()));
+		}
+		return delegate;
+	}
+
+
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2020 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.
@@ -188,7 +188,7 @@ public T deserialize(String topic, Headers headers, byte[] data) {
 			return this.delegate.deserialize(topic, headers, data);
 		}
 		catch (Exception e) {
-			deserializationException(headers, data, e);
+			deserializationException(headers, data, e, this.isForKey);
 			return recoverFromSupplier(topic, headers, data, e);
 		}
 	}
@@ -211,28 +211,36 @@ public void close() {
 		}
 	}
 
-	private void deserializationException(Headers headers, byte[] data, Exception e) {
+	/**
+	 * Populate the record headers with a serialized {@link DeserializationException}.
+	 * @param headers the headers.
+	 * @param data the data.
+	 * @param ex the exception.
+	 * @param isForKeyArg true if this is a key deserialization problem, otherwise value.
+	 * @since 2.8
+	 */
+	public static void deserializationException(Headers headers, byte[] data, Exception ex, boolean isForKeyArg) {
 		ByteArrayOutputStream stream = new ByteArrayOutputStream();
 		DeserializationException exception =
-				new DeserializationException("failed to deserialize", data, this.isForKey, e);
+				new DeserializationException("failed to deserialize", data, isForKeyArg, ex);
 		try (ObjectOutputStream oos = new ObjectOutputStream(stream)) {
 			oos.writeObject(exception);
 		}
-		catch (IOException ex) {
+		catch (IOException ioex) {
 			stream = new ByteArrayOutputStream();
 			try (ObjectOutputStream oos = new ObjectOutputStream(stream)) {
 				exception = new DeserializationException("failed to deserialize",
-						data, this.isForKey, new RuntimeException("Could not deserialize type "
-						+ e.getClass().getName() + " with message " + e.getMessage()
-						+ " failure: " + ex.getMessage()));
+						data, isForKeyArg, new RuntimeException("Could not deserialize type "
+						+ ioex.getClass().getName() + " with message " + ioex.getMessage()
+						+ " failure: " + ioex.getMessage()));
 				oos.writeObject(exception);
 			}
 			catch (IOException ex2) {
 				throw new IllegalStateException("Could not serialize a DeserializationException", ex2); // NOSONAR
 			}
 		}
 		headers.add(
-				new RecordHeader(this.isForKey
+				new RecordHeader(isForKeyArg
 						? KEY_DESERIALIZER_EXCEPTION_HEADER
 						: VALUE_DESERIALIZER_EXCEPTION_HEADER,
 						stream.toByteArray()));