what is the correct way to handle kafka messages asynchronously

[cuichenli]

We currently have one Kafka consumer and we want to handle the events asynchronously. We have attempted to use the following method signature:

public Uni<Void> process(KafkaRecordBatch<Void, Something> records) {
}

However, it seems that the processing is still happening synchronously. In other words, the second batch only starts processing after the first batch is finished.

I have also tried change the signature to

public Uni<Void> process(Multi<KafkaRecord<Void, Something>> records) {
}

but I got the following error

	Suppressed: java.lang.ClassCastException: class java.util.ArrayList cannot be cast to class io.smallrye.mutiny.Multi (java.util.ArrayList is in module java.base of loader 'bootstrap'; io.smallrye.mutiny.Multi is in unnamed module of loader io.quarkus.bootstrap.runner.RunnerClassLoader @2bea5ab4)

Do you have any suggestions on how to address this? Thank you.

[cescoffier]

You need to use @Blocking, with ordered=false:

@Blocking(ordered=false) // Make sure you use the smallrye one
@Incoming("...")
void process(KafkaRecordBatch<Void, Something> records) {
   // ...
}

[cuichenli]

Thanks @cescoffier , I will give it a try. But honestly this is a little bit unintuitive... When specifying @Blocking we usually expected it to be running synchronously.

Besides, do we also need to specify the max-concurrency property to enable the asynchronous? Thanks

[cescoffier]

Async/Sync are unrelated to Blocking/Non-Blocking. You want the ability to run processes in parallel without preserving the order.

@Blocking is for blocking logic, synchronous or not, because we try to always go back to the event loop by default (for ordering and performance reasons), @Blocking is how to fork the pipeline and execute processing in parallel (if your system allows it).

You do not need to set max-concurrency - there is a sensible default. You can override it, of course.