feat(examples): add example using transactional ProducerContainer (#3)

Resolves: GH-3

Author: fhussonnois

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

@@ -74,14 +74,14 @@ class KafkaProducerContainer<K, V> private constructor(
     private lateinit var producer: Producer<K, V>
 
     override fun send(
-        pairs: Collection<Pair<K, V>>,
+        records: Collection<Pair<K, V>>,
         topic: String?,
         partition: Int?,
         timestamp: Instant?,
         onSuccess: OnSendSuccessCallback<K, V>?,
         onError: OnSendErrorCallback<K, V>?
     ): Future<List<SendResult<K?, V?>>> {
-        val futures: List<CompletableFuture<SendResult<K?, V?>>> = pairs.map {
+        val futures: List<CompletableFuture<SendResult<K?, V?>>> = records.map {
             send(it.first, it.second, topic, partition, timestamp, onSuccess, onError) as CompletableFuture
         }
         return  CompletableFuture.allOf(*futures.toTypedArray()).thenApply { futures.map { it.join() }.toList() }

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

@@ -162,7 +162,7 @@ interface ProducerContainer<K, V>: Closeable {
     }
 
     /**
-     * Asynchronously send the given key-value [pair] to the given [topic] (or the default one if null is given)
+     * Asynchronously send the given key-value [record] to the given [topic] (or the default one if null is given)
      * and [partition] with the given [timestamp].
      *
      * Then, optionally invoke the specific given [onSuccess] callback when the record has been acknowledge.
@@ -171,17 +171,17 @@ interface ProducerContainer<K, V>: Closeable {
      * @see Producer.send
      * @return a [Future] of [SendResult]
      */
-    fun send(pair: Pair<K, V?>,
+    fun send(record: Pair<K, V?>,
              topic: String? = null,
              partition: Int? = null,
              timestamp: Instant? = null,
              onSuccess: OnSendSuccessCallback<K, V>? = null,
              onError: OnSendErrorCallback<K, V>? = null)  : Future<SendResult<K?, V?>> {
-        return send(pair.first, pair.second, topic, partition, timestamp, onSuccess, onError)
+        return send(record.first, record.second, topic, partition, timestamp, onSuccess, onError)
     }
 
     /**
-     * Asynchronously send all the given key-value [pairs] to the given [topic] (or the default one if null is given)
+     * Asynchronously send all the given key-value [records] to the given [topic] (or the default one if null is given)
      * and [partition] with the given [timestamp].
      *
      * Then, optionally invoke the specific given [onSuccess] callback when the record has been acknowledge.
@@ -190,7 +190,7 @@ interface ProducerContainer<K, V>: Closeable {
      * @see Producer.send
      * @return a [Future] of [SendResult]
      */
-    fun send(pairs: Collection<Pair<K, V>>,
+    fun send(records: Collection<Pair<K, V>>,
              topic: String? = null,
              partition: Int? = null,
              timestamp: Instant? = null,

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

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2020 StreamThoughts.
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.streamthoughts.kafka.client.examples
+
+import io.streamthoughts.kafka.clients.loadProducerConfigs
+import io.streamthoughts.kafka.clients.producer.Acks
+import io.streamthoughts.kafka.clients.producer.KafkaProducerConfigs
+import io.streamthoughts.kafka.clients.producer.KafkaProducerContainer
+import org.apache.kafka.common.serialization.StringSerializer
+import kotlin.system.exitProcess
+
+
+fun main(args: Array<String>) {
+
+    if (args.size != 2) {
+        println("Missing required command line arguments: configFile topic")
+        exitProcess(1)
+    }
+
+    val (config, topic) = args
+
+
+    // Load properties from file and customize Producer config.
+    val configs: KafkaProducerConfigs = loadProducerConfigs(config)
+        .acks(Acks.Leader)
+        .keySerializer(StringSerializer::class.java.name)
+        .valueSerializer(StringSerializer::class.java.name)
+
+    val producer = KafkaProducerContainer.Builder<Any?, String>(configs)
+        .defaultTopic(topic)
+        .configure {
+            transactionalId("my-tx-id")
+            enableIdempotence(true)
+        }
+        .onSendSuccess { _, _, m ->
+            println("Record was successfully sent (topic=${m.topic()}, partition=${m.partition()}, offset= ${m.offset()})")
+        }
+        .onSendError { _, _, e ->
+            e.printStackTrace()
+        }
+        .build()
+
+    val messages = listOf("I ❤️ Logs", "Making Sense of Stream Processing", "Apache Kafka")
+
+    producer.use {
+        producer.init()
+        producer.runTx {
+            messages.forEach{producer.send(value = it)}
+        }
+    }
+}