Fix issue with multiple partitions halting processing (#52)

* Fix issue with multiple partitions halting processing

Author: mtps

ft-coroutines-kafka/src/main/kotlin/tech/figure/kafka/coroutines/channels/ConsumerSendChannel.kt

@@ -197,7 +197,7 @@ internal class KafkaAckConsumerChannel<K, V>(
         log.trace { "preProcessPollSet(tp:$topicPartition count:${records.size})" }
         val ackChannel =
             Channel<CommitConsumerRecord>(capacity = records.size).also {
-                context["ack-channel"] = it
+                context["ack-channel-$topicPartition"] = it
             }
         return records.map {
             val timestamp = System.currentTimeMillis()
@@ -212,7 +212,7 @@ internal class KafkaAckConsumerChannel<K, V>(
         context: Map<String, Any>
     ) {
         log.trace { "postProcessPollSet(tp:$topicPartition count:${records.size})" }
-        val ackChannel = context["ack-channel"]!! as ReceiveChannel<CommitConsumerRecord>
+        val ackChannel = context["ack-channel-$topicPartition"]!! as ReceiveChannel<CommitConsumerRecord>
         if (records.isEmpty()) {
             log.trace { "empty record set, not waiting for acks" }
             return
@@ -303,20 +303,20 @@ abstract class KafkaConsumerChannel<K, V, R>(
     fun run() {
         consumer.init()
 
-        log.info("starting thread for ${consumer.subscription()}")
+        log.info { "starting thread for ${consumer.subscription()}" }
         runBlocking {
-            log.info("${coroutineContext.job} running consumer ${consumer.subscription()}")
+            log.info { "${coroutineContext.job} running consumer ${consumer.subscription()}" }
             try {
                 while (!sendChannel.isClosedForSend) {
-                    log.trace("poll(topics:${consumer.subscription()}) ...")
+                    log.trace { "poll(topics:${consumer.subscription()}) ..." }
                     val polled =
                         consumer.poll(Duration.ZERO).ifEmpty { consumer.poll(pollInterval) }
                     val polledCount = polled.count()
                     if (polledCount == 0) {
                         continue
                     }
 
-                    log.trace("poll(topics:${consumer.subscription()}) got $polledCount records.")
+                    log.trace { "poll(topics:${consumer.subscription()}) got $polledCount records." }
 
                     // Convert to internal types.
                     val context = mutableMapOf<String, Any>()
@@ -334,7 +334,7 @@ abstract class KafkaConsumerChannel<K, V, R>(
                     }
                 }
             } finally {
-                log.info("${coroutineContext.job} shutting down consumer thread")
+                log.info { "${coroutineContext.job} shutting down consumer thread" }
                 try {
                     sendChannel.cancel(CancellationException("consumer shut down"))
                     consumer.unsubscribe()
@@ -352,7 +352,7 @@ abstract class KafkaConsumerChannel<K, V, R>(
         if (!thread.isAlive) {
             synchronized(thread) {
                 if (!thread.isAlive) {
-                    log.info("starting consumer thread")
+                    log.info { "starting consumer thread" }
                     thread.start()
                 }
             }

ft-coroutines-kafka/src/main/kotlin/tech/figure/kafka/records/AckConsumerRecord.kt

@@ -11,3 +11,5 @@ class AckedConsumerRecordImpl<K, V>(
     record: ConsumerRecord<K, V>,
     override val metadata: OffsetAndMetadata
 ) : AckedConsumerRecord<K, V>, KafkaRecord<K, V> by wrapping(record)
+
+typealias AckedConsumerRecords<K, V> = List<AckedConsumerRecord<K, V>>

ft-coroutines-kafka/src/main/kotlin/tech/figure/kafka/records/UnAckedConsumerRecord.kt

@@ -59,5 +59,4 @@ class UnAckedConsumerRecordImpl<K, V>(
         """.trimIndent().split('\n').joinToString { it.trim() }
 }
 
-@JvmInline
-value class UnAckedConsumerRecords<K, V>(val records: List<UnAckedConsumerRecord<K, V>>)
+typealias UnAckedConsumerRecords<K, V> = List<UnAckedConsumerRecord<K, V>>