Made implementation type safe, no more serde using default config. Added an Avro based serialization example.

Author: debasishg

.travis.yml

@@ -4,7 +4,7 @@ jdk: oraclejdk8
 scala:
 - 2.11.11
 - 2.12.4
-sbt_args: -mem 1500
+sbt_args: -mem 2000
 script:
 - sbt "++ ${TRAVIS_SCALA_VERSION}!" test
 cache:

README.md

@@ -91,18 +91,46 @@ Also, the explicit conversion `asJava` from a Scala `Iterable` to a Java `Iterab
 
 ## Implicit Serdes
 
-One of the areas where the Java APIs' verbosity can be reduced is through a succinct way to pass serializers and de-serializers to the various functions. The library implementation offers implicit serdes to provide the serializers and de-serializers but at the same time also *allows the opt-in to use the default serializers registered in the Kafka Streams config*.
+One of the areas where the Java APIs' verbosity can be reduced is through a succinct way to pass serializers and de-serializers to the various functions. The library implementation offers type safe implicit serdes to provide the serializers and de-serializers. In doing so, the Scala library **does not use configuration based default serdes** which is not type safe and prone to runtime errors.
 
-The optional implicit pattern is implemented with the usual null-default-value trick, but with a difference. The technique used is adopted from [this blog post](http://missingfaktor.blogspot.in/2013/12/optional-implicit-trick-in-scala.html).
+The implementation allows implicits for the `Serde`s or for `Serialized`, `Consumed`, `Produced` and `Joined`. The test examples demonstrate both, though the implicits for Serdes make a cleaner implementation.
 
-The standard way to implement the null-default-value trick could not be applied as Scala [does not allow](https://stackoverflow.com/questions/4652095/why-does-the-scala-compiler-disallow-overloaded-methods-with-default-arguments/4652681#4652681) a mix of default values and function overloads. And we have quite a few examples of such overloaded functions in the Kafka Streams API set.
+### Default Serdes
 
-The implementation allows implicits for the `Serde`s or for `Serialized`, `Consumed` and `Produced`. The test examples demonstrate both, though the implicits for Serdes make a cleaner implementation.
+The library offers a module that contains all the default serdes for the primitives. Importing the object will bring in scope all such primitives and helps reduce implicit hell.
 
-The implementation does a trade-off in using the null-default-value trick as it moves some of the compile time errors to runtime.
+```scala
+object DefaultSerdes {
+  implicit val stringSerde: Serde[String] = Serdes.String()
+  implicit val longSerde: Serde[Long] = Serdes.Long().asInstanceOf[Serde[Long]]
+  implicit val byteArraySerde: Serde[Array[Byte]] = Serdes.ByteArray()
+  implicit val bytesSerde: Serde[org.apache.kafka.common.utils.Bytes] = Serdes.Bytes()
+  implicit val floatSerde: Serde[Float] = Serdes.Float().asInstanceOf[Serde[Float]]
+  implicit val doubleSerde: Serde[Double] = Serdes.Double().asInstanceOf[Serde[Double]]
+  implicit val integerSerde: Serde[Int] = Serdes.Integer().asInstanceOf[Serde[Int]]
+}
+```
+
+### Compile time typesafe
+
+Not only the serdes, but `DefaultSerdes` also brings into scope implicit  `Serialized`, `Produced`, `Consumed` and `Joined` instances. So all APIs that accept `Serialized`, `Produced`, `Consumed` or `Joined` will get these instances automatically with an `import DefaultSerdes._`.
 
-### Examples
+Just one import of `DefaultSerdes._` and the following code does not need a bit of `Serialized`, `Produced`, `Consumed` or `Joined` to be specified explicitly or through the default config. **And the best part is that for any missing instances of these you get a compilation error.** ..
+
+```scala
+val clicksPerRegion: KTableS[String, Long] =
+  userClicksStream
+
+  // Join the stream against the table.
+  .leftJoin(userRegionsTable, (clicks: Long, region: String) => (if (region == null) "UNKNOWN" else region, clicks))
+
+  // Change the stream from <user> -> <region, clicks> to <region> -> <clicks>
+  .map((_, regionWithClicks) => regionWithClicks)
+
+  // Compute the total per region by summing the individual click counts per region.
+  .groupByKey
+  .reduce(_ + _)
 
