+ When using Scala you may optionally include the kafka-streams-scala library. Additional documentation on using the Kafka Streams DSL for Scala is available in the developer guide.
+
+ To use Kafka Streams DSL for Scala for Scala 2.11 you can use the following maven dependency:
+
+
The Connect API allows implementing connectors that continually pull from some source data system into Kafka or push from Kafka into some sink data system.
diff --git a/docs/streams/developer-guide/datatypes.html b/docs/streams/developer-guide/datatypes.html
index 51bd5857eecb0..d8d7b4c985afb 100644
--- a/docs/streams/developer-guide/datatypes.html
+++ b/docs/streams/developer-guide/datatypes.html
@@ -44,12 +44,15 @@
SerDes specified in the Streams configuration via StreamsConfig are used as the default in your Kafka Streams application.
@@ -155,23 +158,27 @@
JSONPageViewTypedDemo
example demonstrates how to use this JSON serde.
-
-
Implementing custom SerDes
-
If you need to implement custom SerDes, your best starting point is to take a look at the source code references of
- existing SerDes (see previous section). Typically, your workflow will be similar to:
Write a serde for T by implementing
- org.apache.kafka.common.serialization.Serde,
- which you either do manually (see existing SerDes in the previous section) or by leveraging helper functions in
- Serdes
- such as Serdes.serdeFrom(Serializer<T>,Deserializer<T>).
-
-
-
+
+
Implementing custom SerDes
+
If you need to implement custom SerDes, your best starting point is to take a look at the source code references of
+ existing SerDes (see previous section). Typically, your workflow will be similar to:
Write a serde for T by implementing
+ org.apache.kafka.common.serialization.Serde,
+ which you either do manually (see existing SerDes in the previous section) or by leveraging helper functions in
+ Serdes
+ such as Serdes.serdeFrom(Serializer<T>,Deserializer<T>).
+
+
+
+
+
Kafka Streams DSL for Scala Implicit SerDes
+
When using the Kafka Streams DSL for Scala you're not required to configure a default SerDes. In fact, it's not supported. SerDes are instead provided implicitly by default implementations for common primitive datatypes. See the Implicit SerDes and User-Defined SerDes sections in the DSL API documentation for details
OverviewKafka Connect API instead. However, if you do use such a sink processor, please be aware that
it is now your responsibility to guarantee message delivery semantics when talking to such external systems (e.g., to
retry on delivery failure or to prevent message duplication).
-
The Kafka Streams DSL Java APIs are based on the Builder design pattern, which allows users to incrementally build the target functionality using lower level compositional fluent APIs. These APIs can be called from Scala, but there are several issues:
+
+
Additional type annotations - The Java APIs use Java generics in a way that are not fully compatible with the type inferencer of the Scala compiler. Hence the user has to add type annotations to the Scala code, which seems rather non-idiomatic in Scala.
+
Verbosity - In some cases the Java APIs appear too verbose compared to idiomatic Scala.
+
Type Unsafety - The Java APIs offer some options where the compile time type safety is sometimes subverted and can result in runtime errors. This stems from the fact that the SerDes defined as part of config are not type checked during compile time. Hence any missing SerDes can result in runtime errors.
+
+
The Kafka Streams DSL for Scala library is a wrapper over the existing Java APIs for Kafka Streams DSL that addresses the concerns raised above.
+ It does not attempt to provide idiomatic Scala APIs that one would implement in a Scala library developed from scratch. The intention is to make the Java APIs more usable in Scala through better type inferencing, enhanced expressiveness, and lesser boilerplates.
+
+
The library wraps Java Stream DSL APIs in Scala thereby providing:
+
+
Better type inference in Scala.
+
Less boilerplate in application code.
+
The usual builder-style composition that developers get with the original Java API.
+
Implicit serializers and de-serializers leading to better abstraction and less verbosity.
+
Better type safety during compile time.
+
+
All functionality provided by Kafka Streams DSL for Scala are under the root package name of org.apache.kafka.streams.scala.
+
Many of the public facing types from the Java API are wrapped. The following Scala abstractions are available to the user:
The library also has several utility abstractions and modules that the user needs to use for proper semantics.
+
+
org.apache.kafka.streams.scala.ImplicitConversions: Module that brings into scope the implicit conversions between the Scala and Java classes.
+
org.apache.kafka.streams.scala.DefaultSerdes: Module that brings into scope the implicit values of all primitive SerDes.
+
org.apache.kafka.streams.scala.ScalaSerde: Base abstraction that can be used to implement custom SerDes in a type safe way.
+
+
The library is cross-built with Scala 2.11 and 2.12. To reference the library compiled against Scala 2.11 include the following in your maven pom.xml add the following:
The library works by wrapping the original Java abstractions of Kafka Streams within a Scala wrapper object and then using implicit conversions between them. All the Scala abstractions are named identically as the corresponding Java abstraction, but they reside in a different package of the library e.g. the Scala class org.apache.kafka.streams.scala.StreamsBuilder is a wrapper around org.apache.kafka.streams.StreamsBuilder, org.apache.kafka.streams.scala.kstream.KStream is a wrapper around org.apache.kafka.streams.kstream.KStream, and so on.
+
Here's an example of the classic WordCount program that uses the Scala StreamsBuilder that builds an instance of KStream which is a wrapper around Java KStream. Then we reify to a table and get a KTable, which, again is a wrapper around Java KTable.
+
The net result is that the following code is structured just like using the Java API, but with Scala and with far fewer type annotations compared to using the Java API directly from Scala. The difference in type annotation usage is more obvious when given an example. Below is an example WordCount implementation that will be used to demonstrate the differences between the Scala and Java API.
+
+import java.util.Properties
+import java.util.concurrent.TimeUnit
+
+import org.apache.kafka.streams.kstream.Materialized
+import org.apache.kafka.streams.{KafkaStreams, StreamsConfig}
+
+object WordCountApplication extends App {
+ import DefaultSerdes._
+ import ImplicitConversions._
+
+ val config: Properties = {
+ val p = new Properties()
+ p.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-application")
+ p.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker1:9092")
+ p
+ }
+
+ val builder = new StreamsBuilder()
+ val textLines = builder.stream[String, String]("TextLinesTopic")
+ val wordCounts = textLines
+ .flatMapValues(textLine => textLine.toLowerCase.split("\\W+"))
+ .groupBy((_, word) => word)
+ .count(Materialized.as("counts-store"))
+ wordCounts.toStream.to("WordsWithCountsTopic")
+
+ val streams: KafkaStreams = new KafkaStreams(builder.build(), config)
+ streams.start()
+
+ sys.ShutdownHookThread {
+ streams.close(10, TimeUnit.SECONDS)
+ }
+}
+
+
In the above code snippet, we don't have to provide any SerDes, Serialized, Produced, Consumed or Joined explicitly. They will also not be dependent on any SerDes specified in the config. In fact all SerDes specified in the config will be ignored by the Scala APIs. All SerDes and Serialized, Produced, Consumed or Joined will be handled through implicit SerDes as discussed later in the Implicit SerDes section. The complete independence from configuration based SerDes is what makes this library completely typesafe. Any missing instances of SerDes, Serialized, Produced, Consumed or Joined will be flagged as a compile time error.
One of the common complaints of Scala users with the Java API has been the repetitive usage of the SerDes in API invocations. Many of the APIs need to take the SerDes through abstractions like Serialized, Produced, Consumed or Joined. And the user has to supply them every time through the with function of these classes.
+
The library uses the power of Scala implicit parameters to alleviate this concern. As a user you can provide implicit SerDes or implicit values of Serialized, Produced, Consumed or Joined once and make your code less verbose. In fact you can just have the implicit SerDes in scope and the library will make the instances of Serialized, Produced, Consumed or Joined available in scope.
+
The library also bundles all implicit SerDes of the commonly used primitive types in a Scala module - so just import the module vals and have all SerDes in scope. A similar strategy of modular implicits can be adopted for any user-defined SerDes as well (User-defined SerDes are discussed in the next section).
+
Here's an example:
+
+// 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._
+
+val builder = new StreamsBuilder()
+
+val userClicksStream: KStream[String, Long] = builder.stream(userClicksTopic)
+
+val userRegionsTable: KTable[String, String] = builder.table(userRegionsTopic)
+
+// The following code fragment does not have a single instance of Serialized,
+// Produced, Consumed or Joined supplied explicitly.
+// All of them are taken care of by the implicit SerDes imported by DefaultSerdes
+val clicksPerRegion: KTable[String, Long] =
+ userClicksStream
+ .leftJoin(userRegionsTable,
+ (clicks: Long, region: String) =>
+ (if (region == null) "UNKNOWN" else region, clicks))
+ .map((_, regionWithClicks) => regionWithClicks)
+ .groupByKey
+ .reduce(_ + _)
+
+clicksPerRegion.toStream.to(outputTopic)
+
+
Quite a few things are going on in the above code snippet that may warrant a few lines of elaboration:
+
+
The code snippet does not depend on any config defined SerDes. In fact any SerDes defined as part of the config will be ignored.
+
All SerDes are picked up from the implicits in scope. And import DefaultSerdes._ brings all necessary SerDes in scope.
+
This is an example of compile time type safety that we don't have in the Java APIs.
+
The code looks less verbose and more focused towards the actual transformation that it does on the data stream.
When the default primitive SerDes are not enough and we need to define custom SerDes, the usage is exactly the same as above. Just define the implicit SerDes and start building the stream transformation. Here's an example with AvroSerde:
+
+// domain object as a case class
+case class UserClicks(clicks: Long)
+
+// An implicit Serde implementation for the values we want to
+// serialize as avro
+implicit val userClicksSerde: Serde[UserClicks] = new AvroSerde
+
+// Primitive SerDes
+import DefaultSerdes._
+
+// And then business as usual ..
+
+val userClicksStream: KStream[String, UserClicks] = builder.stream(userClicksTopic)
+
+val userRegionsTable: KTable[String, String] = builder.table(userRegionsTopic)
+
+// Compute the total per region by summing the individual click counts per region.
+val clicksPerRegion: KTable[String, Long] =
+ userClicksStream
+
+ // Join the stream against the table.
+ .leftJoin(userRegionsTable, (clicks: UserClicks, region: String) => (if (region == null) "UNKNOWN" else region, clicks.clicks))
+
+ // Change the stream from -> to ->
+ .map((_, regionWithClicks) => regionWithClicks)
+
+ // Compute the total per region by summing the individual click counts per region.
+ .groupByKey
+ .reduce(_ + _)
+
+// Write the (continuously updating) results to the output topic.
+clicksPerRegion.toStream.to(outputTopic)
+
+
A complete example of user-defined SerDes can be found in a test class within the library.
Any Java application that makes use of the Kafka Streams library is considered a Kafka Streams application.
+
Any Java or Scala application that makes use of the Kafka Streams library is considered a Kafka Streams application.
The computational logic of a Kafka Streams application is defined as a processor topology,
which is a graph of stream processors (nodes) and streams (edges).
You can define the processor topology with the Kafka Streams APIs:
A high-level API that provides provides the most common data transformation operations such as map, filter, join, and aggregations out of the box. The DSL is the recommended starting point for developers new to Kafka Streams, and should cover many use cases and stream processing needs.
+
A high-level API that provides the most common data transformation operations such as map, filter, join, and aggregations out of the box. The DSL is the recommended starting point for developers new to Kafka Streams, and should cover many use cases and stream processing needs. If you're writing a Scala application then you can use the Kafka Streams DSL for Scala library which removes much of the Java/Scala interoperability boilerplate as opposed to working directly with the Java DSL.
A low-level API that lets you add and connect processors as well as interact directly with state stores. The Processor API provides you with even more flexibility than the DSL but at the expense of requiring more manual work on the side of the application developer (e.g., more lines of code).
(Optional) Kafka Streams DSL for Scala library to write Scala Kafka Streams applications. When not using SBT you will need to suffix the artifact ID with the correct version of Scala your application is using (_2.11, _2.12)
Pinterest uses Apache Kafka and the Kafka Streams at large scale to power the real-time, predictive budgeting system of their advertising infrastructure. With Kafka Streams, spend predictions are more accurate than ever.
Rabobank is one of the 3 largest banks in the Netherlands. Its digital nervous system, the Business Event Bus, is powered by Apache Kafka. It is used by an increasing amount of financial processes and services, one of which is Rabo Alerts. This service alerts customers in real-time upon financial events and is built using Kafka Streams.
The code example below implements a WordCount application that is elastic, highly scalable, fault-tolerant, stateful, and ready to run in production at large scale
\ No newline at end of file
+
diff --git a/docs/streams/upgrade-guide.html b/docs/streams/upgrade-guide.html
index fdc0af1de1fb1..7ffafb547a872 100644
--- a/docs/streams/upgrade-guide.html
+++ b/docs/streams/upgrade-guide.html
@@ -112,9 +112,8 @@
We have deprecated constructors of KafkaStreams that take a StreamsConfig as parameter.
Please use the other corresponding constructors that accept java.util.Properties instead.
For more details, see KIP-245.
@@ -127,6 +126,14 @@
We have added support for methods in ReadOnlyWindowStore which allows for querying WindowStores without the necessity of providing keys.
diff --git a/gradle/dependencies.gradle b/gradle/dependencies.gradle
index d6beba958ddaa..0d05da3764886 100644
--- a/gradle/dependencies.gradle
+++ b/gradle/dependencies.gradle
@@ -60,6 +60,7 @@ versions += [
log4j: "1.2.17",
scalaLogging: "3.8.0",
jaxb: "2.3.0",
+ jfreechart: "1.0.0",
jopt: "5.0.4",
junit: "4.12",
kafka_0100: "0.10.0.1",
@@ -68,6 +69,7 @@ versions += [
kafka_0110: "0.11.0.2",
kafka_10: "1.0.1",
lz4: "1.4.1",
+ mavenArtifact: "3.5.3",
metrics: "2.2.0",
// PowerMock 1.x doesn't support Java 9, so use PowerMock 2.0.0 beta
powermock: "2.0.0-beta.5",
@@ -78,14 +80,11 @@ versions += [
slf4j: "1.7.25",
snappy: "1.1.7.1",
zkclient: "0.10",
- zookeeper: "3.4.10",
- jfreechart: "1.0.0",
- mavenArtifact: "3.5.3"
+ zookeeper: "3.4.10"
]
libs += [
activation: "javax.activation:activation:$versions.activation",
- argparse4j: "net.sourceforge.argparse4j:argparse4j:$versions.argparse4j",
apacheda: "org.apache.directory.api:api-all:$versions.apacheda",
apachedsCoreApi: "org.apache.directory.server:apacheds-core-api:$versions.apacheds",
apachedsInterceptorKerberos: "org.apache.directory.server:apacheds-interceptor-kerberos:$versions.apacheds",
@@ -95,6 +94,7 @@ libs += [
apachedsLdifPartition: "org.apache.directory.server:apacheds-ldif-partition:$versions.apacheds",
apachedsMavibotPartition: "org.apache.directory.server:apacheds-mavibot-partition:$versions.apacheds",
apachedsJdbmPartition: "org.apache.directory.server:apacheds-jdbm-partition:$versions.apacheds",
+ argparse4j: "net.sourceforge.argparse4j:argparse4j:$versions.argparse4j",
bcpkix: "org.bouncycastle:bcpkix-jdk15on:$versions.bcpkix",
easymock: "org.easymock:easymock:$versions.easymock",
jacksonDatabind: "com.fasterxml.jackson.core:jackson-databind:$versions.jackson",
@@ -105,6 +105,7 @@ libs += [
jettyServlet: "org.eclipse.jetty:jetty-servlet:$versions.jetty",
jettyServlets: "org.eclipse.jetty:jetty-servlets:$versions.jetty",
jerseyContainerServlet: "org.glassfish.jersey.containers:jersey-container-servlet:$versions.jersey",
+ jfreechart: "1.0.0",
jmhCore: "org.openjdk.jmh:jmh-core:$versions.jmh",
jmhCoreBenchmarks: "org.openjdk.jmh:jmh-core-benchmarks:$versions.jmh",
jmhGeneratorAnnProcess: "org.openjdk.jmh:jmh-generator-annprocess:$versions.jmh",
diff --git a/gradle/findbugs-exclude.xml b/gradle/findbugs-exclude.xml
index 33702056d83f6..f36f32376ab17 100644
--- a/gradle/findbugs-exclude.xml
+++ b/gradle/findbugs-exclude.xml
@@ -305,4 +305,19 @@ For a detailed description of findbugs bug categories, see http://findbugs.sourc
+
+
+
+
+
+
+
+-
+
+
+
+
+
+
+
diff --git a/settings.gradle b/settings.gradle
index 03136849fd543..19b3ef9f4154f 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -13,7 +13,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
-include 'core', 'examples', 'clients', 'tools', 'streams', 'streams:test-utils', 'streams:examples',
+include 'core', 'examples', 'clients', 'tools', 'streams', 'streams:streams-scala', 'streams:test-utils', 'streams:examples',
'streams:upgrade-system-tests-0100', 'streams:upgrade-system-tests-0101', 'streams:upgrade-system-tests-0102',
'streams:upgrade-system-tests-0110', 'streams:upgrade-system-tests-10', 'log4j-appender',
'connect:api', 'connect:transforms', 'connect:runtime', 'connect:json', 'connect:file', 'jmh-benchmarks'
diff --git a/streams/streams-scala/.gitignore b/streams/streams-scala/.gitignore
new file mode 100644
index 0000000000000..bf11921177ee2
--- /dev/null
+++ b/streams/streams-scala/.gitignore
@@ -0,0 +1 @@
+/logs/
\ No newline at end of file
diff --git a/streams/streams-scala/NOTICE b/streams/streams-scala/NOTICE
new file mode 100644
index 0000000000000..732e373e1c123
--- /dev/null
+++ b/streams/streams-scala/NOTICE
@@ -0,0 +1,3 @@
+Kafka Streams Scala
+Copyright (C) 2018 Lightbend Inc.
+Copyright (C) 2017-2018 Alexis Seigneurin.
diff --git a/streams/streams-scala/src/main/scala/org/apache/kafka/streams/package.scala b/streams/streams-scala/src/main/scala/org/apache/kafka/streams/package.scala
new file mode 100644
index 0000000000000..864fd19f87aaa
--- /dev/null
+++ b/streams/streams-scala/src/main/scala/org/apache/kafka/streams/package.scala
@@ -0,0 +1,27 @@
+/*
+ * Copyright (C) 2018 Lightbend Inc.
+ * Copyright (C) 2017-2018 Alexis Seigneurin.
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams
+
+import org.apache.kafka.streams.state.KeyValueStore
+import org.apache.kafka.common.utils.Bytes
+
+package object scala {
+ type ByteArrayKeyValueStore = KeyValueStore[Bytes, Array[Byte]]
+}
diff --git a/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/DefaultSerdes.scala b/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/DefaultSerdes.scala
new file mode 100644
index 0000000000000..3f2840eeecbed
--- /dev/null
+++ b/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/DefaultSerdes.scala
@@ -0,0 +1,47 @@
+/*
+ * Copyright (C) 2018 Lightbend Inc.
+ * Copyright (C) 2017-2018 Alexis Seigneurin.
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.scala
+
+import java.nio.ByteBuffer
+
+import org.apache.kafka.common.serialization.{Serde, Serdes}
+import org.apache.kafka.common.utils.Bytes
+import org.apache.kafka.streams.kstream.WindowedSerdes
+
+
+/**
+ * Implicit values for default serdes.
+ *
+ * Bring them in scope for default serializers / de-serializers to work.
+ */
+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[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]]
+ implicit val shortSerde: Serde[Short] = Serdes.Short().asInstanceOf[Serde[Short]]
+ implicit val byteBufferSerde: Serde[ByteBuffer] = Serdes.ByteBuffer()
+
+ implicit def timeWindowedSerde[T]: WindowedSerdes.TimeWindowedSerde[T] = new WindowedSerdes.TimeWindowedSerde[T]()
+ implicit def sessionWindowedSerde[T]: WindowedSerdes.SessionWindowedSerde[T] = new WindowedSerdes.SessionWindowedSerde[T]()
+}
diff --git a/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/FunctionConversions.scala b/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/FunctionConversions.scala
new file mode 100644
index 0000000000000..9ce9838c7c9a0
--- /dev/null
+++ b/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/FunctionConversions.scala
@@ -0,0 +1,108 @@
+/*
+ * Copyright (C) 2018 Lightbend Inc.
+ * Copyright (C) 2017-2018 Alexis Seigneurin.
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.scala
+
+import org.apache.kafka.streams.KeyValue
+import org.apache.kafka.streams.kstream._
+import scala.collection.JavaConverters._
+import java.lang.{Iterable => JIterable}
+
+/**
+ * Implicit classes that offer conversions of Scala function literals to
+ * SAM (Single Abstract Method) objects in Java. These make the Scala APIs much
+ * more expressive, with less boilerplate and more succinct.
+ *
+ * For Scala 2.11, most of these conversions need to be invoked explicitly, as Scala 2.11 does not
+ * have full support for SAM types.
+ */
+object FunctionConversions {
+
+ implicit class PredicateFromFunction[K, V](val p: (K, V) => Boolean) extends AnyVal {
+ def asPredicate: Predicate[K, V] = new Predicate[K, V] {
+ override def test(key: K, value: V): Boolean = p(key, value)
+ }
+ }
+
+ implicit class MapperFromFunction[T, U, VR](val f:(T,U) => VR) extends AnyVal {
+ def asKeyValueMapper: KeyValueMapper[T, U, VR] = new KeyValueMapper[T, U, VR] {
+ override def apply(key: T, value: U): VR = f(key, value)
+ }
+ def asValueJoiner: ValueJoiner[T, U, VR] = new ValueJoiner[T, U, VR] {
+ override def apply(value1: T, value2: U): VR = f(value1, value2)
+ }
+ }
+
+ implicit class KeyValueMapperFromFunction[K, V, KR, VR](val f:(K,V) => (KR, VR)) extends AnyVal {
+ def asKeyValueMapper: KeyValueMapper[K, V, KeyValue[KR, VR]] = new KeyValueMapper[K, V, KeyValue[KR, VR]] {
+ override def apply(key: K, value: V): KeyValue[KR, VR] = {
+ val (kr, vr) = f(key, value)
+ KeyValue.pair(kr, vr)
+ }
+ }
+ }
+
+ implicit class ValueMapperFromFunction[V, VR](val f: V => VR) extends AnyVal {
+ def asValueMapper: ValueMapper[V, VR] = new ValueMapper[V, VR] {
+ override def apply(value: V): VR = f(value)
+ }
+ }
+
+ implicit class FlatValueMapperFromFunction[V, VR](val f: V => Iterable[VR]) extends AnyVal {
+ def asValueMapper: ValueMapper[V, JIterable[VR]] = new ValueMapper[V, JIterable[VR]] {
+ override def apply(value: V): JIterable[VR] = f(value).asJava
+ }
+ }
+
+ implicit class ValueMapperWithKeyFromFunction[K, V, VR](val f: (K, V) => VR) extends AnyVal {
+ def asValueMapperWithKey: ValueMapperWithKey[K, V, VR] = new ValueMapperWithKey[K, V, VR] {
+ override def apply(readOnlyKey: K, value: V): VR = f(readOnlyKey, value)
+ }
+ }
+
+ implicit class FlatValueMapperWithKeyFromFunction[K, V, VR](val f: (K, V) => Iterable[VR]) extends AnyVal {
+ def asValueMapperWithKey: ValueMapperWithKey[K, V, JIterable[VR]] = new ValueMapperWithKey[K, V, JIterable[VR]] {
+ override def apply(readOnlyKey: K, value: V): JIterable[VR] = f(readOnlyKey, value).asJava
+ }
+ }
+
+ implicit class AggregatorFromFunction[K, V, VA](val f: (K, V, VA) => VA) extends AnyVal {
+ def asAggregator: Aggregator[K, V, VA] = new Aggregator[K, V, VA] {
+ override def apply(key: K, value: V, aggregate: VA): VA = f(key, value, aggregate)
+ }
+ }
+
+ implicit class MergerFromFunction[K,VR](val f: (K, VR, VR) => VR) extends AnyVal {
+ def asMerger: Merger[K, VR] = new Merger[K, VR] {
+ override def apply(aggKey: K, aggOne: VR, aggTwo: VR): VR = f(aggKey, aggOne, aggTwo)
+ }
+ }
+
+ implicit class ReducerFromFunction[V](val f: (V, V) => V) extends AnyVal {
+ def asReducer: Reducer[V] = new Reducer[V] {
+ override def apply(value1: V, value2: V): V = f(value1, value2)
+ }
+ }
+
+ implicit class InitializerFromFunction[VA](val f: () => VA) extends AnyVal {
+ def asInitializer: Initializer[VA] = new Initializer[VA] {
+ override def apply(): VA = f()
+ }
+ }
+}
diff --git a/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/ImplicitConversions.scala b/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/ImplicitConversions.scala
new file mode 100644
index 0000000000000..000690ba0ab4c
--- /dev/null
+++ b/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/ImplicitConversions.scala
@@ -0,0 +1,76 @@
+/*
+ * Copyright (C) 2018 Lightbend Inc.
+ * Copyright (C) 2017-2018 Alexis Seigneurin.
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.scala
+
+import org.apache.kafka.streams.kstream.{KStream => KStreamJ,
+ KTable => KTableJ,
+ KGroupedStream => KGroupedStreamJ,
+ SessionWindowedKStream => SessionWindowedKStreamJ,
+ TimeWindowedKStream => TimeWindowedKStreamJ,
+ KGroupedTable => KGroupedTableJ, _}
+
+import org.apache.kafka.streams.scala.kstream._
+import org.apache.kafka.streams.{KeyValue, Consumed}
+import org.apache.kafka.common.serialization.Serde
+
+import scala.language.implicitConversions
+
+/**
+ * Implicit conversions between the Scala wrapper objects and the underlying Java
+ * objects.
+ */
+object ImplicitConversions {
+
+ implicit def wrapKStream[K, V](inner: KStreamJ[K, V]): KStream[K, V] =
+ new KStream[K, V](inner)
+
+ implicit def wrapKGroupedStream[K, V](inner: KGroupedStreamJ[K, V]): KGroupedStream[K, V] =
+ new KGroupedStream[K, V](inner)
+
+ implicit def wrapSessionWindowedKStream[K, V](inner: SessionWindowedKStreamJ[K, V]): SessionWindowedKStream[K, V] =
+ new SessionWindowedKStream[K, V](inner)
+
+ implicit def wrapTimeWindowedKStream[K, V](inner: TimeWindowedKStreamJ[K, V]): TimeWindowedKStream[K, V] =
+ new TimeWindowedKStream[K, V](inner)
+
+ implicit def wrapKTable[K, V](inner: KTableJ[K, V]): KTable[K, V] =
+ new KTable[K, V](inner)
+
+ implicit def wrapKGroupedTable[K, V](inner: KGroupedTableJ[K, V]): KGroupedTable[K, V] =
+ new KGroupedTable[K, V](inner)
+
+ implicit def tuple2ToKeyValue[K, V](tuple: (K, V)): KeyValue[K, V] = new KeyValue(tuple._1, tuple._2)
+
+ // 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] =
+ Serialized.`with`(keySerde, valueSerde)
+
+ 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] =
+ Produced.`with`(keySerde, valueSerde)
+
+ implicit def joinedFromKeyValueOtherSerde[K, V, VO]
+ (implicit keySerde: Serde[K], valueSerde: Serde[V], otherValueSerde: Serde[VO]): Joined[K, V, VO] =
+ Joined.`with`(keySerde, valueSerde, otherValueSerde)
+}
diff --git a/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/ScalaSerde.scala b/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/ScalaSerde.scala
new file mode 100644
index 0000000000000..06afcaeed5e06
--- /dev/null
+++ b/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/ScalaSerde.scala
@@ -0,0 +1,70 @@
+/*
+ * Copyright (C) 2018 Lightbend Inc.
+ * Copyright (C) 2017-2018 Alexis Seigneurin.
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.scala
+
+import org.apache.kafka.common.serialization.{Serde, Deserializer => JDeserializer, Serializer => JSerializer}
+
+trait ScalaSerde[T] extends Serde[T] {
+ override def deserializer(): JDeserializer[T]
+
+ override def serializer(): JSerializer[T]
+
+ override def configure(configs: java.util.Map[String, _], isKey: Boolean): Unit = ()
+
+ override def close(): Unit = ()
+}
+
+trait SimpleScalaSerde[T >: Null] extends Serde[T] with ScalaSerde[T] {
+ def serialize(data: T): Array[Byte]
+ def deserialize(data: Array[Byte]): Option[T]
+
+ private def outerSerialize(data: T): Array[Byte] = serialize(data)
+ private def outerDeserialize(data: Array[Byte]): Option[T] = deserialize(data)
+
+ override def deserializer(): Deserializer[T] = new Deserializer[T] {
+ override def deserialize(data: Array[Byte]): Option[T] = outerDeserialize(data)
+ }
+
+ override def serializer(): Serializer[T] = new Serializer[T] {
+ override def serialize(data: T): Array[Byte] = outerSerialize(data)
+ }
+}
+
+trait Deserializer[T >: Null] extends JDeserializer[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]
+}
+
+trait Serializer[T] extends JSerializer[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]
+}
diff --git a/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/StreamsBuilder.scala b/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/StreamsBuilder.scala
new file mode 100644
index 0000000000000..9e6e204a37292
--- /dev/null
+++ b/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/StreamsBuilder.scala
@@ -0,0 +1,179 @@
+/*
+ * Copyright (C) 2018 Lightbend Inc.
+ * Copyright (C) 2017-2018 Alexis Seigneurin.
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.scala
+
+import java.util.regex.Pattern
+
+import org.apache.kafka.streams.kstream.{GlobalKTable, Materialized}
+import org.apache.kafka.streams.processor.{ProcessorSupplier, StateStore}
+import org.apache.kafka.streams.state.StoreBuilder
+import org.apache.kafka.streams.{Consumed, StreamsBuilder => StreamsBuilderJ, Topology}
+
+import org.apache.kafka.streams.scala.kstream._
+import ImplicitConversions._
+import scala.collection.JavaConverters._
+
+/**
+ * Wraps the Java class StreamsBuilder and delegates method calls to the underlying Java object.
+ */
+class StreamsBuilder(inner: StreamsBuilderJ = new StreamsBuilderJ) {
+
+ /**
+ * Create a [[kstream.KStream]] from the specified topic.
+ *
+ * The `implicit Consumed` instance provides the values of `auto.offset.reset` strategy, `TimestampExtractor`,
+ * key and value deserializers etc. If the implicit is not found in scope, compiler error will result.
+ *
+ * A convenient alternative is to have the necessary implicit serdes in scope, which will be implicitly
+ * converted to generate an instance of `Consumed`. @see [[ImplicitConversions]].
+ * {{{
+ * // Brings all implicit conversions in scope
+ * import ImplicitConversions._
+ *
+ * // Bring implicit default serdes in scope
+ * import DefaultSerdes._
+ *
+ * val builder = new StreamsBuilder()
+ *
+ * // stream function gets the implicit Consumed which is constructed automatically
+ * // from the serdes through the implicits in ImplicitConversions#consumedFromSerde
+ * val userClicksStream: KStream[String, Long] = builder.stream(userClicksTopic)
+ * }}}
+ *
+ * @param topic the topic name
+ * @return a [[kstream.KStream]] for the specified topic
+ */
+ def stream[K, V](topic: String)(implicit consumed: Consumed[K, V]): KStream[K, V] =
+ inner.stream[K, V](topic, consumed)
+
+ /**
+ * Create a [[kstream.KStream]] from the specified topics.
+ *
+ * @param topics the topic names
+ * @return a [[kstream.KStream]] for the specified topics
+ * @see #stream(String)
+ * @see `org.apache.kafka.streams.StreamsBuilder#stream`
+ */
+ def stream[K, V](topics: List[String])(implicit consumed: Consumed[K, V]): KStream[K, V] =
+ inner.stream[K, V](topics.asJava, consumed)
+
+ /**
+ * Create a [[kstream.KStream]] from the specified topic pattern.
+ *
+ * @param topics the topic name pattern
+ * @return a [[kstream.KStream]] for the specified topics
+ * @see #stream(String)
+ * @see `org.apache.kafka.streams.StreamsBuilder#stream`
+ */
+ def stream[K, V](topicPattern: Pattern)(implicit consumed: Consumed[K, V]): KStream[K, V] =
+ inner.stream[K, V](topicPattern, consumed)
+
+ /**
+ * Create a [[kstream.KTable]] from the specified topic.
+ *
+ * The `implicit Consumed` instance provides the values of `auto.offset.reset` strategy, `TimestampExtractor`,
+ * key and value deserializers etc. If the implicit is not found in scope, compiler error will result.
+ *
+ * A convenient alternative is to have the necessary implicit serdes in scope, which will be implicitly
+ * converted to generate an instance of `Consumed`. @see [[ImplicitConversions]].
+ * {{{
+ * // Brings all implicit conversions in scope
+ * import ImplicitConversions._
+ *
+ * // Bring implicit default serdes in scope
+ * import DefaultSerdes._
+ *
+ * val builder = new StreamsBuilder()
+ *
+ * // stream function gets the implicit Consumed which is constructed automatically
+ * // from the serdes through the implicits in ImplicitConversions#consumedFromSerde
+ * val userClicksStream: KTable[String, Long] = builder.table(userClicksTopic)
+ * }}}
+ *
+ * @param topic the topic name
+ * @return a [[kstream.KTable]] for the specified topic
+ * @see `org.apache.kafka.streams.StreamsBuilder#table`
+ */
+ def table[K, V](topic: String)(implicit consumed: Consumed[K, V]): KTable[K, V] =
+ inner.table[K, V](topic, consumed)
+
+ /**
+ * Create a [[kstream.KTable]] from the specified topic.
+ *
+ * @param topic the topic name
+ * @param materialized the instance of `Materialized` used to materialize a state store
+ * @return a [[kstream.KTable]] for the specified topic
+ * @see #table(String)
+ * @see `org.apache.kafka.streams.StreamsBuilder#table`
+ */
+ def table[K, V](topic: String, materialized: Materialized[K, V, ByteArrayKeyValueStore])
+ (implicit consumed: Consumed[K, V]): KTable[K, V] =
+ inner.table[K, V](topic, consumed, materialized)
+
+ /**
+ * Create a `GlobalKTable` from the specified topic. The serializers from the implicit `Consumed`
+ * instance will be used. Input records with `null` key will be dropped.
+ *
+ * @param topic the topic name
+ * @return a `GlobalKTable` for the specified topic
+ * @see `org.apache.kafka.streams.StreamsBuilder#globalTable`
+ */
+ def globalTable[K, V](topic: String)(implicit consumed: Consumed[K, V]): GlobalKTable[K, V] =
+ inner.globalTable(topic, consumed)
+
+ /**
+ * Create a `GlobalKTable` from the specified topic. The resulting `GlobalKTable` will be materialized
+ * in a local `KeyValueStore` configured with the provided instance of `Materialized`. The serializers
+ * from the implicit `Consumed` instance will be used.
+ *
+ * @param topic the topic name
+ * @param materialized the instance of `Materialized` used to materialize a state store
+ * @return a `GlobalKTable` for the specified topic
+ * @see `org.apache.kafka.streams.StreamsBuilder#globalTable`
+ */
+ def globalTable[K, V](topic: String, materialized: Materialized[K, V, ByteArrayKeyValueStore])
+ (implicit consumed: Consumed[K, V]): GlobalKTable[K, V] =
+ inner.globalTable(topic, consumed, materialized)
+
+ /**
+ * Adds a state store to the underlying `Topology`. The store must still be "connected" to a `Processor`,
+ * `Transformer`, or `ValueTransformer` before it can be used.
+ *
+ * @param builder the builder used to obtain this state store `StateStore` instance
+ * @return the underlying Java abstraction `StreamsBuilder` after adding the `StateStore`
+ * @throws TopologyException if state store supplier is already added
+ * @see `org.apache.kafka.streams.StreamsBuilder#addStateStore`
+ */
+ def addStateStore(builder: StoreBuilder[_ <: StateStore]): StreamsBuilderJ = inner.addStateStore(builder)
+
+ /**
+ * Adds a global `StateStore` to the topology. Global stores should not be added to `Processor, `Transformer`,
+ * or `ValueTransformer` (in contrast to regular stores).
+ *
+ * @see `org.apache.kafka.streams.StreamsBuilder#addGlobalStore`
+ */
+ def addGlobalStore(storeBuilder: StoreBuilder[_ <: StateStore],
+ topic: String,
+ consumed: Consumed[_, _],
+ stateUpdateSupplier: ProcessorSupplier[_, _]): StreamsBuilderJ =
+ inner.addGlobalStore(storeBuilder, topic, consumed, stateUpdateSupplier)
+
+ def build(): Topology = inner.build()
+}
diff --git a/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/KGroupedStream.scala b/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/KGroupedStream.scala
new file mode 100644
index 0000000000000..8f0ae93c1491d
--- /dev/null
+++ b/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/KGroupedStream.scala
@@ -0,0 +1,145 @@
+/*
+ * Copyright (C) 2018 Lightbend Inc.
+ * Copyright (C) 2017-2018 Alexis Seigneurin.
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.scala
+package kstream
+
+import org.apache.kafka.streams.kstream.{KGroupedStream => KGroupedStreamJ, _}
+import org.apache.kafka.common.serialization.Serde
+import org.apache.kafka.streams.scala.ImplicitConversions._
+import org.apache.kafka.streams.scala.FunctionConversions._
+
+
+/**
+ * Wraps the Java class KGroupedStream and delegates method calls to the underlying Java object.
+ *
+ * @param [K] Type of keys
+ * @param [V] Type of values
+ * @param inner The underlying Java abstraction for KGroupedStream
+ *
+ * @see `org.apache.kafka.streams.kstream.KGroupedStream`
+ */
+class KGroupedStream[K, V](val inner: KGroupedStreamJ[K, V]) {
+
+ /**
+ * Count the number of records in this stream by the grouped key.
+ *
+ * @return a [[KTable]] that contains "update" records with unmodified keys and `Long` values that
+ * represent the latest (rolling) count (i.e., number of records) for each key
+ * @see `org.apache.kafka.streams.kstream.KGroupedStream#count`
+ */
+ def count(): KTable[K, Long] = {
+ val c: KTable[K, java.lang.Long] = inner.count()
+ c.mapValues[Long](Long2long _)
+ }
+
+ /**
+ * Count the number of records in this stream by the grouped key.
+ * The result is written into a local `KeyValueStore` (which is basically an ever-updating materialized view)
+ * provided by the given `materialized`.
+ *
+ * @param materialized an instance of `Materialized` used to materialize a state store.
+ * @return a [[KTable]] that contains "update" records with unmodified keys and `Long` values that
+ * represent the latest (rolling) count (i.e., number of records) for each key
+ * @see `org.apache.kafka.streams.kstream.KGroupedStream#count`
+ */
+ def count(materialized: Materialized[K, Long, ByteArrayKeyValueStore]): KTable[K, Long] = {
+ val c: KTable[K, java.lang.Long] = inner.count(materialized.asInstanceOf[Materialized[K, java.lang.Long, ByteArrayKeyValueStore]])
+ c.mapValues[Long](Long2long _)
+ }
+
+ /**
+ * Combine the values of records in this stream by the grouped key.
+ *
+ * @param reducer a function `(V, V) => V` that computes a new aggregate result.
+ * @return a [[KTable]] that contains "update" records with unmodified keys, and values that represent the
+ * latest (rolling) aggregate for each key
+ * @see `org.apache.kafka.streams.kstream.KGroupedStream#reduce`
+ */
+ def reduce(reducer: (V, V) => V): KTable[K, V] = {
+ inner.reduce(reducer.asReducer)
+ }
+
+ /**
+ * Combine the values of records in this stream by the grouped key.
+ *
+ * @param reducer a function `(V, V) => V` that computes a new aggregate result.
+ * @param materialized an instance of `Materialized` used to materialize a state store.
+ * @return a [[KTable]] that contains "update" records with unmodified keys, and values that represent the
+ * latest (rolling) aggregate for each key
+ * @see `org.apache.kafka.streams.kstream.KGroupedStream#reduce`
+ */
+ def reduce(reducer: (V, V) => V,
+ materialized: Materialized[K, V, ByteArrayKeyValueStore]): KTable[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)
+ }
+
+ /**
+ * Aggregate the values of records in this stream by the grouped key.
+ *
+ * @param initializer an `Initializer` that computes an initial intermediate aggregation result
+ * @param aggregator an `Aggregator` that computes a new aggregate result
+ * @return a [[KTable]] that contains "update" records with unmodified keys, and values that represent the
+ * latest (rolling) aggregate for each key
+ * @see `org.apache.kafka.streams.kstream.KGroupedStream#aggregate`
+ */
+ def aggregate[VR](initializer: () => VR,
+ aggregator: (K, V, VR) => VR): KTable[K, VR] = {
+ inner.aggregate(initializer.asInitializer, aggregator.asAggregator)
+ }
+
+ /**
+ * Aggregate the values of records in this stream by the grouped key.
+ *
+ * @param initializer an `Initializer` that computes an initial intermediate aggregation result
+ * @param aggregator an `Aggregator` that computes a new aggregate result
+ * @param materialized an instance of `Materialized` used to materialize a state store.
+ * @return a [[KTable]] that contains "update" records with unmodified keys, and values that represent the
+ * latest (rolling) aggregate for each key
+ * @see `org.apache.kafka.streams.kstream.KGroupedStream#aggregate`
+ */
+ def aggregate[VR](initializer: () => VR,
+ aggregator: (K, V, VR) => VR,
+ materialized: Materialized[K, VR, ByteArrayKeyValueStore]): KTable[K, VR] = {
+ inner.aggregate(initializer.asInitializer, aggregator.asAggregator, materialized)
+ }
+
+ /**
+ * Create a new [[SessionWindowedKStream]] instance that can be used to perform session windowed aggregations.
+ *
+ * @param windows the specification of the aggregation `SessionWindows`
+ * @return an instance of [[SessionWindowedKStream]]
+ * @see `org.apache.kafka.streams.kstream.KGroupedStream#windowedBy`
+ */
+ def windowedBy(windows: SessionWindows): SessionWindowedKStream[K, V] =
+ inner.windowedBy(windows)
+
+ /**
+ * Create a new [[TimeWindowedKStream]] instance that can be used to perform windowed aggregations.
+ *
+ * @param windows the specification of the aggregation `Windows`
+ * @return an instance of [[TimeWindowedKStream]]
+ * @see `org.apache.kafka.streams.kstream.KGroupedStream#windowedBy`
+ */
+ def windowedBy[W <: Window](windows: Windows[W]): TimeWindowedKStream[K, V] =
+ inner.windowedBy(windows)
+}
diff --git a/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/KGroupedTable.scala b/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/KGroupedTable.scala
new file mode 100644
index 0000000000000..57c44fc8033e6
--- /dev/null
+++ b/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/KGroupedTable.scala
@@ -0,0 +1,138 @@
+/*
+ * Copyright (C) 2018 Lightbend Inc.
+ * Copyright (C) 2017-2018 Alexis Seigneurin.
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.scala
+package kstream
+
+import org.apache.kafka.streams.kstream.{KGroupedTable => KGroupedTableJ, _}
+import org.apache.kafka.streams.scala.ImplicitConversions._
+import org.apache.kafka.streams.scala.FunctionConversions._
+
+/**
+ * Wraps the Java class KGroupedTable and delegates method calls to the underlying Java object.
+ *
+ * @param [K] Type of keys
+ * @param [V] Type of values
+ * @param inner The underlying Java abstraction for KGroupedTable
+ *
+ * @see `org.apache.kafka.streams.kstream.KGroupedTable`
+ */
+class KGroupedTable[K, V](inner: KGroupedTableJ[K, V]) {
+
+ /**
+ * Count number of records of the original [[KTable]] that got [[KTable#groupBy]] to
+ * the same key into a new instance of [[KTable]].
+ *
+ * @return a [[KTable]] that contains "update" records with unmodified keys and `Long` values that
+ * represent the latest (rolling) count (i.e., number of records) for each key
+ * @see `org.apache.kafka.streams.kstream.KGroupedTable#count`
+ */
+ def count(): KTable[K, Long] = {
+ val c: KTable[K, java.lang.Long] = inner.count()
+ c.mapValues[Long](Long2long(_))
+ }
+
+ /**
+ * Count number of records of the original [[KTable]] that got [[KTable#groupBy]] to
+ * the same key into a new instance of [[KTable]].
+ *
+ * @param materialized an instance of `Materialized` used to materialize a state store.
+ * @return a [[KTable]] that contains "update" records with unmodified keys and `Long` values that
+ * represent the latest (rolling) count (i.e., number of records) for each key
+ * @see `org.apache.kafka.streams.kstream.KGroupedTable#count`
+ */
+ def count(materialized: Materialized[K, Long, ByteArrayKeyValueStore]): KTable[K, Long] =
+ inner.count(materialized)
+
+ /**
+ * Combine the value of records of the original [[KTable]] that got [[KTable#groupBy]]
+ * to the same key into a new instance of [[KTable]].
+ *
+ * @param adder a function that adds a new value to the aggregate result
+ * @param subtractor a function that removed an old value from the aggregate result
+ * @return a [[KTable]] that contains "update" records with unmodified keys, and values that represent the
+ * latest (rolling) aggregate for each key
+ * @see `org.apache.kafka.streams.kstream.KGroupedTable#reduce`
+ */
+ def reduce(adder: (V, V) => V,
+ subtractor: (V, V) => V): KTable[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(adder.asReducer, subtractor.asReducer)
+ }
+
+ /**
+ * Combine the value of records of the original [[KTable]] that got [[KTable#groupBy]]
+ * to the same key into a new instance of [[KTable]].
+ *
+ * @param adder a function that adds a new value to the aggregate result
+ * @param subtractor a function that removed an old value from the aggregate result
+ * @param materialized an instance of `Materialized` used to materialize a state store.
+ * @return a [[KTable]] that contains "update" records with unmodified keys, and values that represent the
+ * latest (rolling) aggregate for each key
+ * @see `org.apache.kafka.streams.kstream.KGroupedTable#reduce`
+ */
+ def reduce(adder: (V, V) => V,
+ subtractor: (V, V) => V,
+ materialized: Materialized[K, V, ByteArrayKeyValueStore]): KTable[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(adder.asReducer, subtractor.asReducer, materialized)
+ }
+
+ /**
+ * Aggregate the value of records of the original [[KTable]] that got [[KTable#groupBy]]
+ * to the same key into a new instance of [[KTable]] using default serializers and deserializers.
+ *
+ * @param initializer a function that provides an initial aggregate result value
+ * @param adder a function that adds a new record to the aggregate result
+ * @param subtractor an aggregator function that removed an old record from the aggregate result
+ * @return a [[KTable]] that contains "update" records with unmodified keys, and values that represent the
+ * latest (rolling) aggregate for each key
+ * @see `org.apache.kafka.streams.kstream.KGroupedTable#aggregate`
+ */
+ def aggregate[VR](initializer: () => VR,
+ adder: (K, V, VR) => VR,
+ subtractor: (K, V, VR) => VR): KTable[K, VR] = {
+
+ inner.aggregate(initializer.asInitializer, adder.asAggregator, subtractor.asAggregator)
+ }
+
+ /**
+ * Aggregate the value of records of the original [[KTable]] that got [[KTable#groupBy]]
+ * to the same key into a new instance of [[KTable]] using default serializers and deserializers.
+ *
+ * @param initializer a function that provides an initial aggregate result value
+ * @param adder a function that adds a new record to the aggregate result
+ * @param subtractor an aggregator function that removed an old record from the aggregate result
+ * @param materialized an instance of `Materialized` used to materialize a state store.
+ * @return a [[KTable]] that contains "update" records with unmodified keys, and values that represent the
+ * latest (rolling) aggregate for each key
+ * @see `org.apache.kafka.streams.kstream.KGroupedTable#aggregate`
+ */
+ def aggregate[VR](initializer: () => VR,
+ adder: (K, V, VR) => VR,
+ subtractor: (K, V, VR) => VR,
+ materialized: Materialized[K, VR, ByteArrayKeyValueStore]): KTable[K, VR] = {
+
+ inner.aggregate(initializer.asInitializer, adder.asAggregator, subtractor.asAggregator, materialized)
+ }
+}
diff --git a/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/KStream.scala b/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/KStream.scala
new file mode 100644
index 0000000000000..94c36add172e4
--- /dev/null
+++ b/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/KStream.scala
@@ -0,0 +1,581 @@
+/*
+ * Copyright (C) 2018 Lightbend Inc.
+ * Copyright (C) 2017-2018 Alexis Seigneurin.
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.scala
+package kstream
+
+import org.apache.kafka.streams.KeyValue
+import org.apache.kafka.streams.kstream.{KStream => KStreamJ, _}
+import org.apache.kafka.streams.processor.{Processor, ProcessorContext, ProcessorSupplier}
+import org.apache.kafka.streams.scala.ImplicitConversions._
+import org.apache.kafka.streams.scala.FunctionConversions._
+
+import scala.collection.JavaConverters._
+
+/**
+ * Wraps the Java class [[org.apache.kafka.streams.kstream.KStream]] and delegates method calls to the underlying Java object.
+ *
+ * @param [K] Type of keys
+ * @param [V] Type of values
+ * @param inner The underlying Java abstraction for KStream
+ *
+ * @see `org.apache.kafka.streams.kstream.KStream`
+ */
+class KStream[K, V](val inner: KStreamJ[K, V]) {
+
+ /**
+ * Create a new [[KStream]] that consists all records of this stream which satisfies the given
+ * predicate
+ *
+ * @param predicate a filter that is applied to each record
+ * @return a [[KStream]] that contains only those records that satisfy the given predicate
+ * @see `org.apache.kafka.streams.kstream.KStream#filter`
+ */
+ def filter(predicate: (K, V) => Boolean): KStream[K, V] = {
+ inner.filter(predicate.asPredicate)
+ }
+
+ /**
+ * Create a new [[KStream]] that consists all records of this stream which do not satisfy the given
+ * predicate
+ *
+ * @param predicate a filter that is applied to each record
+ * @return a [[KStream]] that contains only those records that do not satisfy the given predicate
+ * @see `org.apache.kafka.streams.kstream.KStream#filterNot`
+ */
+ def filterNot(predicate: (K, V) => Boolean): KStream[K, V] = {
+ inner.filterNot(predicate.asPredicate)
+ }
+
+ /**
+ * Set a new key (with possibly new type) for each input record.
+ *
+ * The function `mapper` passed is applied to every record and results in the generation of a new
+ * key `KR`. The function outputs a new [[KStream]] where each record has this new key.
+ *
+ * @param mapper a function `(K, V) => KR` that computes a new key for each record
+ * @return a [[KStream]] that contains records with new key (possibly of different type) and unmodified value
+ * @see `org.apache.kafka.streams.kstream.KStream#selectKey`
+ */
+ def selectKey[KR](mapper: (K, V) => KR): KStream[KR, V] = {
+ inner.selectKey[KR](mapper.asKeyValueMapper)
+ }
+
+ /**
+ * Transform each record of the input stream into a new record in the output stream (both key and value type can be
+ * altered arbitrarily).
+ *
+ * The provided `mapper`, a function `(K, V) => (KR, VR)` is applied to each input record and computes a new output record.
+ *
+ * @param mapper a function `(K, V) => (KR, VR)` that computes a new output record
+ * @return a [[KStream]] that contains records with new key and value (possibly both of different type)
+ * @see `org.apache.kafka.streams.kstream.KStream#map`
+ */
+ def map[KR, VR](mapper: (K, V) => (KR, VR)): KStream[KR, VR] = {
+ val kvMapper = mapper.tupled andThen tuple2ToKeyValue
+ inner.map[KR, VR](((k: K, v: V) => kvMapper(k, v)).asKeyValueMapper)
+ }
+
+ /**
+ * Transform the value of each input record into a new value (with possible new type) of the output record.
+ *
+ * The provided `mapper`, a function `V => VR` is applied to each input record value and computes a new value for it
+ *
+ * @param mapper, a function `V => VR` that computes a new output value
+ * @return a [[KStream]] that contains records with unmodified key and new values (possibly of different type)
+ * @see `org.apache.kafka.streams.kstream.KStream#mapValues`
+ */
+ def mapValues[VR](mapper: V => VR): KStream[K, VR] = {
+ inner.mapValues[VR](mapper.asValueMapper)
+ }
+
+ /**
+ * Transform the value of each input record into a new value (with possible new type) of the output record.
+ *
+ * The provided `mapper`, a function `(K, V) => VR` is applied to each input record value and computes a new value for it
+ *
+ * @param mapper, a function `(K, V) => VR` that computes a new output value
+ * @return a [[KStream]] that contains records with unmodified key and new values (possibly of different type)
+ * @see `org.apache.kafka.streams.kstream.KStream#mapValues`
+ */
+ def mapValues[VR](mapper: (K, V) => VR): KStream[K, VR] = {
+ inner.mapValues[VR](mapper.asValueMapperWithKey)
+ }
+
+ /**
+ * Transform each record of the input stream into zero or more records in the output stream (both key and value type
+ * can be altered arbitrarily).
+ *
+ * The provided `mapper`, function `(K, V) => Iterable[(KR, VR)]` is applied to each input record and computes zero or more output records.
+ *
+ * @param mapper function `(K, V) => Iterable[(KR, VR)]` that computes the new output records
+ * @return a [[KStream]] that contains more or less records with new key and value (possibly of different type)
+ * @see `org.apache.kafka.streams.kstream.KStream#flatMap`
+ */
+ def flatMap[KR, VR](mapper: (K, V) => Iterable[(KR, VR)]): KStream[KR, VR] = {
+ val kvMapper = mapper.tupled andThen (iter => iter.map(tuple2ToKeyValue).asJava)
+ inner.flatMap[KR, VR](((k: K, v: V) => kvMapper(k , v)).asKeyValueMapper)
+ }
+
+ /**
+ * Create a new [[KStream]] by transforming the value of each record in this stream into zero or more values
+ * with the same key in the new stream.
+ *
+ * Transform the value of each input record into zero or more records with the same (unmodified) key in the output
+ * stream (value type can be altered arbitrarily).
+ * The provided `mapper`, a function `V => Iterable[VR]` is applied to each input record and computes zero or more output values.
+ *
+ * @param mapper a function `V => Iterable[VR]` that computes the new output values
+ * @return a [[KStream]] that contains more or less records with unmodified keys and new values of different type
+ * @see `org.apache.kafka.streams.kstream.KStream#flatMapValues`
+ */
+ def flatMapValues[VR](mapper: V => Iterable[VR]): KStream[K, VR] = {
+ inner.flatMapValues[VR](mapper.asValueMapper)
+ }
+
+ /**
+ * Create a new [[KStream]] by transforming the value of each record in this stream into zero or more values
+ * with the same key in the new stream.
+ *
+ * Transform the value of each input record into zero or more records with the same (unmodified) key in the output
+ * stream (value type can be altered arbitrarily).
+ * The provided `mapper`, a function `(K, V) => Iterable[VR]` is applied to each input record and computes zero or more output values.
+ *
+ * @param mapper a function `(K, V) => Iterable[VR]` that computes the new output values
+ * @return a [[KStream]] that contains more or less records with unmodified keys and new values of different type
+ * @see `org.apache.kafka.streams.kstream.KStream#flatMapValues`
+ */
+ def flatMapValues[VR](mapper: (K, V) => Iterable[VR]): KStream[K, VR] = {
+ inner.flatMapValues[VR](mapper.asValueMapperWithKey)
+ }
+
+ /**
+ * Print the records of this KStream using the options provided by `Printed`
+ *
+ * @param printed options for printing
+ * @see `org.apache.kafka.streams.kstream.KStream#print`
+ */
+ def print(printed: Printed[K, V]): Unit = inner.print(printed)
+
+ /**
+ * Perform an action on each record of 'KStream`
+ *
+ * @param action an action to perform on each record
+ * @see `org.apache.kafka.streams.kstream.KStream#foreach`
+ */
+ def foreach(action: (K, V) => Unit): Unit = {
+ inner.foreach((k: K, v: V) => action(k, v))
+ }
+
+ /**
+ * Creates an array of {@code KStream} from this stream by branching the records in the original stream based on
+ * the supplied predicates.
+ *
+ * @param predicates the ordered list of functions that return a Boolean
+ * @return multiple distinct substreams of this [[KStream]]
+ * @see `org.apache.kafka.streams.kstream.KStream#branch`
+ */
+ def branch(predicates: ((K, V) => Boolean)*): Array[KStream[K, V]] = {
+ inner.branch(predicates.map(_.asPredicate): _*).map(kstream => wrapKStream(kstream))
+ }
+
+ /**
+ * Materialize this stream to a topic and creates a new [[KStream]] from the topic using the `Produced` instance for
+ * configuration of the `Serde key serde`, `Serde value serde`, and `StreamPartitioner`
+ *
+ * The user can either supply the `Produced` instance as an implicit in scope or she can also provide implicit
+ * key and value serdes that will be converted to a `Produced` instance implicitly.
+ *
+ * {{{
+ * Example:
+ *
+ * // brings implicit serdes in scope
+ * import DefaultSerdes._
+ *
+ * //..
+ * val clicksPerRegion: KTable[String, Long] = //..
+ *
+ * // Implicit serdes in scope will generate an implicit Produced instance, which
+ * // will be passed automatically to the call of through below
+ * clicksPerRegion.through(topic)
+ *
+ * // Similarly you can create an implicit Produced and it will be passed implicitly
+ * // to the through call
+ * }}}
+ *
+ * @param topic the topic name
+ * @param (implicit) produced the instance of Produced that gives the serdes and `StreamPartitioner`
+ * @return a [[KStream]] that contains the exact same (and potentially repartitioned) records as this [[KStream]]
+ * @see `org.apache.kafka.streams.kstream.KStream#through`
+ */
+ def through(topic: String)(implicit produced: Produced[K, V]): KStream[K, V] =
+ inner.through(topic, produced)
+
+ /**
+ * Materialize this stream to a topic using the `Produced` instance for
+ * configuration of the `Serde key serde`, `Serde value serde`, and `StreamPartitioner`
+ *
+ * The user can either supply the `Produced` instance as an implicit in scope or she can also provide implicit
+ * key and value serdes that will be converted to a `Produced` instance implicitly.
+ *
+ * {{{
+ * Example:
+ *
+ * // brings implicit serdes in scope
+ * import DefaultSerdes._
+ *
+ * //..
+ * val clicksPerRegion: KTable[String, Long] = //..
+ *
+ * // Implicit serdes in scope will generate an implicit Produced instance, which
+ * // will be passed automatically to the call of through below
+ * clicksPerRegion.to(topic)
+ *
+ * // Similarly you can create an implicit Produced and it will be passed implicitly
+ * // to the through call
+ * }}}
+ *
+ * @param topic the topic name
+ * @param (implicit) produced the instance of Produced that gives the serdes and `StreamPartitioner`
+ * @see `org.apache.kafka.streams.kstream.KStream#to`
+ */
+ def to(topic: String)(implicit produced: Produced[K, V]): Unit =
+ inner.to(topic, produced)
+
+ /**
+ * Transform each record of the input stream into zero or more records in the output stream (both key and value type
+ * can be altered arbitrarily).
+ * A `Transformer` (provided by the given `TransformerSupplier`) is applied to each input record and
+ * computes zero or more output records. In order to assign a state, the state must be created and registered
+ * beforehand via stores added via `addStateStore` or `addGlobalStore` before they can be connected to the `Transformer`
+ *
+ * @param transformerSupplier a instance of `TransformerSupplier` that generates a `Transformer`
+ * @param stateStoreNames the names of the state stores used by the processor
+ * @return a [[KStream]] that contains more or less records with new key and value (possibly of different type)
+ * @see `org.apache.kafka.streams.kstream.KStream#transform`
+ */
+ def transform[K1, V1](transformerSupplier: Transformer[K, V, (K1, V1)],
+ stateStoreNames: String*): KStream[K1, V1] = {
+ val transformerSupplierJ: TransformerSupplier[K, V, KeyValue[K1, V1]] = new TransformerSupplier[K, V, KeyValue[K1, V1]] {
+ override def get(): Transformer[K, V, KeyValue[K1, V1]] = {
+ new Transformer[K, V, KeyValue[K1, V1]] {
+ override def transform(key: K, value: V): KeyValue[K1, V1] = {
+ transformerSupplier.transform(key, value) match {
+ case (k1, v1) => KeyValue.pair(k1, v1)
+ case _ => null
+ }
+ }
+
+ override def init(context: ProcessorContext): Unit = transformerSupplier.init(context)
+
+ @deprecated ("Please use Punctuator functional interface at https://kafka.apache.org/10/javadoc/org/apache/kafka/streams/processor/Punctuator.html instead", "0.1.3") // scalastyle:ignore
+ override def punctuate(timestamp: Long): KeyValue[K1, V1] = {
+ transformerSupplier.punctuate(timestamp) match {
+ case (k1, v1) => KeyValue.pair[K1, V1](k1, v1)
+ case _ => null
+ }
+ }
+
+ override def close(): Unit = transformerSupplier.close()
+ }
+ }
+ }
+ inner.transform(transformerSupplierJ, stateStoreNames: _*)
+ }
+
+ /**
+ * Transform the value of each input record into a new value (with possible new type) of the output record.
+ * A `ValueTransformer` (provided by the given `ValueTransformerSupplier`) is applied to each input
+ * record value and computes a new value for it.
+ * In order to assign a state, the state must be created and registered
+ * beforehand via stores added via `addStateStore` or `addGlobalStore` before they can be connected to the `Transformer`
+ *
+ * @param valueTransformerSupplier a instance of `ValueTransformerSupplier` that generates a `ValueTransformer`
+ * @param stateStoreNames the names of the state stores used by the processor
+ * @return a [[KStream]] that contains records with unmodified key and new values (possibly of different type)
+ * @see `org.apache.kafka.streams.kstream.KStream#transformValues`
+ */
+ def transformValues[VR](valueTransformerSupplier: ValueTransformerSupplier[V, VR],
+ stateStoreNames: String*): KStream[K, VR] = {
+ inner.transformValues[VR](valueTransformerSupplier, stateStoreNames: _*)
+ }
+
+ /**
+ * Transform the value of each input record into a new value (with possible new type) of the output record.
+ * A `ValueTransformer` (provided by the given `ValueTransformerSupplier`) is applied to each input
+ * record value and computes a new value for it.
+ * In order to assign a state, the state must be created and registered
+ * beforehand via stores added via `addStateStore` or `addGlobalStore` before they can be connected to the `Transformer`
+ *
+ * @param valueTransformerSupplier a instance of `ValueTransformerWithKeySupplier` that generates a `ValueTransformerWithKey`
+ * @param stateStoreNames the names of the state stores used by the processor
+ * @return a [[KStream]] that contains records with unmodified key and new values (possibly of different type)
+ * @see `org.apache.kafka.streams.kstream.KStream#transformValues`
+ */
+ def transformValues[VR](valueTransformerWithKeySupplier: ValueTransformerWithKeySupplier[K, V, VR],
+ stateStoreNames: String*): KStream[K, VR] = {
+ inner.transformValues[VR](valueTransformerWithKeySupplier, stateStoreNames: _*)
+ }
+
+ /**
+ * Process all records in this stream, one record at a time, by applying a `Processor` (provided by the given
+ * `processorSupplier`).
+ * In order to assign a state, the state must be created and registered
+ * beforehand via stores added via `addStateStore` or `addGlobalStore` before they can be connected to the `Transformer`
+ *
+ * @param processorSupplier a function that generates a [[org.apache.kafka.stream.Processor]]
+ * @param stateStoreNames the names of the state store used by the processor
+ * @see `org.apache.kafka.streams.kstream.KStream#process`
+ */
+ def process(processorSupplier: () => Processor[K, V],
+ stateStoreNames: String*): Unit = {
+
+ val processorSupplierJ: ProcessorSupplier[K, V] = new ProcessorSupplier[K, V] {
+ override def get(): Processor[K, V] = processorSupplier()
+ }
+ inner.process(processorSupplierJ, stateStoreNames: _*)
+ }
+
+ /**
+ * Group the records by their current key into a [[KGroupedStream]]
+ *
+ * The user can either supply the `Serialized` instance as an implicit in scope or she can also provide an implicit
+ * serdes that will be converted to a `Serialized` instance implicitly.
+ *
+ * {{{
+ * Example:
+ *
+ * // brings implicit serdes in scope
+ * import DefaultSerdes._
+ *
+ * val clicksPerRegion: KTable[String, Long] =
+ * userClicksStream
+ * .leftJoin(userRegionsTable, (clicks: Long, region: String) => (if (region == null) "UNKNOWN" else region, clicks))
+ * .map((_, regionWithClicks) => regionWithClicks)
+ *
+ * // the groupByKey gets the Serialized instance through an implicit conversion of the
+ * // serdes brought into scope through the import DefaultSerdes._ above
+ * .groupByKey
+ * .reduce(_ + _)
+ *
+ * // Similarly you can create an implicit Serialized and it will be passed implicitly
+ * // to the groupByKey call
+ * }}}
+ *
+ * @param (implicit) serialized the instance of Serialized that gives the serdes
+ * @return a [[KGroupedStream]] that contains the grouped records of the original [[KStream]]
+ * @see `org.apache.kafka.streams.kstream.KStream#groupByKey`
+ */
+ def groupByKey(implicit serialized: Serialized[K, V]): KGroupedStream[K, V] =
+ inner.groupByKey(serialized)
+
+ /**
+ * Group the records of this [[KStream]] on a new key that is selected using the provided key transformation function
+ * and the `Serialized` instance.
+ *
+ * The user can either supply the `Serialized` instance as an implicit in scope or she can also provide an implicit
+ * serdes that will be converted to a `Serialized` instance implicitly.
+ *
+ * {{{
+ * Example:
+ *
+ * // brings implicit serdes in scope
+ * import DefaultSerdes._
+ *
+ * val textLines = streamBuilder.stream[String, String](inputTopic)
+ *
+ * val pattern = Pattern.compile("\\W+", Pattern.UNICODE_CHARACTER_CLASS)
+ *
+ * val wordCounts: KTable[String, Long] =
+ * textLines.flatMapValues(v => pattern.split(v.toLowerCase))
+ *
+ * // the groupBy gets the Serialized instance through an implicit conversion of the
+ * // serdes brought into scope through the import DefaultSerdes._ above
+ * .groupBy((k, v) => v)
+ *
+ * .count()
+ * }}}
+ *
+ * @param selector a function that computes a new key for grouping
+ * @return a [[KGroupedStream]] that contains the grouped records of the original [[KStream]]
+ * @see `org.apache.kafka.streams.kstream.KStream#groupBy`
+ */
+ def groupBy[KR](selector: (K, V) => KR)(implicit serialized: Serialized[KR, V]): KGroupedStream[KR, V] =
+ inner.groupBy(selector.asKeyValueMapper, serialized)
+
+ /**
+ * Join records of this stream with another [[KStream]]'s records using windowed inner equi join with
+ * serializers and deserializers supplied by the implicit `Joined` instance.
+ *
+ * @param otherStream the [[KStream]] to be joined with this stream
+ * @param joiner a function that computes the join result for a pair of matching records
+ * @param windows the specification of the `JoinWindows`
+ * @param joined an implicit `Joined` instance that defines the serdes to be used to serialize/deserialize
+ * inputs and outputs of the joined streams. Instead of `Joined`, the user can also supply
+ * key serde, value serde and other value serde in implicit scope and they will be
+ * converted to the instance of `Joined` through implicit conversion
+ * @return a [[KStream]] that contains join-records for each key and values computed by the given `joiner`,
+ * one for each matched record-pair with the same key and within the joining window intervals
+ * @see `org.apache.kafka.streams.kstream.KStream#join`
+ */
+ def join[VO, VR](otherStream: KStream[K, VO],
+ joiner: (V, VO) => VR,
+ windows: JoinWindows)(implicit joined: Joined[K, V, VO]): KStream[K, VR] =
+ inner.join[VO, VR](otherStream.inner, joiner.asValueJoiner, windows, joined)
+
+ /**
+ * Join records of this stream with another [[KTable]]'s records using inner equi join with
+ * serializers and deserializers supplied by the implicit `Joined` instance.
+ *
+ * @param table the [[KTable]] to be joined with this stream
+ * @param joiner a function that computes the join result for a pair of matching records
+ * @param joined an implicit `Joined` instance that defines the serdes to be used to serialize/deserialize
+ * inputs and outputs of the joined streams. Instead of `Joined`, the user can also supply
+ * key serde, value serde and other value serde in implicit scope and they will be
+ * converted to the instance of `Joined` through implicit conversion
+ * @return a [[KStream]] that contains join-records for each key and values computed by the given `joiner`,
+ * one for each matched record-pair with the same key
+ * @see `org.apache.kafka.streams.kstream.KStream#join`
+ */
+ def join[VT, VR](table: KTable[K, VT],
+ joiner: (V, VT) => VR)(implicit joined: Joined[K, V, VT]): KStream[K, VR] =
+ inner.join[VT, VR](table.inner, joiner.asValueJoiner, joined)
+
+ /**
+ * Join records of this stream with `GlobalKTable`'s records using non-windowed inner equi join.
+ *
+ * @param globalKTable the `GlobalKTable` to be joined with this stream
+ * @param keyValueMapper a function used to map from the (key, value) of this stream
+ * to the key of the `GlobalKTable`
+ * @param joiner a function that computes the join result for a pair of matching records
+ * @return a [[KStream]] that contains join-records for each key and values computed by the given `joiner`,
+ * one output for each input [[KStream]] record
+ * @see `org.apache.kafka.streams.kstream.KStream#join`
+ */
+ def join[GK, GV, RV](globalKTable: GlobalKTable[GK, GV],
+ keyValueMapper: (K, V) => GK,
+ joiner: (V, GV) => RV): KStream[K, RV] =
+ inner.join[GK, GV, RV](
+ globalKTable,
+ ((k: K, v: V) => keyValueMapper(k, v)).asKeyValueMapper,
+ ((v: V, gv: GV) => joiner(v, gv)).asValueJoiner
+ )
+
+ /**
+ * Join records of this stream with another [[KStream]]'s records using windowed left equi join with
+ * serializers and deserializers supplied by the implicit `Joined` instance.
+ *
+ * @param otherStream the [[KStream]] to be joined with this stream
+ * @param joiner a function that computes the join result for a pair of matching records
+ * @param windows the specification of the `JoinWindows`
+ * @param joined an implicit `Joined` instance that defines the serdes to be used to serialize/deserialize
+ * inputs and outputs of the joined streams. Instead of `Joined`, the user can also supply
+ * key serde, value serde and other value serde in implicit scope and they will be
+ * converted to the instance of `Joined` through implicit conversion
+ * @return a [[KStream]] that contains join-records for each key and values computed by the given `joiner`,
+ * one for each matched record-pair with the same key and within the joining window intervals
+ * @see `org.apache.kafka.streams.kstream.KStream#leftJoin`
+ */
+ def leftJoin[VO, VR](otherStream: KStream[K, VO],
+ joiner: (V, VO) => VR,
+ windows: JoinWindows)(implicit joined: Joined[K, V, VO]): KStream[K, VR] =
+ inner.leftJoin[VO, VR](otherStream.inner, joiner.asValueJoiner, windows, joined)
+
+ /**
+ * Join records of this stream with another [[KTable]]'s records using left equi join with
+ * serializers and deserializers supplied by the implicit `Joined` instance.
+ *
+ * @param table the [[KTable]] to be joined with this stream
+ * @param joiner a function that computes the join result for a pair of matching records
+ * @param joined an implicit `Joined` instance that defines the serdes to be used to serialize/deserialize
+ * inputs and outputs of the joined streams. Instead of `Joined`, the user can also supply
+ * key serde, value serde and other value serde in implicit scope and they will be
+ * converted to the instance of `Joined` through implicit conversion
+ * @return a [[KStream]] that contains join-records for each key and values computed by the given `joiner`,
+ * one for each matched record-pair with the same key
+ * @see `org.apache.kafka.streams.kstream.KStream#leftJoin`
+ */
+ def leftJoin[VT, VR](table: KTable[K, VT],
+ joiner: (V, VT) => VR)(implicit joined: Joined[K, V, VT]): KStream[K, VR] =
+ inner.leftJoin[VT, VR](table.inner, joiner.asValueJoiner, joined)
+
+ /**
+ * Join records of this stream with `GlobalKTable`'s records using non-windowed left equi join.
+ *
+ * @param globalKTable the `GlobalKTable` to be joined with this stream
+ * @param keyValueMapper a function used to map from the (key, value) of this stream
+ * to the key of the `GlobalKTable`
+ * @param joiner a function that computes the join result for a pair of matching records
+ * @return a [[KStream]] that contains join-records for each key and values computed by the given `joiner`,
+ * one output for each input [[KStream]] record
+ * @see `org.apache.kafka.streams.kstream.KStream#leftJoin`
+ */
+ def leftJoin[GK, GV, RV](globalKTable: GlobalKTable[GK, GV],
+ keyValueMapper: (K, V) => GK,
+ joiner: (V, GV) => RV): KStream[K, RV] = {
+
+ inner.leftJoin[GK, GV, RV](globalKTable, keyValueMapper.asKeyValueMapper, joiner.asValueJoiner)
+ }
+
+ /**
+ * Join records of this stream with another [[KStream]]'s records using windowed outer equi join with
+ * serializers and deserializers supplied by the implicit `Joined` instance.
+ *
+ * @param otherStream the [[KStream]] to be joined with this stream
+ * @param joiner a function that computes the join result for a pair of matching records
+ * @param windows the specification of the `JoinWindows`
+ * @param joined an implicit `Joined` instance that defines the serdes to be used to serialize/deserialize
+ * inputs and outputs of the joined streams. Instead of `Joined`, the user can also supply
+ * key serde, value serde and other value serde in implicit scope and they will be
+ * converted to the instance of `Joined` through implicit conversion
+ * @return a [[KStream]] that contains join-records for each key and values computed by the given `joiner`,
+ * one for each matched record-pair with the same key and within the joining window intervals
+ * @see `org.apache.kafka.streams.kstream.KStream#outerJoin`
+ */
+ def outerJoin[VO, VR](otherStream: KStream[K, VO],
+ joiner: (V, VO) => VR,
+ windows: JoinWindows)(implicit joined: Joined[K, V, VO]): KStream[K, VR] =
+ inner.outerJoin[VO, VR](otherStream.inner, joiner.asValueJoiner, windows, joined)
+
+ /**
+ * Merge this stream and the given stream into one larger stream.
+ *
+ * There is no ordering guarantee between records from this `KStream` and records from the provided `KStream`
+ * in the merged stream. Relative order is preserved within each input stream though (ie, records within
+ * one input stream are processed in order).
+ *
+ * @param stream a stream which is to be merged into this stream
+ * @return a merged stream containing all records from this and the provided [[KStream]]
+ * @see `org.apache.kafka.streams.kstream.KStream#merge`
+ */
+ def merge(stream: KStream[K, V]): KStream[K, V] = inner.merge(stream.inner)
+
+ /**
+ * Perform an action on each record of {@code KStream}.
+ *
+ * Peek is a non-terminal operation that triggers a side effect (such as logging or statistics collection)
+ * and returns an unchanged stream.
+ *
+ * @param action an action to perform on each record
+ * @see `org.apache.kafka.streams.kstream.KStream#peek`
+ */
+ def peek(action: (K, V) => Unit): KStream[K, V] = {
+ inner.peek((k: K, v: V) => action(k, v))
+ }
+}
diff --git a/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/KTable.scala b/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/KTable.scala
new file mode 100644
index 0000000000000..0369ee5aa589d
--- /dev/null
+++ b/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/KTable.scala
@@ -0,0 +1,292 @@
+/*
+ * Copyright (C) 2018 Lightbend Inc.
+ * Copyright (C) 2017-2018 Alexis Seigneurin.
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.scala
+package kstream
+
+import org.apache.kafka.streams.kstream.{KTable => KTableJ, _}
+import org.apache.kafka.streams.scala.ImplicitConversions._
+import org.apache.kafka.streams.scala.FunctionConversions._
+
+/**
+ * Wraps the Java class [[org.apache.kafka.streams.kstream.KTable]] and delegates method calls to the underlying Java object.
+ *
+ * @param [K] Type of keys
+ * @param [V] Type of values
+ * @param inner The underlying Java abstraction for KTable
+ *
+ * @see `org.apache.kafka.streams.kstream.KTable`
+ */
+class KTable[K, V](val inner: KTableJ[K, V]) {
+
+ /**
+ * Create a new [[KTable]] that consists all records of this [[KTable]] which satisfies the given
+ * predicate
+ *
+ * @param predicate a filter that is applied to each record
+ * @return a [[KTable]] that contains only those records that satisfy the given predicate
+ * @see `org.apache.kafka.streams.kstream.KTable#filter`
+ */
+ def filter(predicate: (K, V) => Boolean): KTable[K, V] = {
+ inner.filter(predicate(_, _))
+ }
+
+ /**
+ * Create a new [[KTable]] that consists all records of this [[KTable]] which satisfies the given
+ * predicate
+ *
+ * @param predicate a filter that is applied to each record
+ * @param materialized a `Materialized` that describes how the `StateStore` for the resulting [[KTable]]
+ * should be materialized.
+ * @return a [[KTable]] that contains only those records that satisfy the given predicate
+ * @see `org.apache.kafka.streams.kstream.KTable#filter`
+ */
+ def filter(predicate: (K, V) => Boolean,
+ materialized: Materialized[K, V, ByteArrayKeyValueStore]): KTable[K, V] = {
+ inner.filter(predicate.asPredicate, materialized)
+ }
+
+ /**
+ * Create a new [[KTable]] that consists all records of this [[KTable]] which do not satisfy the given
+ * predicate
+ *
+ * @param predicate a filter that is applied to each record
+ * @return a [[KTable]] that contains only those records that do not satisfy the given predicate
+ * @see `org.apache.kafka.streams.kstream.KTable#filterNot`
+ */
+ def filterNot(predicate: (K, V) => Boolean): KTable[K, V] = {
+ inner.filterNot(predicate(_, _))
+ }
+
+ /**
+ * Create a new [[KTable]] that consists all records of this [[KTable]] which do not satisfy the given
+ * predicate
+ *
+ * @param predicate a filter that is applied to each record
+ * @param materialized a `Materialized` that describes how the `StateStore` for the resulting [[KTable]]
+ * should be materialized.
+ * @return a [[KTable]] that contains only those records that do not satisfy the given predicate
+ * @see `org.apache.kafka.streams.kstream.KTable#filterNot`
+ */
+ def filterNot(predicate: (K, V) => Boolean,
+ materialized: Materialized[K, V, ByteArrayKeyValueStore]): KTable[K, V] = {
+ inner.filterNot(predicate.asPredicate, materialized)
+ }
+
+ /**
+ * Create a new [[KTable]] by transforming the value of each record in this [[KTable]] into a new value
+ * (with possible new type) in the new [[KTable]].
+ *
+ * The provided `mapper`, a function `V => VR` is applied to each input record value and computes a new value for it
+ *
+ * @param mapper, a function `V => VR` that computes a new output value
+ * @return a [[KTable]] that contains records with unmodified key and new values (possibly of different type)
+ * @see `org.apache.kafka.streams.kstream.KTable#mapValues`
+ */
+ def mapValues[VR](mapper: V => VR): KTable[K, VR] = {
+ inner.mapValues[VR](mapper.asValueMapper)
+ }
+
+ /**
+ * Create a new [[KTable]] by transforming the value of each record in this [[KTable]] into a new value
+ * (with possible new type) in the new [[KTable]].
+ *
+ * The provided `mapper`, a function `V => VR` is applied to each input record value and computes a new value for it
+ *
+ * @param mapper, a function `V => VR` that computes a new output value
+ * @param materialized a `Materialized` that describes how the `StateStore` for the resulting [[KTable]]
+ * should be materialized.
+ * @return a [[KTable]] that contains records with unmodified key and new values (possibly of different type)
+ * @see `org.apache.kafka.streams.kstream.KTable#mapValues`
+ */
+ def mapValues[VR](mapper: V => VR,
+ materialized: Materialized[K, VR, ByteArrayKeyValueStore]): KTable[K, VR] = {
+ inner.mapValues[VR](mapper.asValueMapper, materialized)
+ }
+
+ /**
+ * Create a new [[KTable]] by transforming the value of each record in this [[KTable]] into a new value
+ * (with possible new type) in the new [[KTable]].
+ *
+ * The provided `mapper`, a function `(K, V) => VR` is applied to each input record value and computes a new value for it
+ *
+ * @param mapper, a function `(K, V) => VR` that computes a new output value
+ * @return a [[KTable]] that contains records with unmodified key and new values (possibly of different type)
+ * @see `org.apache.kafka.streams.kstream.KTable#mapValues`
+ */
+ def mapValues[VR](mapper: (K, V) => VR): KTable[K, VR] = {
+ inner.mapValues[VR](mapper.asValueMapperWithKey)
+ }
+
+ /**
+ * Create a new [[KTable]] by transforming the value of each record in this [[KTable]] into a new value
+ * (with possible new type) in the new [[KTable]].
+ *
+ * The provided `mapper`, a function `(K, V) => VR` is applied to each input record value and computes a new value for it
+ *
+ * @param mapper, a function `(K, V) => VR` that computes a new output value
+ * @param materialized a `Materialized` that describes how the `StateStore` for the resulting [[KTable]]
+ * should be materialized.
+ * @return a [[KTable]] that contains records with unmodified key and new values (possibly of different type)
+ * @see `org.apache.kafka.streams.kstream.KTable#mapValues`
+ */
+ def mapValues[VR](mapper: (K, V) => VR,
+ materialized: Materialized[K, VR, ByteArrayKeyValueStore]): KTable[K, VR] = {
+ inner.mapValues[VR](mapper.asValueMapperWithKey)
+ }
+
+ /**
+ * Convert this changelog stream to a [[KStream]].
+ *
+ * @return a [[KStream]] that contains the same records as this [[KTable]]
+ * @see `org.apache.kafka.streams.kstream.KTable#toStream`
+ */
+ def toStream: KStream[K, V] = inner.toStream
+
+ /**
+ * Convert this changelog stream to a [[KStream]] using the given key/value mapper to select the new key
+ *
+ * @param mapper a function that computes a new key for each record
+ * @return a [[KStream]] that contains the same records as this [[KTable]]
+ * @see `org.apache.kafka.streams.kstream.KTable#toStream`
+ */
+ def toStream[KR](mapper: (K, V) => KR): KStream[KR, V] = {
+ inner.toStream[KR](mapper.asKeyValueMapper)
+ }
+
+ /**
+ * Re-groups the records of this [[KTable]] using the provided key/value mapper
+ * and `Serde`s as specified by `Serialized`.
+ *
+ * @param selector a function that computes a new grouping key and value to be aggregated
+ * @param serialized the `Serialized` instance used to specify `Serdes`
+ * @return a [[KGroupedTable]] that contains the re-grouped records of the original [[KTable]]
+ * @see `org.apache.kafka.streams.kstream.KTable#groupBy`
+ */
+ def groupBy[KR, VR](selector: (K, V) => (KR, VR))(implicit serialized: Serialized[KR, VR]): KGroupedTable[KR, VR] = {
+ inner.groupBy(selector.asKeyValueMapper, serialized)
+ }
+
+ /**
+ * Join records of this [[KTable]] with another [[KTable]]'s records using non-windowed inner equi join.
+ *
+ * @param other the other [[KTable]] to be joined with this [[KTable]]
+ * @param joiner a function that computes the join result for a pair of matching records
+ * @return a [[KTable]] that contains join-records for each key and values computed by the given joiner,
+ * one for each matched record-pair with the same key
+ * @see `org.apache.kafka.streams.kstream.KTable#join`
+ */
+ def join[VO, VR](other: KTable[K, VO],
+ joiner: (V, VO) => VR): KTable[K, VR] = {
+
+ inner.join[VO, VR](other.inner, joiner.asValueJoiner)
+ }
+
+ /**
+ * Join records of this [[KTable]] with another [[KTable]]'s records using non-windowed inner equi join.
+ *
+ * @param other the other [[KTable]] to be joined with this [[KTable]]
+ * @param joiner a function that computes the join result for a pair of matching records
+ * @param materialized a `Materialized` that describes how the `StateStore` for the resulting [[KTable]]
+ * should be materialized.
+ * @return a [[KTable]] that contains join-records for each key and values computed by the given joiner,
+ * one for each matched record-pair with the same key
+ * @see `org.apache.kafka.streams.kstream.KTable#join`
+ */
+ def join[VO, VR](other: KTable[K, VO],
+ joiner: (V, VO) => VR,
+ materialized: Materialized[K, VR, ByteArrayKeyValueStore]): KTable[K, VR] = {
+
+ inner.join[VO, VR](other.inner, joiner.asValueJoiner, materialized)
+ }
+
+ /**
+ * Join records of this [[KTable]] with another [[KTable]]'s records using non-windowed left equi join.
+ *
+ * @param other the other [[KTable]] to be joined with this [[KTable]]
+ * @param joiner a function that computes the join result for a pair of matching records
+ * @return a [[KTable]] that contains join-records for each key and values computed by the given joiner,
+ * one for each matched record-pair with the same key
+ * @see `org.apache.kafka.streams.kstream.KTable#leftJoin`
+ */
+ def leftJoin[VO, VR](other: KTable[K, VO],
+ joiner: (V, VO) => VR): KTable[K, VR] = {
+
+ inner.leftJoin[VO, VR](other.inner, joiner.asValueJoiner)
+ }
+
+ /**
+ * Join records of this [[KTable]] with another [[KTable]]'s records using non-windowed left equi join.
+ *
+ * @param other the other [[KTable]] to be joined with this [[KTable]]
+ * @param joiner a function that computes the join result for a pair of matching records
+ * @param materialized a `Materialized` that describes how the `StateStore` for the resulting [[KTable]]
+ * should be materialized.
+ * @return a [[KTable]] that contains join-records for each key and values computed by the given joiner,
+ * one for each matched record-pair with the same key
+ * @see `org.apache.kafka.streams.kstream.KTable#leftJoin`
+ */
+ def leftJoin[VO, VR](other: KTable[K, VO],
+ joiner: (V, VO) => VR,
+ materialized: Materialized[K, VR, ByteArrayKeyValueStore]): KTable[K, VR] = {
+
+ inner.leftJoin[VO, VR](other.inner, joiner.asValueJoiner, materialized)
+ }
+
+ /**
+ * Join records of this [[KTable]] with another [[KTable]]'s records using non-windowed outer equi join.
+ *
+ * @param other the other [[KTable]] to be joined with this [[KTable]]
+ * @param joiner a function that computes the join result for a pair of matching records
+ * @return a [[KTable]] that contains join-records for each key and values computed by the given joiner,
+ * one for each matched record-pair with the same key
+ * @see `org.apache.kafka.streams.kstream.KTable#leftJoin`
+ */
+ def outerJoin[VO, VR](other: KTable[K, VO],
+ joiner: (V, VO) => VR): KTable[K, VR] = {
+
+ inner.outerJoin[VO, VR](other.inner, joiner.asValueJoiner)
+ }
+
+ /**
+ * Join records of this [[KTable]] with another [[KTable]]'s records using non-windowed outer equi join.
+ *
+ * @param other the other [[KTable]] to be joined with this [[KTable]]
+ * @param joiner a function that computes the join result for a pair of matching records
+ * @param materialized a `Materialized` that describes how the `StateStore` for the resulting [[KTable]]
+ * should be materialized.
+ * @return a [[KTable]] that contains join-records for each key and values computed by the given joiner,
+ * one for each matched record-pair with the same key
+ * @see `org.apache.kafka.streams.kstream.KTable#leftJoin`
+ */
+ def outerJoin[VO, VR](other: KTable[K, VO],
+ joiner: (V, VO) => VR,
+ materialized: Materialized[K, VR, ByteArrayKeyValueStore]): KTable[K, VR] = {
+
+ inner.outerJoin[VO, VR](other.inner, joiner.asValueJoiner, materialized)
+ }
+
+ /**
+ * Get the name of the local state store used that can be used to query this [[KTable]].
+ *
+ * @return the underlying state store name, or `null` if this [[KTable]] cannot be queried.
+ */
+ def queryableStoreName: String =
+ inner.queryableStoreName
+}
diff --git a/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/SessionWindowedKStream.scala b/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/SessionWindowedKStream.scala
new file mode 100644
index 0000000000000..7e9fa0713fe29
--- /dev/null
+++ b/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/SessionWindowedKStream.scala
@@ -0,0 +1,125 @@
+/*
+ * Copyright (C) 2018 Lightbend Inc.
+ * Copyright (C) 2017-2018 Alexis Seigneurin.
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.scala
+package kstream
+
+import org.apache.kafka.streams.kstream.{SessionWindowedKStream => SessionWindowedKStreamJ, _}
+import org.apache.kafka.streams.state.SessionStore
+import org.apache.kafka.common.utils.Bytes
+
+import org.apache.kafka.streams.scala.ImplicitConversions._
+import org.apache.kafka.streams.scala.FunctionConversions._
+
+/**
+ * Wraps the Java class SessionWindowedKStream and delegates method calls to the underlying Java object.
+ *
+ * @param [K] Type of keys
+ * @param [V] Type of values
+ * @param inner The underlying Java abstraction for SessionWindowedKStream
+ *
+ * @see `org.apache.kafka.streams.kstream.SessionWindowedKStream`
+ */
+class SessionWindowedKStream[K, V](val inner: SessionWindowedKStreamJ[K, V]) {
+
+ /**
+ * Aggregate the values of records in this stream by the grouped key and defined `SessionWindows`.
+ *
+ * @param initializer the initializer function
+ * @param aggregator the aggregator function
+ * @param sessionMerger the merger function
+ * @return a windowed [[KTable]] that contains "update" records with unmodified keys, and values that represent
+ * the latest (rolling) aggregate for each key within a window
+ * @see `org.apache.kafka.streams.kstream.SessionWindowedKStream#aggregate`
+ */
+ def aggregate[VR](initializer: () => VR,
+ aggregator: (K, V, VR) => VR,
+ merger: (K, VR, VR) => VR): KTable[Windowed[K], VR] = {
+
+ inner.aggregate(initializer.asInitializer, aggregator.asAggregator, merger.asMerger)
+ }
+
+ /**
+ * Aggregate the values of records in this stream by the grouped key and defined `SessionWindows`.
+ *
+ * @param initializer the initializer function
+ * @param aggregator the aggregator function
+ * @param sessionMerger the merger function
+ * @param materialized an instance of `Materialized` used to materialize a state store.
+ * @return a windowed [[KTable]] that contains "update" records with unmodified keys, and values that represent
+ * the latest (rolling) aggregate for each key within a window
+ * @see `org.apache.kafka.streams.kstream.SessionWindowedKStream#aggregate`
+ */
+ def aggregate[VR](initializer: () => VR,
+ aggregator: (K, V, VR) => VR,
+ merger: (K, VR, VR) => VR,
+ materialized: Materialized[K, VR, SessionStore[Bytes, Array[Byte]]]): KTable[Windowed[K], VR] = {
+
+ inner.aggregate(initializer.asInitializer, aggregator.asAggregator, merger.asMerger, materialized)
+ }
+
+ /**
+ * Count the number of records in this stream by the grouped key into `SessionWindows`.
+ *
+ * @return a windowed [[KTable]] that contains "update" records with unmodified keys and `Long` values
+ * that represent the latest (rolling) count (i.e., number of records) for each key within a window
+ * @see `org.apache.kafka.streams.kstream.SessionWindowedKStream#count`
+ */
+ def count(): KTable[Windowed[K], Long] = {
+ val c: KTable[Windowed[K], java.lang.Long] = inner.count()
+ c.mapValues[Long](Long2long(_))
+ }
+
+ /**
+ * Count the number of records in this stream by the grouped key into `SessionWindows`.
+ *
+ * @param materialized an instance of `Materialized` used to materialize a state store.
+ * @return a windowed [[KTable]] that contains "update" records with unmodified keys and `Long` values
+ * that represent the latest (rolling) count (i.e., number of records) for each key within a window
+ * @see `org.apache.kafka.streams.kstream.SessionWindowedKStream#count`
+ */
+ def count(materialized: Materialized[K, Long, SessionStore[Bytes, Array[Byte]]]): KTable[Windowed[K], Long] =
+ inner.count(materialized)
+
+ /**
+ * Combine values of this stream by the grouped key into {@link SessionWindows}.
+ *
+ * @param reducer a reducer function that computes a new aggregate result.
+ * @return a windowed [[KTable]] that contains "update" records with unmodified keys, and values that represent
+ * the latest (rolling) aggregate for each key within a window
+ * @see `org.apache.kafka.streams.kstream.SessionWindowedKStream#reduce`
+ */
+ def reduce(reducer: (V, V) => V): KTable[Windowed[K], V] = {
+ inner.reduce((v1, v2) => reducer(v1, v2))
+ }
+
+ /**
+ * Combine values of this stream by the grouped key into {@link SessionWindows}.
+ *
+ * @param reducer a reducer function that computes a new aggregate result.
+ * @param materialized an instance of `Materialized` used to materialize a state store.
+ * @return a windowed [[KTable]] that contains "update" records with unmodified keys, and values that represent
+ * the latest (rolling) aggregate for each key within a window
+ * @see `org.apache.kafka.streams.kstream.SessionWindowedKStream#reduce`
+ */
+ def reduce(reducer: (V, V) => V,
+ materialized: Materialized[K, V, SessionStore[Bytes, Array[Byte]]]): KTable[Windowed[K], V] = {
+ inner.reduce(reducer.asReducer, materialized)
+ }
+}
diff --git a/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/TimeWindowedKStream.scala b/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/TimeWindowedKStream.scala
new file mode 100644
index 0000000000000..1aa19786d5ab2
--- /dev/null
+++ b/streams/streams-scala/src/main/scala/org/apache/kafka/streams/scala/kstream/TimeWindowedKStream.scala
@@ -0,0 +1,125 @@
+/*
+ * Copyright (C) 2018 Lightbend Inc.
+ * Copyright (C) 2017-2018 Alexis Seigneurin.
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.scala
+package kstream
+
+import org.apache.kafka.streams.kstream.{TimeWindowedKStream => TimeWindowedKStreamJ, _}
+import org.apache.kafka.streams.state.WindowStore
+import org.apache.kafka.common.utils.Bytes
+import org.apache.kafka.common.serialization.Serde
+import org.apache.kafka.streams.scala.ImplicitConversions._
+import org.apache.kafka.streams.scala.FunctionConversions._
+
+/**
+ * Wraps the Java class TimeWindowedKStream and delegates method calls to the underlying Java object.
+ *
+ * @param [K] Type of keys
+ * @param [V] Type of values
+ * @param inner The underlying Java abstraction for TimeWindowedKStream
+ *
+ * @see `org.apache.kafka.streams.kstream.TimeWindowedKStream`
+ */
+class TimeWindowedKStream[K, V](val inner: TimeWindowedKStreamJ[K, V]) {
+
+ /**
+ * Aggregate the values of records in this stream by the grouped key.
+ *
+ * @param initializer an initializer function that computes an initial intermediate aggregation result
+ * @param aggregator an aggregator function that computes a new aggregate result
+ * @return a [[KTable]] that contains "update" records with unmodified keys, and values that represent the
+ * latest (rolling) aggregate for each key
+ * @see `org.apache.kafka.streams.kstream.TimeWindowedKStream#aggregate`
+ */
+ def aggregate[VR](initializer: () => VR,
+ aggregator: (K, V, VR) => VR): KTable[Windowed[K], VR] = {
+
+ inner.aggregate(initializer.asInitializer, aggregator.asAggregator)
+ }
+
+ /**
+ * Aggregate the values of records in this stream by the grouped key.
+ *
+ * @param initializer an initializer function that computes an initial intermediate aggregation result
+ * @param aggregator an aggregator function that computes a new aggregate result
+ * @param materialized an instance of `Materialized` used to materialize a state store.
+ * @return a [[KTable]] that contains "update" records with unmodified keys, and values that represent the
+ * latest (rolling) aggregate for each key
+ * @see `org.apache.kafka.streams.kstream.TimeWindowedKStream#aggregate`
+ */
+ def aggregate[VR](initializer: () => VR,
+ aggregator: (K, V, VR) => VR,
+ materialized: Materialized[K, VR, WindowStore[Bytes, Array[Byte]]]): KTable[Windowed[K], VR] = {
+
+ inner.aggregate(initializer.asInitializer, aggregator.asAggregator, materialized)
+ }
+
+ /**
+ * Count the number of records in this stream by the grouped key and the defined windows.
+ *
+ * @return a [[KTable]] that contains "update" records with unmodified keys and `Long` values that
+ * represent the latest (rolling) count (i.e., number of records) for each key
+ * @see `org.apache.kafka.streams.kstream.TimeWindowedKStream#count`
+ */
+ def count(): KTable[Windowed[K], Long] = {
+ val c: KTable[Windowed[K], java.lang.Long] = inner.count()
+ c.mapValues[Long](Long2long(_))
+ }
+
+ /**
+ * Count the number of records in this stream by the grouped key and the defined windows.
+ *
+ * @param materialized an instance of `Materialized` used to materialize a state store.
+ * @return a [[KTable]] that contains "update" records with unmodified keys and `Long` values that
+ * represent the latest (rolling) count (i.e., number of records) for each key
+ * @see `org.apache.kafka.streams.kstream.TimeWindowedKStream#count`
+ */
+ def count(materialized: Materialized[K, Long, WindowStore[Bytes, Array[Byte]]]): KTable[Windowed[K], Long] = {
+ val c: KTable[Windowed[K], java.lang.Long] =
+ inner.count(materialized.asInstanceOf[Materialized[K, java.lang.Long, WindowStore[Bytes, Array[Byte]]]])
+ c.mapValues[Long](Long2long(_))
+ }
+
+ /**
+ * Combine the values of records in this stream by the grouped key.
+ *
+ * @param reducer a function that computes a new aggregate result
+ * @return a [[KTable]] that contains "update" records with unmodified keys, and values that represent the
+ * latest (rolling) aggregate for each key
+ * @see `org.apache.kafka.streams.kstream.TimeWindowedKStream#reduce`
+ */
+ def reduce(reducer: (V, V) => V): KTable[Windowed[K], V] = {
+ inner.reduce(reducer.asReducer)
+ }
+
+ /**
+ * Combine the values of records in this stream by the grouped key.
+ *
+ * @param reducer a function that computes a new aggregate result
+ * @param materialized an instance of `Materialized` used to materialize a state store.
+ * @return a [[KTable]] that contains "update" records with unmodified keys, and values that represent the
+ * latest (rolling) aggregate for each key
+ * @see `org.apache.kafka.streams.kstream.TimeWindowedKStream#reduce`
+ */
+ def reduce(reducer: (V, V) => V,
+ materialized: Materialized[K, V, WindowStore[Bytes, Array[Byte]]]): KTable[Windowed[K], V] = {
+
+ inner.reduce(reducer.asReducer, materialized)
+ }
+}
diff --git a/streams/streams-scala/src/test/resources/log4j.properties b/streams/streams-scala/src/test/resources/log4j.properties
new file mode 100644
index 0000000000000..93ffc165654a2
--- /dev/null
+++ b/streams/streams-scala/src/test/resources/log4j.properties
@@ -0,0 +1,34 @@
+# Copyright (C) 2018 Lightbend Inc.
+# Copyright (C) 2017-2018 Alexis Seigneurin.
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# Set root logger level to DEBUG and its only appender to A1.
+log4j.rootLogger=INFO, R
+
+# A1 is set to be a ConsoleAppender.
+log4j.appender.A1=org.apache.log4j.ConsoleAppender
+
+log4j.appender.R=org.apache.log4j.RollingFileAppender
+log4j.appender.R.File=logs/kafka-streams-scala.log
+
+log4j.appender.R.MaxFileSize=100KB
+# Keep one backup file
+log4j.appender.R.MaxBackupIndex=1
+
+# A1 uses PatternLayout.
+log4j.appender.R.layout=org.apache.log4j.PatternLayout
+log4j.appender.R.layout.ConversionPattern=%-4r [%t] %-5p %c %x - %m%n
diff --git a/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala b/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
new file mode 100644
index 0000000000000..24974c40f7283
--- /dev/null
+++ b/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
@@ -0,0 +1,237 @@
+/*
+ * Copyright (C) 2018 Lightbend Inc.
+ * Copyright (C) 2017-2018 Alexis Seigneurin.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.scala
+
+import java.util.Properties
+
+import org.scalatest.junit.JUnitSuite
+import org.junit.Assert._
+import org.junit.rules.TemporaryFolder
+import org.junit._
+
+import org.apache.kafka.streams.integration.utils.{EmbeddedKafkaCluster, IntegrationTestUtils}
+import org.apache.kafka.clients.consumer.ConsumerConfig
+import org.apache.kafka.clients.producer.ProducerConfig
+
+import org.apache.kafka.common.serialization._
+import org.apache.kafka.common.utils.MockTime
+import org.apache.kafka.test.TestUtils
+import org.apache.kafka.streams._
+import org.apache.kafka.streams.scala.kstream._
+
+import ImplicitConversions._
+import com.typesafe.scalalogging.LazyLogging
+
+/**
+ * Test suite that does an example to demonstrate stream-table joins in Kafka Streams
+ *
+ * The suite contains the test case using Scala APIs `testShouldCountClicksPerRegion` and the same test case using the
+ * Java APIs `testShouldCountClicksPerRegionJava`. The idea is to demonstrate that both generate the same result.
+ *
+ * Note: In the current project settings SAM type conversion is turned off as it's experimental in Scala 2.11.
+ * Hence the native Java API based version is more verbose.
+ */
+class StreamToTableJoinScalaIntegrationTestImplicitSerdes extends JUnitSuite
+ with StreamToTableJoinTestData with LazyLogging {
+
+ private val privateCluster: EmbeddedKafkaCluster = new EmbeddedKafkaCluster(1)
+
+ @Rule def cluster: EmbeddedKafkaCluster = privateCluster
+
+ final val alignedTime = (System.currentTimeMillis() / 1000 + 1) * 1000
+ val mockTime: MockTime = cluster.time
+ mockTime.setCurrentTimeMs(alignedTime)
+
+ val tFolder: TemporaryFolder = new TemporaryFolder(TestUtils.tempDirectory())
+ @Rule def testFolder: TemporaryFolder = tFolder
+
+ @Before
+ def startKafkaCluster(): Unit = {
+ cluster.createTopic(userClicksTopic)
+ cluster.createTopic(userRegionsTopic)
+ cluster.createTopic(outputTopic)
+ cluster.createTopic(userClicksTopicJ)
+ cluster.createTopic(userRegionsTopicJ)
+ cluster.createTopic(outputTopicJ)
+ }
+
+ @Test def testShouldCountClicksPerRegion(): Unit = {
+
+ // 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._
+
+ val streamsConfiguration: Properties = getStreamsConfiguration()
+
+ val builder = new StreamsBuilder()
+
+ val userClicksStream: KStream[String, Long] = builder.stream(userClicksTopic)
+
+ val userRegionsTable: KTable[String, String] = builder.table(userRegionsTopic)
+
+ // Compute the total per region by summing the individual click counts per region.
+ val clicksPerRegion: KTable[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 -> to ->
+ .map((_, regionWithClicks) => regionWithClicks)
+
+ // Compute the total per region by summing the individual click counts per region.
+ .groupByKey
+ .reduce(_ + _)
+
+ // Write the (continuously updating) results to the output topic.
+ clicksPerRegion.toStream.to(outputTopic)
+
+ val streams: KafkaStreams = new KafkaStreams(builder.build(), streamsConfiguration)
+ streams.start()
+
+
+ val actualClicksPerRegion: java.util.List[KeyValue[String, Long]] =
+ produceNConsume(userClicksTopic, userRegionsTopic, outputTopic)
+
+ streams.close()
+
+ import collection.JavaConverters._
+ assertEquals(actualClicksPerRegion.asScala.sortBy(_.key), expectedClicksPerRegion.sortBy(_.key))
+ }
+
+ @Test def testShouldCountClicksPerRegionJava(): Unit = {
+
+ import org.apache.kafka.streams.{KafkaStreams => KafkaStreamsJ, StreamsBuilder => StreamsBuilderJ}
+ import org.apache.kafka.streams.kstream.{KTable => KTableJ, KStream => KStreamJ, KGroupedStream => KGroupedStreamJ, _}
+ import collection.JavaConverters._
+ import java.lang.{Long => JLong}
+
+ val streamsConfiguration: Properties = getStreamsConfiguration()
+
+ 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 builder: StreamsBuilderJ = new StreamsBuilderJ()
+
+ val userClicksStream: KStreamJ[String, JLong] =
+ builder.stream[String, JLong](userClicksTopicJ, Consumed.`with`(Serdes.String(), Serdes.Long()))
+
+ val userRegionsTable: KTableJ[String, String] =
+ builder.table[String, String](userRegionsTopicJ, Consumed.`with`(Serdes.String(), Serdes.String()))
+
+ // Join the stream against the table.
+ val userClicksJoinRegion: KStreamJ[String, (String, JLong)] = userClicksStream
+ .leftJoin(userRegionsTable,
+ new ValueJoiner[JLong, String, (String, JLong)] {
+ def apply(clicks: JLong, region: String): (String, JLong) =
+ (if (region == null) "UNKNOWN" else region, clicks)
+ },
+ Joined.`with`[String, JLong, String](Serdes.String(), Serdes.Long(), Serdes.String()))
+
+ // Change the stream from -> to ->
+ val clicksByRegion : KStreamJ[String, JLong] = userClicksJoinRegion
+ .map {
+ new KeyValueMapper[String, (String, JLong), KeyValue[String, JLong]] {
+ def apply(k: String, regionWithClicks: (String, JLong)) = new KeyValue[String, JLong](regionWithClicks._1, regionWithClicks._2)
+ }
+ }
+
+ // Compute the total per region by summing the individual click counts per region.
+ val clicksPerRegion: KTableJ[String, JLong] = clicksByRegion
+ .groupByKey(Serialized.`with`(Serdes.String(), Serdes.Long()))
+ .reduce {
+ new Reducer[JLong] {
+ def apply(v1: JLong, v2: JLong) = v1 + v2
+ }
+ }
+
+ // Write the (continuously updating) results to the output topic.
+ clicksPerRegion.toStream.to(outputTopicJ, Produced.`with`(Serdes.String(), Serdes.Long()))
+
+ val streams: KafkaStreamsJ = new KafkaStreamsJ(builder.build(), streamsConfiguration)
+
+ streams.start()
+
+ val actualClicksPerRegion: java.util.List[KeyValue[String, Long]] =
+ produceNConsume(userClicksTopicJ, userRegionsTopicJ, outputTopicJ)
+
+ streams.close()
+ assertEquals(actualClicksPerRegion.asScala.sortBy(_.key), expectedClicksPerRegion.sortBy(_.key))
+ }
+
+ private def getStreamsConfiguration(): Properties = {
+ val streamsConfiguration: Properties = new Properties()
+
+ streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "stream-table-join-scala-integration-test")
+ streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers())
+ streamsConfiguration.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, "10000")
+ streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
+ streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, testFolder.getRoot().getPath())
+
+ streamsConfiguration
+ }
+
+ private def getUserRegionsProducerConfig(): Properties = {
+ val p = new Properties()
+ p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers())
+ p.put(ProducerConfig.ACKS_CONFIG, "all")
+ p.put(ProducerConfig.RETRIES_CONFIG, "0")
+ p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])
+ p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])
+ p
+ }
+
+ private def getUserClicksProducerConfig(): Properties = {
+ val p = new Properties()
+ p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers())
+ p.put(ProducerConfig.ACKS_CONFIG, "all")
+ p.put(ProducerConfig.RETRIES_CONFIG, "0")
+ p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])
+ p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[LongSerializer])
+ p
+ }
+
+ private def getConsumerConfig(): Properties = {
+ val p = new Properties()
+ p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers())
+ p.put(ConsumerConfig.GROUP_ID_CONFIG, "join-scala-integration-test-standard-consumer")
+ p.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
+ p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, classOf[StringDeserializer])
+ p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, classOf[LongDeserializer])
+ p
+ }
+
+ private def produceNConsume(userClicksTopic: String, userRegionsTopic: String, outputTopic: String): java.util.List[KeyValue[String, Long]] = {
+
+ import collection.JavaConverters._
+
+ // Publish user-region information.
+ val userRegionsProducerConfig: Properties = getUserRegionsProducerConfig()
+ IntegrationTestUtils.produceKeyValuesSynchronously(userRegionsTopic, userRegions.asJava, userRegionsProducerConfig, mockTime, false)
+
+ // Publish user-click information.
+ val userClicksProducerConfig: Properties = getUserClicksProducerConfig()
+ IntegrationTestUtils.produceKeyValuesSynchronously(userClicksTopic, userClicks.asJava, userClicksProducerConfig, mockTime, false)
+
+ // consume and verify result
+ val consumerConfig = getConsumerConfig()
+
+ IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(consumerConfig, outputTopic, expectedClicksPerRegion.size)
+ }
+}
+
diff --git a/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinTestData.scala b/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinTestData.scala
new file mode 100644
index 0000000000000..45715a7abe6a5
--- /dev/null
+++ b/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinTestData.scala
@@ -0,0 +1,61 @@
+/*
+ * Copyright (C) 2018 Lightbend Inc.
+ * Copyright (C) 2017-2018 Alexis Seigneurin.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.scala
+
+import org.apache.kafka.streams.KeyValue
+
+trait StreamToTableJoinTestData {
+ val brokers = "localhost:9092"
+
+ val userClicksTopic = s"user-clicks"
+ val userRegionsTopic = s"user-regions"
+ val outputTopic = s"output-topic"
+
+ val userClicksTopicJ = s"user-clicks-j"
+ val userRegionsTopicJ = s"user-regions-j"
+ val outputTopicJ = s"output-topic-j"
+
+ // Input 1: Clicks per user (multiple records allowed per user).
+ val userClicks: Seq[KeyValue[String, Long]] = Seq(
+ new KeyValue("alice", 13L),
+ new KeyValue("bob", 4L),
+ new KeyValue("chao", 25L),
+ new KeyValue("bob", 19L),
+ new KeyValue("dave", 56L),
+ new KeyValue("eve", 78L),
+ new KeyValue("alice", 40L),
+ new KeyValue("fang", 99L)
+ )
+
+ // Input 2: Region per user (multiple records allowed per user).
+ val userRegions: Seq[KeyValue[String, String]] = Seq(
+ new KeyValue("alice", "asia"), /* Alice lived in Asia originally... */
+ new KeyValue("bob", "americas"),
+ new KeyValue("chao", "asia"),
+ new KeyValue("dave", "europe"),
+ new KeyValue("alice", "europe"), /* ...but moved to Europe some time later. */
+ new KeyValue("eve", "americas"),
+ new KeyValue("fang", "asia")
+ )
+
+ val expectedClicksPerRegion: Seq[KeyValue[String, Long]] = Seq(
+ new KeyValue("americas", 101L),
+ new KeyValue("europe", 109L),
+ new KeyValue("asia", 124L)
+ )
+}
+
diff --git a/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/TopologyTest.scala b/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/TopologyTest.scala
new file mode 100644
index 0000000000000..89b2935b9566e
--- /dev/null
+++ b/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/TopologyTest.scala
@@ -0,0 +1,199 @@
+/*
+ * Copyright (C) 2018 Lightbend Inc.
+ * Copyright (C) 2017-2018 Alexis Seigneurin.
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.scala
+
+import java.util.Properties
+import java.util.regex.Pattern
+
+import org.scalatest.junit.JUnitSuite
+import org.junit.Assert._
+import org.junit._
+
+import org.apache.kafka.streams.scala.kstream._
+
+import org.apache.kafka.common.serialization._
+
+import ImplicitConversions._
+import com.typesafe.scalalogging.LazyLogging
+
+import org.apache.kafka.streams.{KafkaStreams => KafkaStreamsJ, StreamsBuilder => StreamsBuilderJ, _}
+import org.apache.kafka.streams.kstream.{KTable => KTableJ, KStream => KStreamJ, KGroupedStream => KGroupedStreamJ, _}
+import collection.JavaConverters._
+
+/**
+ * Test suite that verifies that the topology built by the Java and Scala APIs match.
+ */
+class TopologyTest extends JUnitSuite with LazyLogging {
+
+ val inputTopic = "input-topic"
+ val userClicksTopic = "user-clicks-topic"
+ val userRegionsTopic = "user-regions-topic"
+
+ val pattern = Pattern.compile("\\W+", Pattern.UNICODE_CHARACTER_CLASS)
+
+ @Test def shouldBuildIdenticalTopologyInJavaNScalaSimple() = {
+
+ // build the Scala topology
+ def getTopologyScala(): TopologyDescription = {
+
+ import DefaultSerdes._
+ import collection.JavaConverters._
+
+ val streamBuilder = new StreamsBuilder
+ val textLines = streamBuilder.stream[String, String](inputTopic)
+
+ val _: KStream[String, String] =
+ textLines.flatMapValues(v => pattern.split(v.toLowerCase))
+
+ streamBuilder.build().describe()
+ }
+
+ // build the Java topology
+ def getTopologyJava(): TopologyDescription = {
+
+ val streamBuilder = new StreamsBuilderJ
+ val textLines = streamBuilder.stream[String, String](inputTopic)
+
+ val _: KStreamJ[String, String] = textLines.flatMapValues {
+ new ValueMapper[String, java.lang.Iterable[String]] {
+ def apply(s: String): java.lang.Iterable[String] = pattern.split(s.toLowerCase).toIterable.asJava
+ }
+ }
+ streamBuilder.build().describe()
+ }
+
+ // should match
+ assertEquals(getTopologyScala(), getTopologyJava())
+ }
+
+ @Test def shouldBuildIdenticalTopologyInJavaNScalaAggregate() = {
+
+ // build the Scala topology
+ def getTopologyScala(): TopologyDescription = {
+
+ import DefaultSerdes._
+ import collection.JavaConverters._
+
+ val streamBuilder = new StreamsBuilder
+ val textLines = streamBuilder.stream[String, String](inputTopic)
+
+ val _: KTable[String, Long] =
+ textLines.flatMapValues(v => pattern.split(v.toLowerCase))
+ .groupBy((k, v) => v)
+ .count()
+
+ streamBuilder.build().describe()
+ }
+
+ // build the Java topology
+ def getTopologyJava(): TopologyDescription = {
+
+ val streamBuilder = new StreamsBuilderJ
+ val textLines: KStreamJ[String, String] = streamBuilder.stream[String, String](inputTopic)
+
+ val splits: KStreamJ[String, String] = textLines.flatMapValues {
+ new ValueMapper[String, java.lang.Iterable[String]] {
+ def apply(s: String): java.lang.Iterable[String] = pattern.split(s.toLowerCase).toIterable.asJava
+ }
+ }
+
+ val grouped: KGroupedStreamJ[String, String] = splits.groupBy {
+ new KeyValueMapper[String, String, String] {
+ def apply(k: String, v: String): String = v
+ }
+ }
+
+ val wordCounts: KTableJ[String, java.lang.Long] = grouped.count()
+
+ streamBuilder.build().describe()
+ }
+
+ // should match
+ assertEquals(getTopologyScala(), getTopologyJava())
+ }
+
+ @Test def shouldBuildIdenticalTopologyInJavaNScalaJoin() = {
+
+ // build the Scala topology
+ def getTopologyScala(): TopologyDescription = {
+ import DefaultSerdes._
+
+ val builder = new StreamsBuilder()
+
+ val userClicksStream: KStream[String, Long] = builder.stream(userClicksTopic)
+
+ val userRegionsTable: KTable[String, String] = builder.table(userRegionsTopic)
+
+ val clicksPerRegion: KTable[String, Long] =
+ userClicksStream
+ .leftJoin(userRegionsTable, (clicks: Long, region: String) => (if (region == null) "UNKNOWN" else region, clicks))
+ .map((_, regionWithClicks) => regionWithClicks)
+ .groupByKey
+ .reduce(_ + _)
+
+ builder.build().describe()
+ }
+
+ // build the Java topology
+ def getTopologyJava(): TopologyDescription = {
+
+ import java.lang.{Long => JLong}
+
+ val builder: StreamsBuilderJ = new StreamsBuilderJ()
+
+ val userClicksStream: KStreamJ[String, JLong] =
+ builder.stream[String, JLong](userClicksTopic, Consumed.`with`(Serdes.String(), Serdes.Long()))
+
+ val userRegionsTable: KTableJ[String, String] =
+ builder.table[String, String](userRegionsTopic, Consumed.`with`(Serdes.String(), Serdes.String()))
+
+ // Join the stream against the table.
+ val userClicksJoinRegion: KStreamJ[String, (String, JLong)] = userClicksStream
+ .leftJoin(userRegionsTable,
+ new ValueJoiner[JLong, String, (String, JLong)] {
+ def apply(clicks: JLong, region: String): (String, JLong) =
+ (if (region == null) "UNKNOWN" else region, clicks)
+ },
+ Joined.`with`[String, JLong, String](Serdes.String(), Serdes.Long(), Serdes.String()))
+
+ // Change the stream from -> to ->
+ val clicksByRegion : KStreamJ[String, JLong] = userClicksJoinRegion
+ .map {
+ new KeyValueMapper[String, (String, JLong), KeyValue[String, JLong]] {
+ def apply(k: String, regionWithClicks: (String, JLong)) = new KeyValue[String, JLong](regionWithClicks._1, regionWithClicks._2)
+ }
+ }
+
+ // Compute the total per region by summing the individual click counts per region.
+ val clicksPerRegion: KTableJ[String, JLong] = clicksByRegion
+ .groupByKey(Serialized.`with`(Serdes.String(), Serdes.Long()))
+ .reduce {
+ new Reducer[JLong] {
+ def apply(v1: JLong, v2: JLong) = v1 + v2
+ }
+ }
+
+ builder.build().describe()
+ }
+
+ // should match
+ assertEquals(getTopologyScala(), getTopologyJava())
+ }
+}
diff --git a/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/WordCountTest.scala b/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/WordCountTest.scala
new file mode 100644
index 0000000000000..f71e0cb4a1374
--- /dev/null
+++ b/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/WordCountTest.scala
@@ -0,0 +1,223 @@
+/*
+ * Copyright (C) 2018 Lightbend Inc.
+ * Copyright (C) 2017-2018 Alexis Seigneurin.
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.scala
+
+import java.util.Properties
+import java.util.regex.Pattern
+
+import org.scalatest.junit.JUnitSuite
+import org.junit.Assert._
+import org.junit._
+import org.junit.rules.TemporaryFolder
+
+import org.apache.kafka.streams.KeyValue
+import org.apache.kafka.streams._
+import org.apache.kafka.streams.scala.kstream._
+
+import org.apache.kafka.streams.integration.utils.{EmbeddedKafkaCluster, IntegrationTestUtils}
+import org.apache.kafka.clients.consumer.ConsumerConfig
+import org.apache.kafka.clients.producer.ProducerConfig
+
+import org.apache.kafka.common.serialization._
+import org.apache.kafka.common.utils.MockTime
+import org.apache.kafka.test.TestUtils
+
+import ImplicitConversions._
+import com.typesafe.scalalogging.LazyLogging
+
+/**
+ * Test suite that does a classic word count example.
+ *
+ * The suite contains the test case using Scala APIs `testShouldCountWords` and the same test case using the
+ * Java APIs `testShouldCountWordsJava`. The idea is to demonstrate that both generate the same result.
+ *
+ * Note: In the current project settings SAM type conversion is turned off as it's experimental in Scala 2.11.
+ * Hence the native Java API based version is more verbose.
+ */
+class WordCountTest extends JUnitSuite with WordCountTestData with LazyLogging {
+
+ private val privateCluster: EmbeddedKafkaCluster = new EmbeddedKafkaCluster(1)
+
+ @Rule def cluster: EmbeddedKafkaCluster = privateCluster
+
+ final val alignedTime = (System.currentTimeMillis() / 1000 + 1) * 1000
+ val mockTime: MockTime = cluster.time
+ mockTime.setCurrentTimeMs(alignedTime)
+
+
+ val tFolder: TemporaryFolder = new TemporaryFolder(TestUtils.tempDirectory())
+ @Rule def testFolder: TemporaryFolder = tFolder
+
+
+ @Before
+ def startKafkaCluster(): Unit = {
+ cluster.createTopic(inputTopic)
+ cluster.createTopic(outputTopic)
+ cluster.createTopic(inputTopicJ)
+ cluster.createTopic(outputTopicJ)
+ }
+
+ @Test def testShouldCountWords(): Unit = {
+
+ import DefaultSerdes._
+
+ val streamsConfiguration = getStreamsConfiguration()
+
+ val streamBuilder = new StreamsBuilder
+ val textLines = streamBuilder.stream[String, String](inputTopic)
+
+ val pattern = Pattern.compile("\\W+", Pattern.UNICODE_CHARACTER_CLASS)
+
+ // generate word counts
+ val wordCounts: KTable[String, Long] =
+ textLines.flatMapValues(v => pattern.split(v.toLowerCase))
+ .groupBy((k, v) => v)
+ .count()
+
+ // write to output topic
+ wordCounts.toStream.to(outputTopic)
+
+ val streams: KafkaStreams = new KafkaStreams(streamBuilder.build(), streamsConfiguration)
+ streams.start()
+
+ // produce and consume synchronously
+ val actualWordCounts: java.util.List[KeyValue[String, Long]] = produceNConsume(inputTopic, outputTopic)
+
+ streams.close()
+
+ import collection.JavaConverters._
+ assertEquals(actualWordCounts.asScala.take(expectedWordCounts.size).sortBy(_.key), expectedWordCounts.sortBy(_.key))
+ }
+
+ @Test def testShouldCountWordsJava(): Unit = {
+
+ import org.apache.kafka.streams.{KafkaStreams => KafkaStreamsJ, StreamsBuilder => StreamsBuilderJ}
+ import org.apache.kafka.streams.kstream.{KTable => KTableJ, KStream => KStreamJ, KGroupedStream => KGroupedStreamJ, _}
+ import collection.JavaConverters._
+
+ val streamsConfiguration = getStreamsConfiguration()
+ 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 streamBuilder = new StreamsBuilderJ
+ val textLines: KStreamJ[String, String] = streamBuilder.stream[String, String](inputTopicJ)
+
+ val pattern = Pattern.compile("\\W+", Pattern.UNICODE_CHARACTER_CLASS)
+
+ val splits: KStreamJ[String, String] = textLines.flatMapValues {
+ new ValueMapper[String, java.lang.Iterable[String]] {
+ def apply(s: String): java.lang.Iterable[String] = pattern.split(s.toLowerCase).toIterable.asJava
+ }
+ }
+
+ val grouped: KGroupedStreamJ[String, String] = splits.groupBy {
+ new KeyValueMapper[String, String, String] {
+ def apply(k: String, v: String): String = v
+ }
+ }
+
+ val wordCounts: KTableJ[String, java.lang.Long] = grouped.count()
+
+ wordCounts.toStream.to(outputTopicJ, Produced.`with`(Serdes.String(), Serdes.Long()))
+
+ val streams: KafkaStreamsJ = new KafkaStreamsJ(streamBuilder.build(), streamsConfiguration)
+ streams.start()
+
+ val actualWordCounts: java.util.List[KeyValue[String, Long]] = produceNConsume(inputTopicJ, outputTopicJ)
+
+ streams.close()
+
+ assertEquals(actualWordCounts.asScala.take(expectedWordCounts.size).sortBy(_.key), expectedWordCounts.sortBy(_.key))
+ }
+
+ private def getStreamsConfiguration(): Properties = {
+ val streamsConfiguration: Properties = new Properties()
+
+ streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-test")
+ streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers())
+ streamsConfiguration.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, "10000")
+ streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
+ streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, testFolder.getRoot().getPath())
+ streamsConfiguration
+ }
+
+ private def getProducerConfig(): Properties = {
+ val p = new Properties()
+ p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers())
+ p.put(ProducerConfig.ACKS_CONFIG, "all")
+ p.put(ProducerConfig.RETRIES_CONFIG, "0")
+ p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])
+ p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])
+ p
+ }
+
+ private def getConsumerConfig(): Properties = {
+ val p = new Properties()
+ p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers())
+ p.put(ConsumerConfig.GROUP_ID_CONFIG, "wordcount-scala-integration-test-standard-consumer")
+ p.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
+ p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, classOf[StringDeserializer])
+ p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, classOf[LongDeserializer])
+ p
+ }
+
+ private def produceNConsume(inputTopic: String, outputTopic: String): java.util.List[KeyValue[String, Long]] = {
+
+ val linesProducerConfig: Properties = getProducerConfig()
+
+ import collection.JavaConverters._
+ IntegrationTestUtils.produceValuesSynchronously(inputTopic, inputValues.asJava, linesProducerConfig, mockTime)
+
+ val consumerConfig = getConsumerConfig()
+
+ IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(consumerConfig, outputTopic, expectedWordCounts.size)
+ }
+}
+
+trait WordCountTestData {
+ val inputTopic = s"inputTopic"
+ val outputTopic = s"outputTopic"
+ val inputTopicJ = s"inputTopicJ"
+ val outputTopicJ = s"outputTopicJ"
+
+ val inputValues = List(
+ "Hello Kafka Streams",
+ "All streams lead to Kafka",
+ "Join Kafka Summit",
+ "И теперь пошли русские слова"
+ )
+
+ val expectedWordCounts: List[KeyValue[String, Long]] = List(
+ new KeyValue("hello", 1L),
+ new KeyValue("all", 1L),
+ new KeyValue("streams", 2L),
+ new KeyValue("lead", 1L),
+ new KeyValue("to", 1L),
+ new KeyValue("join", 1L),
+ new KeyValue("kafka", 3L),
+ new KeyValue("summit", 1L),
+ new KeyValue("и", 1L),
+ new KeyValue("теперь", 1L),
+ new KeyValue("пошли", 1L),
+ new KeyValue("русские", 1L),
+ new KeyValue("слова", 1L)
+ )
+}
+