From cb8740dc7ca27f0c009fe8d718aa3a6165a395be Mon Sep 17 00:00:00 2001
From: Soby Chacko <soby.chacko@broadcom.com>
Date: Mon, 21 Oct 2024 12:33:42 -0400
Subject: [PATCH] GH-3581: Address NPE in AbstractKafkaHeaderMapper

Fixes: #3581

https://github.com/spring-projects/spring-kafka/issues/3581

- Optimize headerValueToAddIn method by adding better null checks
- Add unit test to verify

**Auto-cherry-pick to `3.1.x`**

# Conflicts:
#	spring-kafka/src/main/java/org/springframework/kafka/support/AbstractKafkaHeaderMapper.java
---
 .../support/AbstractKafkaHeaderMapper.java     |  9 +++++----
 .../support/DefaultKafkaHeaderMapperTests.java | 18 ++++++++++++++++++
 2 files changed, 23 insertions(+), 4 deletions(-)

diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/AbstractKafkaHeaderMapper.java b/spring-kafka/src/main/java/org/springframework/kafka/support/AbstractKafkaHeaderMapper.java
index d7ac37e200..03f655b299 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/support/AbstractKafkaHeaderMapper.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/support/AbstractKafkaHeaderMapper.java
@@ -43,6 +43,7 @@
  *
  * @author Gary Russell
  * @author Artem Bilan
+ * @author Soby Chacko
  *
  * @since 2.1.3
  *
@@ -267,11 +268,11 @@ else if (value instanceof String) {
 	 * @return the value to add.
 	 */
 	protected Object headerValueToAddIn(Header header) {
-		Object mapped = mapRawIn(header.key(), header.value());
-		if (mapped == null) {
-			mapped = header.value();
+		if (header == null || header.value() == null) {
+			return null;
 		}
-		return mapped;
+		String mapped = mapRawIn(header.key(), header.value());
+		return mapped != null ? mapped : header.value();
 	}
 
 	@Nullable
diff --git a/spring-kafka/src/test/java/org/springframework/kafka/support/DefaultKafkaHeaderMapperTests.java b/spring-kafka/src/test/java/org/springframework/kafka/support/DefaultKafkaHeaderMapperTests.java
index 598b08b084..1d799061cc 100644
--- a/spring-kafka/src/test/java/org/springframework/kafka/support/DefaultKafkaHeaderMapperTests.java
+++ b/spring-kafka/src/test/java/org/springframework/kafka/support/DefaultKafkaHeaderMapperTests.java
@@ -47,6 +47,10 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.entry;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
 
 /**
  * @author Gary Russell
@@ -360,6 +364,20 @@ void deserializationExceptionHeadersAreMappedAsNonByteArray() {
 		assertThat(headers.lastHeader(SerializationUtils.VALUE_DESERIALIZER_EXCEPTION_HEADER)).isNull();
 	}
 
+	@Test
+	void ensureNullHeaderValueHandledGraciously() {
+		DefaultKafkaHeaderMapper mapper = new DefaultKafkaHeaderMapper();
+
+		Header mockHeader = mock(Header.class);
+		given(mockHeader.value()).willReturn(null);
+
+		Object result = mapper.headerValueToAddIn(mockHeader);
+
+		assertThat(result).isNull();
+		verify(mockHeader).value();
+		verify(mockHeader, never()).key();
+	}
+
 	public static final class Foo {
 
 		private String bar = "bar";