Has anyone successfully used kotlin corotines with...

# coroutines

[Dias]

Has anyone successfully used kotlin corotines with kafka consumers? We currently spin up a thread to consume from a topic, we end up spinning 30ish threads so far, which are not too busy most of the time. I want to convert it to a coroutine usage, but when you consume from a topic you need to commit offset from the same thread and committing should be done in order, I think, which will be a problem if I just send them to be executed in a threadpool. What approach can I take?

[codyoss]

I could be wrong, but I don’t think you can. Consuming is not super resource intense though. If you ack your messages right away you could just throw everything into a coroutine pool to process after that?

[Vsevolod Tolstopyatov [JB]]

I’m afraid coroutines can’t help here a lot.
Kafka consumer API is poll-based, so you don’t have a lot of options except to subscribe to literally every topic in a single thread and organize some kind of event loop with suspensions on top of OffsetCommitCallback

[Dias]

you mean: read event, commit offset and then do something with that event in a callback after the commit?

[Vsevolod Tolstopyatov [JB]]

it depends on how and when you are using commits and whether you are pausing consumer when offsets are not commited.
You can have a special kind of event loop (aka dispatcher) which most of the time polls all topics, but also have an actor per topic which commits offsets using suspension over OffsetCommitCallback

[Vsevolod Tolstopyatov [JB]]

the rest is an implementation detail which depends only on how you want it to behave in different situations

[stevecstian]

You could let other thread process your record, but your consumer thread still need to keep calling poll for heartbeat, and you probably need to use pause/resume API, handle rebalance in rebalance listener, adjust configurations for your usage. So it's really depending on your usage. You should find some discussions in Kafka user email group.

[Dias]

@Vsevolod Tolstopyatov [JB] we poll commits -> do some work with all fetched commits -> do a blocking offset commit -> repeat. I am not sure how you propose to use OffsetCommitCallback, because that's the commit that is done after commit

[stevecstian]

So you want to let other thread to process polled records, call commitSync once all record processed?

[Dias]

@stevecstian yeah that's what I was thinking initially, if I understand you correctly

[Dias]

maybe actually the thing that I want to do is simple, and I just confused you all 😛

[stevecstian]

Unless your record process time is very long or hard to determine, otherwise I'll recommend you to just use single thread model.

[Dias]

that's fair point, I don't remember why did we go with thread per topic approach, I will ask about it

[stevecstian]

Kafka's consumer is not thread safe and is designed to be used with single thread...