ref: Migrating Kafka tests to TestContainers

Author: asm0dey

.github/workflows/build.yml

@@ -25,7 +25,7 @@ jobs:
           key: ${{ runner.os }}-m2-${{ hashFiles('**/pom.xml') }}
           restore-keys: ${{ runner.os }}-m2
       - name: Build with Maven
-        run: ./mvnw -B package --file pom.xml -Pscala-2.12 -Dkotest.tags="!Kafka"
+        run: ./mvnw -B package --file pom.xml -Pscala-2.12
 #  qodana:
 #    runs-on: ubuntu-latest
 #    steps:

.github/workflows/publish_dev_version.yml

@@ -24,7 +24,7 @@ jobs:
           key: ${{ runner.os }}-m2-${{ hashFiles('**/pom.xml') }}
           restore-keys: ${{ runner.os }}-m2
       - name: Deploy to GH Packages
-        run: ./mvnw --batch-mode deploy -Dkotest.tags="!Kafka"
+        run: ./mvnw --batch-mode deploy
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
 

kotlin-spark-api/3.2/pom_2.12.xml

@@ -73,6 +73,12 @@
             <version>${kotest-extensions-allure.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>io.kotest.extensions</groupId>
+            <artifactId>kotest-extensions-testcontainers</artifactId>
+            <version>${kotest-extensions-testcontainers.version}</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>io.github.embeddedkafka</groupId>
             <artifactId>embedded-kafka_${scala.compat.version}</artifactId>

kotlin-spark-api/3.2/src/test/kotlin/org/jetbrains/kotlinx/spark/api/KafkaHelper.kt

@@ -7,9 +7,9 @@
  * 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.
@@ -20,7 +20,11 @@
 package org.jetbrains.kotlinx.spark.api
 
 import io.kotest.core.Tag
+import io.kotest.core.extensions.install
 import io.kotest.core.spec.style.ShouldSpec
+import io.kotest.extensions.testcontainers.TestContainerExtension
+import io.kotest.extensions.testcontainers.kafka.createProducer
+import io.kotest.extensions.testcontainers.kafka.createStringStringProducer
 import io.kotest.matchers.collections.shouldBeIn
 import org.apache.kafka.clients.consumer.ConsumerConfig
 import org.apache.kafka.clients.consumer.ConsumerRecord
@@ -32,27 +36,33 @@ import org.apache.spark.streaming.kafka010.ConsumerStrategies
 import org.apache.spark.streaming.kafka010.KafkaUtils
 import org.apache.spark.streaming.kafka010.LocationStrategies
 import org.jetbrains.kotlinx.spark.api.tuples.*
+import org.testcontainers.containers.KafkaContainer
+import org.testcontainers.utility.DockerImageName
 import java.io.Serializable
+import java.util.concurrent.TimeUnit
 
 object Kafka : Tag()
 
 class KafkaStreamingTest : ShouldSpec({
 
-    // making sure it can be skipped on Github actions since it times out
     tags(Kafka)
+    val kafka = install(TestContainerExtension(KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:6.2.4")))) {
+        withEmbeddedZookeeper()
+        withEnv("KAFKA_AUTO_CREATE_TOPICS_ENABLE", "true")
+    }
+
 
     context("kafka") {
-        val port = 9092
-        val broker = "localhost:$port"
         val topic1 = "test1"
         val topic2 = "test2"
-        val kafkaListener = EmbeddedKafkaListener(port)
-        listener(kafkaListener)
 
         should("support kafka streams") {
-            val producer = kafkaListener.stringStringProducer()
-            producer.send(ProducerRecord(topic1, "Hello this is a test test test"))
-            producer.send(ProducerRecord(topic2, "This is also also a test test something"))
+            val producer = kafka.createStringStringProducer()
+            listOf(
+                producer.send(ProducerRecord(topic1, "Hello this is a test test test")),
+                producer.send(ProducerRecord(topic2, "This is also also a test test something")),
+            )
+                .map { it.get(10, TimeUnit.SECONDS) }
             producer.close()
 
             withSparkStreaming(
@@ -62,7 +72,7 @@ class KafkaStreamingTest : ShouldSpec({
             ) {
 
                 val kafkaParams: Map<String, Serializable> = mapOf(
-                    ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG to broker,
+                    ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG to kafka.bootstrapServers,
                     ConsumerConfig.GROUP_ID_CONFIG to "consumer-group",
                     ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG to StringDeserializer::class.java,
                     ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG to StringDeserializer::class.java,
@@ -113,4 +123,4 @@ class KafkaStreamingTest : ShouldSpec({
         }
 
     }
-})
+})

kotlin-spark-api/3.2/src/test/kotlin/org/jetbrains/kotlinx/spark/api/KafkaStreamingTest.kt

@@ -7,9 +7,9 @@
  * 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.
@@ -24,7 +24,6 @@ import io.kotest.extensions.allure.AllureTestReporter
 
 @Suppress("unused")
 object ProjectConfig : AbstractProjectConfig() {
-    override fun listeners() = super.listeners() + AllureTestReporter(true)
-
     override fun extensions() = super.extensions() + AllureTestReporter(true)
+
 }

kotlin-spark-api/3.2/src/test/kotlin/org/jetbrains/kotlinx/spark/api/ProjectConfig.kt

@@ -10,34 +10,35 @@
     <packaging>pom</packaging>
 
     <properties>
-        <kotlin.version>1.6.21</kotlin.version>
-        <dokka.version>1.6.10</dokka.version>
         <atrium.version>0.17.0</atrium.version>
-        <kotest.version>5.2.3</kotest.version>
-        <kotest-extensions-allure.version>1.1.0</kotest-extensions-allure.version>
+        <dokka.version>1.6.10</dokka.version>
         <embedded-kafka.version>3.1.0</embedded-kafka.version>
-        <spark3.version>3.2.1</spark3.version>
+        <hadoop.version>3.3.1</hadoop.version>
+        <kotest-extensions-allure.version>1.1.0</kotest-extensions-allure.version>
+        <kotest-extensions-testcontainers.version>1.3.1</kotest-extensions-testcontainers.version>
+        <kotest.version>5.2.3</kotest.version>
         <kotlin-jupyter-api.version>0.11.0-83</kotlin-jupyter-api.version>
+        <kotlin.version>1.6.21</kotlin.version>
         <kotlinx.html.version>0.7.5</kotlinx.html.version>
-        <hadoop.version>3.3.1</hadoop.version>
+        <spark3.version>3.2.1</spark3.version>
 
         <!-- Plugin versions -->
         <allure-maven.version>2.10.0</allure-maven.version>
         <jacoco-maven-plugin.version>0.8.7</jacoco-maven-plugin.version>
         <klaxon.version>5.5</klaxon.version>
+        <kotlin.code.style>official</kotlin.code.style>
         <license-maven-plugin.version>2.0.0</license-maven-plugin.version>
         <maven-assembly-plugin.version>3.3.0</maven-assembly-plugin.version>
-        <maven-gpg-plugin.version>3.0.1</maven-gpg-plugin.version>
+        <maven-compiler-plugin.version>3.10.1</maven-compiler-plugin.version>
         <maven-deploy-plugin.version>3.0.0-M1</maven-deploy-plugin.version>
         <maven-enforcer-plugin.version>3.0.0-M3</maven-enforcer-plugin.version>
+        <maven-gpg-plugin.version>3.0.1</maven-gpg-plugin.version>
         <maven-jar-plugin.version>3.2.0</maven-jar-plugin.version>
         <maven-site-plugin.version>3.9.1</maven-site-plugin.version>
         <maven-source-plugin.version>3.2.1</maven-source-plugin.version>
-        <maven-compiler-plugin.version>3.10.1</maven-compiler-plugin.version>
         <maven-surefire-plugin.version>3.0.0-M6</maven-surefire-plugin.version>
         <nexus-staging-plugin.version>1.6.8</nexus-staging-plugin.version>
         <scala-maven-plugin.version>4.5.6</scala-maven-plugin.version>
-        <kotlin.code.style>official</kotlin.code.style>
     </properties>
 
     <modules>