title: Control enabling/disabling Kafka consumers in Spring Boot
body: <p>I have configured several Kafka consumers in Spring Boot. This is what the kafka.properties looks like (only listing config for one consumer here):</p>
  
  <pre><code>kafka.topics=
  bootstrap.servers=
  group.id=
  enable.auto.commit=
  auto.commit.interval.ms=
  session.timeout.ms=
  schema.registry.url=
  auto.offset.reset=
  kafka.enabled=
  </code></pre>
  
  <p>Here is the config:</p>
  
  <pre><code>@Configuration
  @PropertySource({"classpath:kafka.properties"})
  public class KafkaConsumerConfig {
  
      @Autowired
      private Environment env;
  
      @Bean
      public ConsumerFactory&lt;String, String&gt; pindropConsumerFactory() {
          Map&lt;String, Object&gt; dataRiverProps = new HashMap&lt;&gt;();
  
          dataRiverProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, env.getProperty("bootstrap.servers"));
          dataRiverProps.put(ConsumerConfig.GROUP_ID_CONFIG, env.getProperty("group.id"));
          dataRiverProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, env.getProperty("enable.auto.commit"));
          dataRiverProps.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, env.getProperty("auto.commit.interval.ms"));
          dataRiverProps.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, env.getProperty("session.timeout.ms"));
  
          dataRiverProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
          dataRiverProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
          dataRiverProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, env.getProperty("auto.offset.reset"));
  
          return new DefaultKafkaConsumerFactory&lt;&gt;(dataRiverProps);
      }
  
      @Bean
      public ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; kafkaListenerContainerFactory() {
          ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
          factory.setConsumerFactory(pindropConsumerFactory());
          return factory;
      }
  }
  </code></pre>
  
  <p>And this is the consumer:</p>
  
  <pre><code>@Component
  public class KafkaConsumer {
  
      @Autowired
      private MessageProcessor messageProcessor;
  
  
      @KafkaListener(topics = "#{'${kafka.topics}'.split(',')}", containerFactory = "kafkaListenerContainerFactory")
      public void consumeJson(String message) {
          // processing message
      }
  }
  </code></pre>
  
  <p>Is there a way for me to use the prop "kafka.enabled" so that I can control the creation or maybe the message retrieval of this consumer? Thanks so much!</p>