From inside of Kafka Consumer, how get the message header

Question

In few words, I want to get the message header listened by consumer but I am getting "Required argument [Header header] not specified".

Here is how I am trying unsuccessfully:

package com.tolearn.consumer

import io.micronaut.configuration.kafka.Acknowledgement
import io.micronaut.configuration.kafka.annotation.KafkaKey
import io.micronaut.configuration.kafka.annotation.KafkaListener
import io.micronaut.configuration.kafka.annotation.OffsetStrategy
import io.micronaut.configuration.kafka.annotation.Topic
import io.micronaut.messaging.annotation.Header

@KafkaListener(
        groupId="myGroup",
        offsetStrategy=OffsetStrategy.SYNC_PER_RECORD
)
class DemoConsumer {
    @Topic("demotopic")
    fun receive(@KafkaKey key: String?,
                msg: String,
                header: Header,
                acknowledgement: Acknowledgement

    ){
        println("Key = $key " +
               "msg = $msg " +
                "header = $header"
       )

        acknowledgement.ack();
    }
}

It prints this error:

00:02:05.890 [consumer-executor-thread-1] ERROR i.m.c.k.e.KafkaListenerExceptionHandler - Error processing record [Optional[ConsumerRecord(topic = demotopic, partition = 0, leaderEpoch = 0, offset = 6, CreateTime = 1607828525818, serialized key size = 36, serialized value size = 14, headers = RecordHeaders(headers = [RecordHeader(key = My-Header, value = [109, 121, 72, 101, 97, 100, 101, 114])], isReadOnly = false), key = 2afe3f0d-40c0-44f6-93a3-cce06678df80, value = name: "Hello"
)]] for Kafka consumer [com.tolearn.consumer.DemoConsumer@4ca2237e] produced error: Required argument [Header header] not specified
io.micronaut.core.bind.exceptions.UnsatisfiedArgumentException: Required argument [Header header] not specified
    at io.micronaut.core.bind.DefaultExecutableBinder.bind(DefaultExecutableBinder.java:88)
    at io.micronaut.configuration.kafka.processor.KafkaConsumerProcessor.lambda$process$8(KafkaConsumerProcessor.java:494)
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
    at java.base/java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:264)
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
    at java.base/java.lang.Thread.run(Thread.java:834)

In case it adds something, here is the producer:

package com.tolearn.producer

import io.micronaut.configuration.kafka.annotation.KafkaClient
import io.micronaut.configuration.kafka.annotation.KafkaKey
import io.micronaut.configuration.kafka.annotation.Topic
import io.micronaut.messaging.annotation.Header

@KafkaClient(
        id = "demo-producer",
        acks = KafkaClient.Acknowledge.ALL)
public interface DemoProducer {

    @Topic("demotopic")
    fun sendDemoMsg(
            @KafkaKey key: String?,
            @Header("My-Header") myHeader: String,
            msg: String?) {
    }

}

The service sending message with header throw producer

package com.tolearn.service

import com.tolearn.producer.DemoProducer
import javax.inject.Inject
import javax.inject.Named
import javax.inject.Singleton

@Singleton
class DemoService {
    @Inject
    @Named("dp")
    lateinit var dp : DemoProducer

    fun postDemo(key: String, msg: String){
        //blocking
        dp.sendDemoMsg(key, "myHeader", msg)
    }
}

I was expecting to print ... My-Header: myHeader

*** edit

For future readers, my soluiton became:

aplication.yml

micronaut:
  application:
    name: demoGrpcKafka
  executors:
    consumer:
      type: fixed
      nThreads: 1
#kafka.bootstrap.servers: localhost:9092
kafka:
  bootstrap:
    servers: localhost:9092
  consumers:
    default:
      auto:
        commit:
          enable: false
  producers:
    #default:
    demo-producer:
      retries: 2
my:
  application:
    token: tokenFromYml

producer

package com.tolearn.producer

import io.micronaut.configuration.kafka.annotation.KafkaClient
import io.micronaut.configuration.kafka.annotation.KafkaKey
import io.micronaut.configuration.kafka.annotation.Topic
import io.micronaut.messaging.annotation.Header

@KafkaClient(
        id = "demo-producer",
        acks = KafkaClient.Acknowledge.ALL)
        @Header(name = "X-Token", value = "\${my.application.token}")
public interface DemoProducer {

    @Topic("demotopic")
    fun sendDemoMsg(
            @KafkaKey key: String?,
            msg: String?) {
    }

}

consumer

package com.tolearn.consumer

import io.micronaut.configuration.kafka.Acknowledgement
import io.micronaut.configuration.kafka.KafkaHeaders
import io.micronaut.configuration.kafka.annotation.KafkaKey
import io.micronaut.configuration.kafka.annotation.KafkaListener
import io.micronaut.configuration.kafka.annotation.OffsetStrategy
import io.micronaut.configuration.kafka.annotation.Topic
import io.micronaut.messaging.MessageHeaders


@KafkaListener(
        groupId="myGroup",
        offsetStrategy=OffsetStrategy.SYNC_PER_RECORD
)
class DemoConsumer {
    @Topic("demotopic")
    fun receive(@KafkaKey key: String?,
                msg: String,
                headers: MessageHeaders,
                acknowledgement: Acknowledgement,
                offset: Long,
                partition: Int,
                topic: String,
                timestamp: Long

    ){
        val h = (headers).get("X-Token")

        println("Key = $key " +
               "msg = $msg " +
                "offset = $offset " +
                "partition = $partition " +
                "topic = $topic " +
                "timestamp = $timestamp " +
                "header = $h"
       )

        acknowledgement.ack();
    }
}

Answer 1

Looks like you should try use io.micronaut.messaging.MessageHeaders and not just kafka Header class.

@Topic("demotopic")
fun receive(@KafkaKey key: String?,
            msg: String,
            headers: MessageHeaders,
            acknowledgement: Acknowledgement)

Also you can just specify value of header like on the producer side with @io.micronaut.messaging.annotation.Header("My-Header") in your consumer.

Don't forget to add @javax.annotation.Nullable if header is optional, or maybe just String? in case of Kotlin.

@Topic("demotopic")
fun receive(@KafkaKey key: String?,
            msg: String,
            @Nullable @Header("My-Header") myHeader: String,
            acknowledgement: Acknowledgement)