JsonSerializer not working properly kafka

Question

I am using kafka I have a Notification class that i am serializing using spring-kafka.

package com.code2hack.notification;

public class Notification {
    private Object message;
    private NotificationType type;

    public static Notification create(NotificationType type, Object message){
        return new Notification(message,type);
    }

    public Notification(){

    }
    public Notification(Object message, NotificationType type){
        this.message = message;
        this.type = type;
    }

    @Override
    public String toString() {
        return "Notification{" +
                "message=" + message +
                ", type=" + type +
                '}';
    }

    public <T> T getMessage(Class<T> type){
        return (T)this.message;
    }
    public NotificationType getType(){
        return this.type;
    }
    public void setType(NotificationType type){
        this.type = type;
    }
    public void setMessage(Object message){
        this.message = message;
    }


}

here is my configuration

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

When i try to consume the notification from consumer my message part is missing in Notification I am able to to receive type.

I even tried kafka console-consumer there also it prints only type field from my notification message is missing here also.

I don't know what i am missing.

My Consumer configuration is

package com.code2hack.booking;

import com.code2hack.notification.Notification;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.support.serializer.JsonDeserializer;

import java.util.HashMap;
import java.util.Map;

@Configuration
@EnableKafka
public class KafkaConfiguration {
    @Value("${spring.kafka.consumer.bootstrap-servers}")
    private String address;

    @Bean
    public ConsumerFactory<String, Notification> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(
                ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                address);
        props.put(
                ConsumerConfig.GROUP_ID_CONFIG,
                "booking");
        JsonDeserializer<Notification> ds = new JsonDeserializer<>();
        ds.addTrustedPackages("*");
        return new DefaultKafkaConsumerFactory<>(props,
                new StringDeserializer(),
                ds);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Notification>
    kafkaListenerContainerFactory() {

        ConcurrentKafkaListenerContainerFactory<String, Notification> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }
}

Here is my consumer

@KafkaListener(topics = "sql-insert",groupId = "booking")
    public void onNotification(@Payload Notification notification){
        handleNotification(notification);
    }

Please help me.

Note: Actually problem is with JsonSerializer in Kafka. i tried below code and it is not serializing the object properly.

public static void main(String[] args) {

        //SpringApplication.run(BookingServiceApplication.class, args);
        Notification notification = Notification.create(NotificationType.NEW_SCHEDULED,"Hellow how are you");
        byte[] serialize = new JsonSerializer<Notification>().serialize("sql-insert", notification);
        System.out.println(new String(serialize));
    }

It is giving me the output.

{"type":"NEW_SCHEDULED"}

Is there any way to fix it.

Answer 1

Unless you create a custom serializer, Jackson only works with JavaBean semantics; there is no getter for message; you need to add a simple getter for the message property.