Implement implicit serde for join and leftJoin - part of Issue #46

Author: debasishg

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

@@ -52,12 +52,17 @@ object ImplicitConversions {
   // we would also like to allow users implicit serdes
   // and these implicits will convert them to `Serialized`, `Produced` or `Consumed`
 
-  implicit def SerializedFromSerde[K,V](implicit keySerde: Serde[K], valueSerde: Serde[V]): Serialized[K,V] = 
+  implicit def SerializedFromSerde[K, V](implicit keySerde: Serde[K], valueSerde: Serde[V]): Serialized[K, V] = 
     Serialized.`with`(keySerde, valueSerde)
 
-  implicit def ConsumedFromSerde[K,V](implicit keySerde: Serde[K], valueSerde: Serde[V]): Consumed[K,V] = 
+  implicit def ConsumedFromSerde[K, V](implicit keySerde: Serde[K], valueSerde: Serde[V]): Consumed[K, V] = 
     Consumed.`with`(keySerde, valueSerde)
 
-  implicit def ProducedFromSerde[K,V](implicit keySerde: Serde[K], valueSerde: Serde[V]): Produced[K,V] = 
+  implicit def ProducedFromSerde[K, V](implicit keySerde: Serde[K], valueSerde: Serde[V]): Produced[K, V] = 
     Produced.`with`(keySerde, valueSerde)
+
+  implicit def JoinedFromKVOSerde[K, V, VO](implicit keySerde: Serde[K], valueSerde: Serde[V], otherValueSerde: Serde[VO]): Joined[K, V, VO] = {
+    println(s"ks: $keySerde vs: $valueSerde ovs: $otherValueSerde")
+    Joined.`with`(keySerde, valueSerde, otherValueSerde)
+  }
 }

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

