streamthoughts/kafka-clients-kotlin

This projects gives Kotlin bindings and several extensions for Apache Kafka Clients.

kafka-clients-kotlin's Issues

Is there any sdk for android ?

Thanks
Allan

ProducerContainer, ConsumerWorker and ConsumerTask should implement Closeable

Add a basic example to show how to send records with a transactional producer

We should provide a simple example to demonstrate how to use the ProducerContainer.runTx() method and optionnaly how to sendOffsetToTransaction using ProducerContainer.execute()

load producer / consumer config from properties file

ref. https://github.com/confluentinc/examples/blob/5.5.1-post/clients/cloud/kotlin/src/main/kotlin/io/confluent/examples/clients/cloud/util/Utils.kt

ConsumerTask should expose a method to execute an action directly on the Consumer.

Motivation

Developers should be able to get an access to the Consumer instance contained in a ConsumerTask in order to perform some operations. For example to seek the consumer to a specific offset.

Proposed Change

New Public API

Add this new method the ConsumerTask class

fun <T> execute(action: (consumer: Consumer<K, V>) -> T): T

Add a new interface ConsumedErrorHandler to handle exception thrown while processing records

Currently the ConsumerTask fails if an exception is thrown while executing the given ConsumerBatchRecordListener.

The project should provide a mecanism to handle thrown exceptions and to decide what to do next (skip records, seek consumer to specifics offsets, etc).

This issue proposes to add a new interface that can be implemented and then set to the ConsumerTask.

new interface proposition

interface ConsumedErrorHandler {
    fun handle(task: ConsumerTask, records: List<ConsumerRecord<*,*>>, thrownException: Exception) 
}