fix(kafka-tests): use non deprecated kafka test container (#99)

Flaxoos · web-flow · commit c8a2a8e8511c · 2025-08-30T21:21:25.000+02:00
* fix(build): removed non existing module entry in settings.gradle.kts

* fix(kafka-tests): use non deprecated kafka test container:

- switched to using the Testcontainers implementation for Confluent Kafka. This required some changes in the tests setup.
- This also exposed an issue where the CommonClientPropertiesBuilder being an object was causing it to be shared across tests, which while cannot be an issue in main code, it still is more correct for it to be a class.

* format: auto applied ktlint formatting

* bump patch version
diff --git a/.gitignore b/.gitignore
@@ -45,4 +45,9 @@ out/
 /.run/publish kafka and run example.run.xml
 /.kotlin
 /buildSrc/.kotlin
-**/node_modules/
+**/node_modules/
+
+### Agents ###
+/CLAUDE.md
+.claude
+.junie/
diff --git a/gradle.properties b/gradle.properties
@@ -3,7 +3,7 @@ github.repository.name=extra-ktor-plugins
 kotlin.native.cacheKind.linuxX64=none
 kotlin.native.ignoreDisabledTargets=true
 gradle.publish.enable.module-metadata=true
-version=2.2.1
+version=2.2.2
 gpr.user=flaxoos
 org.gradle.jvmargs=-Xmx2g -XX:MaxMetaspaceSize=2g
 kotlin.mpp.applyDefaultHierarchyTemplate=false
diff --git a/ktor-server-kafka/src/jvmMain/kotlin/io/github/flaxoos/ktor/server/plugins/kafka/KafkaPluginConfig.kt b/ktor-server-kafka/src/jvmMain/kotlin/io/github/flaxoos/ktor/server/plugins/kafka/KafkaPluginConfig.kt
@@ -245,16 +245,6 @@ internal fun KafkaConfigPropertiesContext.withSslProperties(properties: SslPrope
         }
     }
 
