This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 190fd8946a4 KAFKA-20553 Remove Scala dependency from
streams:integration-tests (#22675)
190fd8946a4 is described below
commit 190fd8946a48f11865011c91797e7a618e0d6547
Author: PoAn Yang <[email protected]>
AuthorDate: Fri Jun 26 04:14:43 2026 +0900
KAFKA-20553 Remove Scala dependency from streams:integration-tests (#22675)
The `streams:integration-tests` module only kept the Scala plugin,
`streams-scala` and `scala-library` for four old Scala test files.
Removing it makes the module pure Java and simplifies the follow-up
testFixtures migration in KAFKA-20553.
Reviewers: Chia-Ping Tsai <[email protected]>
---
build.gradle | 24 --
...bleJoinScalaIntegrationTestImplicitSerdes.scala | 175 ---------------
.../kafka/streams/integration/WordCountTest.scala | 250 ---------------------
...StreamToTableJoinScalaIntegrationTestBase.scala | 144 ------------
.../utils/StreamToTableJoinTestData.scala | 60 -----
.../streams/scala/StreamToTableJoinTest.scala | 106 +++++++++
6 files changed, 106 insertions(+), 653 deletions(-)
diff --git a/build.gradle b/build.gradle
index 5b8c8e8c874..4d44740d9e6 100644
--- a/build.gradle
+++ b/build.gradle
@@ -3039,15 +3039,12 @@ project(':streams:streams-scala') {
}
project(':streams:integration-tests') {
- apply plugin: 'scala'
-
base {
archivesName = "kafka-streams-integration-tests"
}
dependencies {
implementation libs.slf4jApi
- implementation libs.scalaLibrary
testImplementation testFixtures(project(':clients'))
testImplementation project(':group-coordinator')
@@ -3056,7 +3053,6 @@ project(':streams:integration-tests') {
testImplementation testFixtures(project(':server-common'))
testImplementation project(':storage')
testImplementation project(':streams').sourceSets.test.output
- testImplementation project(':streams:streams-scala')
testImplementation project(':test-common:test-common-runtime')
testImplementation project(':tools')
testImplementation project(':transaction-coordinator')
@@ -3070,26 +3066,6 @@ project(':streams:integration-tests') {
testRuntimeOnly runtimeTestLibs
}
-
- sourceSets {
- // Set java/scala source folders in the `scala` block to enable joint
compilation
- main {
- java {
- srcDirs = []
- }
- scala {
- srcDirs = []
- }
- }
- test {
- java {
- srcDirs = []
- }
- scala {
- srcDirs = ["src/test/java", "src/test/scala"]
- }
- }
- }
}
project(':streams:test-utils') {
diff --git
a/streams/integration-tests/src/test/scala/org/apache/kafka/streams/integration/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
b/streams/integration-tests/src/test/scala/org/apache/kafka/streams/integration/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
deleted file mode 100644
index 4cfc811728e..00000000000
---
a/streams/integration-tests/src/test/scala/org/apache/kafka/streams/integration/StreamToTableJoinScalaIntegrationTestImplicitSerdes.scala
+++ /dev/null
@@ -1,175 +0,0 @@
-/*
- * 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.integration
-
-import
org.apache.kafka.streams.integration.utils.StreamToTableJoinScalaIntegrationTestBase
-import org.apache.kafka.streams.scala.ImplicitConversions._
-import org.apache.kafka.streams.scala.StreamsBuilder
-import org.apache.kafka.streams.scala.kstream._
-import org.apache.kafka.streams.scala.serialization.{Serdes => NewSerdes}
-import org.apache.kafka.streams.{KafkaStreams, KeyValue, StreamsConfig}
-import org.junit.jupiter.api.Assertions._
-import org.junit.jupiter.api._
-
-import java.util.Properties
-
-/**
- * Test suite that does an example to demonstrate stream-table joins in Kafka
Streams
- * <p>
- * 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.
- */
-@Tag("integration")
-class StreamToTableJoinScalaIntegrationTestImplicitSerdes extends
StreamToTableJoinScalaIntegrationTestBase {
-
- @Test def testShouldCountClicksPerRegion(): Unit = {
-
- // DefaultSerdes brings into scope implicit serdes (mostly for primitives)
that will set up all Grouped, Produced,
- // Consumed and Joined instances. So all APIs below that accept Grouped,
Produced, Consumed or Joined will
- // get these instances automatically
- import org.apache.kafka.streams.scala.serialization.Serdes._
-
- 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, region) => (if (region == null)
"UNKNOWN" else region, clicks))
-
- // Change the stream from <user> -> <region, clicks> to <region> ->
<clicks>
- .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)
-
- assertTrue(!actualClicksPerRegion.isEmpty, "Expected to process some data")
-
- streams.close()
- }
-
- @Test
- def testShouldCountClicksPerRegionWithNamedRepartitionTopic(): Unit = {
-
- // DefaultSerdes brings into scope implicit serdes (mostly for primitives)
that will set up all Grouped, Produced,
- // Consumed and Joined instances. So all APIs below that accept Grouped,
Produced, Consumed or Joined will
- // get these instances automatically
- import org.apache.kafka.streams.scala.serialization.Serdes._
-
- 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, region) => (if (region == null)
"UNKNOWN" else region, clicks))
-
- // Change the stream from <user> -> <region, clicks> to <region> ->
<clicks>
- .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)
-
- assertTrue(!actualClicksPerRegion.isEmpty, "Expected to process some data")
-
- streams.close()
- }
-
- @Test
- def testShouldCountClicksPerRegionJava(): Unit = {
-
- import org.apache.kafka.streams.kstream.{KStream => KStreamJ, KTable =>
KTableJ, _}
- import org.apache.kafka.streams.{KafkaStreams => KafkaStreamsJ,
StreamsBuilder => StreamsBuilderJ}
-
- import java.lang.{Long => JLong}
-
- val streamsConfiguration: Properties = getStreamsConfiguration()
-
- streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG,
NewSerdes.stringSerde.getClass.getName)
- streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,
NewSerdes.stringSerde.getClass.getName)
-
- val builder: StreamsBuilderJ = new StreamsBuilderJ()
-
- val userClicksStream: KStreamJ[String, JLong] =
- builder.stream[String, JLong](userClicksTopicJ,
Consumed.`with`(NewSerdes.stringSerde, NewSerdes.javaLongSerde))
-
- val userRegionsTable: KTableJ[String, String] =
- builder.table[String, String](userRegionsTopicJ,
Consumed.`with`(NewSerdes.stringSerde, NewSerdes.stringSerde))
-
- // Join the stream against the table.
- val valueJoinerJ: ValueJoiner[JLong, String, (String, JLong)] =
- (clicks: JLong, region: String) => (if (region == null) "UNKNOWN" else
region, clicks)
- val userClicksJoinRegion: KStreamJ[String, (String, JLong)] =
userClicksStream.leftJoin(
- userRegionsTable,
- valueJoinerJ,
- Joined.`with`[String, JLong, String](NewSerdes.stringSerde,
NewSerdes.javaLongSerde, NewSerdes.stringSerde)
- )
-
- // Change the stream from <user> -> <region, clicks> to <region> ->
<clicks>
- val clicksByRegion: KStreamJ[String, JLong] = userClicksJoinRegion.map {
(_, regionWithClicks) =>
- new KeyValue(regionWithClicks._1, regionWithClicks._2)
- }
-
- // Compute the total per region by summing the individual click counts per
region.
- val clicksPerRegion: KTableJ[String, JLong] = clicksByRegion
- .groupByKey(Grouped.`with`(NewSerdes.stringSerde,
NewSerdes.javaLongSerde))
- .reduce((v1, v2) => v1 + v2)
-
- // Write the (continuously updating) results to the output topic.
- clicksPerRegion.toStream.to(outputTopicJ,
Produced.`with`(NewSerdes.stringSerde, NewSerdes.javaLongSerde))
-
- val streams = new KafkaStreamsJ(builder.build(), streamsConfiguration)
-
- streams.start()
- produceNConsume(userClicksTopicJ, userRegionsTopicJ, outputTopicJ)
- streams.close()
- }
-}
diff --git
a/streams/integration-tests/src/test/scala/org/apache/kafka/streams/integration/WordCountTest.scala
b/streams/integration-tests/src/test/scala/org/apache/kafka/streams/integration/WordCountTest.scala
deleted file mode 100644
index 3e9813dda24..00000000000
---
a/streams/integration-tests/src/test/scala/org/apache/kafka/streams/integration/WordCountTest.scala
+++ /dev/null
@@ -1,250 +0,0 @@
-/*
- * 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.integration
-
-import java.util.Properties
-import java.util.regex.Pattern
-import org.junit.jupiter.api.Assertions._
-import org.junit.jupiter.api._
-import org.apache.kafka.streams.scala.serialization.{Serdes => NewSerdes}
-import org.apache.kafka.streams.scala.ImplicitConversions._
-import org.apache.kafka.streams.scala.StreamsBuilder
-import org.apache.kafka.streams.{KafkaStreams, KeyValue, StreamsConfig}
-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.utils.{MockTime, Utils}
-import org.apache.kafka.common.serialization.{LongDeserializer,
StringDeserializer, StringSerializer}
-import org.apache.kafka.test.TestUtils
-import org.junit.jupiter.api.Tag
-
-import java.io.File
-
-/**
- * Test suite that does a classic word count example.
- * <p>
- * 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.
- */
-@Tag("integration")
-class WordCountTest extends WordCountTestData {
-
- private val cluster: EmbeddedKafkaCluster = new EmbeddedKafkaCluster(1)
-
- final private val alignedTime = (System.currentTimeMillis() / 1000 + 1) *
1000
- private val mockTime: MockTime = cluster.time
- mockTime.setCurrentTimeMs(alignedTime)
-
- private val testFolder: File = TestUtils.tempDirectory()
-
- @BeforeEach
- def startKafkaCluster(): Unit = {
- cluster.start()
- cluster.createTopic(inputTopic)
- cluster.createTopic(outputTopic)
- cluster.createTopic(inputTopicJ)
- cluster.createTopic(outputTopicJ)
- }
-
- @AfterEach
- def stopKafkaCluster(): Unit = {
- cluster.stop()
- Utils.delete(testFolder)
- }
-
- @Test
- def testShouldCountWords(): Unit = {
- import org.apache.kafka.streams.scala.serialization.Serdes._
-
- 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((_, v) => v)
- .count()
-
- // write to output topic
- wordCounts.toStream.to(outputTopic)
-
- val streams = 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 scala.jdk.CollectionConverters._
-
assertEquals(actualWordCounts.asScala.take(expectedWordCounts.size).sortBy(_.key),
expectedWordCounts.sortBy(_.key))
- }
-
- @Test
- def testShouldCountWordsMaterialized(): Unit = {
- import org.apache.kafka.streams.scala.serialization.Serdes._
-
- 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()(Materialized.as("word-count"))
-
- // write to output topic
- wordCounts.toStream.to(outputTopic)
-
- val streams = 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 scala.jdk.CollectionConverters._
-
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 scala.jdk.CollectionConverters._
-
- val streamsConfiguration = getStreamsConfiguration()
- streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG,
NewSerdes.stringSerde.getClass.getName)
- streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,
NewSerdes.stringSerde.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 { line =>
- pattern.split(line.toLowerCase).toBuffer.asJava
- }
-
- val grouped: KGroupedStreamJ[String, String] = splits.groupBy { (_, v) =>
- v
- }
-
- val wordCounts: KTableJ[String, java.lang.Long] = grouped.count()
-
- wordCounts.toStream.to(outputTopicJ,
Produced.`with`(NewSerdes.stringSerde, NewSerdes.javaLongSerde))
-
- 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.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.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 scala.jdk.CollectionConverters._
- 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)
- )
-}
diff --git
a/streams/integration-tests/src/test/scala/org/apache/kafka/streams/integration/utils/StreamToTableJoinScalaIntegrationTestBase.scala
b/streams/integration-tests/src/test/scala/org/apache/kafka/streams/integration/utils/StreamToTableJoinScalaIntegrationTestBase.scala
deleted file mode 100644
index f3aec5784cd..00000000000
---
a/streams/integration-tests/src/test/scala/org/apache/kafka/streams/integration/utils/StreamToTableJoinScalaIntegrationTestBase.scala
+++ /dev/null
@@ -1,144 +0,0 @@
-/*
- * 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.integration.utils
-
-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, Utils}
-import org.apache.kafka.streams._
-import org.apache.kafka.test.TestUtils
-import org.junit.jupiter.api._
-
-import java.io.File
-import java.util.Properties
-
-/**
- * Test suite base that prepares Kafka cluster for stream-table joins in Kafka
Streams
- * <p>
- */
-@Tag("integration")
-class StreamToTableJoinScalaIntegrationTestBase extends
StreamToTableJoinTestData {
-
- private val cluster: EmbeddedKafkaCluster = new EmbeddedKafkaCluster(1)
-
- final private val alignedTime = (System.currentTimeMillis() / 1000 + 1) *
1000
- private val mockTime: MockTime = cluster.time
- mockTime.setCurrentTimeMs(alignedTime)
-
- private val testFolder: File = TestUtils.tempDirectory()
-
- @BeforeEach
- def startKafkaCluster(): Unit = {
- cluster.start()
- cluster.createTopic(userClicksTopic)
- cluster.createTopic(userRegionsTopic)
- cluster.createTopic(outputTopic)
- cluster.createTopic(userClicksTopicJ)
- cluster.createTopic(userRegionsTopicJ)
- cluster.createTopic(outputTopicJ)
- }
-
- @AfterEach
- def stopKafkaCluster(): Unit = {
- cluster.stop()
- Utils.delete(testFolder)
- }
-
- 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, "1000")
- streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
"earliest")
- streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG,
testFolder.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.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.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
- }
-
- def produceNConsume(
- userClicksTopic: String,
- userRegionsTopic: String,
- outputTopic: String,
- waitTillRecordsReceived: Boolean = true
- ): java.util.List[KeyValue[String, Long]] = {
-
- import _root_.scala.jdk.CollectionConverters._
-
- // 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
- )
-
- if (waitTillRecordsReceived) {
- // consume and verify result
- val consumerConfig = getConsumerConfig()
-
- IntegrationTestUtils.waitUntilFinalKeyValueRecordsReceived(
- consumerConfig,
- outputTopic,
- expectedClicksPerRegion.asJava
- )
- } else {
- java.util.Collections.emptyList()
- }
- }
-}
diff --git
a/streams/integration-tests/src/test/scala/org/apache/kafka/streams/integration/utils/StreamToTableJoinTestData.scala
b/streams/integration-tests/src/test/scala/org/apache/kafka/streams/integration/utils/StreamToTableJoinTestData.scala
deleted file mode 100644
index 4e8a2f024a7..00000000000
---
a/streams/integration-tests/src/test/scala/org/apache/kafka/streams/integration/utils/StreamToTableJoinTestData.scala
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
- * 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.integration.utils
-
-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/StreamToTableJoinTest.scala
b/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinTest.scala
new file mode 100644
index 00000000000..a104b6cf3f4
--- /dev/null
+++
b/streams/streams-scala/src/test/scala/org/apache/kafka/streams/scala/StreamToTableJoinTest.scala
@@ -0,0 +1,106 @@
+/*
+ * 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.scala.ImplicitConversions._
+import org.apache.kafka.streams.scala.kstream._
+import org.apache.kafka.streams.scala.serialization.Serdes._
+import org.apache.kafka.streams.scala.utils.TestDriver
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+
+import scala.jdk.CollectionConverters._
+
+/**
+ * Test suite that demonstrates a stream-table join in Kafka Streams using the
Scala API with implicit serdes.
+ */
+class StreamToTableJoinTest extends TestDriver {
+
+ private val userClicksTopic = "user-clicks"
+ private val userRegionsTopic = "user-regions"
+ private val outputTopic = "output-topic"
+
+ // Input 1: Clicks per user (multiple records allowed per user).
+ private val userClicks: Seq[(String, Long)] = Seq(
+ "alice" -> 13L,
+ "bob" -> 4L,
+ "chao" -> 25L,
+ "bob" -> 19L,
+ "dave" -> 56L,
+ "eve" -> 78L,
+ "alice" -> 40L,
+ "fang" -> 99L
+ )
+
+ // Input 2: Region per user (multiple records allowed per user).
+ private val userRegions: Seq[(String, String)] = Seq(
+ "alice" -> "asia",
+ "bob" -> "americas",
+ "chao" -> "asia",
+ "dave" -> "europe",
+ "alice" -> "europe",
+ "eve" -> "americas",
+ "fang" -> "asia"
+ )
+
+ private val expectedClicksPerRegion: Map[String, Long] = Map(
+ "americas" -> 101L,
+ "europe" -> 109L,
+ "asia" -> 124L
+ )
+
+ @Test
+ def testShouldCountClicksPerRegionWithImplicitSerdes(): Unit = {
+ // DefaultSerdes brings into scope implicit serdes (mostly for primitives)
that will set up all Grouped, Produced,
+ // Consumed and Joined instances. So all APIs below that accept Grouped,
Produced, Consumed or Joined will
+ // get these instances automatically.
+ 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, region) => (if (region == null)
"UNKNOWN" else region, clicks))
+
+ // Change the stream from <user> -> <region, clicks> to <region> ->
<clicks>
+ .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 testDriver = createTestDriver(builder)
+ val userRegionsInput = testDriver.createInput[String,
String](userRegionsTopic)
+ val userClicksInput = testDriver.createInput[String, Long](userClicksTopic)
+ val output = testDriver.createOutput[String, Long](outputTopic)
+
+ userRegions.foreach { case (user, region) =>
userRegionsInput.pipeInput(user, region) }
+ userClicks.foreach { case (user, clicks) =>
userClicksInput.pipeInput(user, clicks) }
+
+ assertEquals(expectedClicksPerRegion.asJava, output.readKeyValuesToMap())
+
+ testDriver.close()
+ }
+}
