Fix breaking changes

tKe · Apr 6, 2024 · 837947c · 837947c
1 parent d3a091d
commit 837947c
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 3 deletions.
diff --git a/src/main/kotlin/io/github/nomisRev/kafka/Consumer.kt b/src/main/kotlin/io/github/nomisRev/kafka/Consumer.kt
@@ -64,7 +64,6 @@ import org.apache.kafka.clients.consumer.ConsumerRecords
 import org.apache.kafka.clients.consumer.KafkaConsumer
 import org.apache.kafka.clients.consumer.OffsetAndMetadata
 import org.apache.kafka.clients.consumer.RangeAssignor
-import org.apache.kafka.clients.consumer.internals.NoOpConsumerRebalanceListener
 import org.apache.kafka.common.metrics.Sensor
 import org.apache.kafka.common.serialization.Deserializer
 import org.apache.kafka.common.TopicPartition
@@ -195,7 +194,7 @@ public fun <K, V> List<ConsumerRecords<K, V>>.offsets(
 public fun <K, V> Flow<KafkaConsumer<K, V>>.subscribeTo(
   name: String,
   dispatcher: CoroutineDispatcher = IO,
-  listener: ConsumerRebalanceListener = NoOpConsumerRebalanceListener(),
+  listener: ConsumerRebalanceListener = NoOpConsumerRebalanceListener,
   timeout: kotlin.time.Duration = 500.milliseconds,
 ): Flow<ConsumerRecord<K, V>> = flatMapConcat { consumer ->
   consumer.subscribeTo(name, dispatcher, listener, timeout)
@@ -212,7 +211,7 @@ public fun <K, V> Flow<KafkaConsumer<K, V>>.subscribeTo(
 public fun <K, V> KafkaConsumer<K, V>.subscribeTo(
   name: String,
   dispatcher: CoroutineDispatcher = IO,
-  listener: ConsumerRebalanceListener = NoOpConsumerRebalanceListener(),
+  listener: ConsumerRebalanceListener = NoOpConsumerRebalanceListener,
   timeout: kotlin.time.Duration = 500.milliseconds,
 ): Flow<ConsumerRecord<K, V>> = flow {
   subscribe(listOf(name), listener)
@@ -365,3 +364,8 @@ public data class ConsumerSettings<K, V>(
     put(EXCLUDE_INTERNAL_TOPICS_CONFIG, excludeInternalTopics)
   }
 }
+
+private object NoOpConsumerRebalanceListener : ConsumerRebalanceListener {
+  override fun onPartitionsRevoked(partitions: MutableCollection<TopicPartition>?) {}
+  override fun onPartitionsAssigned(partitions: MutableCollection<TopicPartition>?) {}
+}
diff --git a/src/test/kotlin/io/github/nomisrev/kafka/KafkaSpec.kt b/src/test/kotlin/io/github/nomisrev/kafka/KafkaSpec.kt
@@ -34,6 +34,7 @@ import org.apache.kafka.common.Metric
 import org.apache.kafka.common.MetricName
 import org.apache.kafka.common.PartitionInfo
 import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.Uuid
 import org.apache.kafka.common.serialization.StringDeserializer
 import org.apache.kafka.common.serialization.StringSerializer
 import org.junit.jupiter.api.AfterAll
@@ -295,6 +296,8 @@ abstract class KafkaSpec {
     {
       val producer = KafkaProducer(it.properties(), it.keySerializer, it.valueSerializer)
       object : Producer<String, String> {
+        override fun clientInstanceId(p0: Duration?): Uuid =
+          producer.clientInstanceId(p0)
         override fun close() {}
 
         override fun close(timeout: Duration?) {}