UnsatisfiedArgumentException on Kafka Listener using Kotlin

[Chase22]

When defining a message processing method using @Topic i get an "UnsatisfiedArgumentException: Required argument [String key] not specified"

Micronaut-Version: 1.1.3
JVM Version: Java 12

@KafkaListener(offsetStrategy = OffsetStrategy.SYNC, batch = true, threads = 7)
open class Consumer {
    @Topic("\${application.kafka.topic.name}")
    fun processMessage(
        @KafkaKey key: String, 
        message: String, 
        offset: Long, 
        partition: Int, 
        topic: String
    ): Boolean {
        \\...
}

[paulolieuthier]

Have you tried disabling batching (just to help debugging)? I filed a similar issue in #54.

[hmatt1]

To add to paulolieuthier's point,

Check out the batch section of the docs:
https://micronaut-projects.github.io/micronaut-kafka/latest/guide/#kafkaListenerBatch

When batch = true, you receive a list of messages, a list of offsets, etc.

The example method signature looks like this:

@Topic("all-the-books")
public void receive(
        List<Book> books,
        List<Long> offsets,
        List<Integer> partitions,
        List<String> topics,
        Consumer kafkaConsumer) {

[guillermocalvo]

Thanks for for bringing this issue to our attention.

Current documentation does not accurately reflect the way kafka batch listeners work.

The proper way to handle this use case should be by defining a consumer method that receives a list of consumer records:

@Topic("all-the-books")
public void receive(List<ConsumerRecord<String, Book>> records, Consumer kafkaConsumer) {
    /* ... */
}