feat(clients): make ConsumerTask/ProducerContainer Closeaable (#5)

Resolves: GH-5

Author: fhussonnois

clients/src/main/kotlin/io/streamthoughts/kafka/clients/consumer/ConsumerTask.kt

@@ -21,9 +21,10 @@ package io.streamthoughts.kafka.clients.consumer
 import org.apache.kafka.clients.consumer.Consumer
 import org.apache.kafka.clients.consumer.OffsetAndMetadata
 import org.apache.kafka.common.TopicPartition
+import java.io.Closeable
 import java.time.Duration
 
-interface ConsumerTask{
+interface ConsumerTask: Closeable {
 
     enum class State {
         /**
@@ -78,13 +79,13 @@ interface ConsumerTask{
      * Shutdowns the [ConsumerTask] and wait for completion.
      * @see org.apache.kafka.clients.consumer.Consumer.close
      */
-    fun shutdown()
+    override fun close()
 
     /**
      * Shutdowns the [ConsumerTask] and wait for completion until the given [timeout].
      * @see org.apache.kafka.clients.consumer.Consumer.close
      */
-    fun shutdown(timeout: Duration)
+    fun close(timeout: Duration)
 
     /**
      * @return the [State] of this [ConsumerTask].

clients/src/main/kotlin/io/streamthoughts/kafka/clients/consumer/ConsumerWorker.kt

@@ -21,13 +21,15 @@ package io.streamthoughts.kafka.clients.consumer
 import io.streamthoughts.kafka.clients.consumer.error.ConsumedErrorHandler
 import io.streamthoughts.kafka.clients.consumer.error.serialization.DeserializationErrorHandler
 import io.streamthoughts.kafka.clients.consumer.listener.ConsumerBatchRecordsListener
+import java.io.Closeable
+import java.time.Duration
 import java.util.regex.Pattern
 
 /**
  * The [ConsumerWorker] manages one or many concurrent [org.apache.kafka.clients.consumer.Consumer] that belong
  * to the same {@code group.id}.
  */
-interface ConsumerWorker<K, V> {
+interface ConsumerWorker<K, V>: Closeable {
 
 
     interface Builder<K, V> {
@@ -115,7 +117,7 @@ interface ConsumerWorker<K, V> {
     /**
      * Stops all [org.apache.kafka.clients.consumer.Consumer] managed by this [ConsumerWorker].
      */
-    fun stop()
+    override fun close()
 
     /**
      * Pauses all [org.apache.kafka.clients.consumer.Consumer] managed by this [ConsumerWorker].

clients/src/main/kotlin/io/streamthoughts/kafka/clients/consumer/KafkaConsumerTask.kt

@@ -221,14 +221,14 @@ class KafkaConsumerTask<K, V>(
         }
     }
 
-    override fun shutdown() {
+    override fun close() {
         logWithConsumerInfo(Level.INFO, "Closing")
         isShutdown.set(true)
         consumer.wakeup()
         shutdownLatch.await()
     }
 
-    override fun shutdown(timeout: Duration) {
+    override fun close(timeout: Duration) {
         logWithConsumerInfo(Level.INFO, "Closing")
         isShutdown.set(true)
         consumer.wakeup()

clients/src/main/kotlin/io/streamthoughts/kafka/clients/consumer/KafkaConsumerWorker.kt

@@ -139,10 +139,10 @@ class KafkaConsumerWorker<K, V> (
         consumerJobs.joinAll()
     }
 
-    override fun stop() {
+    override fun close() {
         if (isRunning.get()) {
             Log.info("KafkaConsumerWorker(group: $groupId): Stopping all io.streamthoughts.kafka.clients.consumer tasks")
-            consumerTasks.forEach { it.shutdown() }
+            consumerTasks.forEach { it.close() }
             isRunning.set(false)
         }
     }

clients/src/main/kotlin/io/streamthoughts/kafka/clients/consumer/error/ConsumedErrorHandlers.kt

@@ -43,7 +43,7 @@ private object CloseTaskOnConsumedError: ConsumedErrorHandler {
 
     override fun handle(consumerTask: ConsumerTask, records: List<ConsumerRecord<*, *>>, thrownException: Exception) {
         Log.error("Stopping consumerTask after an exception was thrown while processing records", thrownException)
-        consumerTask.shutdown(Duration.ZERO)
+        consumerTask.close(Duration.ZERO)
     }
 }
 

clients/src/main/kotlin/io/streamthoughts/kafka/clients/producer/KafkaProducerContainer.kt

@@ -199,7 +199,7 @@ class KafkaProducerContainer<K, V> private constructor(
     }
 
     override fun close(timeout: Duration) {
-        if (isClosed()) return // silently ignore call if producer is already closed.
+        if (isClosed() || !isInitialized()) return // silently ignore call if producer is already closed.
 
         runOrThrowIfIllegalState {
             state = ProducerContainer.State.PENDING_SHUTDOWN

clients/src/main/kotlin/io/streamthoughts/kafka/clients/producer/ProducerContainer.kt

@@ -28,6 +28,7 @@ import org.apache.kafka.common.Metric
 import org.apache.kafka.common.MetricName
 import org.apache.kafka.common.PartitionInfo
 import org.apache.kafka.common.serialization.Serializer
+import java.io.Closeable
 import java.time.Duration
 import java.time.Instant
 import java.util.concurrent.Future
@@ -52,7 +53,7 @@ data class UnrecoverableErrorTransactionResult( val exception: Exception): Trans
  */
 object CommittedTransactionResult: TransactionResult()
 
-interface ProducerContainer<K, V> {
+interface ProducerContainer<K, V>: Closeable {
 
     enum class State {
         /**
@@ -261,7 +262,7 @@ interface ProducerContainer<K, V> {
      *
      * @see [Producer.close].
      */
-    fun close() {
+    override fun close() {
         close(Duration.ofMillis(Long.MAX_VALUE))
     }
 

clients/src/test/kotlin/io/streamthoughts/kafka/clients/consumer/KafkaConsumerTaskTest.kt

@@ -114,7 +114,7 @@ class KafkaConsumerTaskTest(private val cluster: TestingEmbeddedKafka) {
             try {
                 captureHandler.assertThatEventuallyCapture("Fail to capture processing error before timeout")
             } finally {
-                consumer.shutdown()
+                consumer.close()
             }
         }
     }

examples/src/main/kotlin/io/streamthoughts/kafka/client/examples/ConsumerKotlinDSLExample.kt

@@ -65,12 +65,11 @@ fun main(args: Array<String>) {
         }
     }
 
-    with (consumerWorker) {
-        start("demo-topic", maxParallelHint = 4)
+    consumerWorker.use {
+        consumerWorker.start("demo-topic", maxParallelHint = 4)
         runBlocking {
             println("All consumers started, waiting one minute before stopping")
             delay(Duration.ofMinutes(1).toMillis())
-            stop()
         }
     }
 }

examples/src/main/kotlin/io/streamthoughts/kafka/client/examples/ProducerKotlinDSLExample.kt

@@ -19,7 +19,6 @@
 package io.streamthoughts.kafka.client.examples
 
 import io.streamthoughts.kafka.clients.kafka
-import io.streamthoughts.kafka.clients.loadClientConfigs
 import io.streamthoughts.kafka.clients.producer.Acks
 import io.streamthoughts.kafka.clients.producer.ProducerContainer
 import io.streamthoughts.kafka.clients.producer.callback.closeOnErrorProducerSendCallback
@@ -49,11 +48,11 @@ fun main(args: Array<String>) {
         }
     }
 
-    with(producer) {
-        init()
-        listOf("I ❤️ Logs", "Making Sense of Stream Processing", "Apache Kafka").forEach {
-            send(value = it)
+    val messages = listOf("I ❤️ Logs", "Making Sense of Stream Processing", "Apache Kafka")
+    producer.use {
+        producer.init() // create internal KafkaProducer and eventually call initTransaction if transactional.id is set
+        messages.forEach {
+            producer.send(value = it)
         }
-        close()
     }
 }