Add power-assert, add utilities, clean-up

build.gradle.kts

@@ -1,3 +1,4 @@
+import com.bnorm.power.PowerAssertGradleExtension
 import kotlinx.knit.KnitPluginExtension
 import org.gradle.api.tasks.testing.logging.TestExceptionFormat
 import org.gradle.api.tasks.testing.logging.TestLogEvent
@@ -9,6 +10,7 @@ plugins {
   alias(libs.plugins.spotless)
   alias(libs.plugins.knit)
   alias(libs.plugins.publish)
+  alias(libs.plugins.power.assert)
 }
 
 repositories {
@@ -24,11 +26,15 @@ dependencies {
   api(libs.kafka.clients)
   implementation(libs.slf4j.api)
 
-  testImplementation(libs.bundles.kotest)
+  testImplementation(kotlin("test"))
   testImplementation(libs.testcontainers.kafka)
   testImplementation(libs.slf4j.simple)
 }
 
+configure<PowerAssertGradleExtension> {
+  functions = listOf("kotlin.test.assertEquals")
+}
+
 configure<KnitPluginExtension> {
   siteRoot = "https://nomisrev.github.io/kotlin-kafka/"
 }

gradle/libs.versions.toml

@@ -10,12 +10,10 @@ testcontainers-kafka = "1.19.4"
 slf4j = "2.0.11"
 spotless="6.25.0"
 publish="0.27.0"
+power-assert="0.13.0"
 
 [libraries]
-kotest-assertions = { module = "io.kotest:kotest-assertions-core", version.ref = "kotest" }
-kotest-framework = { module = "io.kotest:kotest-framework-engine", version.ref = "kotest" }
 kotest-property = { module = "io.kotest:kotest-property", version.ref = "kotest" }
-kotest-runner-junit5 = { module = "io.kotest:kotest-runner-junit5", version.ref = "kotest" }
 kafka-connect = { module = "org.apache.kafka:connect-runtime", version.ref = "kafka" }
 kafka-clients = { module = "org.apache.kafka:kafka-clients", version.ref = "kafka" }
 kafka-streams = { module = "org.apache.kafka:kafka-streams", version.ref = "kafka" }
@@ -27,18 +25,11 @@ testcontainers-kafka = { module = "org.testcontainers:kafka", version.ref = "tes
 slf4j-api = { module = "org.slf4j:slf4j-api", version.ref = "slf4j" }
 slf4j-simple = { module = "org.slf4j:slf4j-simple", version.ref = "slf4j" }
 
-[bundles]
-kotest = [
-    "kotest-assertions",
-    "kotest-framework",
-    "kotest-property",
-    "kotest-runner-junit5"
-]
-
 [plugins]
 kotlin-jvm = { id = "org.jetbrains.kotlin.jvm", version.ref = "kotlin" }
 dokka = { id = "org.jetbrains.dokka", version.ref = "dokka" }
 kover = { id = "org.jetbrains.kotlinx.kover", version.ref = "kover" }
 spotless = { id = "com.diffplug.spotless", version.ref = "spotless" }
 publish = { id = "com.vanniktech.maven.publish", version.ref="publish" }
 knit = { id = "org.jetbrains.kotlinx.knit", version.ref="knit" }
+power-assert = { id = "com.bnorm.power.kotlin-power-assert", version.ref="power-assert" }

src/main/kotlin/io/github/nomisRev/kafka/receiver/KafkaReceiver.kt

@@ -22,6 +22,7 @@ public interface KafkaReceiver<K, V> {
 
   public fun receive(topicName: String): Flow<ReceiverRecord<K, V>> =
     receive(setOf(topicName))
+      .buffer()
 
   public fun receiveAutoAck(topicNames: Collection<String>): Flow<Flow<ConsumerRecord<K, V>>>
 

src/main/kotlin/io/github/nomisRev/kafka/receiver/internals/PollLoop.kt

@@ -136,12 +136,14 @@ internal class CommittableOffset<K, V>(
 ) : Offset {
   private val acknowledged = AtomicBoolean(false)
 
+  // TODO should throw if called after Flow finished
   override suspend fun commit(): Unit =
     if (maybeUpdateOffset() > 0) suspendCoroutine { cont ->
       loop.commitBatch.addContinuation(cont)
       loop.scheduleCommitIfRequired()
     } else Unit
 
+  // TODO should throw if called after Flow finished
   override suspend fun acknowledge() {
     val uncommittedCount = maybeUpdateOffset().toLong()
     if (commitBatchSize in 1..uncommittedCount) {

src/test/kotlin/io/github/nomisrev/kafka/KafkaContainer.kt

@@ -0,0 +1,15 @@
+package io.github.nomisrev.kafka
+
+import org.testcontainers.containers.KafkaContainer
+import org.testcontainers.utility.DockerImageName
+
+class Kafka : KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:latest")) {
+
+  fun pause() {
+    dockerClient.pauseContainerCmd(containerId).exec()
+  }
+
+  fun unpause() {
+    dockerClient.unpauseContainerCmd(containerId).exec()
+  }
+}