Skip to content

KAFKA-6670: Implement a Scala wrapper library for Kafka Streams #4756

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
debasishg wants to merge 23 commits into from
Closed

KAFKA-6670: Implement a Scala wrapper library for Kafka Streams #4756

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
23 commits
Select commit Hold shift + click to select a range
dc6181d
[KIP-270] Implementation of Scala Wrapper Library for kafka-streams
debasishg Mar 20, 2018
15a192c
Cleaned up license statement in code
debasishg Mar 22, 2018
6cc6ad1
Removed unnecessary dependencies from build.gradle and dependencies.g…
debasishg Mar 23, 2018
86937f3
Code review feedbacks
debasishg Mar 25, 2018
d46ae4f
More cde review feedbacks - added default serdes for TimeWindowedSerd…
debasishg Mar 26, 2018
a4f92cc
Code review feedback
seglo Mar 26, 2018
fb5dc99
Ignore logs dir from EmbeddedKafka tests
seglo Mar 27, 2018
78f363d
Remove experimental compiler flag and de-sugar SAM conversion sites
seglo Mar 27, 2018
39bc527
Alternate EmbeddedKafka config for WordCountTest
seglo Mar 27, 2018
4e2338d
1. Refactored tests, removed dependencies on embedded-kafka-scalatest
debasishg Mar 28, 2018
d6a9f51
1. Removed KStream#split
debasishg Mar 28, 2018
08c1f91
Fixed a copy/paste error in KStream#join
debasishg Mar 28, 2018
2e796aa
Started writing Scaladoc and added some missing APIs (new ones in 1.1)
debasishg Mar 30, 2018
dbeb4eb
Complete Scaladoc added
debasishg Apr 2, 2018
081a6e4
Added copyright and notice
debasishg Apr 3, 2018
236f0f1
Fixed a bug in KStream#merge - was givingStackOverflowError due to in…
debasishg Apr 3, 2018
ea9a765
Kafka Streams DSL for Scala user docs
seglo Apr 3, 2018
6df2cd3
Fix userdocs typos, formatting bugs, and PR feedback from Guozhang
seglo Apr 4, 2018
4cd3863
Excluded scala-streams/log from rat check
debasishg Apr 7, 2018
3add4c4
Added tests using the native Java APIs in Scala to check matching res…
debasishg Apr 10, 2018
453d4ab
Added TopologyTest which verifies that the topology constructed by th…
debasishg Apr 11, 2018
e91758e
Miscellaneous code review feedbacks
debasishg Apr 17, 2018
ee60ee5
Misc. userdocs updates
seglo Apr 19, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .gitignore
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,4 +50,4 @@ docs/generated/

kafkatest.egg-info/
systest/

*.swp
47 changes: 45 additions & 2 deletions build.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -124,7 +124,8 @@ if (new File('.git').exists()) {
'**/id_rsa',
'**/id_rsa.pub',
'checkstyle/suppressions.xml',
'streams/quickstart/java/src/test/resources/projects/basic/goal.txt'
'streams/quickstart/java/src/test/resources/projects/basic/goal.txt',
'streams/streams-scala/logs/*'
])
}
}
Expand Down Expand Up @@ -518,7 +519,7 @@ for ( sv in availableScalaVersions ) {
}

def connectPkgs = ['connect:api', 'connect:runtime', 'connect:transforms', 'connect:json', 'connect:file']
def pkgs = ['clients', 'examples', 'log4j-appender', 'tools', 'streams', 'streams:test-utils', 'streams:examples'] + connectPkgs
def pkgs = ['clients', 'examples', 'log4j-appender', 'tools', 'streams', 'streams:streams-scala', 'streams:test-utils', 'streams:examples'] + connectPkgs

