title: How multiple consumer can listen to multiple topic in spring boot Kafka?
body: <p>I am not able to listen the kafka topic (my case 2 topics) when there are multiple consumer. 
  In below example, I am have 2 consumer factory which will be accpet 2 different JSON messages (one is user type and other is Event type). Both the messages are posted to different topic. Here when I am trying to access the Event messages from topic1, i am not able to but I can access the user topic messages. </p>
  
  <p>Ex:   </p>
  
  <pre><code>@Configuration
  @EnableKafka
  public class KafkaConsumerConfiguration {      
  @Autowired
  private Environment environment;
  
  @Bean
  public ConsumerFactory&lt;String,User&gt; consumerFactory() {
      Map&lt;String, Object&gt; config = new HashMap&lt;&gt;();
      config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, environment.getProperty("bootstrap.servers"));
      config.put(ConsumerConfig.GROUP_ID_CONFIG, environment.getProperty("user.consumer.group"));
      config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
      config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,JsonDeserializer.class);
  
      return new DefaultKafkaConsumerFactory&lt;&gt;(config, new StringDeserializer(),
              new JsonDeserializer&lt;&gt;(User.class));
  
  }
  @Bean
  public ConcurrentKafkaListenerContainerFactory&lt;String, User&gt; kafkaListenerContainerFactory() {
      ConcurrentKafkaListenerContainerFactory&lt;String, User&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
      factory.setConsumerFactory(consumerFactory());
      return factory;
  }
  
  @Bean
  public ConsumerFactory&lt;String , Event&gt; consumerFactoryEvent(){
      Map&lt;String, Object&gt; config = new HashMap&lt;&gt;();
      config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, environment.getProperty("bootstrap.servers"));
      config.put(ConsumerConfig.GROUP_ID_CONFIG, environment.getProperty("event.consumer.group"));
      config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
      config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,JsonDeserializer.class);
  
      return new DefaultKafkaConsumerFactory&lt;&gt;(config, new StringDeserializer(),
              new JsonDeserializer&lt;&gt;(Event.class));
  }
  
  @Bean
  public ConcurrentKafkaListenerContainerFactory&lt;String, Event&gt; kafkaListenerContainerFactoryEvent() {
      ConcurrentKafkaListenerContainerFactory&lt;String, Event&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
      factory.setConsumerFactory(consumerFactoryEvent());
      return factory;
  }
  }
  </code></pre>
  
  <p>My main application is like below:</p>
  
  <pre><code>@KafkaListener(topics = "${event.topic}")
  public void processEvent(Event event) {
  ..do something..
  ..post the message to User topic
  }
  @KafkaListener(topics = "${user.topic}")
  public void processUser(User user) {
  ..do something..
  }
  </code></pre>
  
  <p>My need is to listen the event topic first and do some massaging to the message and then send it the User topic and I have another method which will listen to the User topic and do something on that message.. 
  I tried to pass different options to the @KafkaListener such as </p>
  
  <pre><code>@KafkaListener(topics="${event.topic}",containerFactory="kafkaListenerContainerFactoryEvent")
  </code></pre>
  
  <p>but it is not working.. I am not sure whats going wrong.. any suggestion is helpful!</p>