title: JsonSerializer not working properly kafka
body: <p>I am using kafka I have a Notification class that i am serializing using spring-kafka.</p>
  
  <pre><code>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 &lt;T&gt; T getMessage(Class&lt;T&gt; 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;
      }
  
  
  }
  </code></pre>
  
  <p>here is my configuration</p>
  
  <pre><code>spring:
    kafka:
      producer:
        bootstrap-servers: localhost:9092
        key-serializer: org.apache.kafka.common.serialization.StringSerializer
        value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
  </code></pre>
  
  <p>When i try to consume the notification from consumer my message part is missing in Notification I am able to to receive type.</p>
  
  <p>I even tried kafka console-consumer there also it prints only type field from my notification message is missing here also.</p>
  
  <p>I don't know what i am missing.</p>
  
  <p>My Consumer configuration is </p>
  
  <pre><code>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&lt;String, Notification&gt; consumerFactory() {
          Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
          props.put(
                  ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                  address);
          props.put(
                  ConsumerConfig.GROUP_ID_CONFIG,
                  "booking");
          JsonDeserializer&lt;Notification&gt; ds = new JsonDeserializer&lt;&gt;();
          ds.addTrustedPackages("*");
          return new DefaultKafkaConsumerFactory&lt;&gt;(props,
                  new StringDeserializer(),
                  ds);
      }
  
      @Bean
      public ConcurrentKafkaListenerContainerFactory&lt;String, Notification&gt;
      kafkaListenerContainerFactory() {
  
          ConcurrentKafkaListenerContainerFactory&lt;String, Notification&gt; factory =
                  new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
          factory.setConsumerFactory(consumerFactory());
          return factory;
      }
  }
  </code></pre>
  
  <p>Here is my consumer</p>
  
  <pre><code>@KafkaListener(topics = "sql-insert",groupId = "booking")
      public void onNotification(@Payload Notification notification){
          handleNotification(notification);
      }
  </code></pre>
  
  <p>Please help me.</p>
  
  <p><strong>Note:</strong> Actually problem is with JsonSerializer in Kafka. i tried below code and it is not serializing the object properly.</p>
  
  <pre><code>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&lt;Notification&gt;().serialize("sql-insert", notification);
          System.out.println(new String(serialize));
      }
  </code></pre>
  
  <p>It is giving me the output.</p>
  
  <pre><code>{"type":"NEW_SCHEDULED"}
  </code></pre>
  
  <p>Is there any way to fix it.</p>