GH-1894: DLPR: Fix Headers with Only Key Exception

Resolves #1894

When there was only a key deserialization exception, the exception info was added
twice, under the key keys and value keys.

**cherry-pick to 2.7.x**

Author: garyrussell

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

@@ -315,21 +315,35 @@ public void accept(ConsumerRecord<?, ?> record, @Nullable Consumer<?, ?> consume
 		DeserializationException kDeserEx = ListenerUtils.getExceptionFromHeader(record,
 					ErrorHandlingDeserializer.KEY_DESERIALIZER_EXCEPTION_HEADER, this.logger);
 		Headers headers = new RecordHeaders(record.headers().toArray());
-		if (kDeserEx != null && !this.retainExceptionHeader) {
-			headers.remove(ErrorHandlingDeserializer.KEY_DESERIALIZER_EXCEPTION_HEADER);
-			addExceptionInfoHeaders(headers, kDeserEx, true);
-		}
-		if (vDeserEx != null && !this.retainExceptionHeader) {
-			headers.remove(ErrorHandlingDeserializer.VALUE_DESERIALIZER_EXCEPTION_HEADER);
-		}
-		enhanceHeaders(headers, record, exception); // NOSONAR headers are never null
+		addAndEnhanceHeaders(record, exception, vDeserEx, kDeserEx, headers);
 		ProducerRecord<Object, Object> outRecord = createProducerRecord(record, tp, headers,
 				kDeserEx == null ? null : kDeserEx.getData(), vDeserEx == null ? null : vDeserEx.getData());
 		KafkaOperations<Object, Object> kafkaTemplate =
 				(KafkaOperations<Object, Object>) this.templateResolver.apply(outRecord);
 		sendOrThrow(outRecord, kafkaTemplate);
 	}
 
+	private void addAndEnhanceHeaders(ConsumerRecord<?, ?> record, Exception exception,
+			@Nullable DeserializationException vDeserEx, @Nullable DeserializationException kDeserEx, Headers headers) {
+
+		if (kDeserEx != null) {
+			if (!this.retainExceptionHeader) {
+				headers.remove(ErrorHandlingDeserializer.KEY_DESERIALIZER_EXCEPTION_HEADER);
+			}
+			addExceptionInfoHeaders(headers, kDeserEx, true);
+		}
+		if (vDeserEx != null) {
+			if (!this.retainExceptionHeader) {
+				headers.remove(ErrorHandlingDeserializer.VALUE_DESERIALIZER_EXCEPTION_HEADER);
+			}
+			addExceptionInfoHeaders(headers, vDeserEx, false);
+		}
+		if (kDeserEx == null && vDeserEx == null) {
+			addExceptionInfoHeaders(headers, exception, false);
+		}
+		enhanceHeaders(headers, record, exception); // NOSONAR headers are never null
+	}
+
 	private void sendOrThrow(ProducerRecord<Object, Object> outRecord,
 			@Nullable KafkaOperations<Object, Object> kafkaTemplate) {
 
@@ -501,7 +515,6 @@ private Duration determineSendTimeout(KafkaOperations<?, ?> template) {
 
 	private void enhanceHeaders(Headers kafkaHeaders, ConsumerRecord<?, ?> record, Exception exception) {
 		maybeAddOriginalHeaders(kafkaHeaders, record);
-		addExceptionInfoHeaders(kafkaHeaders, exception, false);
 		Headers headers = this.headersFunction.apply(record, exception);
 		if (headers != null) {
 			headers.forEach(kafkaHeaders::add);

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

@@ -205,6 +205,29 @@ void keyHeaderStripped() {
 		assertThat(headers.lastHeader(ErrorHandlingDeserializer.KEY_DESERIALIZER_EXCEPTION_HEADER)).isNull();
 	}
 
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	@Test
+	void keyDeserOnly() {
+		KafkaOperations<?, ?> template = mock(KafkaOperations.class);
+		DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template);
+		Headers headers = new RecordHeaders();
+		DeserializationException deserEx = createDeserEx(true);
+		headers.add(
+				new RecordHeader(ErrorHandlingDeserializer.KEY_DESERIALIZER_EXCEPTION_HEADER, header(true, deserEx)));
+		SettableListenableFuture future = new SettableListenableFuture();
+		future.set(new Object());
+		willReturn(future).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, deserEx);
+		ArgumentCaptor<ProducerRecord> captor = ArgumentCaptor.forClass(ProducerRecord.class);
+		verify(template).send(captor.capture());
+		headers = captor.getValue().headers();
+		assertThat(headers.lastHeader(KafkaHeaders.DLT_KEY_EXCEPTION_MESSAGE).value()).isEqualTo("testK".getBytes());
+		assertThat(headers.lastHeader(KafkaHeaders.DLT_EXCEPTION_FQCN)).isNull();
+		assertThat(headers.lastHeader(KafkaHeaders.DLT_EXCEPTION_STACKTRACE)).isNull();
+	}
+
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	@Test
 	void headersNotStripped() {
@@ -225,6 +248,8 @@ void headersNotStripped() {
 		headers = captor.getValue().headers();
 		assertThat(headers.lastHeader(ErrorHandlingDeserializer.VALUE_DESERIALIZER_EXCEPTION_HEADER)).isNotNull();
 		assertThat(headers.lastHeader(ErrorHandlingDeserializer.KEY_DESERIALIZER_EXCEPTION_HEADER)).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({ "unchecked", "rawtypes" })
@@ -262,11 +287,19 @@ void tombstoneWithMultiTemplatesExplicit() {
 	}
 
 	private byte[] header(boolean isKey) {
+		return header(isKey, createDeserEx(isKey));
+	}
+
+	private DeserializationException createDeserEx(boolean isKey) {
+		return new DeserializationException(
+				isKey ? "testK" : "testV",
+				isKey ? "key".getBytes() : "value".getBytes(), isKey, null);
+	}
+
+	private byte[] header(boolean isKey, DeserializationException deserEx) {
 		ByteArrayOutputStream baos = new ByteArrayOutputStream();
 		try {
-			new ObjectOutputStream(baos).writeObject(new DeserializationException(
-					isKey ? "testK" : "testV",
-					isKey ? "key".getBytes() : "value".getBytes(), isKey, null));
+			new ObjectOutputStream(baos).writeObject(deserEx);
 		}
 		catch (IOException e) {
 			throw new UncheckedIOException(e);