BytesJsonMessageConverter with @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY)

[jkupcho]

Given the following @KafkaListener method signature:

public Map<String, Object> listener(@Payload Map<String, Object> value, @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) Map<String, Object> key)

The following MessageConverter:

@Bean
public MessageConverter messageConverter() {
  return new BytesJsonMessageConverter();
}

And the following Deserializer configuration:

spring.kafka.consumer.key-deserializer=org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
spring.kafka.consumer.properties.spring.deserializer.key.delegate.class=org.apache.kafka.common.serialization.BytesDeserializer
spring.kafka.consumer.properties.spring.deserializer.value.delegate.class=org.apache.kafka.common.serialization.BytesDeserializer

If I remove the @Header argument, the @Payload resolves just fine, however, if I include the @Header I receive the following error:

Caused by: org.springframework.core.convert.ConverterNotFoundException: No converter found capable of converting from type [org.apache.kafka.common.utils.Bytes] to type [@org.springframework.messaging.handler.annotation.Header java.util.Map<java.lang.String, java.lang.Object>]

I'm not exactly sure what could be causing this, I'm both key and value objects are being serialized as org.apache.kafka.connect.json.JsonConverter which should come through as an array of bytes from Kafka Connect.

Lastly, if I add @Headers Map<String, Object> headers I can see the key, and it is a JSON string as expected.

Any help would be appreciated!

[garyrussell]

MessageConverters are only applied to payloads, not headers.

You would need to use the JsonDeserializer for the key if you want to convert the JSON (or receive it as Bytes) and do the conversion yourself.

[jkupcho]

Thanks for the reply Gary!