Skip to content

MINOR: Build and code sample updates for Kafka Streams DSL for Scala #4949

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

Merged
guozhangwang merged 4 commits into from
May 7, 2018
Merged

MINOR: Build and code sample updates for Kafka Streams DSL for Scala #4949

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
3f6376b
Use scala major version when producing artifact
seglo Apr 30, 2018
169b031
Update Kafka Streams for Scala code samples in docs
seglo Apr 30, 2018
b4f30fa
Removed dependency on scalaLogging and imports from unit tests
debasishg May 1, 2018
4b451d5
Key serdes no longer required with fixes in #4919
seglo May 6, 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
3 changes: 1 addition & 2 deletions build.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -980,7 +980,7 @@ project(':streams') {
project(':streams:streams-scala') {
println "Building project 'streams-scala' with Scala version ${versions.scala}"
apply plugin: 'scala'
archivesBaseName = "kafka-streams-scala"
archivesBaseName = "kafka-streams-scala_${versions.baseScala}"

@ijuma ijuma Apr 30, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Unrelated question, why is scalaLogging just a test compile dependency?

@seglo seglo Apr 30, 2018

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

It's required by the LazyLogging trait referenced by several tests, but it doesn't look like any logging is actually done. It's probably safe to remove this. @debasishg can you provide some insight here?

@debasishg debasishg Apr 30, 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.

+1 .. it's there, along with log4j.properties as part of logging infrastructure from the tests. Currently not used though. We can remove the imports and do away with the dependency. Do we keep log4j.properties ?

@seglo seglo Apr 30, 2018

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Let's remove the dependency. I think it makes sense to keep log4j.properties so users can see what's happening with the embedded Kafka instance and Kafka streams when the tests are run.


dependencies {
compile project(':streams')
Expand All @@ -991,7 +991,6 @@ project(':streams:streams-scala') {
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
Expand Down
14 changes: 8 additions & 6 deletions docs/streams/developer-guide/dsl-api.html
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3191,11 +3191,13 @@ <h2><a class="toc-backref" href="#id7">Overview</a><a class="headerlink" href="#
import java.util.concurrent.TimeUnit

import org.apache.kafka.streams.kstream.Materialized
import org.apache.kafka.streams.scala.ImplicitConversions._
import org.apache.kafka.streams.scala._
import org.apache.kafka.streams.scala.kstream._
import org.apache.kafka.streams.{KafkaStreams, StreamsConfig}

object WordCountApplication extends App {
import DefaultSerdes._
import ImplicitConversions._

val config: Properties = {
val p = new Properties()
Expand All @@ -3204,9 +3206,9 @@ <h2><a class="toc-backref" href="#id7">Overview</a><a class="headerlink" href="#
p
}

val builder = new StreamsBuilder()
val textLines = builder.stream[String, String]("TextLinesTopic")
val wordCounts = textLines
val builder: StreamsBuilder = new StreamsBuilder
val textLines: KStream[String, String] = builder.stream[String, String]("TextLinesTopic")
val wordCounts: KTable[String, Long] = textLines
.flatMapValues(textLine => textLine.toLowerCase.split("\\W+"))
.groupBy((_, word) => word)
.count(Materialized.as("counts-store"))
Expand All @@ -3216,7 +3218,7 @@ <h2><a class="toc-backref" href="#id7">Overview</a><a class="headerlink" href="#
streams.start()

sys.ShutdownHookThread {
streams.close(10, TimeUnit.SECONDS)
streams.close(10, TimeUnit.SECONDS)
}
}
</pre>
Expand Down Expand Up @@ -3290,7 +3292,7 @@ <h2><a class="toc-backref" href="#id7">Overview</a><a class="headerlink" href="#
// Join the stream against the table.
.leftJoin(userRegionsTable, (clicks: UserClicks, region: String) => (if (region == null) "UNKNOWN" else region, clicks.clicks))

// Change the stream from <user> -> <region, clicks> to <region> -> <clicks>
// Change the stream from &lt;user&gt; -&gt; &lt;region, clicks&gt; to &lt;region&gt; -&gt; &lt;clicks&gt;
.map((_, regionWithClicks) => regionWithClicks)

// Compute the total per region by summing the individual click counts per region.
Expand Down
7 changes: 4 additions & 3 deletions docs/streams/index.html
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -255,12 +255,13 @@ <h3 style="margin-top: 5.3rem;">Hello Kafka Streams</h3>
import java.util.concurrent.TimeUnit

import org.apache.kafka.streams.kstream.Materialized
import org.apache.kafka.streams.scala.ImplicitConversions._
import org.apache.kafka.streams.scala._
import org.apache.kafka.streams.scala.kstream._
import org.apache.kafka.streams.{KafkaStreams, StreamsConfig}

object WordCountApplication extends App {
import DefaultSerdes._
import ImplicitConversions._

val config: Properties = {
val p = new Properties()
Expand All @@ -269,7 +270,7 @@ <h3 style="margin-top: 5.3rem;">Hello Kafka Streams</h3>
p
}

val builder: StreamsBuilder = new StreamsBuilder()
val builder: StreamsBuilder = new StreamsBuilder
val textLines: KStream[String, String] = builder.stream[String, String]("TextLinesTopic")
val wordCounts: KTable[String, Long] = textLines
.flatMapValues(textLine => textLine.toLowerCase.split("\\W+"))
Expand All @@ -281,7 +282,7 @@ <h3 style="margin-top: 5.3rem;">Hello Kafka Streams</h3>
streams.start()

sys.ShutdownHookThread {
streams.close(10, TimeUnit.SECONDS)
streams.close(10, TimeUnit.SECONDS)
}
}
</pre>
Expand Down
3 changes: 1 addition & 2 deletions .../org/apache/kafka/streams/scala/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,6 @@ 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
Expand All @@ -46,7 +45,7 @@ import com.typesafe.scalalogging.LazyLogging
* Hence the native Java API based version is more verbose.
*/
class StreamToTableJoinScalaIntegrationTestImplicitSerdes extends JUnitSuite
with StreamToTableJoinTestData with LazyLogging {
with StreamToTableJoinTestData {

private val privateCluster: EmbeddedKafkaCluster = new EmbeddedKafkaCluster(1)

Expand Down
3 changes: 1 addition & 2 deletions streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/TopologyTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,6 @@ 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, _}
Expand All @@ -40,7 +39,7 @@ import collection.JavaConverters._
/**
* Test suite that verifies that the topology built by the Java and Scala APIs match.
*/
class TopologyTest extends JUnitSuite with LazyLogging {
class TopologyTest extends JUnitSuite {

val inputTopic = "input-topic"
val userClicksTopic = "user-clicks-topic"
Expand Down
3 changes: 1 addition & 2 deletions streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/WordCountTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,6 @@ 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.
Expand All @@ -51,7 +50,7 @@ import com.typesafe.scalalogging.LazyLogging
* 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 {
class WordCountTest extends JUnitSuite with WordCountTestData {

private val privateCluster: EmbeddedKafkaCluster = new EmbeddedKafkaCluster(1)

Expand Down