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. Do you have any suggestions on how to address this? Thank you.

[quarkus-bot[bot]]

/cc @alesj (kafka), @cescoffier (kafka), @ozangunalp (kafka)

[cuichenli]

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)

[cuichenli]

answered in smallrye/smallrye-reactive-messaging#2353 (comment)