concurrent_microunat_kafka_listener.java

@KafkaListener(groupId = "micronaut-group", clientId = "${kafka.consumers.micronaut-group.client-id}", threads = 5)
public class MultithreadedMicronautListener {
    @Topic("${kafka.consumers.micronaut-group.topic}")
    void receive(@KafkaKey String key, String value, int partition) {
        switch (partition) {
            case 0:
                // processing logic comes here
                break;
            case 1:
                // processing logic comes here
                break;
            case 2:
                // processing logic comes here
                break;
            default:
                log.error("Message (key {}, value {}) from unexpected partition ({}) received.", key, value, partition);
        }
    }
}