GH-1605: DeadLetterPublishingRecoverer Improvement

Resolves #1605

When both the key and value fail deserialization, populate both
original values in the record published to the DLT.

* Fix javadocs for new headers.

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.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.
@@ -168,24 +168,21 @@ public void setHeadersFunction(BiFunction<ConsumerRecord<?, ?>, Exception, Heade
 	@Override
 	public void accept(ConsumerRecord<?, ?> record, Exception exception) {
 		TopicPartition tp = this.destinationResolver.apply(record, exception);
-		boolean isKey = false;
-		DeserializationException deserEx = ListenerUtils.getExceptionFromHeader(record,
+		DeserializationException vDeserEx = ListenerUtils.getExceptionFromHeader(record,
 				ErrorHandlingDeserializer.VALUE_DESERIALIZER_EXCEPTION_HEADER, this.logger);
-		if (deserEx == null) {
-			deserEx = ListenerUtils.getExceptionFromHeader(record,
+		DeserializationException kDeserEx = ListenerUtils.getExceptionFromHeader(record,
 					ErrorHandlingDeserializer.KEY_DESERIALIZER_EXCEPTION_HEADER, this.logger);
-			isKey = true;
+		Headers headers = new RecordHeaders(record.headers().toArray());
+		if (kDeserEx != null && !this.retainExceptionHeader) {
+			headers.remove(ErrorHandlingDeserializer.KEY_DESERIALIZER_EXCEPTION_HEADER);
+			addExceptionInfoHeaders(headers, kDeserEx, true);
 		}
-		Headers headers;
-		if (deserEx == null || this.retainExceptionHeader) {
-			headers = new RecordHeaders(record.headers().toArray());
-		}
-		else {
-			headers = deserEx.getHeaders();
+		if (vDeserEx != null && !this.retainExceptionHeader) {
+			headers.remove(ErrorHandlingDeserializer.VALUE_DESERIALIZER_EXCEPTION_HEADER);
 		}
 		enhanceHeaders(headers, record, exception); // NOSONAR headers are never null
 		ProducerRecord<Object, Object> outRecord = createProducerRecord(record, tp, headers,
-				deserEx == null ? null : deserEx.getData(), isKey);
+				kDeserEx == null ? null : kDeserEx.getData(), vDeserEx == null ? null : vDeserEx.getData());
 		KafkaOperations<Object, Object> kafkaTemplate = findTemplateForValue(outRecord.value());
 		if (this.transactional && !kafkaTemplate.inTransaction() && !kafkaTemplate.isAllowNonTransactional()) {
 			kafkaTemplate.executeInTransaction(t -> {
@@ -236,18 +233,18 @@ private KafkaOperations<Object, Object> findTemplateForValue(@Nullable Object va
 	 * @param topicPartition the {@link TopicPartition} returned by the destination
 	 * resolver.
 	 * @param headers the headers - original record headers plus DLT headers.
-	 * @param data the value to use instead of the consumer record value.
-	 * @param isKey true if key deserialization failed.
+	 * @param key the key to use instead of the consumer record key.
+	 * @param value the value to use instead of the consumer record value.
 	 * @return the producer record to send.
 	 * @see KafkaHeaders
 	 */
 	protected ProducerRecord<Object, Object> createProducerRecord(ConsumerRecord<?, ?> record,
-			TopicPartition topicPartition, Headers headers, @Nullable byte[] data, boolean isKey) {
+			TopicPartition topicPartition, Headers headers, @Nullable byte[] key, @Nullable byte[] value) {
 
 		return new ProducerRecord<>(topicPartition.topic(),
 				topicPartition.partition() < 0 ? null : topicPartition.partition(),
-				isKey && data != null ? data : record.key(),
-				data == null || isKey ? record.value() : data, headers);
+				key != null ? key : record.key(),
+				value != null ? value : record.value(), headers);
 	}
 
 	/**
@@ -280,19 +277,27 @@ private void enhanceHeaders(Headers kafkaHeaders, ConsumerRecord<?, ?> record, E
 				ByteBuffer.allocate(Long.BYTES).putLong(record.timestamp()).array()));
 		kafkaHeaders.add(new RecordHeader(KafkaHeaders.DLT_ORIGINAL_TIMESTAMP_TYPE,
 				record.timestampType().toString().getBytes(StandardCharsets.UTF_8)));
-		kafkaHeaders.add(new RecordHeader(KafkaHeaders.DLT_EXCEPTION_FQCN,
+		addExceptionInfoHeaders(kafkaHeaders, exception, false);
+		Headers headers = this.headersFunction.apply(record, exception);
+		if (headers != null) {
+			headers.forEach(header -> kafkaHeaders.add(header));
+		}
+	}
+
+	private void addExceptionInfoHeaders(Headers kafkaHeaders, Exception exception, boolean isKey) {
+		kafkaHeaders.add(new RecordHeader(isKey ? KafkaHeaders.DLT_KEY_EXCEPTION_FQCN : KafkaHeaders.DLT_EXCEPTION_FQCN,
 				exception.getClass().getName().getBytes(StandardCharsets.UTF_8)));
 		String message = exception.getMessage();
 		if (message != null) {
-			kafkaHeaders.add(new RecordHeader(KafkaHeaders.DLT_EXCEPTION_MESSAGE,
+			kafkaHeaders.add(new RecordHeader(isKey
+						? KafkaHeaders.DLT_KEY_EXCEPTION_MESSAGE
+						: KafkaHeaders.DLT_EXCEPTION_MESSAGE,
 					exception.getMessage().getBytes(StandardCharsets.UTF_8)));
 		}
-		kafkaHeaders.add(new RecordHeader(KafkaHeaders.DLT_EXCEPTION_STACKTRACE,
+		kafkaHeaders.add(new RecordHeader(isKey
+						? KafkaHeaders.DLT_KEY_EXCEPTION_STACKTRACE
+						: KafkaHeaders.DLT_EXCEPTION_STACKTRACE,
 				getStackTraceAsString(exception).getBytes(StandardCharsets.UTF_8)));
-		Headers headers = this.headersFunction.apply(record, exception);
-		if (headers != null) {
-			headers.forEach(header -> kafkaHeaders.add(header));
-		}
 	}
 
 	private String getStackTraceAsString(Throwable cause) {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2020 the original author or authors.
+ * Copyright 2017-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.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2020 the original author or authors.
+ * Copyright 2014-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.
@@ -159,6 +159,27 @@ public abstract class KafkaHeaders {
 	 */
 	public static final String DLT_EXCEPTION_MESSAGE = PREFIX + "dlt-exception-message";
 
+	/**
+	 * Exception stack trace for a record published to a dead-letter topic with a key
+	 * deserialization exception.
+	 * @since 2.7
+	 */
+	public static final String DLT_KEY_EXCEPTION_STACKTRACE = PREFIX + "dlt-key-exception-stacktrace";
+
+	/**
+	 * Exception message for a record published to a dead-letter topic with a key
+	 * deserialization exception.
+	 * @since 2.7
+	 */
+	public static final String DLT_KEY_EXCEPTION_MESSAGE = PREFIX + "dlt-key-exception-message";
+
+	/**
+	 * Exception class name for a record published sent to a dead-letter topic with a key
+	 * deserialization exception.
+	 * @since 2.7
+	 */
+	public static final String DLT_KEY_EXCEPTION_FQCN = PREFIX + "dlt-key-exception-fqcn";
+
 	/**
 	 * Original topic for a record published to a dead-letter topic.
 	 * @since 2.2

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020 the original author or authors.
+ * Copyright 2020-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.
@@ -43,6 +43,7 @@
 
 import org.springframework.kafka.core.KafkaOperations;
 import org.springframework.kafka.core.KafkaOperations.OperationsCallback;
+import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.kafka.support.serializer.DeserializationException;
 import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;
 import org.springframework.util.concurrent.SettableListenableFuture;
@@ -130,13 +131,18 @@ void valueHeaderStripped() {
 		willReturn(new SettableListenableFuture<Object>()).given(template).send(any(ProducerRecord.class));
 		ConsumerRecord<String, String> record = new ConsumerRecord<>("foo", 0, 0L, 0L, TimestampType.CREATE_TIME,
 				0L, 0, 0, "bar", "baz", headers);
-		recoverer.accept(record, new RuntimeException());
+		recoverer.accept(record, new RuntimeException("testV"));
 		ArgumentCaptor<ProducerRecord> captor = ArgumentCaptor.forClass(ProducerRecord.class);
 		verify(template).send(captor.capture());
-		headers = captor.getValue().headers();
+		ProducerRecord recovered = captor.getValue();
+		assertThat(recovered.key()).isEqualTo("key".getBytes());
+		assertThat(recovered.value()).isEqualTo("value".getBytes());
+		headers = recovered.headers();
 		assertThat(headers.lastHeader(ErrorHandlingDeserializer.VALUE_DESERIALIZER_EXCEPTION_HEADER)).isNull();
-		assertThat(headers.lastHeader(ErrorHandlingDeserializer.KEY_DESERIALIZER_EXCEPTION_HEADER)).isNotNull();
+		assertThat(headers.lastHeader(ErrorHandlingDeserializer.KEY_DESERIALIZER_EXCEPTION_HEADER)).isNull();
 		assertThat(headers.lastHeader("foo")).isNotNull();
+		assertThat(headers.lastHeader(KafkaHeaders.DLT_KEY_EXCEPTION_MESSAGE).value()).isEqualTo("testK".getBytes());
+		assertThat(headers.lastHeader(KafkaHeaders.DLT_EXCEPTION_MESSAGE).value()).isEqualTo("testV".getBytes());
 	}
 
 	@SuppressWarnings({ "rawtypes", "unchecked" })
@@ -209,7 +215,9 @@ void tombstoneWithMultiTemplatesExplicit() {
 	private byte[] header(boolean isKey) {
 		ByteArrayOutputStream baos = new ByteArrayOutputStream();
 		try {
-			new ObjectOutputStream(baos).writeObject(new DeserializationException("test", new byte[0], isKey, null));
+			new ObjectOutputStream(baos).writeObject(new DeserializationException(
+					isKey ? "testK" : "testV",
+					isKey ? "key".getBytes() : "value".getBytes(), isKey, null));
 		}
 		catch (IOException e) {
 			throw new UncheckedIOException(e);

src/reference/asciidoc/kafka.adoc

@@ -5169,6 +5169,9 @@ The record sent to the dead-letter topic is enhanced with the following headers:
 * `KafkaHeaders.DLT_EXCEPTION_FQCN`: The Exception class name.
 * `KafkaHeaders.DLT_EXCEPTION_STACKTRACE`: The Exception stack trace.
 * `KafkaHeaders.DLT_EXCEPTION_MESSAGE`: The  Exception message.
+* `KafkaHeaders.DLT_KEY_EXCEPTION_FQCN`: The Exception class name (key deserialization errors only).
+* `KafkaHeaders.DLT_KEY_EXCEPTION_STACKTRACE`: The Exception stack trace (key deserialization errors only).
+* `KafkaHeaders.DLT_KEY_EXCEPTION_MESSAGE`: The  Exception message (key deserialization errors only).
 * `KafkaHeaders.DLT_ORIGINAL_TOPIC`: The original topic.
 * `KafkaHeaders.DLT_ORIGINAL_PARTITION`: The original partition.
 * `KafkaHeaders.DLT_ORIGINAL_OFFSET`: The original offset.
@@ -5219,8 +5222,8 @@ By default, the exception type is not considered.
 Starting with version 2.3, the recoverer can also be used with Kafka Streams - see <<streams-deser-recovery>> for more information.
 
 The `ErrorHandlingDeserializer` adds the deserialization exception(s) in headers `ErrorHandlingDeserializer.VALUE_DESERIALIZER_EXCEPTION_HEADER` and `ErrorHandlingDeserializer.KEY_DESERIALIZER_EXCEPTION_HEADER` (using java serialization).
-By default, these headers are not retained in the message published to the dead letter topic, unless both the key and value fail deserialization.
-In that case, the `DLT_*` headers are based on the value deserialization and the key `DeserializationException` is retained in the header.
+By default, these headers are not retained in the message published to the dead letter topic.
+Starting with version 2.7, if both the key and value fail deserialization, the original values of both are populated in the record sent to the DLT.
 
 If incoming records are dependent on each other, but may arrive out of order, it may be useful to republish a failed record to the tail of the original topic (for some number of times), instead of sending it directly to the dead letter topic.
 See https://stackoverflow.com/questions/64646996[this Stack Overflow Question] for an example.

src/reference/asciidoc/whats-new.adoc

@@ -22,3 +22,11 @@ See <<container-props>> for more information.
 
 You can now validate the payload parameter of `@KafkaHandler` methods (class-level listeners).
 See <<kafka-validation>> for more information.
+
+[[27-dlt]]
+==== `DeadLetterPublishingRecover` Changes
+
+Now, if both the key and value fail deserialization, the original values are published to the DLT.
+Previously, the value was populated but the key `DeserializationException` remained in the headers.
+There is a breaking API change, if you subclassed the recoverer and overrode the `createProducerRecord` method.
+See <<dead-letters>> for more information.