-1. The example [StreamToTableJoinScalaIntegrationTestImplicitSerdes](https://github.com/lightbend/kafka-streams-scala/blob/develop/src/test/scala/com/lightbend/kafka/scala/streams/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala) demonstrates how to use the technique of implicit `Serde`s
-2. The example [StreamToTableJoinScalaIntegrationTestImplicitSerialized](https://github.com/lightbend/kafka-streams-scala/blob/develop/src/test/scala/com/lightbend/kafka/scala/streams/StreamToTableJoinScalaIntegrationTestImplicitSerialized.scala) demonstrates how to use the technique of implicit `Serialized`, `Consumed` and `Produced`.
-3. The example [StreamToTableJoinScalaIntegrationTestMixImplicitSerialized](https://github.com/lightbend/kafka-streams-scala/blob/develop/src/test/scala/com/lightbend/kafka/scala/streams/StreamToTableJoinScalaIntegrationTestMixImplicitSerialized.scala) demonstrates how to use the technique of how to use default serdes along with implicit `Serialized`, `Consumed` and `Produced`.
+  // Write the (continuously updating) results to the output topic.
+  clicksPerRegion.toStream.to(outputTopic)
+```

build.sbt

@@ -23,7 +23,8 @@ libraryDependencies ++= Seq(
   minitest % "test",
   minitestLaws % "test",
   algebird % "test",
-  chill % "test"
+  chill % "test",
+  avro4s % "test"
 )
 
 testFrameworks += new TestFramework("minitest.runner.Framework")

project/Dependencies.scala

@@ -21,4 +21,5 @@ object Dependencies {
   val minitestLaws = "io.monix"                       %% "minitest-laws"     % MinitestVersion
   val algebird = "com.twitter"                        %% "algebird-core"     % AlgebirdVersion
   val chill = "com.twitter"                           %% "chill"             % ChillVersion
+  val avro4s = "com.sksamuel.avro4s"                  %% "avro4s-core"       % Avro4sVersion
 }

project/Versions.scala

@@ -9,5 +9,6 @@ object Versions {
   val JDKVersion = "1.8"
   val Scala_2_12_Version = "2.12.4"
   val Scala_2_11_Version = "2.11.11"
+  val Avro4sVersion = "1.8.3"
   val CrossScalaVersions = Seq(Scala_2_12_Version, Scala_2_11_Version )
 }

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

@@ -0,0 +1,21 @@
+/**
+ * Copyright (C) 2018 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package com.lightbend.kafka.scala.streams
+
+import org.apache.kafka.common.serialization.{Serde, Serdes}
+
+
+/**
+ * Implicit values for default serdes
+ */
+object DefaultSerdes {
+  implicit val stringSerde: Serde[String] = Serdes.String()
+  implicit val longSerde: Serde[Long] = Serdes.Long().asInstanceOf[Serde[Long]]
+  implicit val byteArraySerde: Serde[Array[Byte]] = Serdes.ByteArray()
+  implicit val bytesSerde: Serde[org.apache.kafka.common.utils.Bytes] = Serdes.Bytes()
+  implicit val floatSerde: Serde[Float] = Serdes.Float().asInstanceOf[Serde[Float]]
+  implicit val doubleSerde: Serde[Double] = Serdes.Double().asInstanceOf[Serde[Double]]
+  implicit val integerSerde: Serde[Int] = Serdes.Integer().asInstanceOf[Serde[Int]]
+}

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

@@ -36,17 +36,6 @@ object ImplicitConversions {
 
   implicit def tuple2ToKeyValue[K, V](tuple: (K, V)): KeyValue[K, V] = new KeyValue(tuple._1, tuple._2)
 
-  // technique for optional implicits adopted from
-  // http://missingfaktor.blogspot.in/2013/12/optional-implicit-trick-in-scala.html
-
-  case class Perhaps[E](value: Option[E]) {
-    def fold[F](ifAbsent: => F)(ifPresent: E => F): F = {
-      value.fold(ifAbsent)(ifPresent)
-    }
-  }
-
-  //scalastyle:off null
-  implicit def perhaps[E](implicit ev: E = null): Perhaps[E] = Perhaps(Option(ev))
   //scalastyle:on null
   // we would also like to allow users implicit serdes
   // and these implicits will convert them to `Serialized`, `Produced` or `Consumed`

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

@@ -42,11 +42,16 @@ class KGroupedStreamS[K, V](inner: KGroupedStream[K, V]) {
   }
 
   def reduce(reducer: (V, V) => V,
-    storeName: String): KTableS[K, V] = {
+    storeName: String)(implicit keySerde: Serde[K], valueSerde: Serde[V]): KTableS[K, V] = {
 
     // need this explicit asReducer for Scala 2.11 or else the SAM conversion doesn't take place
     // works perfectly with Scala 2.12 though
-    inner.reduce(((v1: V, v2: V) => reducer(v1, v2)).asReducer, Materialized.as[K, V, KeyValueStore[Bytes, Array[Byte]]](storeName))
+    inner.reduce(((v1: V, v2: V) =>
+      reducer(v1, v2)).asReducer,
+      Materialized.as[K, V, KeyValueStore[Bytes, Array[Byte]]](storeName)
+        .withKeySerde(keySerde)
+        .withValueSerde(valueSerde)
+    )
   }
 
   def aggregate[VR](initializer: () => VR,

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

@@ -57,11 +57,11 @@ class KStreamS[K, V](val inner: KStream[K, V]) {
     inner.branch(predicates.map(_.asPredicate): _*).map(kstream => wrapKStream(kstream))
   }
 
-  def through(topic: String)(implicit produced: Perhaps[Produced[K, V]]): KStreamS[K, V] =
-    produced.fold[KStreamS[K, V]] { inner.through(topic) } { ev => inner.through(topic, ev) }
+  def through(topic: String)(implicit produced: Produced[K, V]): KStreamS[K, V] =
+    inner.through(topic, produced)
 
-  def to(topic: String)(implicit produced: Perhaps[Produced[K, V]]): Unit =
-    produced.fold[Unit] { inner.to(topic) } { implicit ev => inner.to(topic, ev) }
+  def to(topic: String)(implicit produced: Produced[K, V]): Unit =
+    inner.to(topic, produced)
 
   //scalastyle:off null
   def transform[K1, V1](transformerSupplier: () => Transformer[K, V, (K1, V1)],
@@ -124,61 +124,34 @@ class KStreamS[K, V](val inner: KStream[K, V]) {
    * implicit val longSerde: Serde[Long] = Serdes.Long().asInstanceOf[Serde[Long]]
    * - .groupByKey
    */
-  def groupByKey(implicit serialized: Perhaps[Serialized[K, V]]): KGroupedStreamS[K, V] =
-    serialized.fold[KGroupedStreamS[K, V]] { inner.groupByKey } { implicit ev => inner.groupByKey(ev) }
-
-  def groupBy[KR](selector: (K, V) => KR)(implicit serialized: Perhaps[Serialized[KR, V]]): KGroupedStreamS[KR, V] = {
-    serialized.fold[KGroupedStreamS[KR, V]] {
-      inner.groupBy(selector.asKeyValueMapper)
-    } { implicit ev =>
-      inner.groupBy(selector.asKeyValueMapper, ev)
-    }
-  }
+  def groupByKey(implicit serialized: Serialized[K, V]): KGroupedStreamS[K, V] =
+    inner.groupByKey(serialized)
+
+  def groupBy[KR](selector: (K, V) => KR)(implicit serialized: Serialized[KR, V]): KGroupedStreamS[KR, V] =
+    inner.groupBy(selector.asKeyValueMapper, serialized)
 
   def join[VO, VR](otherStream: KStreamS[K, VO],
     joiner: (V, VO) => VR,
-    windows: JoinWindows)(implicit joined: Perhaps[Joined[K, V, VO]]): KStreamS[K, VR] = {
-
-    joined.fold[KStreamS[K, VR]] {
-      inner.join[VO, VR](otherStream.inner, joiner.asValueJoiner, windows) } { implicit ev =>
-      inner.join[VO, VR](otherStream.inner, joiner.asValueJoiner, windows, ev)
-    }
-  }
+    windows: JoinWindows)(implicit joined: Joined[K, V, VO]): KStreamS[K, VR] =
+      inner.join[VO, VR](otherStream.inner, joiner.asValueJoiner, windows, joined)
 
   def join[VT, VR](table: KTableS[K, VT],
-    joiner: (V, VT) => VR)(implicit joined: Perhaps[Joined[K, V, VT]]): KStreamS[K, VR] = {
-
-    joined.fold[KStreamS[K, VR]] {
-      inner.leftJoin[VT, VR](table.inner, joiner.asValueJoiner) } { implicit ev =>
-      inner.leftJoin[VT, VR](table.inner, joiner.asValueJoiner, ev)
-    }
-  }
+    joiner: (V, VT) => VR)(implicit joined: Joined[K, V, VT]): KStreamS[K, VR] =
+      inner.leftJoin[VT, VR](table.inner, joiner.asValueJoiner, joined)
 
   def join[GK, GV, RV](globalKTable: GlobalKTable[GK, GV],
     keyValueMapper: (K, V) => GK,
-    joiner: (V, GV) => RV): KStreamS[K, RV] = {
-
-    inner.join[GK, GV, RV](globalKTable, keyValueMapper(_,_), joiner(_,_))
-  }
+    joiner: (V, GV) => RV): KStreamS[K, RV] =
+      inner.join[GK, GV, RV](globalKTable, keyValueMapper(_,_), joiner(_,_))
 
   def leftJoin[VO, VR](otherStream: KStreamS[K, VO],
     joiner: (V, VO) => VR,
-    windows: JoinWindows)(implicit joined: Perhaps[Joined[K, V, VO]]): KStreamS[K, VR] = {
-
-    joined.fold[KStreamS[K, VR]] {
-      inner.leftJoin[VO, VR](otherStream.inner, joiner.asValueJoiner, windows) } { implicit ev =>
-      inner.leftJoin[VO, VR](otherStream.inner, joiner.asValueJoiner, windows, ev)
-    }
-  }
+    windows: JoinWindows)(implicit joined: Joined[K, V, VO]): KStreamS[K, VR] =
+      inner.leftJoin[VO, VR](otherStream.inner, joiner.asValueJoiner, windows, joined)
 
   def leftJoin[VT, VR](table: KTableS[K, VT],
-    joiner: (V, VT) => VR)(implicit joined: Perhaps[Joined[K, V, VT]]): KStreamS[K, VR] = {
-
-    joined.fold[KStreamS[K, VR]] {
-      inner.leftJoin[VT, VR](table.inner, joiner.asValueJoiner) } { implicit ev =>
-      inner.leftJoin[VT, VR](table.inner, joiner.asValueJoiner, ev)
-    }
-  }
+    joiner: (V, VT) => VR)(implicit joined: Joined[K, V, VT]): KStreamS[K, VR] =
+      inner.leftJoin[VT, VR](table.inner, joiner.asValueJoiner, joined)
 
   def leftJoin[GK, GV, RV](globalKTable: GlobalKTable[GK, GV],
     keyValueMapper: (K, V) => GK,
@@ -189,13 +162,8 @@ class KStreamS[K, V](val inner: KStream[K, V]) {
 
   def outerJoin[VO, VR](otherStream: KStreamS[K, VO],
     joiner: (V, VO) => VR,
-    windows: JoinWindows)(implicit joined: Perhaps[Joined[K, V, VO]]): KStreamS[K, VR] = {
-
-    joined.fold[KStreamS[K, VR]] {
-      inner.outerJoin[VO, VR](otherStream.inner, joiner.asValueJoiner, windows) } { implicit ev =>
-      inner.outerJoin[VO, VR](otherStream.inner, joiner.asValueJoiner, windows, ev)
-    }
-  }
+    windows: JoinWindows)(implicit joined: Joined[K, V, VO]): KStreamS[K, VR] =
+      inner.outerJoin[VO, VR](otherStream.inner, joiner.asValueJoiner, windows, joined)
 
   def merge(stream: KStreamS[K, V]): KStreamS[K, V] = inner.merge(stream)
 

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

@@ -48,12 +48,8 @@ class KTableS[K, V](val inner: KTable[K, V]) {
     inner.toStream[KR](mapper.asKeyValueMapper)
   }
 
-  def groupBy[KR, VR](selector: (K, V) => (KR, VR))(implicit serialized: Perhaps[Serialized[KR, VR]]): KGroupedTableS[KR, VR] = {
-    serialized.fold[KGroupedTableS[KR, VR]] {
-      inner.groupBy(selector.asKeyValueMapper)
-    } { implicit ev =>
-      inner.groupBy(selector.asKeyValueMapper, ev)
-    }
+  def groupBy[KR, VR](selector: (K, V) => (KR, VR))(implicit serialized: Serialized[KR, VR]): KGroupedTableS[KR, VR] = {
+    inner.groupBy(selector.asKeyValueMapper, serialized)
   }
 
   def join[VO, VR](other: KTableS[K, VO],