Make Serdes less confusing

src/main/scala/com/lightbend/kafka/scala/streams/ImplicitConversions.scala

@@ -7,7 +7,6 @@ package com.lightbend.kafka.scala.streams
 import org.apache.kafka.streams.kstream._
 import org.apache.kafka.streams.{Consumed, KeyValue}
 import org.apache.kafka.common.serialization.Serde
-
 import scala.language.implicitConversions
 
 /**

src/main/scala/com/lightbend/kafka/scala/streams/Serdes.scala

@@ -0,0 +1,104 @@
+/**
+  * Copyright (C) 2018 Lightbend Inc. <https://www.lightbend.com>
+  * Copyright 2017-2018 Alexis Seigneurin.
+  */
+package com.lightbend.kafka.scala.streams
+
+import java.util
+import org.apache.kafka.common.serialization.{
+  Serde,
+  Serdes => JavaSerdes,
+  Deserializer => JavaDeserializer,
+  Serializer => JavaSerializer
+}
+
+trait Serdes {
+  implicit val string: Serde[String]                             = JavaSerdes.String()
+  implicit val long: Serde[Long]                                 = JavaSerdes.Long().asInstanceOf[Serde[Long]]
+  implicit val javaLong: Serde[java.lang.Long]                   = JavaSerdes.Long()
+  implicit val byteArray: Serde[Array[Byte]]                     = JavaSerdes.ByteArray()
+  implicit val bytes: Serde[org.apache.kafka.common.utils.Bytes] = JavaSerdes.Bytes()
+  implicit val float: Serde[Float]                               = JavaSerdes.Float().asInstanceOf[Serde[Float]]
+  implicit val javaFloat: Serde[java.lang.Float]                 = JavaSerdes.Float()
+  implicit val double: Serde[Double]                             = JavaSerdes.Double().asInstanceOf[Serde[Double]]
+  implicit val javaDouble: Serde[java.lang.Double]               = JavaSerdes.Double()
+  implicit val integer: Serde[Int]                               = JavaSerdes.Integer().asInstanceOf[Serde[Int]]
+  implicit val javaInteger: Serde[java.lang.Integer]             = JavaSerdes.Integer()
+
+  def fromFn[T >: Null](serializer: T => Array[Byte], deserializer: Array[Byte] => Option[T]): Serde[T] =
+    JavaSerdes.serdeFrom(
+      new JavaSerializer[T] {
+        override def serialize(topic: String, data: T): Array[Byte]                = serializer(data)
+        override def configure(configs: util.Map[String, _], isKey: Boolean): Unit = ()
+        override def close(): Unit                                                 = ()
+      },
+      new JavaDeserializer[T] {
+        override def deserialize(topic: String, data: Array[Byte]): T              = deserializer(data).orNull
+        override def configure(configs: util.Map[String, _], isKey: Boolean): Unit = ()
+        override def close(): Unit                                                 = ()
+      }
+    )
+
+  def fromFn[T >: Null](serializer: (String, T) => Array[Byte],
+                        deserializer: (String, Array[Byte]) => Option[T]): Serde[T] =
+    JavaSerdes.serdeFrom(
+      new JavaSerializer[T] {
+        override def serialize(topic: String, data: T): Array[Byte]                = serializer(topic, data)
+        override def configure(configs: util.Map[String, _], isKey: Boolean): Unit = ()
+        override def close(): Unit                                                 = ()
+      },
+      new JavaDeserializer[T] {
+        override def deserialize(topic: String, data: Array[Byte]): T              = deserializer(topic, data).orNull
+        override def configure(configs: util.Map[String, _], isKey: Boolean): Unit = ()
+        override def close(): Unit                                                 = ()
+      }
+    )
+}
+
+/**
+  * Implicit values for default serdes
+  */
+object Serdes extends Serdes
+
+@deprecated("Use com.lightbend.kafka.scala.streams.Serdes instead", "")
+object DefaultSerdes extends Serdes
+
+@deprecated("Use com.lightbend.kafka.scala.streams.Serdes.fromFn instead", "")
+trait StatelessSerde[T >: Null] extends Serde[T] {
+  def serialize(data: T): Array[Byte]
+  def deserialize(data: Array[Byte]): Option[T]
+
+  override def deserializer(): Deserializer[T] =
+    (data: Array[Byte]) => deserialize(data)
+
+  override def serializer(): Serializer[T] =
+    (data: T) => serialize(data)
+
+  override def configure(configs: java.util.Map[String, _], isKey: Boolean): Unit = ()
+
+  override def close(): Unit = ()
+}
+
+@deprecated("Use org.apache.kafka.common.serialization.Deserializer instead", "")
+trait Deserializer[T >: Null] extends JavaDeserializer[T] {
+  override def configure(configs: java.util.Map[String, _], isKey: Boolean): Unit = ()
+
+  override def close(): Unit = ()
+
+  override def deserialize(topic: String, data: Array[Byte]): T =
+    Option(data).flatMap(deserialize).orNull
+
+  def deserialize(data: Array[Byte]): Option[T]
+}
+
+@deprecated("Use org.apache.kafka.common.serialization.Serializer instead", "")
+trait Serializer[T] extends JavaSerializer[T] {
+  override def configure(configs: java.util.Map[String, _], isKey: Boolean): Unit = ()
+
+  override def close(): Unit = ()
+
+  override def serialize(topic: String, data: T): Array[Byte] =
+    Option(data).map(serialize).orNull
+
+  def serialize(data: T): Array[Byte]
+}

