GH-4205: Allow serializing/deserializing null values in Kotlin

- Replaced byte[] with byte @nullable [] to be able to return and
  receive null values

Signed-off-by: Trond Ziarkowski <[email protected]>

Author: trondgzi

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

@@ -566,7 +566,7 @@ private void doAddTrustedPackages(String... packages) {
 	}
 
 	@Override
-	public @Nullable T deserialize(String topic, Headers headers, byte[] data) {
+	public @Nullable T deserialize(String topic, Headers headers, byte @Nullable [] data) {
 		if (data == null) {
 			return null;
 		}
@@ -597,7 +597,7 @@ private void doAddTrustedPackages(String... packages) {
 	}
 
 	@Override
-	public @Nullable T deserialize(String topic, byte[] data) {
+	public @Nullable T deserialize(String topic, byte @Nullable [] data) {
 		if (data == null) {
 			return null;
 		}

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

@@ -202,9 +202,8 @@ protected static Map<String, Class<?>> createMappings(String mappings) {
 		return mappingsMap;
 	}
 
-	@SuppressWarnings("NullAway") // Dataflow analysis limitation
 	@Override
-	public byte[] serialize(String topic, Headers headers, @Nullable T data) {
+	public byte @Nullable [] serialize(String topic, Headers headers, @Nullable T data) {
 		if (data == null) {
 			return null;
 		}
@@ -214,9 +213,8 @@ public byte[] serialize(String topic, Headers headers, @Nullable T data) {
 		return serialize(topic, data);
 	}
 
-	@SuppressWarnings("NullAway") // Dataflow analysis limitation
 	@Override
-	public byte[] serialize(String topic, @Nullable T data) {
+	public byte @Nullable [] serialize(String topic, @Nullable T data) {
 		if (data == null) {
 			return null;
 		}

spring-kafka/src/test/kotlin/org/springframework/kafka/support/serializer/JacksonJsonDeserializerTests.kt

@@ -0,0 +1,29 @@
+package org.springframework.kafka.support.serializer
+
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.assertNull
+import tools.jackson.core.type.TypeReference
+import kotlin.text.Charsets.UTF_8
+
+class JacksonJsonDeserializerTests {
+    private val targetType = object : TypeReference<DeserializedData>() {}
+    private val deserializer = JacksonJsonDeserializer(targetType).ignoreTypeHeaders()
+
+    @Test
+    fun `Expect deserializing a non-null value to work`() {
+        val dataToDeserialize = """{"value":"test-data"}""".toByteArray(UTF_8)
+        val deserializedData = deserializer.deserialize("topic", dataToDeserialize)
+
+        assertEquals(DeserializedData("test-data"), deserializedData)
+    }
+
+    @Test
+    fun `Expect deserializing a null value to work`() {
+        val deserializedData = deserializer.deserialize("topic", null)
+        assertNull(deserializedData)
+    }
+
+    private data class DeserializedData(val value: String)
+
+}

spring-kafka/src/test/kotlin/org/springframework/kafka/support/serializer/JacksonJsonSerializerTests.kt

@@ -0,0 +1,28 @@
+package org.springframework.kafka.support.serializer
+
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.assertNull
+import tools.jackson.core.type.TypeReference
+import kotlin.text.Charsets.UTF_8
+
+class JacksonJsonSerializerTests {
+    private val targetType = object : TypeReference<DataToSerialize>() {}
+    private val serializer = JacksonJsonSerializer(targetType).noTypeInfo()
+
+    @Test
+    fun `Expect serializing a non-null value to work`() {
+        val dataToSerialize = DataToSerialize("test-data")
+        val bytes = serializer.serialize("topic", dataToSerialize)
+
+        assertEquals("""{"value":"test-data"}""", bytes?.toString(UTF_8))
+    }
+
+    @Test
+    fun `Expect serializing a null value to work`() {
+        val bytes = serializer.serialize("topic", null)
+        assertNull(bytes)
+    }
+
+    private data class DataToSerialize(val value: String)
+}