GH-4205: Allow JSON (de)serialize null values in Kotlin

Fixes: #4205

- Replaced `byte[]` with `byte @nullable []` for `JacksonJson(De)serializer` to be able to return and
  receive null values

Signed-off-by: Trond Ziarkowski <trondgzi@gmail.com>

Author: trondgzi

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

@@ -65,6 +65,7 @@
  * @author Ivan Ponomarev
  * @author Omer Celik
  * @author Soby Chacko
+ * @author Trond Ziarkowski
  *
  * @since 4.0
  */
@@ -566,7 +567,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 +598,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

@@ -54,6 +54,7 @@
  * @author Wang Zhiyang
  * @author Omer Celik
  * @author Soby Chacko
+ * @author Trond Ziarkowski
  *
  * @since 4.0
  */
@@ -202,9 +203,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 +214,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,49 @@
+/*
+ * Copyright 2016-present 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.support.serializer
+
+import org.assertj.core.api.Assertions.assertThat
+import org.junit.jupiter.api.Test
+import tools.jackson.core.type.TypeReference
+import kotlin.text.Charsets.UTF_8
+
+/**
+ * @author Trond Ziarkowski
+ *
+ * @since 4.0.1
+ */
+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)
+
+        assertThat(deserializedData).isEqualTo(DeserializedData("test-data"))
+    }
+
+    @Test
+    fun `Expect deserializing a null value to work`() {
+        val deserializedData = deserializer.deserialize("topic", null)
+        assertThat(deserializedData).isNull()
+    }
+
+    private data class DeserializedData(val value: String)
+
+}

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

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2016-present 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.support.serializer
+
+import org.assertj.core.api.Assertions.assertThat
+import org.junit.jupiter.api.Test
+import tools.jackson.core.type.TypeReference
+import kotlin.text.Charsets.UTF_8
+
+/**
+ * @author Trond Ziarkowski
+ *
+ * @since 4.0.1
+ */
+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)
+
+        assertThat(bytes?.toString(UTF_8)).isEqualTo("""{"value":"test-data"}""")
+    }
+
+    @Test
+    fun `Expect serializing a null value to work`() {
+        val bytes = serializer.serialize("topic", null)
+        assertThat(bytes).isNull()
+    }
+
+    private data class DataToSerialize(val value: String)
+}