Is there a functioning Kafka lib that works well with coroutines? Nothing fancy, just need to send s...

# coroutines

[João Gabriel Zó]

Is there a functioning Kafka lib that works well with coroutines? Nothing fancy, just need to send some messages to my notifications service

[simon.vergauwen]

Hey @João Gabriel Zó,

I've been experimenting with building a Kafka library, and recently I wrote a Alpakka inspired eventloop to stream messages from Kafka using KotlinX Coroutines Flow.

The producer code is not 100% finished, but hope to put some more love into this project soon-ish.

You can pull in the library and use it. I've used it in some smaller projects, or you can check the internals to see how I used the Kafka SDK 😉

https://github.com/nomisRev/kotlin-kafka

[João Gabriel Zó]

Oh I have seen your lib, I just wasn’t sure if it was production ready. The flow aspect of it will come in handy. I’ll spend a little more time with the code and present it to my team.
Thanks!

[simon.vergauwen]

Some of the APIs might change a little bit. I would say that everything under the io.github.nomisrev.kafka.receiver package is stable, but the API of the publisher might still slightly change.

I am planning to also organise it similarly to receiver package with interfaces, and more a bit more structure.

[simon.vergauwen]

Besides that it is production ready to use. If you have any feedback or remarks please share on a Github ticket 🙏

[hfhbd]

I created a similar wrapper for internal use. Biggest problem was not the wrapper itself, but as soon as using Flows, you are doing the operations in memory, eg windowing. Kafka uses a persistent database instead. Also, I used Sharedflows to make it clear, that it never returns normally, but each map operation returns a normal flow, so you lose this information. But this is a normal coroutine issue.

[simon.vergauwen]

Yes, streaming Kafka is more complex than it originally looks like. You need to take into account backpressure, and pause partitions or Kafka might consider your consumer dead, etc.

That's why Reactor build dedicated support, and same for Akka. Or FS2 in Scala.

This project aims to do the same for KotlinX Coroutines Flow so you can stream Kafka message with a lot of guarantees out-of-the-box.

[João Gabriel Zó]

Simon, do you think there is a way to use a Avro/GenericRecord serializer with the producer for your lib?

[simon.vergauwen]

Yes, I am actually doing that. I'm using Avro4k. Disclaimer: ran into some issues when a different service was using KafkaStreams, and Confluent deserializers and it tripped over the magic bytes 😅 It can be fixed by writing a small intermediate deserializer, but I'd need to look it up.

[simon.vergauwen]

If you're not using KafkaStreams, or Confluent serializer (which breaks the Avro spec with the magic bytes), you can just use this. https://github.com/avro-kotlin/avro4k
With this:

[simon.vergauwen]

Or well, you wouldn't even need this. You would just need to use Avro.toGenericRecord, or toRecord (not 100% sure on the syntax on top of my head).

[João Gabriel Zó]

I'm trying to use it with the Confluent serializer because for some reason the people at my company built everything around those magic bytes and it breaks if we don't send them 🙃
I'm running into some troubles while trying to pass the KafkaAvroSerializer to the ProducerSettings tho

[simon.vergauwen]

I don't have the solution on hand, but I can share it on Monday when I get back to work.

[simon.vergauwen]

Hoping to contribute it to Avro4k soon

[simon.vergauwen]

Hey @João Gabriel Zó,
This is the solution we are using for this. Hope that helps.