Merge branch 'develop'

Author: debasishg

build.sbt

@@ -4,7 +4,7 @@ name := "kafka-streams-scala"
 
 organization := "com.lightbend"
 
-version := "0.1.1"
+version := "0.1.2"
 
 scalaVersion := Versions.Scala_2_12_Version
 

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

@@ -36,13 +36,17 @@ class KGroupedStreamS[K, V](inner: KGroupedStream[K, V]) {
   def reduce(reducer: (V, V) => V,
     materialized: Materialized[K, V, KeyValueStore[Bytes, Array[Byte]]]): KTableS[K, V] = {
 
-    inner.reduce((v1: V, v2: V) => reducer(v1, v2), materialized)
+    // 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)
   }
 
   def reduce(reducer: (V, V) => V,
     storeName: String): KTableS[K, V] = {
 
-    inner.reduce((v1: V, v2: V) => reducer(v1, v2), Materialized.as[K, V, KeyValueStore[Bytes, Array[Byte]]](storeName))
+    // 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))
   }
 
   def aggregate[VR](initializer: () => VR,

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

@@ -28,14 +28,18 @@ class KGroupedTableS[K, V](inner: KGroupedTable[K, V]) {
   def reduce(adder: (V, V) => V,
     subtractor: (V, V) => V): KTableS[K, V] = {
 
-    inner.reduce((v1, v2) => adder(v1, v2), (v1, v2) => subtractor(v1, v2))
+    // 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, v2) => adder(v1, v2)).asReducer, ((v1, v2) => subtractor(v1, v2)).asReducer)
   }
 
   def reduce(adder: (V, V) => V,
     subtractor: (V, V) => V,
     materialized: Materialized[K, V, ByteArrayKVStore]): KTableS[K, V] = {
 
-    inner.reduce((v1, v2) => adder(v1, v2), (v1, v2) => subtractor(v1, v2), materialized)
+    // 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, v2) => adder(v1, v2)).asReducer, ((v1, v2) => subtractor(v1, v2)).asReducer, materialized)
   }
 
   def aggregate[VR](initializer: () => VR,
@@ -50,7 +54,6 @@ class KGroupedTableS[K, V](inner: KGroupedTable[K, V]) {
     subtractor: (K, V, VR) => VR,
     materialized: Materialized[K, VR, ByteArrayKVStore]): KTableS[K, VR] = {
 
-
     inner.aggregate(initializer.asInitializer, adder.asAggregator, subtractor.asAggregator, materialized)
   }
 }

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

@@ -73,7 +73,7 @@ object StreamToTableJoinScalaIntegrationTest extends TestSuite[KafkaLocalServer]
       p.put(StreamsConfig.CLIENT_ID_CONFIG, "join-scala-integration-test-standard-consumer")
       p.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
       p.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String.getClass.getName)
-      p.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String.getClass.getName)
+      p.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.Long.getClass.getName)
       p.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, "100")
       p.put(StreamsConfig.STATE_DIR_CONFIG, localStateDir)
       p
@@ -97,7 +97,9 @@ object StreamToTableJoinScalaIntegrationTest extends TestSuite[KafkaLocalServer]
 
         // Compute the total per region by summing the individual click counts per region.
         .groupByKey(Serialized.`with`(stringSerde, longSerde))
-        .reduce(_ + _)
+
+        // .reduce(_ + _, "local_state_data") // doesn't work in Scala 2.11, works with Scala 2.12
+        .reduce((firstClicks: Long, secondClicks: Long) => firstClicks + secondClicks, "local_state_data")
 
     // Write the (continuously updating) results to the output topic.
     clicksPerRegion.toStream.to(outputTopic, Produced.`with`(stringSerde, longSerde))