refactor(clients): enhance configs classes

Additional changes:
 - update readme with new features

Author: fhussonnois

README.md

@@ -46,9 +46,40 @@ Just add **Kafka Clients for Kotlin** to the dependencies of your projects.
 
 ## Getting Started
 
-### Kafka Producer
+### Writing messages to Kafka
 
-See the full code-snippet : [ProducerExample.kt](https://github.com/streamthoughts/kafka-clients-kotlin/blob/master/examples/src/main/kotlin/io/streamthoughts/kafka/client/examples/ProducerKotlinDSLExample.kt)
+**Example: How to create `KafkaProducer` config ?**
+
+```kotlin
+val configs = producerConfigsOf()
+    .client { bootstrapServers("localhost:9092") }
+    .acks(Acks.Leader)
+    .keySerializer(StringSerializer::class.java.name)
+    .valueSerializer(StringSerializer::class.java.name)
+```
+
+**Example with standard `KafkaProducer` (i.e : using java `kafka-clients`)**
+
+```kotlin
+val producer = KafkaProducer<String, String>(configs)
+
+val messages = listOf("I ❤️ Logs", "Making Sense of Stream Processing", "Apache Kafka")
+producer.use {
+    messages.forEach {value ->
+        val record = ProducerRecord<String, String>(topic, value)
+        producer.send(record) { m: RecordMetadata, e: Exception? ->
+            when (e) {
+                null -> println("Record was successfully sent (topic=${m.topic()}, partition=${m.partition()}, offset= ${m.offset()})")
+                else -> e.printStackTrace()
+            }
+        }
+    }
+}
+```
+
+N.B: See the full source code: [ProducerClientExample.kt](https://github.com/streamthoughts/kafka-clients-kotlin/blob/master/examples/src/main/kotlin/io/streamthoughts/kafka/client/examples/ProducerClientExample.kt)
+
+**Example with Kotlin DSL**
 
 ```kotlin
 val producer: ProducerContainer<String, String> = kafka("localhost:9092") {
@@ -65,24 +96,59 @@ val producer: ProducerContainer<String, String> = kafka("localhost:9092") {
 
         defaultTopic("demo-topic")
 
+        onSendError {_, _, error ->
+            e.printStackTrace()
+        }
+
         onSendSuccess{ _, _, metadata ->
             println("Record was sent successfully: topic=${metadata.topic()}, partition=${metadata.partition()}, offset=${metadata.offset()} ")
         }
     }
 }
 
-with(producer) {
-    init()
-    listOf("I ❤️ Logs", "Making Sense of Stream Processing", "Apache Kafka").forEach {
-        send(value = it)
-    }
-    close()
+val messages = listOf("I ❤️ Logs", "Making Sense of Stream Processing", "Apache Kafka")
+producer.use {
+    producer.init() // create internal producer and call initTransaction() if `transactional.id` is set
+    messages.forEach { producer.send(value = it) }
 }
 ```
 
-### Kafka Consumer
+N.B: See the full source code: [ProducerKotlinDSLExample.kt](https://github.com/streamthoughts/kafka-clients-kotlin/blob/master/examples/src/main/kotlin/io/streamthoughts/kafka/client/examples/ProducerKotlinDSLExample.kt)
+
+### Consuming messages from a Kafka topic
+
+
+**Example: How to create `KafkaConsumer` config ?**
+
+```kotlin
+val configs = consumerConfigsOf()
+    .client { bootstrapServers("localhost:9092") }
+    .groupId("demo-consumer-group")
+    .keyDeserializer(StringDeserializer::class.java.name)
+    .valueDeserializer(StringDeserializer::class.java.name)
+```
+
+**Example with standard `KafkaConsumer` (i.e : using java `kafka-clients`)**
+
+```kotlin
+val consumer = KafkaConsumer<String, String>(configs)
+
+consumer.use {
+    consumer.subscribe(listOf(topic))
+    while(true) {
+        consumer
+            .poll(Duration.ofMillis(500))
+            .forEach { record ->
+                println(
+                    "Received record with key ${record.key()} " +
+                    "and value ${record.value()} from topic ${record.topic()} and partition ${record.partition()}"
+                )
+            }
+    }
+}
+```
 
-See the full code-snippet : [ConsumerExample.kt](https://github.com/streamthoughts/kafka-clients-kotlin/blob/master/examples/src/main/kotlin/io/streamthoughts/kafka/client/examples/ConsumerKotlinDSLExample.kt)
+N.B: See the full source code: [ConsumerClientExample.kt](https://github.com/streamthoughts/kafka-clients-kotlin/blob/master/examples/src/main/kotlin/io/streamthoughts/kafka/client/examples/ConsumerClientExample.kt)
 
 ```kotlin
 val consumerWorker: ConsumerWorker<String, String> = kafka("localhost:9092") {
@@ -97,7 +163,7 @@ val consumerWorker: ConsumerWorker<String, String> = kafka("localhost:9092") {
             autoOffsetReset(AutoOffsetReset.Earliest)
         }
 
-        onDeserializationError(DeserializationErrorHandlers.silentlyReplaceWithNull())
+        onDeserializationError(silentlyReplaceWithNull())
 
         onPartitionsAssigned { _: Consumer<*, *>, partitions ->
             println("Partitions assigned: $partitions")
@@ -111,12 +177,23 @@ val consumerWorker: ConsumerWorker<String, String> = kafka("localhost:9092") {
             println("consumed record-value: $value")
         }
 
+        onConsumedError(closeTaskOnConsumedError())
+
         Runtime.getRuntime().addShutdownHook(Thread { run { stop() } })
     }
 }
-consumerWorker.start("topic-test", 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())
+    }
+}
 ```
 
+N.B: See the full source code: [ConsumerKotlinDSLExample.kt](https://github.com/streamthoughts/kafka-clients-kotlin/blob/master/examples/src/main/kotlin/io/streamthoughts/kafka/client/examples/ConsumerKotlinDSLExample.kt)
+
 ## How to build project ?
 
 Kafka Clients for Kotlin uses [maven-wrapper](https://github.com/takari/maven-wrapper).

clients/src/main/kotlin/io/streamthoughts/kafka/clients/Configs.kt

@@ -28,68 +28,68 @@ import kotlin.collections.HashMap
  * @see io.streamthoughts.kafka.clients.consumer.KafkaConsumerConfigs
  * @see io.streamthoughts.kafka.clients.producer.KafkaProducerConfigs
  */
-open class Configs protected constructor(backed: Map<String, Any?> = emptyMap()) : MutableMap<String, Any?> {
+open class Configs protected constructor(props: Map<String, Any?> = emptyMap()) : MutableMap<String, Any?> {
 
-    private val mutableMap = HashMap(backed)
+    private val props = HashMap(props)
 
     override val entries: MutableSet<MutableMap.MutableEntry<String, Any?>>
-        get() = mutableMap.entries
+        get() = props.entries
 
     override val keys: MutableSet<String>
-        get() = mutableMap.keys
+        get() = props.keys
 
     override val size: Int
-        get() = mutableMap.size
+        get() = props.size
 
     override val values: MutableCollection<Any?>
-        get() = mutableMap.values
+        get() = props.values
 
     override fun containsKey(key: String): Boolean {
-        return mutableMap.containsKey(key)
+        return props.containsKey(key)
     }
 
     override fun containsValue(value: Any?): Boolean {
-        return mutableMap.containsValue(value)
+        return props.containsValue(value)
     }
 
     override fun get(key: String): Any? {
-        return mutableMap[key]
+        return props[key]
     }
 
     override fun isEmpty(): Boolean {
-        return mutableMap.isEmpty()
+        return props.isEmpty()
     }
 
     open fun with(key: String, value: Any?) = apply { this[key] = value }
 
     operator fun set(key: String, value: Any?) {
-        mutableMap[key] = value
+        props[key] = value
     }
 
     override fun equals(other: Any?): Boolean {
         if (this === other) return true
         if (other !is Configs) return false
 
-        if (mutableMap != other.mutableMap) return false
+        if (props != other.props) return false
 
         return true
     }
 
     override fun hashCode(): Int {
-        return mutableMap.hashCode()
+        return props.hashCode()
     }
 
     override fun toString(): String {
-        return "Configs[$mutableMap]"
+        return "Configs[$props]"
     }
 
     override fun clear() {
-       mutableMap.clear()
+        props.clear()
     }
 
-    override fun put(key: String, value: Any?): Any? = mutableMap.put(key, value)
+    override fun put(key: String, value: Any?): Any? = props.put(key, value)
 
-    override fun putAll(from: Map<out String, Any?>) = mutableMap.putAll(from)
+    override fun putAll(from: Map<out String, Any?>) = props.putAll(from)
 
-    override fun remove(key: String): Any?  = mutableMap.remove(key)
+    override fun remove(key: String): Any?  = props.remove(key)
 }

clients/src/main/kotlin/io/streamthoughts/kafka/clients/KafkaClientConfigs.kt

@@ -18,34 +18,24 @@
  */
 package io.streamthoughts.kafka.clients
 
-import io.streamthoughts.kafka.clients.consumer.KafkaConsumerConfigs
-import io.streamthoughts.kafka.clients.producer.KafkaProducerConfigs
 import org.apache.kafka.clients.CommonClientConfigs
 import java.io.InputStream
 import java.util.Properties
+import kotlin.collections.HashMap
+import kotlin.collections.Map
+import kotlin.collections.emptyMap
+import kotlin.collections.joinToString
+import kotlin.collections.mutableMapOf
+
 
 open class KafkaClientConfigs constructor(props: Map<String, Any?> = emptyMap()): Configs(props) {
 
-    constructor(kafka : Kafka): this(mapOf(
-        Pair(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, kafka.bootstrapServers.joinToString())
-    ))
+    constructor(kafka : Kafka): this(bootstrapServersConfig(kafka))
 
     companion object {
-
-        /**
-         * Creates a new [KafkaClientConfigs] with no properties.
-         */
-        fun empty() = KafkaClientConfigs()
-
-        /**
-         * Creates a new [KafkaClientConfigs] with the given [props].
-         */
-        fun of(props: Map<String, Any?>) = KafkaClientConfigs(props)
-
-        /**
-         * Creates a new [KafkaClientConfigs] with the given [props].
-         */
-        fun of(props: Properties) = of(props.toStringMap())
+        private fun bootstrapServersConfig(kafka: Kafka) = mutableMapOf<String, Any?>(
+            Pair(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, kafka.bootstrapServers.joinToString())
+        )
     }
 
     /**
@@ -72,35 +62,24 @@ open class KafkaClientConfigs constructor(props: Map<String, Any?> = emptyMap())
 /**
  * Convenient method to create and populate a new [KafkaClientConfigs] from a [configFile].
  */
-fun loadClientConfigs(configFile: String): KafkaClientConfigs
-        = KafkaClientConfigs().load(configFile)
+fun loadClientConfigs(configFile: String): KafkaClientConfigs = KafkaClientConfigs().load(configFile)
 
 /**
  * Convenient method to create and populate a new [KafkaClientConfigs] from an [inputStream].
  */
-fun loadClientConfigs(inputStream: InputStream): KafkaClientConfigs
-        = KafkaClientConfigs().load(inputStream)
+fun loadClientConfigs(inputStream: InputStream): KafkaClientConfigs = KafkaClientConfigs().load(inputStream)
 
 /**
- * Convenient method to create and populate a new [KafkaProducerConfigs] from a [configFile].
- */
-fun loadProducerConfigs(configFile: String): KafkaProducerConfigs
-        = KafkaProducerConfigs().load(configFile)
-
-/**
- * Convenient method to create and populate a new [KafkaClientConfigs] from an [inputStream].
+ * Creates a new [KafkaClientConfigs] with no properties.
  */
-fun loadProducerConfigs(inputStream: InputStream): KafkaProducerConfigs
-        = KafkaProducerConfigs().load(inputStream)
+fun emptyClientConfigs(): KafkaClientConfigs = KafkaClientConfigs()
 
 /**
- * Convenient method to create and populate a new [KafkaConsumerConfigs] from a [configFile].
+ * Creates a new [KafkaClientConfigs] with the given [props].
  */
-fun loadConsumerConfigs(configFile: String): KafkaConsumerConfigs
-        = KafkaConsumerConfigs().load(configFile)
+fun clientConfigsOf(props: Map<String, Any?>): KafkaClientConfigs = KafkaClientConfigs(HashMap(props))
 
 /**
- * Convenient method to create and populate new [KafkaConsumerConfigs]  from an [inputStream].
+ * Creates a new [KafkaClientConfigs] with the given [props].
  */
-fun loadConsumerConfigs(inputStream: InputStream): KafkaConsumerConfigs
-        = KafkaConsumerConfigs().load(inputStream)
+fun clientConfigsOf(props: Properties): KafkaClientConfigs = clientConfigsOf(props.toStringMap())

clients/src/main/kotlin/io/streamthoughts/kafka/clients/KafkaClients.kt

@@ -19,11 +19,11 @@
 package io.streamthoughts.kafka.clients
 
 import io.streamthoughts.kafka.clients.consumer.ConsumerWorker
-import io.streamthoughts.kafka.clients.consumer.KafkaConsumerConfigs
 import io.streamthoughts.kafka.clients.consumer.KafkaConsumerWorker
-import io.streamthoughts.kafka.clients.producer.KafkaProducerConfigs
+import io.streamthoughts.kafka.clients.consumer.consumerConfigsOf
 import io.streamthoughts.kafka.clients.producer.KafkaProducerContainer
 import io.streamthoughts.kafka.clients.producer.ProducerContainer
+import io.streamthoughts.kafka.clients.producer.producerConfigsOf
 import org.apache.kafka.common.serialization.Deserializer
 
 /**
@@ -46,7 +46,7 @@ class KafkaClients(private val configs: KafkaClientConfigs) {
                         keyDeserializer: Deserializer<K>,
                         valueDeserializer: Deserializer<V>,
                         init: KafkaConsumerWorker.Builder<K, V>.() -> Unit): ConsumerWorker<K, V> {
-        val configs = KafkaConsumerConfigs(configs).groupId(groupId)
+        val configs = consumerConfigsOf(configs).groupId(groupId)
         return KafkaConsumerWorker.Builder(configs, keyDeserializer, valueDeserializer).also(init).build()
     }
 
@@ -56,7 +56,7 @@ class KafkaClients(private val configs: KafkaClientConfigs) {
      * @return a new [ProducerContainer] instance.
      */
     fun<K, V> producer(init: ProducerContainer.Builder<K, V>.() -> Unit): ProducerContainer<K, V> {
-        val configs = KafkaProducerConfigs(configs)
+        val configs = producerConfigsOf(configs)
         return KafkaProducerContainer.Builder<K, V>(configs).also(init).build()
     }
 }