Hi everyone, I try to call the run method of Kafka...

# javascript

[Patrick Doering]

Hi everyone,
I try to call the run method of KafkaJS from Kotlin. For that I did the following steps:
1. Add kotlinx-nodejs dependency
2. Run task nodeJsGenerateExternals
3. Copy from externals to sources
4. Add to index.module_kafkajs.kt on top @file:JsModule("kafkajs")
5. Move all typealiases (non external) to other file (in my case index.module_kafkajs2.kt
In plain JS I have to call it that way:

await Consumer.run({
   eachMessage: async ({ topic, partition, message }) => {
     console.log({
       value: message.value.toString(),
     })
   },
 });

In the created index.module_kafkajs.kt file I found that:

external interface Consumer {
    fun connect(): Promise<Unit>
    fun disconnect(): Promise<Unit>
    fun subscribe(topic: ConsumerSubscribeTopic): Promise<Unit>
    fun stop(): Promise<Unit>
    fun run(config: ConsumerRunConfig = definedExternally): Promise<Unit>
    fun commitOffsets(topicPartitions: Array<TopicPartitionOffsetAndMedata>): Promise<Unit>
    fun seek(topicPartition: `T$28`)
    fun describeGroup(): Promise<GroupDescription>
    fun pause(topics: Array<`T$29`>)
    fun paused(): Array<TopicPartitions>
    fun resume(topics: Array<`T$29`>)
    fun on(eventName: ValueOf<ConsumerEvents>, listener: (args: Any) -> Unit): RemoveInstrumentationEventListener<Any>
    fun logger(): Logger
    var events: ConsumerEvents
}

and that: 

external interface ConsumerRunConfig {
    var autoCommit: Boolean?
        get() = definedExternally
        set(value) = definedExternally
    var autoCommitInterval: Number?
        get() = definedExternally
        set(value) = definedExternally
    var autoCommitThreshold: Number?
        get() = definedExternally
        set(value) = definedExternally
    var eachBatchAutoResolve: Boolean?
        get() = definedExternally
        set(value) = definedExternally
    var partitionsConsumedConcurrently: Number?
        get() = definedExternally
        set(value) = definedExternally
    var eachBatch: ((payload: EachBatchPayload) -> Promise<Unit>)?
        get() = definedExternally
        set(value) = definedExternally
    var eachMessage: ((payload: EachMessagePayload) -> Promise<Unit>)?
        get() = definedExternally
        set(value) = definedExternally
}

Actual I don` understand how to use that from Kotlin with eachMessage. I hope somebody can help me.

[andylamax]

Can you post the external interface EachMessagePayload Are you using this with kotlinx-coroutines?

[andylamax]

Consumer.run(jsObject{
  eachMessage = { payload ->
    val topic = payload.topic
    val partition = payload.partion
    val message = payload.message
    console.log(js{
      value = message.value.toString()
    })
  }
}).then({res->
   console.log(res)
})

[Patrick Doering]

Here is the EachMessagePayload interface:

external interface EachMessagePayload {
    var topic: String
    var partition: Number
    var message: KafkaMessage
}

[Patrick Doering]

Ok I will try it.

[Patrick Doering]

Yeah I am using it with kotlinx-coroutines

[andylamax]

Then you can do it this way

coroutineScope {
  val res = Consumer.run(jsObject{
    eachMessage = { payload ->
      val topic = payload.topic
      val partition = payload.partion
      val message = payload.message
      console.log(js{
        value = message.value.toString()
      })
    }
  }).await() // suspends here
}

[Patrick Doering]

That is not working. As I can see I have to call the run with a config object.
Like that:
consumer.run(config)

For that I have to define:

val config: ConsumerRunConfig = js("({})")
config.autoCommit = true
config.autoCommitInterval = null
config.autoCommitThreshold = null
config.eachBatchAutoResolve = true
config.partitionsConsumedConcurrently = 1

but then I cant call consumer.run(config{

[andylamax]

for the config object, you should be able to do

val config = jsObject<ConsumerRunConfig> {
  autoCommit = true
  autoCommit = true
  autoCommitInterval = null
  autoCommitThreshold = null
  eachBatchAutoResolve = true
  partitionsConsumedConcurrently = 1
}

[andylamax]

How do you get/create and instance of a the Consumer ?

[Patrick Doering]

I create my consumer in this way.

private var consumer: dynamic
consumer = Kafka(kafkaOptions).consumer(consumerOptions)

[Patrick Doering]

Then I do that:

fun consume() {
    val consumerSubscription: ConsumerSubscribeTopic = js("({})")
    consumerSubscription.topic = getTopic()
    consumerSubscription.fromBeginning = fromBeginning

    // default values set
    val config: ConsumerRunConfig = js("({})")
    config.autoCommit = true
    config.autoCommitInterval = null
    config.autoCommitThreshold = null
    config.eachBatchAutoResolve = true
    config.partitionsConsumedConcurrently = 1

    consumer.connect()
    GlobalScope.launch ( context = Dispatchers.Default ) {
        delay(1000)
        consumer.subscribe(consumerSubscription)
        
        // Here comes my consumer.run(config) 
        
        //Here is the pain: 
         val res = consumer.run(config{
                eachMessage = { payload ->
                    val topic = payload.topic
                    val partition = payload.partion
                    val message = payload.message
                    console.log(js{
                        value = message.value.toString()
                    })
                }
            }).await() // suspends here

[andylamax]

Try this

fun consume() {
    val consumerSubscription: ConsumerSubscribeTopic = js("({})")
    consumerSubscription.topic = getTopic()
    consumerSubscription.fromBeginning = fromBeginning
    // default values set
    val config: ConsumerRunConfig = js("({})")
    config.autoCommit = true
    config.autoCommitInterval = null
    config.autoCommitThreshold = null
    config.eachBatchAutoResolve = true
    config.partitionsConsumedConcurrently = 1
    consumer.connect()
    GlobalScope.launch ( context = Dispatchers.Default ) {
        delay(1000)
        consumer.subscribe(consumerSubscription)
        // Here comes my consumer.run(config) 
        //Here is the pain: 
         val res = consumer.run(config.apply {
                eachMessage = { payload ->
                    val topic = payload.topic
                    val partition = payload.partion
                    val message = payload.message
                    console.log(js{
                        value = message.value.toString()
                    })
                }
            }).await() // suspends here

[Patrick Doering]

Cool. But now for this three lines I become a:

Expected a value of type Promise<Unit> error.
Any idea how to solve that?

[andylamax]

which three lines? Sorry?

[Patrick Doering]

Oh sorry.

val topic = payload.topic
                    val partition = payload.partion
                    val message = payload.message

[andylamax]

try

val res = consumer.run(config.apply {
            eachMessage = { payload ->
               val topic = payload.topic
               val partition = payload.partion
               val message = payload.message
               console.log(js{
                  value = message.value.toString()
               })
               promise{ Unit }
            }
         }).await() // suspends here

[Patrick Doering]

Ok, it worked for the first message. Great cool.

But I get some errors.
await is not a function.

[Patrick Doering]

Do you know why it is only working with the first message but not with the other 40 messages?

[Patrick Doering]

So the problem could be with this line:

promise{ Unit } --> that breaks the consumption. But when I have more messages, it stops after the first message.

And JB mentioned:
Unfortunately, we don’t able to work with js async/await.

[andylamax]

I don't think if that's a kotlin/js issue. What is Kafka's requirement for you to get multiple messages? Can you post the js equivalent?

[Patrick Doering]

Here is the equivalent:

await consumer.run({

eachMessage: async ({ topic, partition, message }) => {

console.log({

value: message.value.toString(),

})

},

});

Well I wanna consume 1 to n messages when new ones are produced.

[Patrick Doering]

The js is running. I get every message of the topic

[andylamax]

What is Kafka's requirement for you to get multiple messages?
I have a feeling there is something needed to return but I am not sure which one.
try

promise { undefined } // in the end

[Patrick Doering]

That is interesting.
When I use:

promise { undefined } // in the end

The consumer throws a lot of errors.

But when I use:

Promise { resolve, reject ->
}

then I get no errors. But also only the first message.