title: Kafka 2 consumer factories listeners are not connected constantly
body: <p>We are using Spring Kafka 2.1.4.RELEASE version in our project and we have the following configuration:</p>
  
  <pre class="lang-java prettyprint-override"><code>@EnableKafka
  public class KafkaConfig {
  
      @Value("${spring.kafka.bootstrap-servers}")
      private String bootstrapServers;
  
      @Configuration
      class ProducerConfig {
          @Bean
          public Map&lt;String, Object&gt; producerConfigs() {
              Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
              props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
              props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
              props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ASerializer.class);
              return props;
          }
  
          @Bean
          public ProducerFactory&lt;String, A&gt; producerFactory() {
              return new DefaultKafkaProducerFactory&lt;&gt;(producerConfigs());
          }
  
          @Bean
          public KafkaTemplate&lt;String, A&gt; kafkaTemplate() {
              return new KafkaTemplate&lt;&gt;(producerFactory());
          }
      }
  
      @Configuration
      class ConsumerConfig {
          @Value("${spring.kafka.consumer.group-id}")
          private String groupId;
          @Value("${spring.kafka.consumer.auto-offset-reset}")
          private String autoOffsetReset;
          @Value("${spring.kafka.consumer.enable-auto-commit}")
          private boolean enableAutoCommit;
          @Value("${spring.kafka.consumer.max-poll-records}")
          private Integer maxPollRecords;
  
          @Bean
          public Map&lt;String, Object&gt; firstConsumerConfig() {
              Map&lt;String, Object&gt; props = getCommonConsumerConfig();
              props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
              props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ADeserializer.class);
              return props;
          }
  
          @Bean
          public Map&lt;String, Object&gt; secondConsumerConfig() {
              Map&lt;String, Object&gt; props = getCommonConsumerConfig();
              props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
              props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, BDeserializer.class);
              return props;
          }
  
          @Bean
          public ConsumerFactory&lt;String, A&gt; firstConsumerFactory() {
              return new DefaultKafkaConsumerFactory&lt;&gt;(firstConsumerConfig());
          }
  
          @Bean
          public KafkaListenerContainerFactory&lt;ConcurrentMessageListenerContainer&lt;String, A&gt;&gt; firsttContainerFactory() {
              ConcurrentKafkaListenerContainerFactory&lt;String, A&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
              factory.setConsumerFactory(firstConsumerFactory());
              return factory;
          }
  
          @Bean
          public ConsumerFactory&lt;String, B&gt; secondConsumerFactory() {
              return new DefaultKafkaConsumerFactory&lt;&gt;(secondConsumerConfig());
          }
  
          @Bean
          public KafkaListenerContainerFactory&lt;ConcurrentMessageListenerContainer&lt;String, B&gt;&gt; outputTopicContainerFactory() {
              ConcurrentKafkaListenerContainerFactory&lt;String, B&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
              factory.setConsumerFactory(secondConsumerFactory());
              return factory;
          }
  
  
          private Map&lt;String, Object&gt; getCommonConsumerConfig() {
              Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
              props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
              props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
              props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
              props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
              props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
              return props;
          }
      }
  }
  </code></pre>
  
  
  
  <p>So as you can see we use 2 consumer factories. And we have following consumer class:</p>
  
  <pre class="lang-java prettyprint-override"><code>@ConfigurationProperties
  @Service
  public class Listener {
  
      @KafkaListener(containerFactory = "firstContainerFactory", topics = "someTopic")
      public void firstListener(@Payload A first) {
          //some logic
      }
  
      @KafkaListener(containerFactory = "secondContainerFactory", topics = {
       //topic list       
      })
      public void secondTopicListener(@Payload B second) {
         //some logic
  
      }
  
  
  }
  </code></pre>
  
  
  
  <p>So what we are noticing during starting of this app is that it's not connected to both topics all the time. Sometimes it is connected only to the second topic or only to the first topic and possibly it might connect to the first and second topics (which is correct). So can you please help to understand what is misconfigured here?</p>