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()
            }
        }
    }
}

val producer: ProducerContainer<String, String> = kafka("localhost:9092") {
    client {
        clientId("my-client")
    }

    producer {
        configure {
            acks(Acks.InSyncReplicas)
        }
        keySerializer(StringSerializer())
        valueSerializer(StringSerializer())

        defaultTopic("demo-topic")

        onSendError {_, _, error ->
            e.printStackTrace()
        }

        onSendSuccess{ _, _, metadata ->
            println("Record was sent successfully: topic=${metadata.topic()}, partition=${metadata.partition()}, offset=${metadata.offset()} ")
        }
    }
}

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) }
}

val configs = consumerConfigsOf()
    .client { bootstrapServers("localhost:9092") }
    .groupId("demo-consumer-group")
    .keyDeserializer(StringDeserializer::class.java.name)
    .valueDeserializer(StringDeserializer::class.java.name)

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()}"
                )
            }
    }
}

val consumerWorker: ConsumerWorker<String, String> = kafka("localhost:9092") {
    client {
        clientId("my-client")
    }

    val stringDeserializer: Deserializer<String> = StringDeserializer()
    consumer("my-group", stringDeserializer, stringDeserializer) {
        configure {
            maxPollRecords(1000)
            autoOffsetReset(AutoOffsetReset.Earliest)
        }

        onDeserializationError(replaceWithNullOnInvalidRecord())

        onPartitionsAssigned { _: Consumer<*, *>, partitions ->
            println("Partitions assigned: $partitions")
        }

        onPartitionsRevokedAfterCommit { _: Consumer<*, *>, partitions ->
            println("Partitions revoked: $partitions")
        }

        onConsumed { _: Consumer<*, *>, value: String? ->
            println("consumed record-value: $value")
        }

        onConsumedError(closeTaskOnConsumedError())

        Runtime.getRuntime().addShutdownHook(Thread { run { stop() } })
    }
}

consumerWorker.use {
    consumerWorker.start("demo-topic", maxParallelHint = 4)
    runBlocking {
        println("All consumers started, waiting one minute before stopping")
        delay(Duration.ofMinutes(1).toMillis())
    }
}