src/test/scala/com/lightbend/kafka/scala/streams/KafkaStreamsMergeTest.scala

@@ -33,7 +33,7 @@ object KafkaStreamsMergeTest extends TestSuite[KafkaLocalServer] with WordCountM
     //
     // Step 1: Configure and start the processor topology.
     //
-    import DefaultSerdes._
+    import Serdes._
 
     val streamsConfiguration = new Properties()
     streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, s"wordcount-${scala.util.Random.nextInt(100)}")

src/test/scala/com/lightbend/kafka/scala/streams/KafkaStreamsTest.scala

@@ -32,7 +32,7 @@ object KafkaStreamsTest extends TestSuite[KafkaLocalServer] with WordCountTestDa
     //
     // Step 1: Configure and start the processor topology.
     //
-    import DefaultSerdes._
+    import Serdes._
 
     val streamsConfiguration = new Properties()
     streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, s"wordcount-${scala.util.Random.nextInt(100)}")

src/test/scala/com/lightbend/kafka/scala/streams/ProbabilisticCountingScalaIntegrationTest.scala

@@ -90,8 +90,8 @@ object ProbabilisticCountingScalaIntegrationTest
       p.put(StreamsConfig.APPLICATION_ID_CONFIG,
             s"probabilistic-counting-scala-integration-test-${scala.util.Random.nextInt(100)}")
       p.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
-      p.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.ByteArray.getClass.getName)
-      p.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String.getClass.getName)
+      p.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.byteArray.getClass.getName)
+      p.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.string.getClass.getName)
       p.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, "10000")
       p.put(StreamsConfig.STATE_DIR_CONFIG, localStateDir)
       p
@@ -107,7 +107,7 @@ object ProbabilisticCountingScalaIntegrationTest
         cfg.put("segment.bytes", segmentSizeBytes)
         cfg
       }
-      new CMSStoreBuilder[String](cmsStoreName, Serdes.String())
+      new CMSStoreBuilder[String](cmsStoreName, Serdes.string)
         .withLoggingEnabled(changelogConfig)
     }
     builder.addStateStore(cmsStoreBuilder)
@@ -137,9 +137,9 @@ object ProbabilisticCountingScalaIntegrationTest
       override def close(): Unit = {}
     }
 
-    implicit val stringSerde: Serde[String]         = Serdes.String()
-    implicit val byteArraySerde: Serde[Array[Byte]] = Serdes.ByteArray()
-    implicit val longSerde: Serde[Long]             = Serdes.Long().asInstanceOf[Serde[Long]]
+    implicit val stringSerde: Serde[String]         = Serdes.string
+    implicit val byteArraySerde: Serde[Array[Byte]] = Serdes.byteArray
+    implicit val longSerde: Serde[Long]             = Serdes.long
 
     // Read the input from Kafka.
     val textLines: KStreamS[Array[Byte], String] = builder.stream(inputTopic)

src/test/scala/com/lightbend/kafka/scala/streams/PunctuateTest.scala

