handle null in genericserde for kafka tombstones handling (#34)

kaa/src/main/scala/com/davideicardi/kaa/kafka/GenericSerde.scala

@@ -18,19 +18,22 @@ class GenericSerde[T >: Null : SchemaFor : Encoder : Decoder]
   def deserializer: Deserializer[T] = this
 
   override def deserialize(topic: String, data: Array[Byte]): T = {
-    avroSerializer.deserialize(data)
+    if (data == null || data.length == 0) {
+      null
+    } else {
+      avroSerializer.deserialize(data)
+    }
   }
 
   override def serialize(topic: String, data: T): Array[Byte] = {
-    avroSerializer.serialize(data)
+    if (data == null) {
+      Array()
+    } else {
+      avroSerializer.serialize(data)
+    }
   }
 
   override def close(): Unit = ()
 
   override def configure(configs: java.util.Map[String, _], isKey: Boolean): Unit = ()
 }
-
-
-
-
-

kaa/src/test/scala/kaa/kafka/GenericSerdeSpec.scala

@@ -0,0 +1,53 @@
+package com.davideicardi.kaa.darwin
+
+import org.scalatest._
+import flatspec._
+import matchers._
+import com.davideicardi.kaa.kafka.GenericSerde
+import com.davideicardi.kaa.test.TestSchemaRegistry
+
+class GenericSerdeSpec extends AnyFlatSpec with should.Matchers {
+
+  val registry = new TestSchemaRegistry
+
+  "GenericSerde" should "serialize and deserialize a case class" in {
+    val target = new GenericSerde[FooUser](registry)
+
+    val expected = FooUser("foo")
+    val bytes = target.serialize("topic", expected)
+    val result = target.deserialize("topic", bytes)
+
+    result should equal (expected)
+  }
+
+  it should "serialize and deserialize an Option Some" in {
+    val target = new GenericSerde[Option[FooUser]](registry)
+
+    val expected = Some(FooUser("foo"))
+    val bytes = target.serialize("topic", expected)
+    val result = target.deserialize("topic", bytes)
+    result should equal (expected)
+  }
+
+  it should "serialize and deserialize an Option None" in {
+    val target = new GenericSerde[Option[FooUser]](registry)
+
+    val expected: Option[FooUser] = None
+    val bytes = target.serialize("topic", expected)
+    val result = target.deserialize("topic", bytes)
+    result should equal (expected)
+  }
+
+  it should "serialize and deserialize a null" in {
+    val target = new GenericSerde[FooUser](registry)
+
+    val expected: FooUser = null
+    val bytes = target.serialize("topic", expected)
+    val result = target.deserialize("topic", bytes)
+
+    result should equal (expected)
+  }
+
+  case class FooUser (name: String) {
+  }
+}