title: Spring-Kafka @KafkaHandlers not consuming respective java Objects
body: <p>I know it has been asked multiple time, however, i cant find the solution yet. I am able to consume a specific Java Object by my @KafkaListener(On a class level) and works perfectly fine, however, When i try to consume multiple different JSON Objects from the same topic( I use @KafkaListener at class level and @KafkaHandler at the method level, each @KafkaHandler method expecting different Object), it always produces LinkedHashMap. I can parse this and get the data and do a factory pattern to generate different Instances based on a json field, however, i do not want to do that when Spring can auto detect a specific @KafkaHandler for routing the message. 
  How do i consume different JSON object from a single topic with single @KafkaListener.</p>
  
  <p>I am using below configs:</p>
  
  <pre><code>public ConsumerFactory&lt;String, Object&gt; abcConsumerFactory() {       
      Map&lt;String, Object&gt; config = new HashMap&lt;&gt;();       
      config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
      config.put(ConsumerConfig.GROUP_ID_CONFIG,"group-1");
      config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
      config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,JsonDeserializer.class );
      config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
      config.put(JsonDeserializer.TRUSTED_PACKAGES, "*");
  
      return new DefaultKafkaConsumerFactory&lt;&gt;(config, new StringDeserializer(), new JsonDeserializer&lt;&gt;(Object.class, false)); 
  
  @Bean
  public ConcurrentKafkaListenerContainerFactory&lt;String, Object&gt; abcListenerContainerFactory() {      
      ConcurrentKafkaListenerContainerFactory&lt;String, Object&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
      factory.setConsumerFactory(abcConsumerFactory());
      return factory;
      }
  </code></pre>
  
  <p>If i use the actual class(Foo or Bar) instead of Object in above configs it works fine for that particular objects, However, when i try to generalize, it does not go to the specific @KafkaHandler, and instead goes to a @KafkaHandler with Payload type LinkedHashMap(I was trying to check if it gets delivered to a @KafkaHandler)</p>
  
  <pre><code>@Bean
  public ConcurrentKafkaListenerContainerFactory&lt;String, Object&gt; abcListenerContainerFactory() {      
      ConcurrentKafkaListenerContainerFactory&lt;String, Object&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
      factory.setConsumerFactory(abcConsumerFactory());
      return factory;
      }
  
  Class:  
  
      @KafkaListener(topics={"abc_gh"}, containerFactory = "abcListenerContainerFactory")
      @Service
      public class MyListener {
  
      @KafkaHandler
      public void consumeMessage(@Payload Foo f) {
          //only comes here when i use Foo in my configs instead of Object
  
      }
  
      @KafkaHandler
      public void consumeMessage22(@Payload Bar b) {
          //only comes here when i use Bar in my configs instead of Object
  
      }   
  
      @KafkaHandler
      public void consumeMessage77(@Payload LinkedHashMap bc) {
  
          //comes here when i use Object in the configs, even if i expect a Foo or a Bar object
  
      }               
      }
  </code></pre>
  
  <p>One thing i want to share is the Producer is not using Spring-Kafka.</p>
  
  <p>I dont know what i am missing, i tried a lot of things but no luck. </p>