@@ -135,30 +135,21 @@ class KStreamS[K, V](val inner: KStream[K, V]) {
 
   def join[VO, VR](otherStream: KStreamS[K, VO],
     joiner: (V, VO) => VR,
-    windows: JoinWindows): KStreamS[K, VR] = {
+    windows: JoinWindows)(implicit joined: Perhaps[Joined[K, V, VO]]): KStreamS[K, VR] = {
 
-    inner.join[VO, VR](otherStream.inner, joiner.asValueJoiner, windows)
-  }
-
-  def join[VO, VR](otherStream: KStreamS[K, VO],
-    joiner: (V, VO) => VR,
-    windows: JoinWindows,
-    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): KStreamS[K, VR] = {
-
-    inner.join[VT, VR](table.inner, joiner.asValueJoiner)
+    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) 
+    }
   }
 
   def join[VT, VR](table: KTableS[K, VT],
-    joiner: (V, VT) => VR,
-    joined: Joined[K, V, VT]): KStreamS[K, VR] = {
+    joiner: (V, VT) => VR)(implicit joined: Perhaps[Joined[K, V, VT]]): KStreamS[K, VR] = {
 
-    inner.join[VT, VR](table.inner, joiner.asValueJoiner, joined)
+    joined.fold[KStreamS[K, VR]] { 
+      inner.leftJoin[VT, VR](table.inner, joiner.asValueJoiner) } { implicit ev => 
+      inner.leftJoin[VT, VR](table.inner, joiner.asValueJoiner, ev) 
+    }
   }
 
   def join[GK, GV, RV](globalKTable: GlobalKTable[GK, GV],
@@ -170,30 +161,21 @@ class KStreamS[K, V](val inner: KStream[K, V]) {
 
   def leftJoin[VO, VR](otherStream: KStreamS[K, VO],
     joiner: (V, VO) => VR,
-    windows: JoinWindows): KStreamS[K, VR] = {
-
-    inner.leftJoin[VO, VR](otherStream.inner, joiner.asValueJoiner, windows)
-  }
-
-  def leftJoin[VO, VR](otherStream: KStreamS[K, VO],
-    joiner: (V, VO) => VR,
-    windows: JoinWindows,
-    joined: Joined[K, V, VO]): KStreamS[K, VR] = {
-
-    inner.leftJoin[VO, VR](otherStream.inner, joiner.asValueJoiner, windows, joined)
-  }
+    windows: JoinWindows)(implicit joined: Perhaps[Joined[K, V, VO]]): KStreamS[K, VR] = {
 
-  def leftJoin[VT, VR](table: KTableS[K, VT],
-    joiner: (V, VT) => VR): KStreamS[K, VR] = {
-
-    inner.leftJoin[VT, VR](table.inner, joiner.asValueJoiner)
+    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) 
+    }
   }
 
   def leftJoin[VT, VR](table: KTableS[K, VT],
-    joiner: (V, VT) => VR,
-    joined: Joined[K, V, VT]): KStreamS[K, VR] = {
+    joiner: (V, VT) => VR)(implicit joined: Perhaps[Joined[K, V, VT]]): KStreamS[K, VR] = {
 
-    inner.leftJoin[VT, VR](table.inner, joiner.asValueJoiner, joined)
+    joined.fold[KStreamS[K, VR]] { 
+      inner.leftJoin[VT, VR](table.inner, joiner.asValueJoiner) } { implicit ev => 
+      inner.leftJoin[VT, VR](table.inner, joiner.asValueJoiner, ev) 
+    }
   }
 
   def leftJoin[GK, GV, RV](globalKTable: GlobalKTable[GK, GV],
@@ -205,17 +187,12 @@ class KStreamS[K, V](val inner: KStream[K, V]) {
 
   def outerJoin[VO, VR](otherStream: KStreamS[K, VO],
     joiner: (V, VO) => VR,
-    windows: JoinWindows): KStreamS[K, VR] = {
-
-    inner.outerJoin[VO, VR](otherStream.inner, joiner.asValueJoiner, windows)
-  }
+    windows: JoinWindows)(implicit joined: Perhaps[Joined[K, V, VO]]): KStreamS[K, VR] = {
 
-  def outerJoin[VO, VR](otherStream: KStreamS[K, VO],
-    joiner: (V, VO) => VR,
-    windows: JoinWindows,
-    joined: Joined[K, V, VO]): KStreamS[K, VR] = {
-
-    inner.outerJoin[VO, VR](otherStream.inner, joiner.asValueJoiner, windows, joined)
+    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) 
+    }
   }
 
   def merge(stream: KStreamS[K, V]): KStreamS[K, V] = inner.merge(stream)

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

@@ -26,6 +26,7 @@ import com.lightbend.kafka.scala.server.{ KafkaLocalServer, MessageSender, Messa
 
 import org.apache.kafka.common.serialization._
 import org.apache.kafka.streams._
+import org.apache.kafka.streams.kstream.Joined
 
 import org.apache.kafka.clients.consumer.ConsumerRecord
 import ImplicitConversions._
@@ -79,6 +80,22 @@ object StreamToTableJoinScalaIntegrationTestImplicitSerdes extends TestSuite[Kaf
       p
     }
 
+    // this will make the leftJoin use the key and value serde from this implicit, while use the default
+    // from config for otherValueSerde
+    implicit val joined: Joined[String, Long, String] = Joined.keySerde(stringSerde).withValueSerde(longSerde)
+
+    /**
+     * Patterns for handling serdes in leftJoin (similar will be the handling of all join functions that accept a Joined argument)
+     *
+     * a. For `Joined[K, V, VO`], just make the implicit serdes available for `K`, `V` and `VO`. In the following
+     *    example of `leftJoin`, we have `Joined[String, Long, String]` and have the implicits `stringSerde` and
+     *    `longSerdes` available in scope. This should be enough to make an implicit `Joined` for `leftJoin`.
+     *
+     * b. Want to use default serdes from config for key and otherValue. Add the implicit `Joined` in scope as:
+     *    `implicit val joined: Joined[String, Long, String] = Joined.valueSerde(longSerde)`. The other serdes will
+     *    be picked up as `null` and used from the config.
+     */ 
+
     val builder = new StreamsBuilderS()
 
     val userClicksStream: KStreamS[String, Long] = builder.stream(userClicksTopic)