GH-1494: Fix NPE in DeadLetterPublishingRecoverer

Resolves #1494

Author: garyrussell

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2020 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.
@@ -98,7 +98,9 @@ public DeadLetterPublishingRecoverer(KafkaTemplate<? extends Object, ? extends O
 	 * original topic. The templates map keys are classes and the value the corresponding
 	 * template to use for objects (producer record values) of that type. A
 	 * {@link java.util.LinkedHashMap} is recommended when there is more than one
-	 * template, to ensure the map is traversed in order.
+	 * template, to ensure the map is traversed in order. To send records with a null
+	 * value, add a template with the {@link Void} class as a key; otherwise the first
+	 * template from the map values iterator will be used.
 	 * @param templates the {@link KafkaTemplate}s to use for publishing.
 	 */
 	public DeadLetterPublishingRecoverer(Map<Class<?>, KafkaTemplate<? extends Object, ? extends Object>> templates) {
@@ -112,7 +114,9 @@ public DeadLetterPublishingRecoverer(Map<Class<?>, KafkaTemplate<? extends Objec
 	 * 0, no partition is set when publishing to the topic. The templates map keys are
 	 * classes and the value the corresponding template to use for objects (producer
 	 * record values) of that type. A {@link java.util.LinkedHashMap} is recommended when
-	 * there is more than one template, to ensure the map is traversed in order.
+	 * there is more than one template, to ensure the map is traversed in order. To send
+	 * records with a null value, add a template with the {@link Void} class as a key;
+	 * otherwise the first template from the map values iterator will be used.
 	 * @param templates the {@link KafkaTemplate}s to use for publishing.
 	 * @param destinationResolver the resolving function.
 	 */
@@ -161,10 +165,19 @@ public void accept(ConsumerRecord<?, ?> record, Exception exception) {
 	}
 
 	@SuppressWarnings("unchecked")
-	private KafkaTemplate<Object, Object> findTemplateForValue(Object value) {
+	private KafkaTemplate<Object, Object> findTemplateForValue(@Nullable Object value) {
 		if (this.template != null) {
 			return this.template;
 		}
+		if (value == null) {
+			KafkaOperations<?, ?> operations = this.templates.get(Void.class);
+			if (operations == null) {
+				return (KafkaTemplate<Object, Object>) this.templates.values().iterator().next();
+			}
+			else {
+				return (KafkaTemplate<Object, Object>) operations;
+			}
+		}
 		Optional<Class<?>> key = this.templates.keySet()
 			.stream()
 			.filter((k) -> k.isAssignableFrom(value.getClass()))
@@ -235,8 +248,11 @@ private void enhanceHeaders(RecordHeaders kafkaHeaders, ConsumerRecord<?, ?> rec
 				record.timestampType().toString().getBytes(StandardCharsets.UTF_8)));
 		kafkaHeaders.add(new RecordHeader(KafkaHeaders.DLT_EXCEPTION_FQCN,
 				exception.getClass().getName().getBytes(StandardCharsets.UTF_8)));
-		kafkaHeaders.add(new RecordHeader(KafkaHeaders.DLT_EXCEPTION_MESSAGE,
-				exception.getMessage().getBytes(StandardCharsets.UTF_8)));
+		String message = exception.getMessage();
+		if (message != null) {
+			kafkaHeaders.add(new RecordHeader(KafkaHeaders.DLT_EXCEPTION_MESSAGE,
+					exception.getMessage().getBytes(StandardCharsets.UTF_8)));
+		}
 		kafkaHeaders.add(new RecordHeader(KafkaHeaders.DLT_EXCEPTION_STACKTRACE,
 				getStackTraceAsString(exception).getBytes(StandardCharsets.UTF_8)));
 	}

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

@@ -0,0 +1,71 @@
+/*
+ * Copyright 2020 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.listener;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.util.concurrent.SettableListenableFuture;
+
+/**
+ * @author Gary Russell
+ * @since 2.3.9
+ *
+ */
+public class DeadLetterPublishingRecovererTests {
+
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	@Test
+	void tombstoneWithMultiTemplates() {
+		KafkaTemplate<?, ?> template1 = mock(KafkaTemplate.class);
+		given(template1.send(any(ProducerRecord.class))).willReturn(new SettableListenableFuture());
+		KafkaTemplate<?, ?> template2 = mock(KafkaTemplate.class);
+		Map<Class<?>, KafkaTemplate<?, ?>> templates = new LinkedHashMap<>();
+		templates.put(String.class, template1);
+		templates.put(Integer.class, template2);
+		DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(templates);
+		ConsumerRecord<String, String> record = new ConsumerRecord<>("foo", 0, 0L, "bar", null);
+		recoverer.accept(record, new RuntimeException());
+		verify(template1).send(any(ProducerRecord.class));
+	}
+
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	@Test
+	void tombstoneWithMultiTemplatesExplicit() {
+		KafkaTemplate<?, ?> template1 = mock(KafkaTemplate.class);
+		KafkaTemplate<?, ?> template2 = mock(KafkaTemplate.class);
+		given(template2.send(any(ProducerRecord.class))).willReturn(new SettableListenableFuture());
+		Map<Class<?>, KafkaTemplate<?, ?>> templates = new LinkedHashMap<>();
+		templates.put(String.class, template1);
+		templates.put(Void.class, template2);
+		DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(templates);
+		ConsumerRecord<String, String> record = new ConsumerRecord<>("foo", 0, 0L, "bar", null);
+		recoverer.accept(record, new RuntimeException());
+		verify(template2).send(any(ProducerRecord.class));
+	}
+
+}

src/reference/asciidoc/kafka.adoc

@@ -3830,6 +3830,8 @@ public DeadLetterPublishingRecoverer publisher(KafkaTemplate<?, ?> stringTemplat
 The publisher uses the map keys to locate a template that is suitable for the `value()` about to be published.
 A `LinkedHashMap` is recommended so that the keys are examined in order.
 
+When publishing `null` values, when there are multiple templates, the recoverer will look for a template for the `Void` class; if none is present, the first template from the `values().iterator()` will be used.
+
 IMPORTANT: If the recoverer fails (throws an exception), the record will be included in the seeks and recovery will be attempted again during the next delivery.
 
 Starting with version 2.3, the recoverer can also be used with Kafka Streams - see <<streams-deser-recovery>> for more information.