GH-2779: ParseStringDeserializer: Add Null Check

Resolves #2779

Handle tombstone records.

**cherry-pick to 2.9.x**

* Delegate null handling to the supplied function.

spring-kafka/src/main/java/org/springframework/kafka/support/serializer/ParseStringDeserializer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2020 the original author or authors.
+ * Copyright 2016-2023 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.
@@ -67,15 +67,17 @@ public ParseStringDeserializer() {
 	}
 
 	/**
-	 * Construct an instance with the supplied parser function.
+	 * Construct an instance with the supplied parser function. The function may receive
+	 * null as the input value, for example for a tombstone record in a compacted topic.
 	 * @param parser the function.
 	 */
 	public ParseStringDeserializer(Function<String, T> parser) {
 		this.parser = (message, ignoredHeaders) -> parser.apply(message);
 	}
 
 	/**
-	 * Construct an instance with the supplied parser function.
+	 * Construct an instance with the supplied parser function. The function may receive
+	 * null as the input value, for example for a tombstone record in a compacted topic.
 	 * @param parser the function.
 	 */
 	public ParseStringDeserializer(BiFunction<String, Headers, T> parser) {
@@ -100,7 +102,7 @@ public T deserialize(String topic, byte[] data) {
 
 	@Override
 	public T deserialize(String topic, Headers headers, byte[] data) {
-		return this.parser.apply(new String(data, this.charset), headers);
+		return this.parser.apply(data == null ? null : new String(data, this.charset), headers);
 	}
 
 	/**

spring-kafka/src/test/java/org/springframework/kafka/support/serializer/ToStringSerializationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2020 the original author or authors.
+ * Copyright 2016-2023 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.
@@ -29,6 +29,7 @@
 import org.junit.jupiter.api.Test;
 
 import org.springframework.kafka.support.serializer.testentities.DummyEntity;
+import org.springframework.lang.Nullable;
 
 /**
  *
@@ -212,6 +213,13 @@ public void testSerialization_usingHeaders() {
 				.hasFieldOrPropertyWithValue("intValue", 123);
 	}
 
+	@Test
+	void nullValue() {
+		ParseStringDeserializer<Object> deserializer =
+				new ParseStringDeserializer<>(ToStringSerializationTests::parseWithHeaders);
+		assertThat(deserializer.deserialize("foo", new RecordHeaders(), null)).isNull();
+	}
+
 	@Test
 	@DisplayName("Test deserialization using headers via config")
 	public void testSerialization_usingHeadersViaConfig() {
@@ -272,7 +280,10 @@ public void testSerializationDeserializationWithCharset() {
 				.isNotEqualTo("tôtô");
 	}
 
-	public static Object parseWithHeaders(String str, Headers headers) {
+	public static Object parseWithHeaders(@Nullable String str, Headers headers) {
+		if (str == null) {
+			return null;
+		}
 		byte[] header = headers.lastHeader(ToStringSerializer.VALUE_TYPE).value();
 		String entityType = new String(header);