How to configure DelegatingByTopicSerialization from application.yml

[jnernest]

Hi,

I am trying to make use of DelegatingByTopicSerialization as i need different serializer for different topic but to consume it i am trying to configure it with VALUE_SERIALIZATION_TOPIC_CONFIG on application.yml but the constructor is not called.

I dont know what i am missing in configuration.

spring.kafka.consumer:
properties:
spring.deserializer.value.delegate.class: org.apache.kafka.common.serialization.BytesDeserializer
"spring.kafka.value.serialization.bytopic.config": *-error:org.apache.kafka.common.serialization.BytesDeserializer
"spring.kafka.value.serialization.bytopic.default": org.apache.kafka.common.serialization.BytesDeserializer

I also tried to do an bean

    @Bean
    public ConsumerFactory<String, Object> consumerFactory(Map<String, Object> config) {

        return new DefaultKafkaConsumerFactory<>(config, null, new DelegatingByTopicDeserializer(
                Map.of(Pattern.compile("*-error"), new BytesDeserializer()), new BytesDeserializer()));
    }

Both it didn't worked valueserializer was still using StringDeserializer

Can you help me how to configure the DelegatingByTopicSerialization through applucation.yml.

@garyrussell

[artembilan]

You probably are missing to set:
spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.DelegatingByTopicDeserializer.

Please, consider to share with us more of your YAML config and in more readable format.
Of course if value-deserializer doesn't help.
There is another way: share with us a simple project to reproduce and play with.

[jnernest]

Thanks @artembilan that was missing from my side and it worked :) thanks a lot

can you add an example to have this somewhere, or can you guide me to the documentation so i can use for reference :)

[artembilan]

What you are asking about is out of Spring for Apache Kafka scope. That configuration property is a part of Spring Boot: https://docs.spring.io/spring-boot/docs/current/reference/htmlsingle/#messaging.kafka.

As you see Spring Boot doesn't provide too much details about consumer configuration.
Apparently with an assumption that you are going to read JavaDocs for configuration properties.
And since there is not too much details for any similar auto-configuration feature in Spring Boot, I don't see what and how you'd like to be documented: we cannot add this into Spring for Apache Kafka docs since it is not its feature and we cannot add it into Spring Boot since it does not document too much.

[jnernest]

yeah it makes sense but can we have one example(spring Kafka code repository) where we use serialisation by topic which is obtained from application.properties and also if possible other ways to configure it as you mentioned earlier.

Thanks

[artembilan]

Feel free to contribute one: sorry, we are short-handed here to add sample for every single singe feature we have in the framework.

[monsdroid]

this ⬆️ would be a nice hint to have in the doc's, somewhere around here
i know if you know its obvious but if you dont this can be confusing.

just a short, "if you plan on using property based Delegating Serializer and Deserializer you have to set the base? value/key serializers to the corresponding e.g. DelegatingByTopicDeserializer"

spring:
     kafka:
        ...
        consumer:
        ...
            key-deserializer: "org.springframework.kafka.support.serializer.DelegatingByTopicDeserializer"
            value-deserializer: "org.springframework.kafka.support.serializer.DelegatingByTopicDeserializer"
            properties:
            ...
                spring.kafka.key.serialization.bytopic.config: "${kafka.consumer.topic_one}:org.apache.kafka.common.serialization.StringDeserializer"
                spring.kafka.value.serialization.bytopic.config: "${kafka.consumer.topic_one}:com.example.custom.KafkaSomethiungDeserializer"

link to spring boot enhancment request: spring-projects/spring-boot#39066