GH-3581: Address NPE in AbstractKafkaHeaderMapper (#3582)

Fixes: #3581

#3581

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

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

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

@@ -44,6 +44,7 @@
  * @author Gary Russell
  * @author Artem Bilan
  * @author Sanghyeok An
+ * @author Soby Chacko
  *
  * @since 2.1.3
  *
@@ -268,11 +269,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

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";