@@ -41,8 +41,8 @@ object PunctuateTest extends TestSuite[KafkaLocalServer] with PunctuateTestData
     streamsConfiguration.put(StreamsConfig.CLIENT_ID_CONFIG, "punctuategroup")
 
     streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
-    streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName())
-    streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName())
+    streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.string.getClass.getName)
+    streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.string.getClass.getName)
 
     val topology = new Topology
     // Data input streams

src/test/scala/com/lightbend/kafka/scala/streams/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala

@@ -66,7 +66,7 @@ object StreamToTableJoinScalaIntegrationTestImplicitSerdes
     // DefaultSerdes brings into scope implicit serdes (mostly for primitives) that will set up all Serialized, Produced,
     // Consumed and Joined instances. So all APIs below that accept Serialized, Produced, Consumed or Joined will
     // get these instances automatically
-    import DefaultSerdes._
+    import Serdes._
 
     // we don't have any serde declared as part of configuration. Even if they are declared here, the
     // Scala APIs will ignore them. But it's possible to declare serdes here and use them through

src/test/scala/com/lightbend/kafka/scala/streams/StreamToTableJoinScalaIntegrationTestImplicitSerdesWithAvro.scala

@@ -36,28 +36,22 @@ object StreamToTableJoinScalaIntegrationTestImplicitSerdesWithAvro
 
   case class UserClicks(clicks: Long)
 
-  // adopted from Openshine implementation
-  class AvroSerde[T >: Null: SchemaFor: FromRecord: ToRecord] extends StatelessScalaSerde[T] {
-
-    override def serialize(data: T): Array[Byte] = {
+  /** Our implicit Serde implementation for the values we want to serialize
+    * as avro
+    */
+  implicit def avroSerde[T >: Null: SchemaFor: FromRecord: ToRecord]: Serde[T] = Serdes.fromFn(
+    { data: T =>
       val baos   = new ByteArrayOutputStream()
       val output = AvroOutputStream.binary[T](baos)
       output.write(data)
       output.close()
       baos.toByteArray
-    }
-
-    override def deserialize(data: Array[Byte]): Option[T] = {
+    }, { data =>
       val in    = new ByteArrayInputStream(data)
       val input = AvroInputStream.binary[T](in)
       input.iterator.toSeq.headOption
     }
-  }
-
-  /** Our implicit Serde implementation for the values we want to serialize
-    * as avro
-    */
-  implicit val userClicksSerde: Serde[UserClicks] = new AvroSerde
+  )
 
   /**
     * End-to-end integration test that demonstrates how to perform a join
@@ -95,7 +89,7 @@ object StreamToTableJoinScalaIntegrationTestImplicitSerdesWithAvro
     // DefaultSerdes brings into scope implicit serdes (mostly for primitives) that will set up all Serialized, Produced,
     // Consumed and Joined instances. So all APIs below that accept Serialized, Produced, Consumed or Joined will
     // get these instances automatically
-    import DefaultSerdes._
+    import Serdes._
 
     //
     // Step 1: Configure and start the processor topology.
@@ -147,11 +141,11 @@ object StreamToTableJoinScalaIntegrationTestImplicitSerdesWithAvro
             println(s"Stream terminated because of uncaught exception .. Shutting " +
                       s"down app",
                     e)
-            e.printStackTrace
+            e.printStackTrace()
             val closed = streams.close()
             println(s"Exiting application after streams close ($closed)")
           } catch {
-            case x: Exception => x.printStackTrace
+            case x: Exception => x.printStackTrace()
           } finally {
             println("Exiting application ..")
             System.exit(-1)
@@ -180,13 +174,12 @@ object StreamToTableJoinScalaIntegrationTestImplicitSerdesWithAvro
                                                      classOf[StringSerializer].getName,
                                                      classOf[ByteArraySerializer].getName)
     userClicks
-      .map(
-        kv =>
-          new KeyValue[String, Array[Byte]](
-            kv.key,
-            new AvroSerde[UserClicks].serialize(UserClicks(kv.value))
+      .map { kv =>
+        new KeyValue[String, Array[Byte]](
+          kv.key,
+          avroSerde[UserClicks].serializer().serialize("", UserClicks(kv.value))
         )
-      )
+      }
       .foreach(r => sender2.writeKeyValue(userClicksTopic, r.key, r.value))
 
     //