Kafka - Deserializing the object in Consumer

Question

We are considering to use Kafka in our for messaging and our applications are developed using Spring. So, we have planned to use spring-kafka.

The producer puts the message as HashMap object into the queue. We have JSON serializer and we assumed that the map will be serialized and put into the queue. And here is the producer config.

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
        key-serializer: org.springframework.kafka.support.serializer.JsonSerializer
        value-serializer: org.springframework.kafka.support.serializer.JsonSerializer

On the other hand, we have a listener which listens to the same topic where the producer has published the message. Here is the consumer config:

spring:
   kafka:
       consumer:
            group-id: xyz
            key-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
            value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer

Our listener method:

  public void listener(SomeClass abx)

We were expecting the json will be de-serialized and an object of type "SomeClass" will be generated. But apparently, it throws de-serialization exception.

We saw few articles and the suggestion was to do something like:

 @Bean
  public ConsumerFactory<String, Car> consumerFactory() {
    return new DefaultKafkaConsumerFactory<>(consumerConfigs(), new StringDeserializer(),
        new JsonDeserializer<>(Car.class));
  }

We don't want to write some code for creating the Deserializer. Is there any boilerplate thing which we are missing? Any help will be appreciated!!

Kacykaczer

Docs for serialization/deserialization: docs.spring.io/spring-kafka/reference/htmlsingle/#serdes Looks like it should work just with @KafkaListener annotation – Paillette

They are suggesting to do something like new JsonDeserializer<>(Bar.class); which I dont want todo, as it will grow when we have more listeners with different types of object – Kacykaczer

I don't think it's a problem to have a single configuration class for Kafka. it's just a part of app configuration. But I will watch this issue, cause it's interesting if the solution exists. – Paillette

Sure. Thanks for suggestions!! – Kacykaczer

You can set the required type using a property - see my answer. – Potomac

In case anyone comes here with the same question, I recommend reading answers in this question: #55110008 – Fran

Answer 1

See the boot documentation. In particular:

You can also configure the Spring Kafka JsonDeserializer as follows:

spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer

spring.kafka.consumer.properties.spring.json.value.default.type=com.example.Invoice

spring.kafka.consumer.properties.spring.json.trusted.packages=com.example,org.acme

Potomac

What if I had multiple types? for ex: Invoice, Payment, This and That etc. How does Spring know how to deserialize ? – Earthshaking

There are several options. You can override the properties at the listener level - see docs.spring.io/spring-kafka/docs/current/reference/html/… - just omit the spring.kafka.consumer.properties part. Or you can add a function - see docs.spring.io/spring-kafka/docs/current/reference/html/… Or, use a message converter instead and the framework will detect the type from the listener method signature - see docs.spring.io/spring-kafka/docs/current/reference/html/… – Potomac

@GaryRussell - Could you please guide here: #78348660? – Galligaskins

Answer 2

I was consuming remote Kafka producer event and facing Class not found exception.

so finally I removed configuration form .properties file added below config class in consumer.

Here is my application.properties.

spring.application.name=payment-service
server.port=8082
spring.kafka.payment.bootstrap-servers= localhost:9092
spring.kafka.order.consumer.group-id.notification= group-id
spring.kafka.consumer.auto-offset-reset= latest
spring.kafka.order.topic.create-order=new_order1    

Code:

        @EnableKafka
        @Configuration("NotificationConfiguration")
        public class CreateOrderConsumerConfig {
            @Value("${spring.kafka.payment.bootstrap-servers}")
            private String bootstrapServers;
            @Value("${spring.kafka.order.consumer.group-id.notification}")
            private String groupId;
            @Bean("NotificationConsumerFactory")
            public ConsumerFactory<String, OrderEvent> createOrderConsumerFactory() {
                Map<String, Object> props = new HashMap<>();
                props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
                props.put(ErrorHandlingDeserializer.KEY_DESERIALIZER_CLASS, ErrorHandlingDeserializer.class);
                props.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, ErrorHandlingDeserializer.class);
                props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
                props.put(ConsumerConfig.CLIENT_ID_CONFIG, UUID.randomUUID().toString());
                props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringSerializer.class);
                props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonSerializer.class);
                props.put(JsonDeserializer.VALUE_DEFAULT_TYPE, "com.swiggy.payment.event.OrderEvent");// this my consumer event class
                props.put(JsonDeserializer.USE_TYPE_INFO_HEADERS,false);
                props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
                props.put(JsonDeserializer.TRUSTED_PACKAGES, "*");
        
                return new DefaultKafkaConsumerFactory<>(props, new StringDeserializer(),
                        new JsonDeserializer<>(OrderEvent.class));
            }
            @Bean("NotificationContainerFactory")
            public ConcurrentKafkaListenerContainerFactory<String, OrderEvent> createOrderKafkaListenerContainerFactory() {
                ConcurrentKafkaListenerContainerFactory<String, OrderEvent> factory = new ConcurrentKafkaListenerContainerFactory<>();
                factory.setConsumerFactory(createOrderConsumerFactory());
                
    factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
                return factory;
            }
        }

Mana