Merge pull request #58 from lightbend/fix-reduce-issue

Fix KGroupedStream#reduce to ensure proper serialization

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:

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,