title: Confluent Cloud Apache Kafka Consumer - Topic(s) [test-1] is/are not present and missingTopicsFatal is true
body: <p>I'm a newbie trying to make the communication work between two Spring Boot microservices using Confluent Cloud Apache Kafka. </p>
  
  <p>When using Kafka on Confluent Cloud, I'm getting the following error on my consumer(ServiceB) after ServiceA publishes the message to the topic. However, when I login to my Confluent Cloud, I see that the message has been successfully published to the topic. </p>
  
  <pre><code> org.springframework.context.ApplicationContextException: Failed to start bean 
  'org.springframework.kafka.config.internalKafkaListenerEndpointRegistry'; nested exception is 
   java.lang.IllegalStateException: Topic(s) [topic-1] is/are not present and 
   missingTopicsFatal is true 
  </code></pre>
  
  <p>I do not face this issue when I run Kafka on my local server. ServiceA is able to publish the message to the topic on my local Kafka server and ServiceB is successfully able to consume that message. </p>
  
  <p><strong>I have mentioned my local Kafka server configuration in application.properties(as commented out code)</strong></p>
  
  <p><strong>Service A: PRODUCER</strong></p>
  
  <p><strong>application.properties</strong></p>
  
  <pre><code>app.topic=test-1
  #Remote
  ssl.endpoint.identification.algorithm=https
  security.protocol=SASL_SSL
  sasl.mechanism=PLAIN
  request.timeout.ms=20000
  bootstrap.servers=pkc-4kgmg.us-west-2.aws.confluent.cloud:9092
  retry.backoff.ms=500
  sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule
  requiredusername="*******"
  password="****"
  
  #Local
  #ssl.endpoint.identification.algorithm=https
  #security.protocol=SASL_SSL
  #sasl.mechanism=PLAIN
  #request.timeout.ms=20000
  #bootstrap.servers=localhost:9092
  #retry.backoff.ms=500
  #sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule
  </code></pre>
  
  <p><strong>Sender.java</strong></p>
  
  <pre><code>public class Sender {
  
  @Autowired
  private KafkaTemplate&lt;String, String&gt; kafkaTemplate;
  
  @Value("${app.topic}")
  private String topic;
  
  public void send(String data){
      Message&lt;String&gt; message = MessageBuilder
              .withPayload(data)
              .setHeader(KafkaHeaders.TOPIC, topic)
              .build();
      kafkaTemplate.send(message);
    }
  }
  </code></pre>
  
  <p><strong>KafkaProducerConfig.java</strong></p>
  
  <pre><code>@Configuration
  @EnableKafka
  public class KafkaProducerConfig {
  
  @Value("${bootstrap.servers}")
  private String bootstrapServers;
  
  @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, StringSerializer.class);
      return props;
  }
  
  @Bean
  public ProducerFactory&lt;String, String&gt; producerFactory() {
      return new DefaultKafkaProducerFactory(producerConfigs());
  }
  
  @Bean
  public KafkaTemplate&lt;String, String&gt; kafkaTemplate() {
      return new KafkaTemplate(producerFactory());
   }
  
  }
  </code></pre>
  
  <p><strong>Service B: CONSUMER</strong> </p>
  
  <p><strong>application.properties</strong></p>
  
  <pre><code>app.topic=test-1
  #Remote
  ssl.endpoint.identification.algorithm=https
  security.protocol=SASL_SSL
  sasl.mechanism=PLAIN
  request.timeout.ms=20000
  bootstrap.servers=pkc-4kgmg.us-west-2.aws.confluent.cloud:9092
  retry.backoff.ms=500
  sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule
  requiredusername="*******"
  password="****"
  
  #Local
  #ssl.endpoint.identification.algorithm=https
  #security.protocol=SASL_SSL
  #sasl.mechanism=PLAIN
  #request.timeout.ms=20000
  #bootstrap.servers=localhost:9092
  #retry.backoff.ms=500
  #sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule
  </code></pre>
  
  <p><strong>KafkaConsumerConfig.java</strong></p>
  
  <pre><code>@Configuration
  @EnableKafka
  public class KafkaConsumerConfig {
    @Value("${bootstrap.servers}")
  private String bootstrapServers;
  
  @Bean
  public Map&lt;String, Object&gt; consumerConfigs() {
      Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
      props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
      props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
      props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
      props.put(ConsumerConfig.GROUP_ID_CONFIG, "confluent_cli_consumer_040e5c14-0c18-4ae6-a10f-8c3ff69cbc1a"); // confluent cloud consumer group-id
      props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
      return props;
  }
  
  @Bean
  public ConsumerFactory&lt;String, String&gt; consumerFactory() {
      return new DefaultKafkaConsumerFactory(
              consumerConfigs(),
              new StringDeserializer(), new StringDeserializer());
  }
  
  @Bean(name = "kafkaListenerContainerFactory")
  public ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; kafkaListenerContainerFactory() {
      ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; factory =
              new ConcurrentKafkaListenerContainerFactory();
      factory.setConsumerFactory(consumerFactory());
      return factory;
   }
  }
  </code></pre>
  
  <p><strong>KafkaConsumer.java</strong></p>
  
  <pre><code>@Service
  public class KafkaConsumer {
  private static final Logger LOG = LoggerFactory.getLogger(KafkaListener.class);
  
  @Value("{app.topic}")
  private String kafkaTopic;
  
    @KafkaListener(topics = "${app.topic}", containerFactory = "kafkaListenerContainerFactory")
    public void receive(@Payload String data) {
      LOG.info("received data='{}'", data);
    }
  }
  </code></pre>