Merge pull request #185 from nomisRev/beta5

2.0.0-Beta5 build failure

Author: nomisRev

build.gradle.kts

@@ -1,22 +1,18 @@
-import com.bnorm.power.PowerAssertGradleExtension
 import kotlinx.knit.KnitPluginExtension
-import org.gradle.api.tasks.testing.logging.TestExceptionFormat
 import org.gradle.api.tasks.testing.logging.TestExceptionFormat.*
-import org.gradle.api.tasks.testing.logging.TestLogEvent
 import org.gradle.api.tasks.testing.logging.TestLogEvent.FAILED
-import org.gradle.api.tasks.testing.logging.TestLogEvent.PASSED
 import org.gradle.api.tasks.testing.logging.TestLogEvent.SKIPPED
 import org.gradle.api.tasks.testing.logging.TestLogEvent.STANDARD_ERROR
-import org.gradle.api.tasks.testing.logging.TestLogEvent.STANDARD_OUT
 import org.jetbrains.dokka.gradle.DokkaTask
+import org.jetbrains.kotlin.gradle.dsl.KotlinVersion.KOTLIN_2_0
 
 plugins {
   alias(libs.plugins.kotlin.jvm)
+  alias(libs.plugins.kotlin.assert)
   alias(libs.plugins.dokka)
   alias(libs.plugins.spotless)
   alias(libs.plugins.knit)
   alias(libs.plugins.publish)
-  alias(libs.plugins.power.assert)
 }
 
 repositories {
@@ -38,10 +34,15 @@ dependencies {
   testImplementation(libs.kotlinx.coroutines.test)
 }
 
-configure<PowerAssertGradleExtension> {
+@Suppress("OPT_IN_USAGE")
+powerAssert {
   functions = listOf("kotlin.test.assertEquals")
 }
 
+//configure<PowerAssertGradleExtension> {
+//  functions = listOf("kotlin.test.assertEquals")
+//}
+
 configure<KnitPluginExtension> {
   siteRoot = "https://nomisrev.github.io/kotlin-kafka/"
 }
@@ -54,6 +55,10 @@ configure<JavaPluginExtension> {
 
 kotlin {
   explicitApi()
+  compilerOptions {
+    languageVersion.set(KOTLIN_2_0)
+    apiVersion.set(KOTLIN_2_0)
+  }
 }
 
 tasks {

gradle/libs.versions.toml

@@ -1,7 +1,7 @@
 [versions]
 kotest = "5.8.1"
 kafka = "3.7.0"
-kotlin = "1.9.23"
+kotlin = "2.0.0-RC2"
 kotlinx-coroutines = "1.8.0"
 dokka = "1.9.20"
 knit = "0.5.0"
@@ -10,7 +10,6 @@ testcontainers-kafka = "1.19.7"
 slf4j = "2.0.12"
 spotless="6.25.0"
 publish="0.28.0"
-power-assert="0.13.0"
 
 [libraries]
 kotest-property = { module = "io.kotest:kotest-property", version.ref = "kotest" }
@@ -28,9 +27,9 @@ slf4j-simple = { module = "org.slf4j:slf4j-simple", version.ref = "slf4j" }
 
 [plugins]
 kotlin-jvm = { id = "org.jetbrains.kotlin.jvm", version.ref = "kotlin" }
+kotlin-assert = { id = "org.jetbrains.kotlin.plugin.power-assert", 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/internal/FlowTimeChunked.kt

@@ -10,8 +10,9 @@ import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.ReceiveChannel
 import kotlinx.coroutines.channels.getOrElse
 import kotlinx.coroutines.channels.produce
+import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.internal.scopedFlow
+import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.selects.whileSelect
 import kotlinx.coroutines.selects.onTimeout
 import kotlin.time.Duration
@@ -45,30 +46,32 @@ public fun <T> Flow<T>.chunked(
 ): Flow<List<T>> {
   require(size > 0) { "Cannot create chunks smaller than 0 but found $size" }
   require(!duration.isNegative() && duration != Duration.ZERO) { "Chunk duration should be positive non-zero duration" }
-  return scopedFlow { downstream ->
-    val emitNowAndMaybeContinue = Channel<Boolean>(capacity = Channel.RENDEZVOUS)
-    val elements = produce(capacity = size) {
-      collect { element ->
-        val hasCapacity = channel.trySend(element).isSuccess
-        if (!hasCapacity) {
-          emitNowAndMaybeContinue.send(true)
-          channel.send(element)
+  return flow {
+    coroutineScope {
+      val emitNowAndMaybeContinue = Channel<Boolean>(capacity = Channel.RENDEZVOUS)
+      val elements = produce(capacity = size) {
+        collect { element ->
+          val hasCapacity = channel.trySend(element).isSuccess
+          if (!hasCapacity) {
+            emitNowAndMaybeContinue.send(true)
+            channel.send(element)
+          }
         }
+        emitNowAndMaybeContinue.send(false)
       }
-      emitNowAndMaybeContinue.send(false)
-    }
 
-    whileSelect {
-      emitNowAndMaybeContinue.onReceive { shouldContinue ->
-        val chunk = elements.drain(maxElements = size)
-        if (chunk.isNotEmpty()) downstream.emit(chunk)
-        shouldContinue
-      }
+      whileSelect {
+        emitNowAndMaybeContinue.onReceive { shouldContinue ->
+          val chunk = elements.drain(maxElements = size)
+          if (chunk.isNotEmpty()) emit(chunk)
+          shouldContinue
+        }
 
-      onTimeout(duration) {
-        val chunk: List<T> = elements.drain(maxElements = size)
-        if (chunk.isNotEmpty()) downstream.emit(chunk)
-        true
+        onTimeout(duration) {
+          val chunk: List<T> = elements.drain(maxElements = size)
+          if (chunk.isNotEmpty()) emit(chunk)
+          true
+        }
       }
     }
   }

src/main/kotlin/io/github/nomisRev/kafka/publisher/PublisherScope.kt

@@ -1,5 +1,4 @@
-@file:JvmMultifileClass @file:JvmName("PublisherScope.kt")
-
+//@file:JvmName("PublisherScope")
 package io.github.nomisRev.kafka.publisher
 
 import kotlinx.coroutines.CoroutineScope

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

@@ -93,7 +93,7 @@ internal class CommittableBatch {
   fun getAndClearOffsets(): CommitArgs {
     val offsetMap: MutableMap<TopicPartition, OffsetAndMetadata> = HashMap()
     if (outOfOrderCommits) {
-      deferred.forEach { (tp: TopicPartition, offsets: List<Long>) ->
+      deferred.forEach { (tp: TopicPartition, offsets: MutableList<Long>) ->
         if (offsets.size > 0) {
           offsets.sort()
           val uncomittedThisPart: MutableList<Long> = uncommitted[tp]!!