Как обойти сбой брокера kafka при запуске сервера Springboot?

Question

У меня проблема с сервером приложений spring-boot (2.1.7.RELEASE), который использует библиотеку интеграции spring-boot-Kafka ('spring-Kafka' -'.2.2.7.RELEASE ') для доступа к моему Kafkaтемы.

Когда мои брокеры Kafka не работают, мое приложение не может запуститься.

Вот что я получаю:

2019-10-06 02:41:02.764  WARN -- [main           ] org.apache.kafka.clients.NetworkClient                       [] : [Consumer clientId=consumer-1, groupId=caas] Connection to node -1 could not be established. Broker may not be available.

Error starting ApplicationContext. To display the conditions report re-run your application with 'debug' enabled.
2019-10-06 02:41:31.777 ERROR -- [main           ] org.springframework.boot.SpringApplication                   [] : Application run failed

org.springframework.context.ApplicationContextException: Failed to start bean 'org.springframework.kafka.config.internalKafkaListenerEndpointRegistry'; nested exception is org.apache.kafka.common.errors.TimeoutException: Timeout expired while fetching topic metadata
    at org.springframework.context.support.DefaultLifecycleProcessor.doStart(DefaultLifecycleProcessor.java:185)
    at org.springframework.context.support.DefaultLifecycleProcessor.access$200(DefaultLifecycleProcessor.java:53)
    at org.springframework.context.support.DefaultLifecycleProcessor$LifecycleGroup.start(DefaultLifecycleProcessor.java:360)
    at org.springframework.context.support.DefaultLifecycleProcessor.startBeans(DefaultLifecycleProcessor.java:158)
    at org.springframework.context.support.DefaultLifecycleProcessor.onRefresh(DefaultLifecycleProcessor.java:122)
    at org.springframework.context.support.AbstractApplicationContext.finishRefresh(AbstractApplicationContext.java:893)
    at org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext.finishRefresh(ServletWebServerApplicationContext.java:162)
    at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:552)
    at org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext.refresh(ServletWebServerApplicationContext.java:141)
    at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:743)
    at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:390)
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:312)
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:1214)
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:1203)
    at com.zinka.fmsclients.FmsClientsApplication.main(FmsClientsApplication.java:14)
Caused by: org.apache.kafka.common.errors.TimeoutException: Timeout expired while fetching topic metadata

Есть ли способ обойти эту зависимость отКафки брокеры? Я хочу, чтобы мое приложение запускалось, даже если брокеры недоступны / не работают. Сервер приложений должен быть в состоянии подключиться, когда брокеры подключены.

Answer 1

Существует свойство контейнера missingTopicsFatal: https://docs.spring.io/spring-kafka/api/org/springframework/kafka/listener/ContainerProperties.html#isMissingTopicsFatal--, которое позволяет запускать приложение, даже если тема недоступна.

@Bean(name = "kafkaListenerContainerFactory")
public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
    ConsumerFactory<Object, Object> kafkaConsumerFactor,
    ConcurrentKafkaListenerContainerFactoryConfigurer configurer) {

  ConcurrentKafkaListenerContainerFactory<Object, Object> factory =
      new ConcurrentKafkaListenerContainerFactory<>();
  configurer.configure(factory, kafkaConsumerFactor);

  ContainerProperties containerProperties = factory.getContainerProperties();
  containerProperties.setMissingTopicsFatal(false);

  ...
  return factory;
}