/** Create one task per default Scala version */
def withDefScalaVersions(taskName) {
Expand Down Expand Up @@ -733,6 +734,8 @@ project(':core') {
from(project(':connect:file').configurations.runtime) { into("libs/") }
from(project(':streams').jar) { into("libs/") }
from(project(':streams').configurations.runtime) { into("libs/") }
from(project(':streams:streams-scala').jar) { into("libs/") }
from(project(':streams:streams-scala').configurations.runtime) { into("libs/") }
from(project(':streams:test-utils').jar) { into("libs/") }
from(project(':streams:test-utils').configurations.runtime) { into("libs/") }
from(project(':streams:examples').jar) { into("libs/") }
Expand Down Expand Up @@ -965,6 +968,46 @@ project(':streams') {
}
}

project(':streams:streams-scala') {
println "Building project 'streams-scala' with Scala version ${versions.scala}"
apply plugin: 'scala'
archivesBaseName = "kafka-streams-scala"

dependencies {
compile project(':streams')

compile libs.scalaLibrary

testCompile project(':core')
testCompile project(':core').sourceSets.test.output
testCompile project(':streams').sourceSets.test.output
testCompile project(':clients').sourceSets.test.output
testCompile libs.scalaLogging

testCompile libs.junit
testCompile libs.scalatest

testRuntime libs.slf4jlog4j
}

javadoc {
include "**/org/apache/kafka/streams/scala/**"
}

tasks.create(name: "copyDependantLibs", type: Copy) {
from (configurations.runtime) {
exclude('kafka-streams*')
}
into "$buildDir/dependant-libs-${versions.scala}"
duplicatesStrategy 'exclude'
}

jar {
dependsOn 'copyDependantLibs'
}

}

project(':streams:test-utils') {
archivesBaseName = "kafka-streams-test-utils"

Expand Down
13 changes: 13 additions & 0 deletions docs/api.html
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,19 @@ <h3><a id="streamsapi" href="#streamsapi">2.3 Streams API</a></h3>
&lt;/dependency&gt;
</pre>

<p>
When using Scala you may optionally include the <code>kafka-streams-scala</code> library. Additional documentation on using the Kafka Streams DSL for Scala is available <a href="/{{version}}/documentation/streams/developer-guide/dsl-api.html#scala-dsl">in the developer guide</a>.
<p>
To use Kafka Streams DSL for Scala for Scala 2.11 you can use the following maven dependency:

<pre class="brush: xml;">
&lt;dependency&gt;
&lt;groupId&gt;org.apache.kafka&lt;/groupId&gt;
&lt;artifactId&gt;kafka-streams-scala_2.11&lt;/artifactId&gt;
&lt;version&gt;{{fullDotVersion}}&lt;/version&gt;
&lt;/dependency&gt;
</pre>

<h3><a id="connectapi" href="#connectapi">2.4 Connect API</a></h3>

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.
Expand Down
43 changes: 25 additions & 18 deletions docs/streams/developer-guide/datatypes.html
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,12 +44,15 @@
<ul class="simple">
<li><a class="reference internal" href="#configuring-serdes" id="id1">Configuring SerDes</a></li>
<li><a class="reference internal" href="#overriding-default-serdes" id="id2">Overriding default SerDes</a></li>
<li><a class="reference internal" href="#available-serdes" id="id3">Available SerDes</a><ul>
<li><a class="reference internal" href="#available-serdes" id="id3">Available SerDes</a></li>
<ul>
<li><a class="reference internal" href="#primitive-and-basic-types" id="id4">Primitive and basic types</a></li>
<li><a class="reference internal" href="#avro" id="id5">Avro</a></li>
<li><a class="reference internal" href="#json" id="id6">JSON</a></li>
<li><a class="reference internal" href="#further-serdes" id="id7">Further serdes</a></li>
</ul>
<li><a class="reference internal" href="#scala-dsl-serdes" id="id8">Kafka Streams DSL for Scala Implicit SerDes</a></li>
</ul>
<div class="section" id="configuring-serdes">
<h2>Configuring SerDes<a class="headerlink" href="#configuring-serdes" title="Permalink to this headline"></a></h2>
<p>SerDes specified in the Streams configuration via <code class="docutils literal"><span class="pre">StreamsConfig</span></code> are used as the default in your Kafka Streams application.</p>
Expand Down Expand Up @@ -155,23 +158,27 @@ <h3>JSON<a class="headerlink" href="#json" title="Permalink to this headline"></
<a class="reference external" href="https://github.com/apache/kafka/blob/1.0/streams/examples/src/main/java/org/apache/kafka/streams/examples/pageview/PageViewTypedDemo.java">PageViewTypedDemo</a>
example demonstrates how to use this JSON serde.</p>
</div>
<div class="section" id="implementing-custom-serdes">
<span id="streams-developer-guide-serdes-custom"></span><h2>Implementing custom SerDes<a class="headerlink" href="#implementing-custom-serdes" title="Permalink to this headline"></a></h2>
<p>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:</p>
<ol class="arabic simple">
<li>Write a <em>serializer</em> for your data type <code class="docutils literal"><span class="pre">T</span></code> by implementing
<a class="reference external" href="https://github.com/apache/kafka/blob/1.0/clients/src/main/java/org/apache/kafka/common/serialization/Serializer.java">org.apache.kafka.common.serialization.Serializer</a>.</li>
<li>Write a <em>deserializer</em> for <code class="docutils literal"><span class="pre">T</span></code> by implementing
<a class="reference external" href="https://github.com/apache/kafka/blob/1.0/clients/src/main/java/org/apache/kafka/common/serialization/Deserializer.java">org.apache.kafka.common.serialization.Deserializer</a>.</li>
<li>Write a <em>serde</em> for <code class="docutils literal"><span class="pre">T</span></code> by implementing
<a class="reference external" href="https://github.com/apache/kafka/blob/1.0/clients/src/main/java/org/apache/kafka/common/serialization/Serde.java">org.apache.kafka.common.serialization.Serde</a>,
which you either do manually (see existing SerDes in the previous section) or by leveraging helper functions in
<a class="reference external" href="https://github.com/apache/kafka/blob/1.0/clients/src/main/java/org/apache/kafka/common/serialization/Serdes.java">Serdes</a>
such as <code class="docutils literal"><span class="pre">Serdes.serdeFrom(Serializer&lt;T&gt;,</span> <span class="pre">Deserializer&lt;T&gt;)</span></code>.</li>
</ol>
</div>
</div>
<div class="section" id="implementing-custom-serdes">

@guozhangwang guozhangwang Apr 16, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

cc @joel-hamill @ewencp we are adding a few new sections in web docs regarding the streams scala API, which may be affecting #4536.

<span id="streams-developer-guide-serdes-custom"></span><h2>Implementing custom SerDes<a class="headerlink" href="#implementing-custom-serdes" title="Permalink to this headline"></a></h2>
<p>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:</p>
<ol class="arabic simple">
<li>Write a <em>serializer</em> for your data type <code class="docutils literal"><span class="pre">T</span></code> by implementing
<a class="reference external" href="https://github.com/apache/kafka/blob/1.0/clients/src/main/java/org/apache/kafka/common/serialization/Serializer.java">org.apache.kafka.common.serialization.Serializer</a>.</li>
<li>Write a <em>deserializer</em> for <code class="docutils literal"><span class="pre">T</span></code> by implementing
<a class="reference external" href="https://github.com/apache/kafka/blob/1.0/clients/src/main/java/org/apache/kafka/common/serialization/Deserializer.java">org.apache.kafka.common.serialization.Deserializer</a>.</li>
<li>Write a <em>serde</em> for <code class="docutils literal"><span class="pre">T</span></code> by implementing
<a class="reference external" href="https://github.com/apache/kafka/blob/1.0/clients/src/main/java/org/apache/kafka/common/serialization/Serde.java">org.apache.kafka.common.serialization.Serde</a>,
which you either do manually (see existing SerDes in the previous section) or by leveraging helper functions in
<a class="reference external" href="https://github.com/apache/kafka/blob/1.0/clients/src/main/java/org/apache/kafka/common/serialization/Serdes.java">Serdes</a>
such as <code class="docutils literal"><span class="pre">Serdes.serdeFrom(Serializer&lt;T&gt;,</span> <span class="pre">Deserializer&lt;T&gt;)</span></code>.</li>
</ol>
</div>
</div>
<div class="section" id="scala-dsl-serdes">
<h2>Kafka Streams DSL for Scala Implicit SerDes<a class="headerlink" href="scala-dsl-serdes" title="Permalink to this headline"></a></h2>
<p>When using the <a href="dsl-api.html#scala-dsl">Kafka Streams DSL for Scala</a> 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 <a href="dsl-api.html#scala-dsl-implicit-serdes">Implicit SerDes</a> and <a href="dsl-api.html#scala-dsl-user-defined-serdes">User-Defined SerDes</a> sections in the DSL API documentation for details</p>
</div>


</div>
Expand Down
Loading