-internal fun KafkaConfigPropertiesContext.withSslProperties(properties: SslPropertiesBuilder?) =
-    apply {
-        properties?.let { kafkaProperties.putAll(it.build()) }
-    }
-
-internal fun KafkaConfigPropertiesContext.withSaslProperties(properties: SaslPropertiesBuilder?) =
-    apply {
-        properties?.let { kafkaProperties.putAll(it.build()) }
-    }
-
 @KafkaDsl
 fun KafkaConsumerConfig.consumerRecordHandler(
     topicName: TopicName,
@@ -264,9 +254,9 @@ fun KafkaConsumerConfig.consumerRecordHandler(
 }
 
 @KafkaDsl
-fun KafkaConfig.common(configuration: CommonClientPropertiesBuilder.() -> Unit = { CommonClientPropertiesBuilder }) {
+fun KafkaConfig.common(configuration: CommonClientPropertiesBuilder.() -> Unit = { CommonClientPropertiesBuilder() }) {
     commonPropertiesBuilder =
-        CommonClientPropertiesBuilder.apply(configuration)
+        CommonClientPropertiesBuilder().apply(configuration)
 }
 
 @KafkaDsl
@@ -591,7 +581,7 @@ sealed class ClientPropertiesBuilder : KafkaPropertiesBuilder() {
 /**
  * Concrete implementation of [ClientPropertiesBuilder] to represent the common properties
  */
-data object CommonClientPropertiesBuilder : ClientPropertiesBuilder()
+class CommonClientPropertiesBuilder : ClientPropertiesBuilder()
 
 class AdminPropertiesBuilder : ClientPropertiesBuilder()
 
diff --git a/ktor-server-kafka/src/jvmTest/kotlin/io/github/flaxoos/ktor/server/plugins/kafka/BaseKafkaIntegrationTest.kt b/ktor-server-kafka/src/jvmTest/kotlin/io/github/flaxoos/ktor/server/plugins/kafka/BaseKafkaIntegrationTest.kt
@@ -103,7 +103,7 @@ abstract class BaseKafkaIntegrationTest : FunSpec() {
                 }
                 isConnected = true
                 break
-            } catch (e: Exception) {
+            } catch (_: Exception) {
                 logger.info(
                     "Attempt $i to connect to Kafka broker at bootstrap.servers: " +
                         "$bootstrapServers failed, retrying",
diff --git a/ktor-server-kafka/src/jvmTest/kotlin/io/github/flaxoos/ktor/server/plugins/kafka/KafkaContainers.kt b/ktor-server-kafka/src/jvmTest/kotlin/io/github/flaxoos/ktor/server/plugins/kafka/KafkaContainers.kt
@@ -1,53 +1,67 @@
 package io.github.flaxoos.ktor.server.plugins.kafka
 
-import io.github.flaxoos.ktor.server.plugins.kafka.KafkaGenericContainer.Companion.kafkaImage
 import io.ktor.util.logging.KtorSimpleLogger
 import io.ktor.util.logging.Logger
 import org.testcontainers.containers.DockerComposeContainer
 import org.testcontainers.containers.GenericContainer
-import org.testcontainers.containers.KafkaContainer
 import org.testcontainers.containers.Network
 import org.testcontainers.containers.output.Slf4jLogConsumer
 import org.testcontainers.containers.wait.strategy.Wait
+import org.testcontainers.kafka.ConfluentKafkaContainer
 import org.testcontainers.utility.DockerImageName
 import org.testcontainers.utility.MountableFile
 import java.io.File
 import kotlin.time.Duration.Companion.seconds
 import kotlin.time.toJavaDuration
 
 private const val CONFLUENT_PLATFORM_VERSION = "7.7.1"
+private const val KAFKA_IMAGE_NAME = "confluentinc/cp-kafka"
+private const val SCHEMA_REGISTRY_IMAGE_NAME = "confluentinc/cp-schema-registry"
+private const val KAFKA_IMAGE = "$KAFKA_IMAGE_NAME:$CONFLUENT_PLATFORM_VERSION"
+private const val SCHEMA_REGISTRY_IMAGE = "$SCHEMA_REGISTRY_IMAGE_NAME:$CONFLUENT_PLATFORM_VERSION"
 private val logger: Logger = KtorSimpleLogger(BaseKafkaIntegrationTest::class.java.simpleName)
 private val kafkaNetwork = Network.builder().driver("bridge").build()
 private const val YML_FILE_PATH = "/Kafka Cluster SSL.yml"
 private val yml = object {}::class.java.getResource(YML_FILE_PATH)?.toURI() ?: error("$YML_FILE_PATH not found")
-
 internal const val LOCALHOST = "localhost"
 internal const val KAFKA_BROKER_PORT = 19092
 internal const val BOOTSTRAP_SERVERS: String = "$LOCALHOST:$KAFKA_BROKER_PORT"
 internal const val SCHEMA_REGISTRY_PORT = 8081
 internal const val SCHEMA_REGISTRY_URL: String = "https://$LOCALHOST:$SCHEMA_REGISTRY_PORT"
 internal const val PASSWORD = "test_password"
 
-internal fun newKafkaContainer() = KafkaContainer(kafkaImage)
+internal fun newKafkaContainer() =
+    ConfluentKafkaContainer(
+        DockerImageName
+            .parse(KAFKA_IMAGE)
+            .asCompatibleSubstituteFor(KAFKA_IMAGE_NAME),
+    ).withStartupTimeout(CONTAINER_STARTUP_TIMEOUT_S.seconds.toJavaDuration())
 
-internal fun KafkaContainer.config() {
-    if (System.getProperty("os.name").lowercase().contains("mac")) {
-        withCreateContainerCmdModifier { it.withPlatform("linux/amd64") }
-    }
-    withNetworkAliases("kafka")
+internal fun ConfluentKafkaContainer.config() {
     withNetwork(kafkaNetwork)
-    withEnv("KAFKA_AUTO_CREATE_TOPIC_ENABLE", "false")
+    withNetworkAliases("kafka")
+    withEnv("KAFKA_AUTO_CREATE_TOPICS_ENABLE", "true")
     withEnv("KAFKA_TRANSACTION_STATE_LOG_MIN_ISR", "1")
     withEnv("KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR", "1")
+    withListener("kafka:9095")
 }
 
-internal fun SchemaRegistryContainer.config(kafka: KafkaContainer) {
-    waitingFor(Wait.forHttp("/subjects").forStatusCode(200).withStartupTimeout(120.seconds.toJavaDuration()))
+private const val CONTAINER_STARTUP_TIMEOUT_S = 30
+
+internal fun SchemaRegistryContainer.config(kafka: ConfluentKafkaContainer) {
+    waitingFor(
+        Wait
+            .forHttp("/subjects")
+            .forStatusCode(200)
+            .withStartupTimeout(CONTAINER_STARTUP_TIMEOUT_S.seconds.toJavaDuration()),
+    )
     withExposedPorts(SCHEMA_REGISTRY_PORT)
     withNetwork(kafka.network)
+    withNetworkAliases("schema-registry")
     withEnv("SCHEMA_REGISTRY_HOST_NAME", "schema-registry")
     withEnv("SCHEMA_REGISTRY_LISTENERS", "http://0.0.0.0:$SCHEMA_REGISTRY_PORT")
-    withEnv("SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS", "PLAINTEXT://${kafka.networkAliases[0]}:9092")
+    withEnv("SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS", "PLAINTEXT://kafka:9095")
+    withEnv("SCHEMA_REGISTRY_KAFKASTORE_SECURITY_PROTOCOL", "PLAINTEXT")
 }
 
 internal val zookeeperContainer =
@@ -125,6 +139,7 @@ internal fun SchemaRegistryContainer.sslConfig() {
     withCopyToContainer(MountableFile.forClasspathResource("/secrets"), "/etc/kafka/secrets")
 }
 
+@Suppress("unused")
 internal val dockerComposeContainer: DockerComposeContainer<*> =
     logErrors { DockerComposeContainer("kafka-ssl", File(yml)) }
 
@@ -146,28 +161,31 @@ internal abstract class FixedPortsContainer<SELF : GenericContainer<SELF>>(
         }
 }
 
-internal class SchemaRegistryContainer private constructor() :
-    FixedPortsContainer<SchemaRegistryContainer>("$schemaRegistryImage:$CONFLUENT_PLATFORM_VERSION") {
-        companion object {
-            val schemaRegistryImage: DockerImageName = DockerImageName.parse("confluentinc/cp-schema-registry")
+internal class SchemaRegistryContainer private constructor() : FixedPortsContainer<SchemaRegistryContainer>(SCHEMA_REGISTRY_IMAGE) {
+    companion object {
+        @Suppress("unused")
+        val schemaRegistryImage: DockerImageName = DockerImageName.parse(SCHEMA_REGISTRY_IMAGE)
 
-            fun new() = SchemaRegistryContainer()
-        }
+        fun new() = SchemaRegistryContainer()
     }
+}
 
-internal class KafkaGenericContainer private constructor() :
-    FixedPortsContainer<KafkaGenericContainer>("$kafkaImage:$CONFLUENT_PLATFORM_VERSION") {
-        companion object {
-            val kafkaImage: DockerImageName = DockerImageName.parse("confluentinc/cp-kafka")
+internal class KafkaGenericContainer private constructor() : FixedPortsContainer<KafkaGenericContainer>(KAFKA_IMAGE) {
+    companion object {
+        @Suppress("unused")
+        val kafkaImage: DockerImageName = DockerImageName.parse(KAFKA_IMAGE)
 
-            fun new() = KafkaGenericContainer()
-        }
+        fun new() = KafkaGenericContainer()
     }
+}
 
-internal fun KafkaContainer.withConsumeLogs() = withLogConsumer(kafkaLogConsumer)
+@Suppress("unused")
+internal fun ConfluentKafkaContainer.withConsumeLogs() = withLogConsumer(kafkaLogConsumer)
 
+@Suppress("unused")
 internal fun KafkaGenericContainer.withConsumeLogs() = withLogConsumer(kafkaLogConsumer)
 
+@Suppress("unused")
 internal fun SchemaRegistryContainer.withConsumeLogs() = withLogConsumer(schemaRegistryLogConsumer)
 
 private val kafkaLogger = KtorSimpleLogger("kafka")
diff --git a/ktor-server-kafka/src/jvmTest/kotlin/io/github/flaxoos/ktor/server/plugins/kafka/KafkaIntegrationTest.kt b/ktor-server-kafka/src/jvmTest/kotlin/io/github/flaxoos/ktor/server/plugins/kafka/KafkaIntegrationTest.kt
@@ -8,9 +8,10 @@ import io.kotest.extensions.testcontainers.ContainerExtension
 import io.kotest.extensions.testcontainers.ContainerLifecycleMode
 import io.ktor.server.application.install
 import kotlinx.serialization.Serializable
+import org.apache.kafka.clients.CommonClientConfigs
 import org.apache.kafka.clients.CommonClientConfigs.CLIENT_ID_CONFIG
 import org.apache.kafka.clients.consumer.ConsumerConfig.GROUP_ID_CONFIG
-import org.testcontainers.containers.KafkaContainer
+import org.testcontainers.kafka.ConfluentKafkaContainer
 import kotlin.time.Duration.Companion.seconds
 
 const val CODE_CONFIGURED_CLIENT_ID = "code-configured-client-id"
@@ -19,13 +20,13 @@ const val CODE_CONFIGURED_GROUP_ID = "code-configured-group-id"
 @OptIn(ExperimentalKotest::class)
 class KafkaIntegrationTest : BaseKafkaIntegrationTest() {
     init {
-        val kafkaContainer: KafkaContainer = newKafkaContainer()
+        val kafkaContainer: ConfluentKafkaContainer = newKafkaContainer()
         val schemaRegistryContainer = SchemaRegistryContainer.new()
         val kafka =
             install(
                 ContainerExtension(
                     kafkaContainer.apply { config() },
-                    mode = ContainerLifecycleMode.Project,
+                    mode = ContainerLifecycleMode.Spec,
                     beforeTest = {
                         bootstrapServers = kafkaContainer.bootstrapServers
                     },
@@ -41,7 +42,7 @@ class KafkaIntegrationTest : BaseKafkaIntegrationTest() {
                 },
             ),
         )
-        context("should pr oduce and consume records").config(timeout = 120.seconds) {
+        context("should produce and consume records").config(timeout = 20.seconds) {
             test("With default config path") {
                 editConfigurationFile()
                 testKafkaApplication {
@@ -66,7 +67,7 @@ class KafkaIntegrationTest : BaseKafkaIntegrationTest() {
                     install(Kafka) {
                         schemaRegistryUrl = resolvedSchemaRegistryUrl
                         setupTopics()
-                        common { bootstrapServers = listOf(kafka.bootstrapServers) }
+                        common { bootstrapServers = listOf(this@KafkaIntegrationTest.bootstrapServers) }
                         admin { clientId = CODE_CONFIGURED_CLIENT_ID }
                         producer { clientId = CODE_CONFIGURED_CLIENT_ID }
                         consumer {
@@ -84,7 +85,7 @@ class KafkaIntegrationTest : BaseKafkaIntegrationTest() {
                         setupTopics()
                         common {
                             additional {
-                                (listOf(kafka.bootstrapServers))
+                                CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG(this@KafkaIntegrationTest.bootstrapServers)
                             }
                         }
                         admin {
diff --git a/ktor-server-kafka/src/jvmTest/kotlin/io/github/flaxoos/ktor/server/plugins/kafka/KafkaSSLIntegrationTest.kt b/ktor-server-kafka/src/jvmTest/kotlin/io/github/flaxoos/ktor/server/plugins/kafka/KafkaSSLIntegrationTest.kt
@@ -8,6 +8,7 @@ import org.apache.kafka.clients.CommonClientConfigs
 import org.apache.kafka.common.config.SslConfigs
 import java.io.File
 import java.nio.file.Paths
+import kotlin.time.Duration.Companion.seconds
 
 class KafkaSSLIntegrationTest : BaseKafkaIntegrationTest() {
     override val additionalProducerProperties: Map<String, Any> by lazy {
@@ -30,7 +31,7 @@ class KafkaSSLIntegrationTest : BaseKafkaIntegrationTest() {
             install(
                 ContainerExtension(
                     zookeeperContainer,
-                    ContainerLifecycleMode.Project,
+                    ContainerLifecycleMode.Spec,
                     beforeStart = { generateCertificates() },
                 ),
             )
@@ -41,10 +42,10 @@ class KafkaSSLIntegrationTest : BaseKafkaIntegrationTest() {
                         sslConfig()
                         dependsOn(zooKeeper)
                     },
-                    ContainerLifecycleMode.Project,
-                    beforeStart = { zooKeeper.start() },
-                    beforeTest = {
+                    ContainerLifecycleMode.Spec,
+                    beforeStart = {
                         bootstrapServers = BOOTSTRAP_SERVERS
+                        zooKeeper.start()
                     },
                 ),
             )
@@ -54,7 +55,7 @@ class KafkaSSLIntegrationTest : BaseKafkaIntegrationTest() {
                     sslConfig()
                     dependsOn(kafka)
                 },
-                ContainerLifecycleMode.Project,
+                ContainerLifecycleMode.Spec,
                 beforeStart = {
                     kafkaContainer.start()
                 },
@@ -64,13 +65,13 @@ class KafkaSSLIntegrationTest : BaseKafkaIntegrationTest() {
             ),
         )
 
-        test("With code configuration") {
+        test("With code configuration").config(timeout = 20.seconds) {
             testKafkaApplication(waitSecondsAfterApplicationStart = 10) {
                 install(Kafka) {
                     schemaRegistryUrl = resolvedSchemaRegistryUrl
                     setupTopics()
                     common {
-                        bootstrapServers = listOf(BOOTSTRAP_SERVERS)
+                        bootstrapServers = listOf(this@KafkaSSLIntegrationTest.bootstrapServers)
                         securityProtocol = "SSL"
                     }
                     admin {
diff --git a/ktor-server-task-scheduling/ktor-server-task-scheduling-core/test/src/jvmMain/kotlin/io/github/flaxoos/ktor/server/plugins/taskscheduling/TaskSchedulingPluginTest.kt b/ktor-server-task-scheduling/ktor-server-task-scheduling-core/test/src/jvmMain/kotlin/io/github/flaxoos/ktor/server/plugins/taskscheduling/TaskSchedulingPluginTest.kt
@@ -19,12 +19,12 @@ import io.ktor.server.testing.TestApplication
 import io.ktor.server.testing.TestApplicationBuilder
 import korlibs.time.DateTime
 import korlibs.time.minutes
-import kotlin.math.pow
-import kotlin.time.Duration.Companion.milliseconds
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.joinAll
 import kotlinx.coroutines.launch
+import kotlin.math.pow
+import kotlin.time.Duration.Companion.milliseconds
 
 @Suppress("UNUSED")
 val logger = KotlinLogging.logger { }
@@ -77,10 +77,10 @@ abstract class TaskSchedulingPluginTest : FunSpec() {
                                     kronTaskSchedule = kronTaskSchedule(freqMs),
                                 ).map { executionsAndApp ->
                                     executionsAndApp to
-                                            launch {
-                                                val app = executionsAndApp.second
-                                                app.start()
-                                            }
+                                        launch {
+                                            val app = executionsAndApp.second
+                                            app.start()
+                                        }
                                 }.also { appsAndJobs ->
                                     // wait for all app engines to start
                                     appsAndJobs.map { appAndJob -> appAndJob.second }.joinAll()
@@ -196,12 +196,12 @@ abstract class TaskSchedulingPluginTest : FunSpec() {
                 }
             }
             executionRecords to
-                    TestApplication {
-                        engine {
-                            shutdownGracePeriod = freqMs * 10
-                        }
-                        block()
+                TestApplication {
+                    engine {
+                        shutdownGracePeriod = freqMs * 10
                     }
+                    block()
+                }
         }
 
     private fun exponentialScheduleGenerator(
diff --git a/ktor-server-task-scheduling/ktor-server-task-scheduling-redis/src/commonMain/kotlin/io/github/flaxoos/ktor/server/plugins/taskscheduling/managers/lock/redis/RedisLockManager.kt b/ktor-server-task-scheduling/ktor-server-task-scheduling-redis/src/commonMain/kotlin/io/github/flaxoos/ktor/server/plugins/taskscheduling/managers/lock/redis/RedisLockManager.kt
@@ -13,8 +13,8 @@ import io.github.flaxoos.ktor.server.plugins.taskscheduling.tasks.TaskLock
 import io.github.oshai.kotlinlogging.KotlinLogging
 import io.ktor.server.application.Application
 import korlibs.time.DateTime
-import kotlin.jvm.JvmInline
 import kotlinx.coroutines.runBlocking
+import kotlin.jvm.JvmInline
 
 internal val logger = KotlinLogging.logger { }
 
@@ -28,7 +28,6 @@ public class RedisLockManager(
     private val lockExpirationMs: Long,
     private val connectionAcquisitionTimeoutMs: Long,
 ) : TaskLockManager<RedisTaskLock>() {
-
     override suspend fun init(tasks: List<Task>) {}
 
     override suspend fun acquireLockKey(
diff --git a/settings.gradle.kts b/settings.gradle.kts
@@ -29,4 +29,3 @@ include("ktor-server-task-scheduling:ktor-server-task-scheduling-core:test")
 include("ktor-server-task-scheduling:ktor-server-task-scheduling-jdbc")
 include("ktor-server-task-scheduling:ktor-server-task-scheduling-mongodb")
 include("ktor-server-task-scheduling:ktor-server-task-scheduling-redis")
-include("ktor-server-task-scheduling:common-test")

Original file line number	Diff line number	Diff line change
`@@ -103,7 +103,7 @@ abstract class BaseKafkaIntegrationTest : FunSpec() {`
`103`	`103`	`}`
`104`	`104`	`isConnected = true`
`105`	`105`	`break`
`106`		`- } catch (e: Exception) {`
	`106`	`+ } catch (_: Exception) {`
`107`	`107`	`logger.info(`
`108`	`108`	`"Attempt $i to connect to Kafka broker at bootstrap.servers: " +`
`109`	`109`	`"$bootstrapServers failed, retrying",`